109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Mock ELM327 v2 — совместим с AndrOBD-паттернами (elm_proto.py).
|
|
|
|
Эмулирует:
|
|
- '>' как разделитель строк (AndrOBD StreamHandler)
|
|
- Адаптивный таймаут ATST (AndrOBD AdaptiveTiming)
|
|
- Ошибки: SEARCHING, NO DATA, BUS BUSY, UNABLE
|
|
- Побайтовую отправку с микро-задержками
|
|
|
|
Запуск: python tools/mock_elm327_v2.py
|
|
"""
|
|
|
|
import argparse
|
|
import random
|
|
import socketserver
|
|
import time
|
|
|
|
FAKE_VIN = "WVWZZZ1KZAW123456"
|
|
FAKE_DTC = ["43011300000000", "43013300000000"]
|
|
FAKE_NO_DTC = "43000000000000"
|
|
FAKE_PIDS = {
|
|
"05": "41055A", "0C": "410C1AF8", "0D": "410D00",
|
|
"11": "411180", "0B": "410B64", "0F": "410F46",
|
|
"1F": "411F0258", "04": "410440", "06": "410680", "07": "410780",
|
|
}
|
|
|
|
|
|
class ElmHandler(socketserver.StreamRequestHandler):
|
|
|
|
def handle(self):
|
|
print(f"\n🔌 {self.client_address}")
|
|
buf = b""
|
|
while True:
|
|
try:
|
|
ch = self.rfile.read(1)
|
|
if not ch: break
|
|
if ch in (b"\n", b"\r"):
|
|
if buf:
|
|
cmd = buf.decode(errors="ignore").strip().upper()
|
|
buf = b""
|
|
if cmd: self._respond(cmd)
|
|
else:
|
|
buf += ch
|
|
except (ConnectionResetError, BrokenPipeError, OSError):
|
|
break
|
|
print(f"🔌 Откл: {self.client_address}")
|
|
|
|
def _respond(self, cmd: str):
|
|
print(f" 📥 {cmd}")
|
|
|
|
if cmd.startswith("ATST"):
|
|
self._send(b"OK\r\n>"); return
|
|
|
|
if random.random() < 0.03 and cmd not in ("ATZ",):
|
|
err = random.choice([b"BUS BUSY\r\n>", b"BUS BUSY\r\n>",
|
|
b"UNABLE TO CONNECT\r\n>"])
|
|
self._send(err); return
|
|
|
|
if cmd == "ATSP0":
|
|
self._send(b"SEARCHING...\r\n")
|
|
self._sleep(1.8)
|
|
self._send(b"AUTO\r\n>")
|
|
elif cmd in ("ATAT1", "ATE0", "ATL0", "ATS0", "ATH0", "ATH1", "ATPC"):
|
|
self._send(b"OK\r\n>")
|
|
elif cmd == "0902":
|
|
self._sleep(0.5)
|
|
v = "".join(f"{ord(c):02X}" for c in FAKE_VIN)
|
|
self._send(f"014\r\n0:490201{v}\r\n>".encode())
|
|
elif cmd == "03":
|
|
self._sleep(0.3)
|
|
for line in FAKE_DTC:
|
|
self._send(line.encode() + b"\r\n")
|
|
time.sleep(0.05)
|
|
self._send(b">")
|
|
elif cmd == "07":
|
|
self._sleep(0.3)
|
|
self._send(FAKE_NO_DTC.encode() + b"\r\n>")
|
|
elif cmd.startswith("01"):
|
|
pid = cmd[2:4]; self._sleep(0.2)
|
|
self._send((FAKE_PIDS.get(pid, "NODATA") + "\r\n>").encode())
|
|
else:
|
|
self._send(b"?\r\n>")
|
|
|
|
def _send(self, data: bytes):
|
|
for byte in data:
|
|
self.wfile.write(bytes([byte]))
|
|
self.wfile.flush()
|
|
time.sleep(0.0005)
|
|
|
|
def _sleep(self, s: float):
|
|
time.sleep(s * random.uniform(0.7, 1.3))
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument("--port", type=int, default=35000)
|
|
p.add_argument("--host", default="0.0.0.0")
|
|
args = p.parse_args()
|
|
class S(socketserver.ThreadingTCPServer):
|
|
allow_reuse_address = True
|
|
server = S((args.host, args.port), ElmHandler)
|
|
print(f"🚗 Mock ELM327 v2 (AndrOBD) на {args.host}:{args.port}")
|
|
try: server.serve_forever()
|
|
except KeyboardInterrupt: print("\n⏹️ Стоп"); server.shutdown()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|