“Naeel”
aa37a1a70c
- obd/probe.py: трехуровневый каскад (L0/L1/L2) - obd/commands.py: каталог всех AT-команд с метаданными - obd/classifier.py: классификация ответов + определение уровня - obd/connection.py: транспортный слой (SerialTransport) - obd/protocol.py: init() только база, без ATAT1/ATST - api/db.py: таблица device_profiles по BT MAC - api/scripts.py: три уровня скриптов (l0/l1/l2) - api/routes.py: /elm/probe, /elm/profile/<mac>, /script?level= - web/templates/index.html: v0.48.0 - CHANGELOG.md, doc/architecture.md, resume.txt: версии
106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
"""
|
|
obd/connection.py — Транспортный слой ELM327 (serial/Bluetooth).
|
|
|
|
Чтение байт-за-байтом с поллингом 1мс, запись + flush.
|
|
Выделен из protocol.py чтобы разделить I/O и логику протокола.
|
|
"""
|
|
|
|
import logging
|
|
import time
|
|
|
|
logger = logging.getLogger("elm.connection")
|
|
|
|
|
|
class SerialTransport:
|
|
"""Транспорт: открыть порт, читать/писать байты, закрыть.
|
|
|
|
Не знает протокол. Только I/O.
|
|
"""
|
|
|
|
def __init__(self, port: str, baudrate: int = 38400):
|
|
self.port = port
|
|
self.baudrate = baudrate
|
|
self._ser = None
|
|
|
|
def connect(self):
|
|
"""Открыть serial-соединение."""
|
|
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"SerialTransport: connected {self.port}")
|
|
|
|
def close(self):
|
|
"""Закрыть serial-соединение."""
|
|
if self._ser and self._ser.is_open:
|
|
self._ser.close()
|
|
logger.info("SerialTransport: closed")
|
|
|
|
@property
|
|
def is_open(self) -> bool:
|
|
"""Открыт ли порт."""
|
|
return self._ser is not None and self._ser.is_open
|
|
|
|
def write(self, data: str):
|
|
"""Отправить данные в ELM: data + CR → flush."""
|
|
self._ser.write((data + "\r").encode())
|
|
self._ser.flush()
|
|
logger.debug(f"→ {data}")
|
|
|
|
def read(self, timeout_ms: int) -> str:
|
|
"""Прочитать ответ ELM327 байт-за-байтом.
|
|
|
|
Читает до символа '>' (промпт) или до таймаута.
|
|
Возвращает сырой ответ без '>'.
|
|
|
|
Raises:
|
|
TimeoutError: если '>' не получен за timeout_ms.
|
|
"""
|
|
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"read 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()
|