docs: эталонный анализ 5 проектов — ELM327/BT паттерны

tools/mock_elm327_v2.py

@@ -0,0 +1,166 @@
+#!/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()