elmer/tools/mock_elm327.py

#!/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}")

        # Реальный ELM327 НЕ шлёт приветствие при BT-подключении — ждёт ATZ
        buf = b""
        while True:
            try:
                # Читаем побайтово — принимает \r или \n как конец строки
                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:
                            response = self._process(cmd)
                            if response:
                                self.wfile.write(response)
                                self.wfile.flush()
                else:
                    buf += ch

            except (ConnectionResetError, BrokenPipeError, OSError):
                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":
            self._sleep(2.5)  # реальный ELM перезагружается ~3 сек
            resp = b"\r\n\r\nELM327 v1.5\r\n>"

        elif cmd == "ATSP0":
            # Имитация SEARCHING как у реального ELM
            self.wfile.write(b"SEARCHING...\r\n")
            self.wfile.flush()
            self._sleep(1.8)
            resp = b"AUTO, ISO 15765-4 (CAN 11/500)\r\n>"

        elif cmd.startswith("ATE"):
            resp = b"OK\r\n>"
        elif cmd.startswith("ATL"):
            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 — ответ длинный, пауза побольше
            self._sleep(0.5)
            resp = self._vin_response()

        elif cmd == "03":  # Stored DTC
            self._sleep(0.3)
            resp = (FAKE_DTC[0] + "\r\n" + FAKE_DTC[1] + "\r\n>").encode()

        elif cmd == "07":  # Pending DTC
            self._sleep(0.3)
            resp = (FAKE_NO_DTC + "\r\n>").encode()

        elif cmd.startswith("01"):  # PID запрос
            pid = cmd[2:4]
            self._sleep(0.2)
            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 _sleep(self, seconds: float):
        """Пауза с случайным разбросом ±30%, как у реального железа."""
        import random
        time.sleep(seconds * random.uniform(0.7, 1.3))

    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()

    class ReuseTCPServer(socketserver.ThreadingTCPServer):
        allow_reuse_address = True

    server = ReuseTCPServer((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()