""" obd/protocol.py — ELM327 стейт-машина (AndrOBD). Точная копия логики из AndrOBD (ElmProt.java, github.com/fr3ts0n/AndrOBD). ## Архитектура ┌──────────┐ команда ┌──────────┐ │ READY │──────────────▶│ BUSY │ └──────────┘ └────┬─────┘ ▲ │ ответ получен │ ┌────────────────┘ │ ▼ ┌────┴─────┐ ошибка ┌──────────┐ │ ERROR │◀─────────│ (любое) │ └────┬─────┘ └──────────┘ │ восстановление ▲ └─────────────────────┘ ## Использование elm = AndrOBD("/dev/rfcomm0", 38400) elm.connect() elm.init() vin = elm.send("0902") rpm = elm.send("010C") elm.close() ## Ключевые особенности - Байт-за-байтом чтение с 1мс поллингом - `>` как разделитель ответов (промпт ELM327) - Адаптивный таймаут (200мс ± 4мс, ATST) - Восстановление после BUS ERROR (ATPC → ATSP0) - Не тот ответ → переход в ERROR → восстановление """ import logging import time from typing import Optional from obd.state import State, Rsp from obd.timing import AdaptiveTiming logger = logging.getLogger("androbd") class AndrOBD: """Стейт-машина ELM327 — 1:1 копия AndrOBD (ElmProt.java). Управляет жизненным циклом ELM327: 1. connect() — открыть serial/Bluetooth порт 2. init() — инициализация (ATSP0, ATAT1, ATST, ATS0, ATL0, ATE0) 3. send(cmd) — отправить OBD-команду, получить ответ 4. close() — закрыть порт Автоматически обрабатывает: таймауты, BUS ERROR, восстановление. """ INIT_TMO = 10000 # мс — таймаут для команд инициализации DEF_TMO = 200 # мс — начальный таймаут (заменяется AdaptiveTiming) def __init__(self, port: str, baudrate: int = 38400): """port — устройство (напр. /dev/rfcomm0), baudrate — скорость.""" self.port = port self.baudrate = baudrate self._ser = None self._timing = AdaptiveTiming() self._state = State.UNDEFINED self._last_cmd: Optional[str] = None def connect(self): """Открыть serial-соединение с ELM327.""" 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) logger.info(f"AndrOBD: connected {self.port}") def close(self): """Закрыть serial-соединение.""" if self._ser and self._ser.is_open: self._ser.close() def init(self): """Инициализация ELM327 — 6 AT-команд. ATSP0 — авто-протокол ATAT1 — адаптивный таймаут вкл ATSTxx — установить таймаут ATS0 — без пробелов в ответах ATL0 — без перевода строки ATE0 — без эха """ logger.info("AndrOBD: init") self._state = State.INITIALIZING self._exec("ATSP0", self.INIT_TMO) self._exec("ATAT1", self.DEF_TMO * 5) self._update_atst() self._exec("ATS0", self.DEF_TMO * 5) self._exec("ATL0", self.DEF_TMO * 5) self._exec("ATE0", self.DEF_TMO * 5) self._state = State.READY logger.info("AndrOBD: ready") def send(self, cmd: str) -> str: """Отправить OBD-команду и получить ответ. cmd — команда (напр. '0105', '0902', '03'). Возвращает сырой ответ ELM327. При ошибке — авто-восстановление. """ if self._state == State.ERROR: self._recover() self._state = State.BUSY result = self._exec(cmd, self._timing.ms) if self._state == State.BUSY: self._state = State.READY return result # ── Приватные методы ────────────────────────────────── def _exec(self, cmd: str, timeout: int) -> str: """Выполнить команду ELM327 с ретраями (до 10 попыток). Архитектура AndrOBD: ELM327 может не ответить, если: - Адаптер ещё не инициализирован (SEARCHING) - ЭБУ медленно отвечает (увеличиваем AdaptiveTiming) - Плохой контакт / потеря связи (Busy→Timeout→Error) 10 ретраев — жёсткий лимит из AndrOBD, после — ERROR. """ self._last_cmd = cmd self._write(cmd) t = timeout for _ in range(10): try: return self._handle(self._read(t)) except TimeoutError: # Инициализация: каждый ретрай +1с (ELM может быть занят init) if self._state == State.INITIALIZING: t += 1000 else: # Активный режим: растём таймаут адаптивно self._timing.increase() t = self._timing.ms logger.error(f"AndrOBD: no response for {cmd}") self._state = State.ERROR return "" def _handle(self, raw: str) -> str: """Обработать ответ ELM327: классификация → обновление таймаута. Возвращает raw как есть — декодирование данных на уровень выше. """ t = Rsp.identify(raw) if t == Rsp.SEARCHING: return raw if t == Rsp.OK: self._timing.decrease() return raw if t == Rsp.NODATA: # NODATA = ЭБУ жив, но не может ответить (например, PID недоступен) # Увеличиваем таймаут — возможно ЭБУ медленный self._timing.increase() self._update_atst() return raw # BUS ERROR — сброс протокола if t in (Rsp.UNABLE, Rsp.BUS_BUSY, Rsp.BUS_ERROR, Rsp.CAN_ERROR, Rsp.BUS_INIT, Rsp.STOPPED): logger.warning(f"AndrOBD: BUS ERROR ({t})") self._state = State.DISCONNECTED self._timing.reset() self._update_atst() self._write("ATPC") # закрыть протокол self._try_read() self._write("ATSP0") # переоткрыть авто-протокол self._try_read() return raw # Другие ошибки — warm start if t in (Rsp.ERROR, Rsp.DATA_ERROR, Rsp.BUFFER_FULL, Rsp.RX_ERROR): logger.warning(f"AndrOBD: {t} — warm start") self._state = State.ERROR self._write("ATWS") self._try_read() return raw # Данные — успешный ответ self._timing.decrease() return raw def _recover(self): """Восстановление после ошибки: ATWS → ATSP0 → ATE0.""" """Восстановление после ошибки: ATWS → ATSP0 → ATE0.""" logger.info("AndrOBD: recovering...") self._state = State.INITIALIZING self._write("ATWS") self._try_read() self._write("ATSP0") self._try_read() self._write("ATE0") self._try_read() self._state = State.READY def _write(self, cmd: str): """Отправить команду в ELM327: сброс буфера → write + CR → flush.""" self._ser.reset_input_buffer() # сбросить хвосты предыдущего ответа self._ser.write((cmd + "\r").encode()) self._ser.flush() logger.debug(f"AndrOBD → {cmd}") def _read(self, timeout_ms: int) -> str: """Прочитать ответ ELM327 байт-за-байтом. Читает до символа `>` (промпт) или до таймаута. Возвращает сырой ответ без `>`. """ dl = time.monotonic() + timeout_ms / 1000.0 lines, cur = [], [] got_prompt = False while time.monotonic() < dl: if self._ser.in_waiting > 0: ch = self._ser.read(1) if not ch: continue cp = ch[0] if cp == 62: # '>' — промпт ELM327 self._push(cur, lines) got_prompt = True break elif cp == 13: # CR — конец строки self._push(cur, lines) elif cp in (10, 32): # LF и пробел — игнорируем pass else: cur.append(chr(cp)) else: time.sleep(0.001) # поллинг 1мс self._push(cur, lines) if not got_prompt: raise TimeoutError(f"timeout {timeout_ms}ms") return "\n".join(lines) def _try_read(self, timeout: int = 5000): """Прочитать и проигнорировать ответ (для команд восстановления).""" try: self._read(timeout) except TimeoutError: pass @staticmethod def _push(cur, lines): """Добавить накопленные байты как строку в lines.""" if cur: lines.append("".join(cur)) cur.clear() def _update_atst(self): """Отправить ATST с текущим значением адаптивного таймаута.""" self._write(f"ATST{self._timing.atst:02X}") self._try_read()