219 lines
7.9 KiB
Python
219 lines
7.9 KiB
Python
"""
|
|
AndrOBD Protocol — ПОЛНАЯ копия стейт-машины AndrOBD.
|
|
|
|
Источник: github.com/fr3ts0n/AndrOBD, ElmProt.java
|
|
|
|
Состояния:
|
|
UNDEFINED → INITIALIZING → READY
|
|
Любое → BUSY (команда) → READY
|
|
Любое → ERROR → RECOVERING → READY
|
|
BUS ERROR → DISCONNECTED → RECONNECTING → READY
|
|
|
|
Каждый ответ проверяется — не тот ответ → переход в ошибку → восстановление.
|
|
"""
|
|
|
|
import logging
|
|
import time
|
|
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:
|
|
DEFAULT = 500; MIN = 50; MAX = 2000; STEP = 20; RES = 4
|
|
|
|
def __init__(self):
|
|
self._t = self.DEFAULT; self._min = self.MIN
|
|
|
|
@property
|
|
def ms(self) -> int: return self._t
|
|
@property
|
|
def atst(self) -> int: return max(1, self._t // self.RES)
|
|
|
|
def increase(self):
|
|
if self._t + self.STEP < self.MAX: self._t += self.STEP
|
|
def decrease(self):
|
|
if self._t - self.STEP >= self._min: self._t -= self.STEP
|
|
def reset(self): self._t = self.DEFAULT
|
|
|
|
|
|
# ── Протокол (AndrOBD ElmProt) ──────────────────────────────
|
|
|
|
class AndrOBD:
|
|
"""Стейт-машина ELM327 — 1:1 копия AndrOBD."""
|
|
|
|
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._state = State.UNDEFINED; self._last_cmd: 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); logger.info(f"AndrOBD: connected {self.port}")
|
|
|
|
def close(self):
|
|
if self._ser and self._ser.is_open: self._ser.close()
|
|
|
|
# ── Инициализация ───────────────────────────────────
|
|
|
|
def init(self):
|
|
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")
|
|
|
|
# ── OBD-команда ─────────────────────────────────────
|
|
|
|
def send(self, cmd: str) -> str:
|
|
if self._state == State.ERROR:
|
|
self._recover()
|
|
self._state = State.BUSY
|
|
result = self._exec(cmd, self._timing.ms)
|
|
self._state = State.READY
|
|
return result
|
|
|
|
# ── Выполнение ──────────────────────────────────────
|
|
|
|
def _exec(self, cmd: str, timeout: int) -> str:
|
|
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:
|
|
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()
|
|
logger.debug(f"AndrOBD → {cmd}")
|
|
|
|
def _read(self, timeout_ms: int) -> str:
|
|
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: 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 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):
|
|
if cur: lines.append("".join(cur)); cur.clear()
|
|
|
|
def _update_atst(self):
|
|
self._write(f"ATST{self._timing.atst:02X}"); self._try_read()
|