167 lines
5.5 KiB
Python
167 lines
5.5 KiB
Python
#!/usr/bin/env python3
|
||
"""mock_elm327_v2.py — эмулятор ELM327 с '>' промптами и симуляцией ошибок.
|
||
|
||
Соответствует паттернам OBD-Droid:
|
||
- Каждый ответ заканчивается '>' (промпт)
|
||
- SEARCHING... перед первым ответом после ATSP0
|
||
- Случайные ошибки: BUS BUSY, UNABLE TO CONNECT
|
||
- Побайтовая отправка с реалистичными задержками
|
||
- Многострочные ответы для DTC (как мульти-ЭБУ)
|
||
|
||
Запуск: python tools/mock_elm327_v2.py
|
||
"""
|
||
|
||
import argparse
|
||
import random
|
||
import socketserver
|
||
import time
|
||
|
||
|
||
FAKE_VIN = "WVWZZZ1KZAW123456"
|
||
|
||
# Многострочные DTC — имитация нескольких ЭБУ
|
||
FAKE_DTC_MULTI = [
|
||
"43011300000000", # P0301 (ЭБУ 1)
|
||
"43013300000000", # P0303 (ЭБУ 1)
|
||
"47500047000000", # P0047 (ЭБУ 2, pending)
|
||
]
|
||
|
||
FAKE_NO_DTC = "43000000000000"
|
||
|
||
FAKE_PIDS = {
|
||
"05": "4105 5A",
|
||
"0C": "410C 1A F8",
|
||
"0D": "410D 00",
|
||
"11": "4111 80",
|
||
"0B": "410B 64",
|
||
"0F": "410F 46",
|
||
"1F": "411F 02 58",
|
||
"04": "4104 40",
|
||
"06": "4106 80",
|
||
"07": "4107 80",
|
||
}
|
||
|
||
|
||
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("utf-8", 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}")
|
||
|
||
# Случайная симуляция ошибок (3%)
|
||
if random.random() < 0.03 and cmd not in ("ATZ", "ATE0", "ATL0", "ATH0"):
|
||
error = random.choice([
|
||
b"BUS BUSY\r\n>",
|
||
b"BUS BUSY\r\n>",
|
||
b"UNABLE TO CONNECT\r\n>",
|
||
])
|
||
self._send_bytes(error)
|
||
print(f" 📤 (ERROR) {error.decode().strip()}")
|
||
return
|
||
|
||
# ── AT команды ───────────────────────
|
||
if cmd == "ATZ":
|
||
self._sleep(2.5)
|
||
self._send_bytes(b"\r\nELM327 v1.5\r\n>")
|
||
|
||
elif cmd == "ATSP0":
|
||
self._send_bytes(b"SEARCHING...\r\n")
|
||
self._sleep(1.8)
|
||
self._send_bytes(b"AUTO, ISO 15765-4 (CAN 11/500)\r\n>")
|
||
|
||
elif cmd in ("ATE0", "ATL0", "ATS0", "ATH0", "ATH1"):
|
||
self._send_bytes(b"OK\r\n>")
|
||
|
||
elif cmd == "ATI":
|
||
self._send_bytes(b"ELM327 v1.5\r\n>")
|
||
|
||
# ── OBD запросы ─────────────────────
|
||
elif cmd == "0902":
|
||
self._sleep(0.5)
|
||
resp = self._vin_bytes()
|
||
self._send_bytes(resp)
|
||
|
||
elif cmd == "03":
|
||
self._sleep(0.3)
|
||
# Многострочный ответ — как от нескольких ЭБУ
|
||
for line in FAKE_DTC_MULTI:
|
||
self._send_bytes(line.encode() + b"\r\n")
|
||
time.sleep(0.05)
|
||
self._send_bytes(b">")
|
||
|
||
elif cmd == "07":
|
||
self._sleep(0.3)
|
||
self._send_bytes(FAKE_NO_DTC.encode() + b"\r\n>")
|
||
|
||
elif cmd.startswith("01"):
|
||
pid = cmd[2:4]
|
||
self._sleep(0.2)
|
||
if pid in FAKE_PIDS:
|
||
self._send_bytes(FAKE_PIDS[pid].encode() + b"\r\n>")
|
||
else:
|
||
self._send_bytes(b"NO DATA\r\n>")
|
||
|
||
else:
|
||
self._send_bytes(b"?\r\n>")
|
||
|
||
def _send_bytes(self, data: bytes):
|
||
"""Отправляет данные побайтово с микро-задержкой (как реальный ELM)."""
|
||
for byte in data:
|
||
self.wfile.write(bytes([byte]))
|
||
self.wfile.flush()
|
||
time.sleep(0.0005) # 0.5 мс между байтами
|
||
|
||
def _sleep(self, seconds: float):
|
||
"""Пауза ±30%, как у реального железа."""
|
||
time.sleep(seconds * random.uniform(0.7, 1.3))
|
||
|
||
def _vin_bytes(self) -> bytes:
|
||
vin_hex = "".join(f"{ord(c):02X}" for c in FAKE_VIN)
|
||
return f"014\r\n0:49 02 01 {vin_hex}\r\n>".encode()
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Mock ELM327 v2 (OBD-Droid совместимый)")
|
||
parser.add_argument("--port", type=int, default=35000)
|
||
parser.add_argument("--host", default="0.0.0.0")
|
||
args = parser.parse_args()
|
||
|
||
class ReuseTCPServer(socketserver.ThreadingTCPServer):
|
||
allow_reuse_address = True
|
||
|
||
server = ReuseTCPServer((args.host, args.port), ElmHandler)
|
||
print(f"🚗 Mock ELM327 v2 на {args.host}:{args.port}")
|
||
print(f" Промпт '>' после каждого ответа")
|
||
print(f" Случайные ошибки: 3%")
|
||
print(f" Ctrl+C = стоп")
|
||
try:
|
||
server.serve_forever()
|
||
except KeyboardInterrupt:
|
||
print("\n⏹️ Стоп")
|
||
server.shutdown()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|