diff --git a/elmer/androbd.py b/elmer/androbd.py new file mode 100644 index 0000000..d75d1e1 --- /dev/null +++ b/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 diff --git a/elmer/elm_proto.py b/elmer/elm_proto.py index 0387c9c..cb00188 100644 --- a/elmer/elm_proto.py +++ b/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._write("ATAT1"); self._update_timeout(); time.sleep(0.2) # adaptive timing - self._write("ATS0"); time.sleep(0.1) # пробелы выкл - self._write("ATL0"); time.sleep(0.1) # line feeds выкл - self._write("ATE0"); time.sleep(0.1) # эхо выкл + self.send_command("ATSP0") # авто-протокол + self.send_command("ATAT1") # adaptive timing + self._update_timeout() + self.send_command("ATS0") # пробелы выкл + self.send_command("ATL0") # line feeds выкл + self.send_command("ATE0") # эхо выкл - self._drain() logger.info("ELM: init done") return True diff --git a/tools/test_androbd.py b/tools/test_androbd.py new file mode 100644 index 0000000..50cd984 --- /dev/null +++ b/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)