refactor: архитектура сервера — obd/ brain/ api/
- obd/protocol.py — ELM327 стейт-машина AndrOBD (подробные докстринги) - brain/client.py — LLM-клиент Diagnoser - brain/prompts.py — системные промпты - api/db.py — SQLite (sessions + старые таблицы) - api/scripts.py — сборка диагностических скриптов - api/parser.py — парсинг ответов ELM327 - api/routes.py — все REST-эндпоинты - api/config.py — загрузка конфига - web/app.py — точка входа Flask (упрощена) Удалено: elmer/elm.py, elmer/elm_proto.py, web/raw_endpoint.py
This commit is contained in:
+130
@@ -0,0 +1,130 @@
|
||||
"""Парсинг батча ответов ELM327 в структуру для LLM.
|
||||
|
||||
Поддерживает:
|
||||
- VIN (mode 09 PID 02) — из decoded и fallback из raw HEX
|
||||
- DTC stored/pending (mode 03/07) — из decoded и fallback из raw HEX
|
||||
- PID параметры (mode 01) — из decoded
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger("elmer.parser")
|
||||
|
||||
|
||||
def parse_batch(responses: list[dict]) -> dict:
|
||||
"""Парсит батч ответов в структуру для LLM.
|
||||
|
||||
Returns:
|
||||
{"vin": str|None, "dtc_stored": [str], "dtc_pending": [str],
|
||||
"parameters": [{"name": str, "value": str}], "raw_log": [str]}
|
||||
"""
|
||||
result = {
|
||||
"vin": None,
|
||||
"dtc_stored": [],
|
||||
"dtc_pending": [],
|
||||
"parameters": [],
|
||||
"raw_log": [],
|
||||
}
|
||||
|
||||
for r in responses:
|
||||
cmd = (r.get("cmd") or "").strip()
|
||||
raw = (r.get("raw") or "").strip()
|
||||
decoded = (r.get("decoded") or "").strip()
|
||||
|
||||
result["raw_log"].append(f"→ {cmd}\n← {raw}")
|
||||
|
||||
_parse_vin(result, raw, decoded)
|
||||
_parse_dtc(result, raw, decoded, mode="03", key="dtc_stored", prefix="DTC stored:")
|
||||
_parse_dtc(result, raw, decoded, mode="47", key="dtc_pending", prefix="DTC pending:")
|
||||
_parse_pid(result, decoded, cmd)
|
||||
|
||||
logger.info(f"[{cmd}] decoded={decoded[:60]}")
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def _parse_vin(result: dict, raw: str, decoded: str):
|
||||
if decoded.startswith("VIN:"):
|
||||
vin = decoded.replace("VIN:", "").strip()
|
||||
if len(vin) == 17:
|
||||
result["vin"] = vin
|
||||
return
|
||||
|
||||
# Fallback: парсим VIN из raw HEX
|
||||
if "49" in raw and ("02" in raw or "4902" in raw.replace(" ", "")):
|
||||
clean = raw.replace(":", "").replace(" ", "").upper()
|
||||
if "490201" in clean:
|
||||
hex_str = clean.split("490201")[-1][:34]
|
||||
vin = ""
|
||||
for i in range(0, len(hex_str) - 1, 2):
|
||||
try:
|
||||
vin += chr(int(hex_str[i:i+2], 16))
|
||||
except (ValueError, OverflowError):
|
||||
pass
|
||||
if len(vin) == 17:
|
||||
result["vin"] = vin
|
||||
|
||||
|
||||
def _parse_dtc(result: dict, raw: str, decoded: str, *, mode: str, key: str, prefix: str):
|
||||
if decoded.startswith(prefix):
|
||||
codes = decoded.replace(prefix, "").strip()
|
||||
if codes != "none":
|
||||
result[key] = [c.strip() for c in codes.split()]
|
||||
return
|
||||
|
||||
# Fallback: парсим DTC из raw HEX (43XX... или 47XX...)
|
||||
clean = raw.replace(" ", "").upper()
|
||||
if clean.startswith(mode) and len(clean) >= 4:
|
||||
codes = _decode_dtc_bytes(clean[2:])
|
||||
if codes:
|
||||
result[key] = codes
|
||||
|
||||
|
||||
def _decode_dtc_bytes(hex_str: str) -> list[str]:
|
||||
"""Декодирует HEX-строку DTC (после 43/47) в коды."""
|
||||
codes = []
|
||||
i = 2 # skip byte count
|
||||
while i + 3 < len(hex_str):
|
||||
try:
|
||||
a = int(hex_str[i:i+2], 16)
|
||||
b = int(hex_str[i+2:i+4], 16)
|
||||
p = {0: "P", 1: "C", 2: "B", 3: "U"}.get(a >> 6, "?")
|
||||
code = f"{p}{(a>>4)&3}{a&15}{b>>4:X}{b&15:X}"
|
||||
if code != "P0000":
|
||||
codes.append(code)
|
||||
except Exception:
|
||||
pass
|
||||
i += 4
|
||||
return codes
|
||||
|
||||
|
||||
def _parse_pid(result: dict, decoded: str, cmd: str):
|
||||
if ":" not in decoded:
|
||||
return
|
||||
if decoded.startswith(("VIN", "DTC", "ELM", "Protocol")):
|
||||
return
|
||||
parts = decoded.split(":", 1)
|
||||
if len(parts) == 2:
|
||||
result["parameters"].append({
|
||||
"name": parts[0].strip(),
|
||||
"value": parts[1].strip(),
|
||||
})
|
||||
|
||||
|
||||
def format_no_llm(parsed: dict) -> str:
|
||||
"""Форматирует ответ без LLM."""
|
||||
lines = []
|
||||
if parsed["vin"]:
|
||||
lines.append(f"VIN: {parsed['vin']}")
|
||||
if parsed["dtc_stored"]:
|
||||
lines.append(f"Ошибки: {', '.join(parsed['dtc_stored'])}")
|
||||
if parsed["dtc_pending"]:
|
||||
lines.append(f"Pending: {', '.join(parsed['dtc_pending'])}")
|
||||
if parsed["parameters"]:
|
||||
lines.append("Параметры:")
|
||||
for p in parsed["parameters"]:
|
||||
lines.append(f" {p['name']}: {p['value']}")
|
||||
if not lines:
|
||||
lines.append("Данные не распознаны.")
|
||||
lines.append("\n(LLM не настроен — только сырые данные)")
|
||||
return "\n".join(lines)
|
||||
Reference in New Issue
Block a user