v0.48.0 — пробинг ELM327, трехуровневый профиль, рефакторинг obd/
- 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: версии
This commit is contained in:
@@ -0,0 +1,86 @@
|
||||
"""
|
||||
obd/probe.py — Пробинг ELM327: определение уровня устройства.
|
||||
|
||||
Трехуровневый каскад (использует obd/commands.py + obd/classifier.py):
|
||||
Уровень 0 (база, все клоны):
|
||||
ATE0 ATL0 ATS0 ATH1 ATSP0 ATDPN ATRV ATI
|
||||
→ хоть одна не ответила → НЕИСПРАВЕН (-1)
|
||||
|
||||
Уровень 1 (хорошие клоны):
|
||||
ATAT1 → уровень 1
|
||||
|
||||
Уровень 2 (настоящие ELM):
|
||||
ATCAF1 ATCFC1 → уровень 2
|
||||
|
||||
Принцип:
|
||||
- Каждая команда: ОДИН запрос, короткий таймаут (500мс)
|
||||
- Нет ретраев на неизвестные команды
|
||||
- Команды и классификация — в отдельных модулях
|
||||
"""
|
||||
|
||||
import logging
|
||||
import time
|
||||
|
||||
from obd.commands import L0_NAMES, L1_NAMES, L2_NAMES
|
||||
from obd.classifier import classify, determine_level
|
||||
|
||||
logger = logging.getLogger("elmer.probe")
|
||||
|
||||
# Таймауты
|
||||
PROBE_TMO = 500 # мс — таймаут на одну команду при пробинге
|
||||
INTER_CMD_DELAY = 0.05 # с — пауза между командами
|
||||
|
||||
|
||||
def probe(elm) -> dict:
|
||||
"""Пробинг ELM327: определяет уровень и профиль устройства.
|
||||
|
||||
Args:
|
||||
elm: AndrOBD с открытым connect().
|
||||
|
||||
Returns:
|
||||
{"level": 0|1|2|-1, "elm_version": str|None, ...}
|
||||
level = -1 → устройство нерабочее.
|
||||
"""
|
||||
# Собираем ответы на все команды пробинга
|
||||
responses = _probe_commands(elm, L0_NAMES + L1_NAMES + L2_NAMES)
|
||||
|
||||
# Классифицируем — определяем уровень
|
||||
result = determine_level(responses)
|
||||
|
||||
if result["level"] >= 0:
|
||||
logger.info(f"probe: device level={result['level']}")
|
||||
else:
|
||||
logger.warning("probe: device UNUSABLE")
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def _probe_commands(elm, cmds: list[str]) -> dict[str, str]:
|
||||
"""Посылает список команд и собирает ответы.
|
||||
|
||||
Returns:
|
||||
{cmd_name: raw_response}
|
||||
"""
|
||||
responses = {}
|
||||
for cmd in cmds:
|
||||
ok, raw = _try_one(elm, cmd)
|
||||
responses[cmd] = raw if ok else ""
|
||||
return responses
|
||||
|
||||
|
||||
def _try_one(elm, cmd: str) -> tuple[bool, str]:
|
||||
"""Посылает ОДНУ команду в ELM, ждёт PROBE_TMO мс.
|
||||
|
||||
Returns:
|
||||
(ok, raw_response)
|
||||
"""
|
||||
elm._write(cmd)
|
||||
time.sleep(INTER_CMD_DELAY)
|
||||
try:
|
||||
raw = elm._read(PROBE_TMO)
|
||||
except TimeoutError:
|
||||
return False, ""
|
||||
|
||||
_, ok = classify(raw)
|
||||
return (ok, raw)
|
||||
|
||||
Reference in New Issue
Block a user