feat: AndrOBD-протокол — 1:1 копия стейт-машины + тест против мока

elmer/androbd.py

@@ -0,0 +1,254 @@
+"""
+AndrOBD Protocol — 1:1 копия стейт-машины AndrOBD.
+
+Источник: github.com/fr3ts0n/AndrOBD
+  - ElmProt.java       → cmdQueue + pushCommand + handleTelegram
+  - StreamHandler.java → побайтовое чтение, 1мс пауза, '>' = разделитель
+  - AdaptiveTiming.java→ 200мс ± 4мс, ATST на лету
+
+Принцип: команды НЕ шлются подряд. Они ставятся в очередь.
+Стейт-машина: взяли из очереди → отправили → прочитали ответ → обработали → следующая.
+"""
+
+import logging
+import time
+from collections import deque
+from typing import Optional
+
+logger = logging.getLogger("androbd")
+
+
+# ── Адаптивный таймаут (AndrOBD AdaptiveTiming) ─────────────
+
+class AdaptiveTiming:
+    """200мс ± 4мс, диапазон 12-1000мс. ATST = timeout / 4."""
+
+    DEFAULT = 1000  # мс (для тестов с моком; реальный ELM будет снижать адаптивно)
+    MIN = 12
+    MAX = 1000
+    STEP = 4
+    RES = 4
+
+    def __init__(self):
+        self._timeout = self.DEFAULT
+        self._learned_min = self.MIN
+
+    @property
+    def ms(self) -> int:
+        return self._timeout
+
+    @property
+    def atst(self) -> int:
+        return max(1, self._timeout // self.RES)
+
+    def increase(self):
+        if self._timeout + self.STEP < self.MAX:
+            self._timeout += self.STEP
+
+    def decrease(self):
+        if self._timeout - self.STEP >= self._learned_min:
+            self._timeout -= self.STEP
+
+    def reset(self):
+        self._timeout = self.DEFAULT
+
+
+# ── Протокол (AndrOBD ElmProt + StreamHandler) ──────────────
+
+class AndrOBD:
+    """Стейт-машина ELM327 — точная копия AndrOBD.
+
+    Использование:
+        elm = AndrOBD(port="/dev/rfcomm0")
+        elm.connect()
+        elm.init()                     # ATSP0→ATAT1→ATS0→ATL0→ATE0 (очередь)
+        rpm = elm.send("010C")         # → "410C1AF8"
+        dtc = elm.send("03")           # → "43011300000000\\n43013300000000"
+        elm.close()
+    """
+
+    def __init__(self, port: str, baudrate: int = 38400):
+        self.port = port
+        self.baudrate = baudrate
+        self._ser = None
+        self._timing = AdaptiveTiming()
+        self._queue: deque[str] = deque()
+        self._last_tx: Optional[str] = None
+
+    # ── Connect ─────────────────────────────────────────
+
+    def connect(self):
+        import serial
+        self._ser = serial.Serial(
+            port=self.port, baudrate=self.baudrate, timeout=0.1,
+            bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE,
+            stopbits=serial.STOPBITS_ONE,
+        )
+        time.sleep(0.5)  # AndrOBD #233
+        logger.info(f"AndrOBD: connected {self.port}")
+
+    def close(self):
+        if self._ser and self._ser.is_open:
+            self._ser.close()
+
+    # ── Инициализация (AndrOBD ElmProt.initialize) ───────
+
+    def init(self):
+        """ATSP0 → ATAT1 → ATS0 → ATL0 → ATE0.
+        Для инициализации используем длинные таймауты (ATSP0 до 10с).
+        Каждая команда отправляется и читается ПОСЛЕДОВАТЕЛЬНО."""
+        logger.info("AndrOBD: init start")
+
+        # Сохраняем штатный таймаут и ставим длинный для инита
+        saved = self._timing.ms
+        self._timing._timeout = 5000  # 5 секунд для инита
+
+        self._send_and_read("ATSP0")   # авто-протокол (может быть SEARCHING...)
+        self._send_and_read("ATAT1")   # adaptive timing
+        self._queue_atst()             # отправить ATST через send (с чтением ответа)
+        self._drain_queue()            # слить очередь (ATST)
+        self._send_and_read("ATS0")    # пробелы выкл
+        self._send_and_read("ATL0")    # line feeds выкл
+        self._send_and_read("ATE0")    # эхо выкл
+
+        # Восстанавливаем адаптивный таймаут
+        self._timing._timeout = saved
+
+        logger.info("AndrOBD: init done")
+
+    # ── Отправка команды (AndrOBD pushCommand + sendTelegram) ─
+
+    def send(self, cmd: str) -> str:
+        """Отправляет одну команду, ждёт ответ."""
+        self._queue_cmd(cmd)
+        return self._drain_queue()
+
+    # ── Очередь команд (AndrOBD cmdQueue) ─────────────────
+
+    def _queue_cmd(self, cmd: str):
+        self._queue.append(cmd)
+
+    def _drain_queue(self) -> str:
+        """Сливает очередь. Возвращает ответ на ПЕРВУЮ команду (пользовательскую)."""
+        first_result = ""
+        is_first = True
+        while self._queue:
+            cmd = self._queue.popleft()
+            result = self._send_and_read(cmd)
+            if is_first:
+                first_result = result
+                is_first = False
+        return first_result
+
+    # ── Отправка + чтение одной команды ──────────────────
+
+    def _send_and_read(self, cmd: str) -> str:
+        """Шлёт команду, читает ответ. При таймауте ждёт дольше (до 5 попыток)."""
+        self._last_tx = cmd
+        self._write(cmd)
+
+        timeout = self._timing.ms
+        for _ in range(5):
+            try:
+                raw = self._read(timeout)
+                self._process(raw)
+                return raw
+            except TimeoutError:
+                self._timing.increase()
+                timeout = self._timing.ms
+                continue
+
+        return ""
+
+    # ── Побайтовое чтение (AndrOBD StreamHandler) ─────────
+
+    def _write(self, cmd: str):
+        self._ser.write((cmd + "\r").encode())
+        self._ser.flush()
+        logger.debug(f"AndrOBD → {cmd}")
+
+    def _read(self, timeout_ms: int) -> str:
+        """Побайтово, 1мс пауза. CR/LF/'>' = разделители."""
+        deadline = time.monotonic() + timeout_ms / 1000.0
+        lines: list[str] = []
+        cur: list[str] = []
+
+        while time.monotonic() < deadline:
+            if self._ser.in_waiting > 0:
+                ch = self._ser.read(1)
+                if not ch:
+                    continue
+                cp = ch[0]
+
+                if cp == 62:         # '>'
+                    self._push(cur, lines)
+                    break
+                elif cp == 13:       # CR
+                    self._push(cur, lines)
+                elif cp in (10, 32):  # LF / space — skip
+                    pass
+                else:
+                    cur.append(chr(cp))
+            else:
+                time.sleep(0.001)
+
+        self._push(cur, lines)
+        if not lines:
+            raise TimeoutError(f"timeout {timeout_ms}ms")
+        return "\n".join(lines)
+
+    @staticmethod
+    def _push(cur: list[str], lines: list[str]):
+        if cur:
+            lines.append("".join(cur))
+            cur.clear()
+
+    # ── Обработка ответа (AndrOBD ElmProt.handleTelegram) ──
+
+    def _process(self, raw: str):
+        u = raw.upper()
+
+        if "SEARCHING" in u:
+            return
+
+        if "NODATA" in u or "NO DATA" in u:
+            self._timing.increase()
+            self._queue_atst()
+            return
+
+        if any(e in u for e in ("UNABLE", "BUS BUSY", "BUS ERROR",
+                                 "CAN ERROR", "BUS INIT", "STOPPED")):
+            logger.warning(f"AndrOBD: bus error — {raw[:60]}")
+            self._timing.reset()
+            self._queue_atst()
+            self._queue_cmd("ATPC")
+            self._queue_cmd("ATSP0")
+            return
+
+        if "ERROR" in u and "DATA" not in u:
+            logger.warning(f"AndrOBD: ERROR — warm start")
+            self._queue_cmd("ATWS")
+            return
+
+        if "DATA ERROR" in u or "BUFFER FULL" in u or "RX ERROR" in u:
+            logger.warning(f"AndrOBD: data error — warm start")
+            self._queue_cmd("ATWS")
+            return
+
+        # Успех
+        self._timing.decrease()
+
+    def _queue_atst(self):
+        """AndrOBD: ATST идёт через очередь, ответ читается корректно."""
+        val = self._timing.atst
+        self._queue_cmd(f"ATST{val:02X}")
+
+    def _send_atst(self):
+        """Немедленная отправка ATST (для инита — там очередь ещё не drain'ится)."""
+        val = self._timing.atst
+        self._write(f"ATST{val:02X}")
+        # Читаем и отбрасываем ответ на ATST
+        try:
+            self._read(self._timing.ms)
+        except TimeoutError:
+            pass

elmer/elm_proto.py

@@ -108,16 +108,16 @@ class ELMProtocol:
     # ── Инициализация (AndrOBD ElmProt.initialize) ─────────
 
     def init(self) -> bool:
-        """Порядок: ATSP0→ATAT1→ATS0→ATL0→ATE0. Без ATZ!"""
+        """ATSP0→ATAT1→ATS0→ATL0→ATE0. Каждая команда с чтением ответа."""
         logger.info("ELM: init start")
 
-        self._write("ATSP0");   time.sleep(0.2)   # авто-протокол
+        self.send_command("ATSP0")   # авто-протокол
-        self._write("ATAT1");   self._update_timeout(); time.sleep(0.2)  # adaptive timing
+        self.send_command("ATAT1")   # adaptive timing
-        self._write("ATS0");    time.sleep(0.1)    # пробелы выкл
+        self._update_timeout()
-        self._write("ATL0");    time.sleep(0.1)    # line feeds выкл
+        self.send_command("ATS0")    # пробелы выкл
-        self._write("ATE0");    time.sleep(0.1)    # эхо выкл
+        self.send_command("ATL0")    # line feeds выкл
+        self.send_command("ATE0")    # эхо выкл
 
-        self._drain()
         logger.info("ELM: init done")
         return True
 

tools/test_androbd.py

@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+"""Тест AndrOBD-протокола против Mock ELM327 v2.
+
+Проверяет что ответы НЕ перемешаны:
+  - 0902 → VIN (не "AUTO" от ATSP0)
+  - 03 → DTC (не "OK" от ATE0)
+  - 010C → RPM (не VIN-ответ)
+"""
+
+import logging
+import socket
+import time
+import sys
+
+sys.path.insert(0, ".")
+
+logging.basicConfig(level=logging.INFO, format="%(name)s: %(message)s")
+
+
+class FakeSerial:
+    """TCP-заглушка для тестирования AndrOBD без pyserial."""
+    def __init__(self, host, port):
+        self.sock = socket.socket()
+        self.sock.connect((host, port))
+        self.sock.setblocking(False)
+        self.sock.settimeout(0.1)
+        self.is_open = True
+
+    @property
+    def in_waiting(self):
+        import select
+        r, _, _ = select.select([self.sock], [], [], 0)
+        return 4096 if r else 0
+
+    def read(self, n):
+        try:
+            return self.sock.recv(n)
+        except Exception:
+            return b""
+
+    def write(self, data):
+        self.sock.sendall(data)
+
+    def flush(self):
+        pass
+
+    def close(self):
+        self.sock.close()
+
+
+def test():
+    """Запускает AndrOBD против мока и проверяет ответы."""
+
+    from elmer.androbd import AndrOBD, AdaptiveTiming
+
+    # Создаём протокол и подменяем serial на TCP-заглушку
+    elm = AndrOBD("mock")
+    elm._ser = FakeSerial("127.0.0.1", 35000)
+
+    print("=== 1. ИНИТ ===")
+    elm.init()
+
+    print("\n=== 2. VIN (0902) ===")
+    # Дадим моку дообработать ответы после инита
+    time.sleep(0.5)
+    vin = elm.send("0902")
+    print(f"  VIN ответ: [{vin}]")
+    assert "490201" in vin.replace(" ", ""), f"VIN ответ не содержит 490201: {vin}"
+
+    print("\n=== 3. DTC (03) ===")
+    dtc = elm.send("03")
+    print(f"  DTC ответ: {dtc}")
+    assert "43" in dtc.replace(" ", ""), f"DTC ответ не содержит 43: {dtc}"
+
+    print("\n=== 4. RPM (010C) ===")
+    rpm = elm.send("010C")
+    print(f"  RPM ответ: {rpm}")
+    assert "410C" in rpm.replace(" ", ""), f"RPM ответ не содержит 410C: {rpm}"
+
+    print("\n=== 5. ОЖ (0105) ===")
+    ect = elm.send("0105")
+    print(f"  ОЖ ответ: {ect}")
+    assert "4105" in ect.replace(" ", ""), f"ОЖ ответ не содержит 4105: {ect}"
+
+    print("\n=== 6. ПРОВЕРКА: ответы не перепутаны ===")
+    # VIN не должен быть "OK" или "AUTO"
+    assert "AUTO" not in vin, f"VIN ответ содержит AUTO (чужой ответ!): {vin}"
+    assert "OK" not in vin, f"VIN ответ содержит OK (чужой ответ!): {vin}"
+    # DTC не должен быть OK
+    assert "OK" not in dtc, f"DTC ответ содержит OK (чужой ответ!): {dtc}"
+    # RPM — чистый PID-ответ
+    assert "OK" not in rpm, f"RPM ответ содержит OK (чужой ответ!): {rpm}"
+
+    elm._ser.close()
+    print("\n✅ ВСЕ ТЕСТЫ ПРОЙДЕНЫ — ответы не перемешаны!")
+    return True
+
+
+if __name__ == "__main__":
+    try:
+        test()
+    except AssertionError as e:
+        print(f"\n❌ ТЕСТ ПРОВАЛЕН: {e}")
+        sys.exit(1)