131 lines
4.3 KiB
Python
131 lines
4.3 KiB
Python
"""Парсинг батча ответов 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)
|