“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: версии
108 lines
3.3 KiB
Python
108 lines
3.3 KiB
Python
"""
|
|
obd/classifier.py — Классификация ответов ELM327 и определение уровня.
|
|
|
|
Отдельный сервис:
|
|
- classify(raw) → (tag: str, is_ok: bool)
|
|
- determine_level(responses: dict) → int (-1/0/1/2)
|
|
|
|
Использует obd/commands.py для списков команд по уровням.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict, Optional, Tuple
|
|
|
|
from obd.state import Rsp
|
|
from obd.commands import L0_NAMES, L1_NAMES, L2_NAMES
|
|
|
|
logger = logging.getLogger("elmer.classifier")
|
|
|
|
|
|
def classify(raw: str) -> Tuple[str, bool]:
|
|
"""Классифицирует сырой ответ ELM327.
|
|
|
|
Returns:
|
|
(tag, is_ok)
|
|
tag: Rsp.OK / Rsp.ERROR / Rsp.UNKNOWN / ...
|
|
is_ok: True если устройство ответило нормально (OK или данные)
|
|
"""
|
|
tag = Rsp.identify(raw)
|
|
is_ok = tag in (Rsp.OK, Rsp.UNKNOWN, Rsp.NODATA, Rsp.SEARCHING)
|
|
return (tag, is_ok)
|
|
|
|
|
|
def determine_level(responses: Dict[str, str]) -> dict:
|
|
"""Определяет уровень устройства по ответам на пробинг-команды.
|
|
|
|
Args:
|
|
responses: {cmd_name: raw_response} — ответы на команды пробинга.
|
|
|
|
Returns:
|
|
Профиль: {level, elm_version, protocol, voltage, supported, unsupported, errors}
|
|
"""
|
|
result: dict = {
|
|
"level": -1,
|
|
"elm_version": None,
|
|
"protocol": None,
|
|
"voltage": None,
|
|
"supported": [],
|
|
"unsupported": [],
|
|
"errors": [],
|
|
}
|
|
|
|
# ── Уровень 0 ──────────────────────────────────
|
|
l0_ok = True
|
|
for cmd in L0_NAMES:
|
|
raw = responses.get(cmd, "")
|
|
tag, ok = classify(raw)
|
|
if ok and raw:
|
|
result["supported"].append(cmd)
|
|
if cmd == "ATI":
|
|
result["elm_version"] = raw.strip()
|
|
elif cmd == "ATDPN":
|
|
result["protocol"] = raw.strip()
|
|
elif cmd == "ATRV":
|
|
result["voltage"] = raw.strip()
|
|
else:
|
|
result["unsupported"].append(cmd)
|
|
result["errors"].append(f"{cmd}: {tag if tag else 'no response'}")
|
|
l0_ok = False
|
|
|
|
if not l0_ok:
|
|
logger.warning("classifier: L0 failed")
|
|
return result
|
|
|
|
result["level"] = 0
|
|
|
|
# ── Уровень 1 ──────────────────────────────────
|
|
l1_ok = True
|
|
for cmd in L1_NAMES:
|
|
raw = responses.get(cmd, "")
|
|
tag, ok = classify(raw)
|
|
if ok and raw:
|
|
result["supported"].append(cmd)
|
|
else:
|
|
result["unsupported"].append(cmd)
|
|
l1_ok = False
|
|
|
|
if not l1_ok:
|
|
return result
|
|
|
|
result["level"] = 1
|
|
|
|
# ── Уровень 2 ──────────────────────────────────
|
|
l2_ok = True
|
|
for cmd in L2_NAMES:
|
|
raw = responses.get(cmd, "")
|
|
tag, ok = classify(raw)
|
|
if ok and raw:
|
|
result["supported"].append(cmd)
|
|
else:
|
|
result["unsupported"].append(cmd)
|
|
l2_ok = False
|
|
|
|
if l2_ok:
|
|
result["level"] = 2
|
|
|
|
logger.info(f"classifier: level={result['level']}")
|
|
return result
|