tools: mock ELM327 for testing without real scanner

2026-05-25 22:40:05 +04:00
parent e4f32f3e85
commit 0240322e35
1 changed files with 161 additions and 0 deletions
@@ -0,0 +1,161 @@
+#!/usr/bin/env python3
+"""mock_elm327.py — эмулятор ELM327 для тестирования без реального сканера.
+
+Отвечает на стандартные OBD2-команды, как настоящий ELM327 v1.5.
+Работает через TCP — не нужен реальный Bluetooth.
+
+Запуск:
+  python tools/mock_elm327.py                # порт 35000 по умолчанию
+  python tools/mock_elm327.py --port 9999    # свой порт
+
+Подключение из клиента (Android или Python):
+  telnet localhost 35000
+  или TCP-режим в ElmForwardService (DEBUG_HOST=192.168.x.x)
+
+Поддерживаемые команды:
+  ATZ, ATE0, ATL0, ATSP0, ATH1, ATI, ATDP
+  0100..01FF (PID запросы)
+  03 (коды ошибок), 07 (pending)
+  0902 (VIN)
+"""
+
+import argparse
+import random
+import socketserver
+import time
+
+
+# ── Fake car data ──────────────────────────────────────
+
+FAKE_VIN = "WVWZZZ1KZAW123456"  # VW Golf
+
+FAKE_DTC = [
+    "43011300000000",  # P0301 — пропуск зажигания цилиндр 1
+    "43013300000000",  # P0303 — пропуск зажигания цилиндр 3
+]
+
+FAKE_NO_DTC = "43000000000000"  # нет ошибок
+
+# PID ответы: байты для режима 01
+FAKE_PIDS = {
+    "05": "4105 5A",       # температура ОЖ: 0x5A = 90 → 90-40 = 50°C
+    "0C": "410C 1A F8",   # RPM: (0x1A*256+0xF8)/4 = 1726
+    "0D": "410D 00",       # скорость: 0 км/ч (стоим)
+    "11": "4111 80",       # дроссель: 0x80 = 50%
+    "0B": "410B 64",       # MAP: 100 кПа
+    "0F": "410F 46",       # IAT: 0x46-40 = 30°C
+    "1F": "411F 02 58",    # runtime: 600 сек
+    "04": "4104 40",       # нагрузка: 25%
+    "06": "4106 80",       # STFT: 0%
+    "07": "4107 80",       # LTFT: 0%
+}
+
+
+class ElmHandler(socketserver.StreamRequestHandler):
+    """Обработчик TCP-соединения. Эмулирует ELM327."""
+
+    def handle(self):
+        print(f"\n🔌 Подключение: {self.client_address}")
+
+        # Отправляем приветствие
+        self.wfile.write(b"ELM327 v1.5\r\n>")
+
+        while True:
+            try:
+                data = self.rfile.readline()
+                if not data:
+                    break
+
+                cmd = data.decode("utf-8", errors="ignore").strip().upper()
+                if not cmd:
+                    continue
+
+                response = self._process(cmd)
+                if response:
+                    self.wfile.write(response)
+                    self.wfile.flush()
+
+            except (ConnectionResetError, BrokenPipeError):
+                break
+            except Exception as e:
+                print(f"   ⚠️ Ошибка: {e}")
+                break
+
+        print(f"🔌 Отключение: {self.client_address}")
+
+    def _process(self, cmd: str) -> bytes:
+        """Обрабатывает AT/OBD команду, возвращает ответ."""
+        print(f"   📥 {cmd}")
+
+        # ── AT команды ───────────────────────
+        if cmd == "ATZ":
+            resp = b"\r\nELM327 v1.5\r\n>"
+        elif cmd.startswith("ATE"):
+            resp = b"OK\r\n>"
+        elif cmd.startswith("ATL"):
+            resp = b"OK\r\n>"
+        elif cmd == "ATSP0":
+            resp = b"OK\r\n>"
+        elif cmd.startswith("ATH"):
+            resp = b"OK\r\n>"
+        elif cmd == "ATI":
+            resp = b"ELM327 v1.5\r\n>"
+        elif cmd == "ATDP":
+            resp = b"AUTO, ISO 15765-4 (CAN 11/500)\r\n>"
+
+        # ── OBD запросы ─────────────────────
+        elif cmd == "0902":  # VIN
+            resp = self._vin_response()
+
+        elif cmd == "03":  # Stored DTC
+            resp = (FAKE_DTC[0] + "\r\n" + FAKE_DTC[1] + "\r\n>").encode()
+
+        elif cmd == "07":  # Pending DTC
+            resp = (FAKE_NO_DTC + "\r\n>").encode()
+
+        elif cmd.startswith("01"):  # PID запрос
+            pid = cmd[2:4]
+            if pid in FAKE_PIDS:
+                resp = (FAKE_PIDS[pid] + "\r\n>").encode()
+            else:
+                resp = b"NO DATA\r\n>"
+
+        else:
+            resp = b"?\r\n>"
+
+        print(f"   📤 {resp.decode('utf-8', errors='ignore').strip()[:60]}")
+        return resp
+
+    def _vin_response(self) -> bytes:
+        """Формирует ответ VIN (mode 09 PID 02)."""
+        vin_hex = "".join(f"{ord(c):02X}" for c in FAKE_VIN)
+        # Формат: 014\r\n0:49 02 01 <hex bytes>
+        lines = [
+            "014\r\n",
+            f"0:49 02 01 {vin_hex}\r\n",
+            ">",
+        ]
+        return "".join(lines).encode()
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Mock ELM327")
+    parser.add_argument("--port", type=int, default=35000)
+    parser.add_argument("--host", default="0.0.0.0")
+    args = parser.parse_args()
+
+    server = socketserver.ThreadingTCPServer((args.host, args.port), ElmHandler)
+    print(f"🚗 Mock ELM327 на {args.host}:{args.port}")
+    print(f"   Подключиться: telnet localhost {args.port}")
+    print(f"   Ctrl+C чтобы остановить")
+    print()
+
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        print("\n⏹️  Стоп")
+        server.shutdown()
+
+
+if __name__ == "__main__":
+    main()