“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: версии
87 lines
2.6 KiB
Python
87 lines
2.6 KiB
Python
"""
|
||
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)
|
||
|