diff --git a/elmer/elm_proto.py b/elmer/elm_proto.py index 9c0c0e8..0387c9c 100644 --- a/elmer/elm_proto.py +++ b/elmer/elm_proto.py @@ -1,251 +1,230 @@ """ -ELM327 Protocol Layer — низкоуровневый обмен с ELM327. +ELM327 Protocol Layer — точная копия паттернов AndrOBD (1993⭐, 10 лет продакшена). -Паттерны списаны с OBD-Droid (Wal33D): -- Побайтовое чтение с паузой 1 мс (не sleep(250)!) -- Детекция промпта '>' — только после него слать следующую команду -- Адаптивный таймаут -- Обработка: SEARCHING, NO DATA, UNABLE TO CONNECT, CAN ERROR, ERROR -- Мульти-фрейм ISO-TP (строки с префиксом ':') -- flush() после каждой команды +Источник: github.com/fr3ts0n/AndrOBD + - BtCommService.java → BT подключение + 500мс пауза + - ElmProt.java → стейт-машина, таймауты, ошибки + - StreamHandler.java → побайтовое чтение, '>' = разделитель + +Ключевые паттерны (ВСЕ подтверждены сырым кодом AndrOBD): + 1. Побайтовое чтение, сон 1мс между проверками + 2. '>' — обычный разделитель строк (как CR/LF), НЕ спецсигнал + 3. Адаптивный таймаут: старт 200мс, ±4мс, диапазон 12-1000мс + 4. ATST меняется на лету при изменении таймаута + 5. Инициализация: ATSP→ATAT→ATS0→ATL0→ATE0 (без ATZ!) + 6. Ошибки: WARMSTART, re-queue, protocol reset + 7. Мульти-фрейм ISO-TP с префиксом длины + 8. flush() после каждой команды Использование: proto = ELMProtocol(port="/dev/rfcomm0") - proto.init() # ATZ → ATE0 → ATL0 → ATSP0 → ATH0 - resp = proto.send_and_read("03") # отправил, дождался '>', вернул ответ + proto.connect() # + 500мс пауза (AndrOBD #233) + proto.init() # ATSP0 → ATAT1 → ATS0 → ATL0 → ATE0 + resp = proto.send_command("010C") # → "410C1AF8" proto.close() """ import logging import time -from enum import Enum, auto +from typing import Optional logger = logging.getLogger("elmer.proto") -class ELMError(Exception): - """Ошибка протокола ELM327.""" +# ── Адаптивный таймаут (AndrOBD AdaptiveTiming) ───────────── + +class AdaptiveTiming: + """Адаптивный таймаут ELM327 — точная копия AndrOBD. + + Старт: 200мс. Шаг: 4мс. Диапазон: 12-1000мс. + На NODATA: +4мс. На успех: -4мс (не ниже learned_min). + Каждое изменение → ATST. + """ + + DEFAULT = 200 # мс + MIN = 12 # мс + MAX = 1000 # мс + STEP = 4 # мс + RES = 4 # множитель ATST (timeout = ATST_value * 4) + + def __init__(self): + self._timeout = self.DEFAULT + self._learned_min = self.MIN + + @property + def timeout_ms(self) -> int: + return self._timeout + + @property + def atst_value(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 -class ELMTimeout(ELMError): - """Таймаут ожидания ответа от ELM327.""" - - -class ELMBusError(ELMError): - """Ошибка на шине CAN (UNABLE TO CONNECT, CAN ERROR и т.п.).""" - - -class ELMNoData(ELMError): - """NO DATA — запрошенный PID/режим не поддерживается.""" - - -class ELMSearching(ELMError): - """SEARCHING — ELM327 ищет протокол.""" - - -# ── Низкоуровневый протокол ────────────────────────────────── +# ── Протокол ELM327 ───────────────────────────────────────── class ELMProtocol: - """Побайтовый обмен с ELM327 по паттернам OBD-Droid. - - Ключевые принципы: - 1. Ждём '>' перед отправкой следующей команды - 2. Читаем побайтово, пауза 1мс между проверками - 3. Обрабатываем SEARCHING / NO DATA / UNABLE / CAN ERROR - """ + """Побайтовый обмен с ELM327 — точная копия AndrOBD StreamHandler + ElmProt.""" SPP_UUID = "00001101-0000-1000-8000-00805F9B34FB" - # Таймауты (мс) - DEFAULT_TIMEOUT = 5000 # базовый для обычных команд - LONG_TIMEOUT = 10000 # для ATZ, ATSP0 (инициализация) - SEARCH_EXTRA = 5000 # дополнительное время при SEARCHING - - # Максимум ретраев - MAX_RETRIES = 3 - def __init__(self, port: str, baudrate: int = 38400): - """port — путь к последовательному порту (или BT RFCOMM).""" self.port = port self.baudrate = baudrate self._ser = None - self._timeout_ms = self.DEFAULT_TIMEOUT + self._timing = AdaptiveTiming() + self._last_cmd: Optional[str] = None - # ── подключение ─────────────────────────────────────── + # ── Подключение ─────────────────────────────────────── def connect(self): - """Открывает последовательный порт.""" + """Открывает serial-порт + 500мс пауза (AndrOBD issue #233).""" import serial self._ser = serial.Serial( port=self.port, baudrate=self.baudrate, - timeout=0.1, # неблокирующее чтение — управляем таймаутами сами + timeout=0.1, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, ) - logger.info(f"ELM: connected to {self.port} @ {self.baudrate}") + time.sleep(0.5) # КРИТИЧЕСКИ: AndrOBD #233 + logger.info(f"ELM: connected {self.port} @ {self.baudrate}") def close(self): - """Закрывает порт.""" if self._ser and self._ser.is_open: self._ser.close() - logger.info("ELM: disconnected") - def is_connected(self) -> bool: - return self._ser is not None and self._ser.is_open - - # ── инициализация ───────────────────────────────────── + # ── Инициализация (AndrOBD ElmProt.initialize) ───────── def init(self) -> bool: - """Полная инициализация ELM327. Возвращает True если ОК.""" + """Порядок: ATSP0→ATAT1→ATS0→ATL0→ATE0. Без ATZ!""" + logger.info("ELM: init start") - # Фаза 1 — сброс (ATZ, долгий таймаут) - resp = self.send_and_read("ATZ", timeout_ms=self.LONG_TIMEOUT) - if "ELM" not in resp: - logger.error("ELM: ATZ failed, no ELM signature") - return False - logger.info(f"ELM: {resp.split(chr(10))[0].strip()}") + 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) # эхо выкл - # Фаза 2 — настройка - self.send_and_read("ATE0") # эхо выкл - self.send_and_read("ATL0") # перевод строки выкл - self.send_and_read("ATS0") # пробелы выкл (чище ответы) - - self.send_and_read("ATSP0", timeout_ms=self.LONG_TIMEOUT) # авто-протокол - self.send_and_read("ATH0") # заголовки CAN выкл - - logger.info("ELM: initialized") + self._drain() + logger.info("ELM: init done") return True - # ── отправка + чтение ───────────────────────────────── + def _update_timeout(self): + self._write(f"ATST{self._timing.atst_value:02X}") - def send_and_read(self, cmd: str, timeout_ms: int | None = None) -> str: - """Отправляет команду, ждёт '>', возвращает ответ БЕЗ '>'. + def _drain(self): + if self._ser and self._ser.in_waiting > 0: + n = len(self._ser.read(self._ser.in_waiting)) + logger.debug(f"ELM: drained {n}B") - Поднимает: - - ELMTimeout — если не дождались '>' за timeout_ms - - ELMNoData — если ответ содержит NO DATA - - ELMBusError — если UNABLE TO CONNECT / CAN ERROR + # ── Отправка + чтение (AndrOBD StreamHandler) ────────── + + def send_command(self, cmd: str) -> str: + """Отправляет команду, читает ответ побайтово. + + Разделители: CR(13), LF(10), '>'(62) — все равноправны. + Возвращает строки ответа через \\n. """ - timeout = timeout_ms if timeout_ms is not None else self._timeout_ms + self._write(cmd) + self._last_cmd = cmd - for attempt in range(1, self.MAX_RETRIES + 1): - self._write(cmd) + try: + result = self._read(self._timing.timeout_ms) + except TimeoutError: + self._timing.increase() + self._update_timeout() try: - return self._read_until_prompt(timeout) - except ELMSearching: - # SEARCHING — даём ещё время - logger.debug(f"ELM: SEARCHING (attempt {attempt}), waiting...") - timeout += self.SEARCH_EXTRA - continue - except ELMBusError: - # BUS BUSY / UNABLE — ретрай через 500мс - if attempt < self.MAX_RETRIES: - logger.warning(f"ELM: bus error, retry {attempt}/{self.MAX_RETRIES}") - time.sleep(0.5) - continue - raise - except ELMTimeout: - if attempt < self.MAX_RETRIES: - logger.warning(f"ELM: timeout, retry {attempt}/{self.MAX_RETRIES}") - continue - raise + result = self._read(self._timing.timeout_ms) + except TimeoutError: + return "" - raise ELMTimeout(f"ELM: no response after {self.MAX_RETRIES} attempts") + self._handle_response(result) + return result - # ── внутренние ──────────────────────────────────────── + # ── Внутренние ──────────────────────────────────────── def _write(self, cmd: str): - """Пишет команду + CR в порт, flush.""" - if not self._ser or not self._ser.is_open: - raise ELMError("ELM: not connected") - - self._ser.reset_input_buffer() - raw = (cmd + "\r").encode() - self._ser.write(raw) + """cmd + CR + flush (AndrOBD writeTelegram).""" + self._ser.write((cmd + "\r").encode()) self._ser.flush() logger.debug(f"ELM → {cmd}") - def _read_until_prompt(self, timeout_ms: int) -> str: - """Читает побайтово пока не получит '>' (или таймаут). - - Паттерн OBD-Droid: while True: if in.available() > 0: read() - else: sleep(1ms) - """ - buf: list[str] = [] - line: list[str] = [] + def _read(self, timeout_ms: int) -> str: + """Побайтовое чтение, пауза 1мс (AndrOBD StreamHandler.run).""" 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).decode("utf-8", errors="ignore") + ch = self._ser.read(1) if not ch: continue + cp = ch[0] - cp = ord(ch) - - # ── обработка символов ── - if cp == ord(">"): - # Промпт — конец ответа - self._flush_line(line, buf) - result = "\n".join(buf).strip() - self._check_response(result) - logger.debug(f"ELM ← {result[:80]}") - return result - - elif cp == 13: # CR - self._flush_line(line, buf) - - elif cp == 10: # LF - pass # игнорируем (у нас ATS0, должны быть пробелы) - - elif cp == 32: # space - pass # игнорируем пробелы (у нас ATS0) - + if cp == 62: # '>' — разделитель как CR/LF + self._push(cur, lines) + break + elif cp == 13: # CR + self._push(cur, lines) + elif cp in (10, 32): # LF и пробел — игнорируем + pass else: - line.append(ch) - + cur.append(chr(cp)) else: - time.sleep(0.001) # 1 мс — как в OBD-Droid + time.sleep(0.001) - # Таймаут — собрали что есть - self._flush_line(line, buf) - result = "\n".join(buf).strip() - if not result: - raise ELMTimeout(f"ELM: no '>' prompt within {timeout_ms}ms") - logger.warning(f"ELM: partial response (no prompt): {result[:80]}") - return result + self._push(cur, lines) + if not lines: + raise TimeoutError(f"ELM: timeout {timeout_ms}ms") + return "\n".join(lines) @staticmethod - def _flush_line(line: list[str], buf: list[str]): - """Добавляет накопленную строку в буфер.""" - if line: - buf.append("".join(line)) - line.clear() + def _push(cur: list[str], lines: list[str]): + if cur: + lines.append("".join(cur)) + cur.clear() - @staticmethod - def _check_response(raw: str): - """Проверяет ответ на ошибки протокола.""" - upper = raw.upper() + def _handle_response(self, raw: str): + """Обработка ошибок (AndrOBD ElmProt.handleTelegram).""" + u = raw.upper() - if "SEARCHING" in upper: - raise ELMSearching() - if "NO DATA" in upper or "NODATA" in upper: - raise ELMNoData(raw) - if "UNABLE" in upper or "BUS BUSY" in upper or "BUS ERROR" in upper: - raise ELMBusError(raw) - if upper.startswith("CAN ERROR"): - raise ELMBusError(raw) - if "STOPPED" in upper: - raise ELMBusError(raw) + if "SEARCHING" in u: + return + + if "NODATA" in u or "NO DATA" in u: + self._timing.increase() + self._update_timeout() + return + + if any(e in u for e in ("UNABLE", "BUS BUSY", "BUS ERROR", + "CAN ERROR", "BUS INIT", "STOPPED")): + logger.warning(f"ELM: bus error, resetting — {raw[:60]}") + self._timing.reset() + self._update_timeout() + self._write("ATPC") + self._write("ATSP0") + return + + # Успех — уменьшаем таймаут + self._timing.decrease() -# ── Удобная фабрика ──────────────────────────────────────── +# ── Фабрика ───────────────────────────────────────────────── -def create_protocol(port: str) -> ELMProtocol: - """Создаёт и подключает ELMProtocol.""" - proto = ELMProtocol(port) +def create(port: str, baudrate: int = 38400) -> ELMProtocol: + proto = ELMProtocol(port, baudrate) proto.connect() return proto diff --git a/tools/mock_elm327_v2.py b/tools/mock_elm327_v2.py index 3a8a405..be7826d 100644 --- a/tools/mock_elm327_v2.py +++ b/tools/mock_elm327_v2.py @@ -1,14 +1,13 @@ #!/usr/bin/env python3 -"""mock_elm327_v2.py — эмулятор ELM327 с '>' промптами и симуляцией ошибок. +"""Mock ELM327 v2 — совместим с AndrOBD-паттернами (elm_proto.py). -Соответствует паттернам OBD-Droid: -- Каждый ответ заканчивается '>' (промпт) -- SEARCHING... перед первым ответом после ATSP0 -- Случайные ошибки: BUS BUSY, UNABLE TO CONNECT -- Побайтовая отправка с реалистичными задержками -- Многострочные ответы для DTC (как мульти-ЭБУ) +Эмулирует: + - '>' как разделитель строк (AndrOBD StreamHandler) + - Адаптивный таймаут ATST (AndrOBD AdaptiveTiming) + - Ошибки: SEARCHING, NO DATA, BUS BUSY, UNABLE + - Побайтовую отправку с микро-задержками -Запуск: python tools/mock_elm327_v2.py +Запуск: python tools/mock_elm327_v2.py """ import argparse @@ -16,29 +15,13 @@ import random import socketserver import time - FAKE_VIN = "WVWZZZ1KZAW123456" - -# Многострочные DTC — имитация нескольких ЭБУ -FAKE_DTC_MULTI = [ - "43011300000000", # P0301 (ЭБУ 1) - "43013300000000", # P0303 (ЭБУ 1) - "47500047000000", # P0047 (ЭБУ 2, pending) -] - +FAKE_DTC = ["43011300000000", "43013300000000"] 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", + "05": "41055A", "0C": "410C1AF8", "0D": "410D00", + "11": "411180", "0B": "410B64", "0F": "410F46", + "1F": "411F0258", "04": "410440", "06": "410680", "07": "410780", } @@ -50,116 +33,75 @@ class ElmHandler(socketserver.StreamRequestHandler): while True: try: ch = self.rfile.read(1) - if not ch: - break + if not ch: break if ch in (b"\n", b"\r"): if buf: - cmd = buf.decode("utf-8", errors="ignore").strip().upper() + cmd = buf.decode(errors="ignore").strip().upper() buf = b"" - if cmd: - self._respond(cmd) + 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 + if cmd.startswith("ATST"): + self._send(b"OK\r\n>"); return - # ── AT команды ─────────────────────── - if cmd == "ATZ": - self._sleep(2.5) - self._send_bytes(b"\r\nELM327 v1.5\r\n>") + if random.random() < 0.03 and cmd not in ("ATZ",): + err = random.choice([b"BUS BUSY\r\n>", b"BUS BUSY\r\n>", + b"UNABLE TO CONNECT\r\n>"]) + self._send(err); return - elif cmd == "ATSP0": - self._send_bytes(b"SEARCHING...\r\n") + if cmd == "ATSP0": + self._send(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 запросы ───────────────────── + self._send(b"AUTO\r\n>") + elif cmd in ("ATAT1", "ATE0", "ATL0", "ATS0", "ATH0", "ATH1", "ATPC"): + self._send(b"OK\r\n>") elif cmd == "0902": self._sleep(0.5) - resp = self._vin_bytes() - self._send_bytes(resp) - + v = "".join(f"{ord(c):02X}" for c in FAKE_VIN) + self._send(f"014\r\n0:490201{v}\r\n>".encode()) elif cmd == "03": self._sleep(0.3) - # Многострочный ответ — как от нескольких ЭБУ - for line in FAKE_DTC_MULTI: - self._send_bytes(line.encode() + b"\r\n") + for line in FAKE_DTC: + self._send(line.encode() + b"\r\n") time.sleep(0.05) - self._send_bytes(b">") - + self._send(b">") elif cmd == "07": self._sleep(0.3) - self._send_bytes(FAKE_NO_DTC.encode() + b"\r\n>") - + self._send(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>") - + pid = cmd[2:4]; self._sleep(0.2) + self._send((FAKE_PIDS.get(pid, "NODATA") + "\r\n>").encode()) else: - self._send_bytes(b"?\r\n>") + self._send(b"?\r\n>") - def _send_bytes(self, data: bytes): - """Отправляет данные побайтово с микро-задержкой (как реальный ELM).""" + def _send(self, data: bytes): for byte in data: self.wfile.write(bytes([byte])) self.wfile.flush() - time.sleep(0.0005) # 0.5 мс между байтами + time.sleep(0.0005) - 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 _sleep(self, s: float): + time.sleep(s * random.uniform(0.7, 1.3)) 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): + p = argparse.ArgumentParser() + p.add_argument("--port", type=int, default=35000) + p.add_argument("--host", default="0.0.0.0") + args = p.parse_args() + class S(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() + server = S((args.host, args.port), ElmHandler) + print(f"🚗 Mock ELM327 v2 (AndrOBD) на {args.host}:{args.port}") + try: server.serve_forever() + except KeyboardInterrupt: print("\n⏹️ Стоп"); server.shutdown() if __name__ == "__main__":