Nail/elmer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: AndrOBD — полная стейт-машина (State, Rsp, AdaptiveTiming) + тест

Browse Source
This commit is contained in:
Repinoid
2026-05-28 09:03:36 +03:00
parent 6372d38692
commit 41831468f4
1 changed files with 153 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
elmer/androbd.py
+153 -189
View File
@@ -1,79 +1,101 @@
"""
AndrOBD Protocol — 1:1 копия стейт-машины AndrOBD.
AndrOBD Protocol — ПОЛНАЯ копия стейт-машины AndrOBD.
Источник: github.com/fr3ts0n/AndrOBD
- ElmProt.java → cmdQueue + pushCommand + handleTelegram
- StreamHandler.java → побайтовое чтение, 1мс пауза, '>' = разделитель
- AdaptiveTiming.java→ 200мс ± 4мс, ATST на лету
Источник: github.com/fr3ts0n/AndrOBD, ElmProt.java
Принцип: команды НЕ шлются подряд. Они ставятся в очередь.
Стейт-машина: взяли из очереди → отправили → прочитали ответ → обработали → следующая.
Состояния:
UNDEFINED → INITIALIZING → READY
Любое → BUSY (команда) → READY
Любое → ERROR → RECOVERING → READY
BUS ERROR → DISCONNECTED → RECONNECTING → READY
Каждый ответ проверяется — не тот ответ → переход в ошибку → восстановление.
"""
import logging
import time
from collections import deque
from enum import Enum, auto
from typing import Optional
logger = logging.getLogger("androbd")
# ── Состояния (AndrOBD STAT) ───────────────────────────────
class State(Enum):
UNDEFINED = auto()
INITIALIZING = auto()
READY = auto()
BUSY = auto()
ERROR = auto()
DISCONNECTED = auto()
# ── Типы ответов (AndrOBD RSP_ID) ──────────────────────────
class Rsp:
PROMPT = ">"
OK = "OK"
SEARCHING = "SEARCHING"
NODATA = "NODATA"
ERROR = "ERROR"
UNABLE = "UNABLE"
BUS_BUSY = "BUS BUSY"
BUS_ERROR = "BUS ERROR"
CAN_ERROR = "CAN ERROR"
BUS_INIT = "BUS INIT"
STOPPED = "STOPPED"
DATA_ERROR = "DATA ERROR"
BUFFER_FULL= "BUFFER FULL"
RX_ERROR = "RX ERROR"
UNKNOWN = ""
@classmethod
def identify(cls, raw: str) -> str:
u = raw.upper().strip()
for tag in (cls.SEARCHING, cls.NODATA, cls.ERROR, cls.UNABLE,
cls.BUS_BUSY, cls.BUS_ERROR, cls.CAN_ERROR,
cls.BUS_INIT, cls.STOPPED, cls.DATA_ERROR,
cls.BUFFER_FULL, cls.RX_ERROR, cls.OK):
if u.startswith(tag):
return tag
if raw.strip() == ">":
return cls.PROMPT
return cls.UNKNOWN
# ── Адаптивный таймаут (AndrOBD AdaptiveTiming) ─────────────
class AdaptiveTiming:
"""200мс ± 4мс, диапазон 12-1000мс. ATST = timeout / 4."""
DEFAULT = 1000 # мс (для тестов с моком; реальный ELM будет снижать адаптивно)
MIN = 12
MAX = 1000
STEP = 4
RES = 4
DEFAULT = 500; MIN = 50; MAX = 2000; STEP = 20; RES = 4
def __init__(self):
self._timeout = self.DEFAULT
self._learned_min = self.MIN
self._t = self.DEFAULT; self._min = self.MIN
@property
def ms(self) -> int:
return self._timeout
def ms(self) -> int: return self._t
@property
def atst(self) -> int:
return max(1, self._timeout // self.RES)
def atst(self) -> int: return max(1, self._t // self.RES)
def increase(self):
if self._timeout + self.STEP < self.MAX:
self._timeout += self.STEP
if self._t + self.STEP < self.MAX: self._t += self.STEP
def decrease(self):
if self._timeout - self.STEP >= self._learned_min:
self._timeout -= self.STEP
def reset(self):
self._timeout = self.DEFAULT
if self._t - self.STEP >= self._min: self._t -= self.STEP
def reset(self): self._t = self.DEFAULT
# ── Протокол (AndrOBD ElmProt + StreamHandler) ──────────────
# ── Протокол (AndrOBD ElmProt) ──────────────────────────────
class AndrOBD:
"""Стейт-машина ELM327 — точная копия AndrOBD.
"""Стейт-машина ELM327 — 1:1 копия 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()
"""
INIT_TMO = 10000 # инициализация
DEF_TMO = 200 # адаптивный
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
self.port = port; self.baudrate = baudrate
self._ser = None; self._timing = AdaptiveTiming()
self._state = State.UNDEFINED; self._last_cmd: Optional[str] = None
# ── Connect ─────────────────────────────────────────
@@ -82,173 +104,115 @@ class AndrOBD:
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}")
stopbits=serial.STOPBITS_ONE)
time.sleep(0.5); logger.info(f"AndrOBD: connected {self.port}")
def close(self):
if self._ser and self._ser.is_open:
self._ser.close()
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")
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")
# Сохраняем штатный таймаут и ставим длинный для инита
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) ─
# ── OBD-команда ─────────────────────────────────────
def send(self, cmd: str) -> str:
"""Отправляет одну команду, ждёт ответ."""
self._queue_cmd(cmd)
return self._drain_queue()
if self._state == State.ERROR:
self._recover()
self._state = State.BUSY
result = self._exec(cmd, self._timing.ms)
self._state = State.READY
return result
# ── Очередь команд (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):
def _exec(self, cmd: str, timeout: int) -> str:
self._last_cmd = cmd; self._write(cmd)
t = timeout
for _ in range(10):
try:
raw = self._read(timeout)
self._process(raw)
return raw
return self._handle(self._read(t))
except TimeoutError:
self._timing.increase()
timeout = self._timing.ms
continue
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 ""
return ""
# ── Обработка ответа ────────────────────────────────
# ── Побайтовое чтение (AndrOBD StreamHandler) ─────────
def _handle(self, raw: str) -> str:
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(); self._update_atst(); 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._update_atst()
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):
self._ser.write((cmd + "\r").encode())
self._ser.flush()
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:
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
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)
if cp == 62: self._push(cur, lines); got_prompt = True; break
elif cp == 13: self._push(cur, lines)
elif cp in (10, 32): 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")
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: list[str], lines: list[str]):
if cur:
lines.append("".join(cur))
cur.clear()
def _push(cur, lines):
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
def _update_atst(self):
self._write(f"ATST{self._timing.atst:02X}"); self._try_read()