217 lines
8.1 KiB
Python
217 lines
8.1 KiB
Python
"""
|
||
obd/protocol.py — ELM327 стейт-машина (AndrOBD).
|
||
|
||
Основана на AndrOBD (ElmProt.java, github.com/fr3ts0n/AndrOBD).
|
||
|
||
## Архитектура
|
||
┌──────────┐ команда ┌──────────┐
|
||
│ READY │──────────────▶│ BUSY │
|
||
└──────────┘ └────┬─────┘
|
||
▲ │ ответ получен
|
||
│ ┌────────────────┘
|
||
│ ▼
|
||
┌────┴─────┐ ошибка ┌──────────┐
|
||
│ ERROR │◀─────────│ (любое) │
|
||
└────┬─────┘ └──────────┘
|
||
│ восстановление ▲
|
||
└─────────────────────┘
|
||
|
||
## Зависимости
|
||
obd/connection.py — транспорт (SerialTransport)
|
||
obd/state.py — состояния/ответы (State, Rsp)
|
||
obd/timing.py — адаптивный таймаут (AdaptiveTiming)
|
||
obd/commands.py — каталог команд
|
||
obd/classifier.py — классификация ответов
|
||
|
||
## Использование
|
||
elm = AndrOBD("/dev/rfcomm0", 38400)
|
||
elm.connect()
|
||
elm.init()
|
||
vin = elm.send("0902")
|
||
elm.close()
|
||
"""
|
||
|
||
import logging
|
||
import time
|
||
from typing import Optional
|
||
|
||
from obd.connection import SerialTransport
|
||
from obd.state import State, Rsp
|
||
from obd.timing import AdaptiveTiming
|
||
|
||
logger = logging.getLogger("androbd")
|
||
|
||
|
||
class AndrOBD:
|
||
"""Стейт-машина ELM327.
|
||
|
||
Управляет жизненным циклом ELM327:
|
||
1. connect() — открыть serial/Bluetooth порт
|
||
2. init() — базовая инициализация (уровень 0)
|
||
3. send(cmd) — отправить OBD-команду, получить ответ
|
||
4. close() — закрыть порт
|
||
|
||
Автоматически обрабатывает: таймауты, BUS ERROR, восстановление.
|
||
"""
|
||
|
||
INIT_TMO = 10000 # мс — таймаут для команд инициализации
|
||
DEF_TMO = 200 # мс — начальный таймаут
|
||
ATST_CLONE = 0x96 # 150×4=600ms — фиксированный для клонов
|
||
|
||
def __init__(self, port: str, baudrate: int = 38400):
|
||
self._transport = SerialTransport(port, baudrate)
|
||
self._timing = AdaptiveTiming()
|
||
self._state = State.UNDEFINED
|
||
self._last_cmd: Optional[str] = None
|
||
|
||
def connect(self):
|
||
"""Открыть serial-соединение с ELM327."""
|
||
self._transport.connect()
|
||
logger.info(f"AndrOBD: connected {self._transport.port}")
|
||
|
||
def close(self):
|
||
"""Закрыть serial-соединение."""
|
||
self._transport.close()
|
||
|
||
def init(self):
|
||
"""Канонический init: ATE0→ATL0→ATS0→ATI→[ветвление]→ATSP0.
|
||
|
||
Эхо ПЕРВЫМ. ATI ДО протокола. Клон: ATST фикс, без ATAT1.
|
||
"""
|
||
logger.info("AndrOBD: init (L0)")
|
||
self._state = State.INITIALIZING
|
||
|
||
# Шаг 1: clone-safe, без протокола
|
||
self._exec("ATE0", self.DEF_TMO * 5)
|
||
self._transport.try_read(500) # drain после AT-команды
|
||
self._exec("ATL0", self.DEF_TMO * 5)
|
||
self._transport.try_read(500)
|
||
self._exec("ATS0", self.DEF_TMO * 5)
|
||
self._transport.try_read(500)
|
||
|
||
# Шаг 2: ATI → detectClone ДО ветвления
|
||
ati = self._exec("ATI", self.DEF_TMO * 5)
|
||
self._transport.try_read(500)
|
||
is_clone = "v1.5" in ati.lower()
|
||
if is_clone:
|
||
logger.info("AndrOBD: clone v1.5 — ATST fixed, no ATAT1")
|
||
self._exec(f"ATST{self.ATST_CLONE:02X}", self.DEF_TMO * 5)
|
||
self._transport.try_read(500)
|
||
else:
|
||
self._exec("ATAT1", self.DEF_TMO * 5)
|
||
self._transport.try_read(500)
|
||
|
||
# Шаг 3: протокол
|
||
self._exec("ATSP0", self.INIT_TMO)
|
||
self._transport.try_read(500)
|
||
|
||
self._state = State.READY
|
||
logger.info(f"AndrOBD: ready (L0, clone={is_clone})")
|
||
|
||
def init_l1(self):
|
||
"""Инициализация уровня 1: база + адаптивный тайминг."""
|
||
logger.info("AndrOBD: init L1 (+ATAT1)")
|
||
self._state = State.INITIALIZING
|
||
self._exec("ATAT1", self.DEF_TMO * 5)
|
||
self._state = State.READY
|
||
logger.info("AndrOBD: ready (L1)")
|
||
|
||
def init_l2(self):
|
||
"""Инициализация уровня 2: L1 + CAN автоформат + flow control."""
|
||
logger.info("AndrOBD: init L2 (+ATCAF1 +ATCFC1)")
|
||
self._state = State.INITIALIZING
|
||
self._exec("ATCAF1", self.DEF_TMO * 5)
|
||
self._exec("ATCFC1", self.DEF_TMO * 5)
|
||
self._state = State.READY
|
||
logger.info("AndrOBD: ready (L2)")
|
||
|
||
def send(self, cmd: str) -> str:
|
||
"""Отправить OBD-команду и получить ответ."""
|
||
if self._state == State.ERROR:
|
||
self._recover()
|
||
self._state = State.BUSY
|
||
result = self._exec(cmd, self._timing.ms)
|
||
self._transport.try_read(500) # drain после read — клон v1.5
|
||
if self._state == State.BUSY:
|
||
self._state = State.READY
|
||
return result
|
||
|
||
# ── Приватные методы ──────────────────────────────
|
||
|
||
def _exec(self, cmd: str, timeout: int) -> str:
|
||
"""Выполнить команду с ретраями (до 10)."""
|
||
self._last_cmd = cmd
|
||
self._write(cmd)
|
||
t = timeout
|
||
for _ in range(10):
|
||
try:
|
||
return self._handle(self._read(t))
|
||
except TimeoutError:
|
||
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."""
|
||
t = Rsp.identify(raw)
|
||
|
||
if t == Rsp.SEARCHING:
|
||
return raw
|
||
if t == Rsp.OK:
|
||
self._timing.decrease()
|
||
return raw
|
||
if t == Rsp.NODATA:
|
||
self._timing.increase()
|
||
return raw
|
||
|
||
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._write("ATPC")
|
||
self._try_read()
|
||
self._write("ATSP0")
|
||
self._try_read()
|
||
return raw
|
||
|
||
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):
|
||
"""Восстановление после ошибки."""
|
||
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."""
|
||
self._transport.write(cmd)
|
||
|
||
def _read(self, timeout_ms: int) -> str:
|
||
"""Прочитать ответ ELM327."""
|
||
return self._transport.read(timeout_ms)
|
||
|
||
def _try_read(self, timeout: int = 5000):
|
||
"""Прочитать и проигнорировать ответ."""
|
||
self._transport.try_read(timeout)
|
||
|