From 75d690b4eaa28002796393d8db70ca393df424bb Mon Sep 17 00:00:00 2001 From: Repinoid Date: Wed, 27 May 2026 10:55:37 +0300 Subject: [PATCH] =?UTF-8?q?refactor:=20script=5Fendpoint.py=20=D1=80=D0=B0?= =?UTF-8?q?=D0=B7=D0=B1=D0=B8=D1=82=20=D0=BD=D0=B0=20script=5Fbuilder=20+?= =?UTF-8?q?=20script=5Fparser=20+=20endpoint?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- web/script_builder.py | 26 +++++ web/script_endpoint.py | 218 +++++------------------------------------ web/script_parser.py | 130 ++++++++++++++++++++++++ 3 files changed, 183 insertions(+), 191 deletions(-) create mode 100644 web/script_builder.py create mode 100644 web/script_parser.py diff --git a/web/script_builder.py b/web/script_builder.py new file mode 100644 index 0000000..68be13c --- /dev/null +++ b/web/script_builder.py @@ -0,0 +1,26 @@ +"""Сборка диагностических скриптов.""" + + +def build_default_script() -> dict: + """Стандартный скрипт диагностики.""" + return { + "version": 1, + "title": "Базовая диагностика", + "steps": [ + {"id": "prompt_engine", "prompt": "Заведите двигатель (холостой ход) и нажмите «Далее»", + "wait_for_user": True}, + {"id": "vin", "cmd": "0902", "desc": "VIN"}, + {"id": "dtc_stored", "cmd": "03", "desc": "Ошибки (сохранённые)"}, + {"id": "dtc_pending", "cmd": "07", "desc": "Ошибки (ожидающие)"}, + {"id": "pid_05", "cmd": "0105", "desc": "Температура ОЖ"}, + {"id": "pid_0C", "cmd": "010C", "desc": "Обороты"}, + {"id": "pid_0D", "cmd": "010D", "desc": "Скорость"}, + {"id": "pid_11", "cmd": "0111", "desc": "Дроссель"}, + {"id": "pid_0B", "cmd": "010B", "desc": "MAP"}, + {"id": "pid_0F", "cmd": "010F", "desc": "IAT"}, + {"id": "pid_1F", "cmd": "011F", "desc": "Время работы"}, + {"id": "pid_04", "cmd": "0104", "desc": "Нагрузка"}, + {"id": "pid_06", "cmd": "0106", "desc": "STFT"}, + {"id": "pid_07", "cmd": "0107", "desc": "LTFT"}, + ], + } diff --git a/web/script_endpoint.py b/web/script_endpoint.py index ec4b696..314f3d2 100644 --- a/web/script_endpoint.py +++ b/web/script_endpoint.py @@ -4,148 +4,13 @@ GET /api/v1/script — выдача скрипта диагностик POST /api/v1/session/upload — приём батча, LLM-анализ, возврат диагноза """ -import json import logging -from datetime import datetime, timezone +from web.script_builder import build_default_script +from web.script_parser import parse_batch, format_no_llm logger = logging.getLogger("elmer.script") -def build_default_script() -> dict: - """Стандартный скрипт диагностики (без серверных подсказок).""" - return { - "version": 1, - "title": "Базовая диагностика", - "steps": [ - {"id": "init_atz", "cmd": "ATZ", "desc": "Сброс ELM327"}, - {"id": "init_ate0", "cmd": "ATE0", "desc": "Эхо выкл"}, - {"id": "init_atl0", "cmd": "ATL0", "desc": "Перевод строки выкл"}, - {"id": "init_atsp0", "cmd": "ATSP0", "desc": "Авто-протокол"}, - {"id": "init_ath0", "cmd": "ATH0", "desc": "Заголовки выкл"}, - {"id": "prompt_engine", "prompt": "Заведите двигатель (холостой ход) и нажмите «Далее»", - "wait_for_user": True}, - {"id": "vin", "cmd": "0902", "desc": "VIN"}, - {"id": "dtc_stored", "cmd": "03", "desc": "Ошибки (сохранённые)"}, - {"id": "dtc_pending", "cmd": "07", "desc": "Ошибки (ожидающие)"}, - {"id": "pid_05", "cmd": "0105", "desc": "Температура ОЖ"}, - {"id": "pid_0C", "cmd": "010C", "desc": "Обороты"}, - {"id": "pid_0D", "cmd": "010D", "desc": "Скорость"}, - {"id": "pid_11", "cmd": "0111", "desc": "Дроссель"}, - {"id": "pid_0B", "cmd": "010B", "desc": "MAP"}, - {"id": "pid_0F", "cmd": "010F", "desc": "IAT"}, - {"id": "pid_1F", "cmd": "011F", "desc": "Время работы"}, - {"id": "pid_04", "cmd": "0104", "desc": "Нагрузка"}, - {"id": "pid_06", "cmd": "0106", "desc": "STFT"}, - {"id": "pid_07", "cmd": "0107", "desc": "LTFT"}, - ], - } - - -def _parse_batch(responses: list[dict]) -> dict: - """Парсит батч ответов в структуру для LLM.""" - import re - - 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}") - - # ── VIN ── - if decoded.startswith("VIN:"): - vin = decoded.replace("VIN:", "").strip() - if len(vin) == 17: - result["vin"] = vin - elif "49" in raw and ("02" in raw or "4902" in raw.replace(" ", "")): - # fallback: парсим VIN из raw - # формат: "014 0:49 02 01 57 56 57..." или "490201575657..." - 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 - decoded = f"VIN: {vin}" # для лога - - # ── DTC ── - if decoded.startswith("DTC stored:"): - codes = decoded.replace("DTC stored:", "").strip() - if codes != "none": - result["dtc_stored"] = [c.strip() for c in codes.split()] - elif raw.replace(" ", "").startswith("43"): - # fallback: парсим DTC из raw 43XX... - clean = raw.replace(" ", "").upper() - if clean.startswith("43") and len(clean) >= 4: - codes = [] - i = 2 # skip byte count - while i + 3 < len(clean): - try: - a = int(clean[i:i+2], 16) - b = int(clean[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 - if codes: - result["dtc_stored"] = codes - decoded = f"DTC stored: {' '.join(codes)}" - - if decoded.startswith("DTC pending:"): - codes = decoded.replace("DTC pending:", "").strip() - if codes != "none": - result["dtc_pending"] = [c.strip() for c in codes.split()] - elif raw.replace(" ", "").startswith("47"): - # fallback: pending DTC - clean = raw.replace(" ", "").upper() - if clean.startswith("47") and len(clean) >= 4: - codes = [] - i = 2 - while i + 3 < len(clean): - try: - a = int(clean[i:i+2], 16) - b = int(clean[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 - if codes: - result["dtc_pending"] = codes - decoded = f"DTC pending: {' '.join(codes)}" - - # ── PID ── - if ":" in decoded and not decoded.startswith(("VIN", "DTC", "ELM", "Protocol")): - parts = decoded.split(":", 1) - if len(parts) == 2: - result["parameters"].append({ - "name": parts[0].strip(), - "value": parts[1].strip(), - }) - - logger.info(f"[{cmd}] decoded={decoded[:60]}") - - return result - - def _build_diagnosis_prompt(data: dict) -> str: """Строит промпт для LLM из распарсенных данных.""" parts = ["## Данные диагностики\n"] @@ -166,21 +31,21 @@ def _build_diagnosis_prompt(data: dict) -> str: if not data["vin"] and not data["dtc_stored"] and not data["parameters"]: parts.append("\n(данные не распознаны)") - parts.append(f"\n**Сырые ответы ЭБУ:**") - for line in data["raw_log"]: - parts.append(line) + parts.append("\n**Сырые ответы ЭБУ:**") + parts.extend(data["raw_log"]) parts.append("\n---") parts.append("## Запрос на анализ") - parts.append("Дай ГЛУБОКИЙ, РАЗВЁРНУТЫЙ анализ на основе этих данных. " - "Не ограничивайся кратким резюме — мне нужен полный технический разбор.\n" - "1. Разбери КАЖДУЮ ошибку: все возможные причины, от частых к редким.\n" - "2. Проанализируй КАЖДЫЙ параметр: норма/отклонение, на что влияет.\n" - "3. Найди ВЗАИМОСВЯЗИ между ошибками и параметрами.\n" - "4. Предложи КОНКРЕТНЫЙ план проверок: что сначала, что потом, как проверять.\n" - "5. Если данных мало — скажи какие PID'ы досчитать и зачем.\n" - "6. Дай оценки уверенности в ПРОЦЕНТАХ для каждой версии.") - + parts.append( + "Дай ГЛУБОКИЙ, РАЗВЁРНУТЫЙ анализ на основе этих данных. " + "Не ограничивайся кратким резюме — мне нужен полный технический разбор.\n" + "1. Разбери КАЖДУЮ ошибку: все возможные причины, от частых к редким.\n" + "2. Проанализируй КАЖДЫЙ параметр: норма/отклонение, на что влияет.\n" + "3. Найди ВЗАИМОСВЯЗИ между ошибками и параметрами.\n" + "4. Предложи КОНКРЕТНЫЙ план проверок: что сначала, что потом, как проверять.\n" + "5. Если данных мало — скажи какие PID'ы досчитать и зачем.\n" + "6. Дай оценки уверенности в ПРОЦЕНТАХ для каждой версии." + ) return "\n".join(parts) @@ -189,16 +54,11 @@ def register(app): @app.route("/api/v1/script", methods=["GET"]) def get_script(): - """Возвращает скрипт диагностики для клиента.""" from flask import jsonify - - # В будущем — выбирать скрипт в зависимости от VIN / истории - script = build_default_script() - return jsonify(script) + return jsonify(build_default_script()) @app.route("/api/v1/session/upload", methods=["POST"]) def upload_session(): - """Принимает батч сессии, анализирует через LLM, возвращает диагноз.""" from flask import request, jsonify from elmer.config import load from elmer.diagnose import Diagnoser @@ -210,23 +70,15 @@ def register(app): responses = data["responses"] logger.info(f"Upload: {len(responses)} responses") + parsed = parse_batch(responses) - parsed = _parse_batch(responses) - - # LLM cfg = load() api_key = cfg["llm"]["api_key"] if not api_key: - # Без LLM — возвращаем распарсенные данные return jsonify({ - "diagnosis": _format_no_llm(parsed), - "parsed": { - "vin": parsed["vin"], - "dtc_stored": parsed["dtc_stored"], - "dtc_pending": parsed["dtc_pending"], - "parameters": parsed["parameters"], - }, + "diagnosis": format_no_llm(parsed), + "parsed": _summary(parsed), }) diagnoser = Diagnoser( @@ -234,34 +86,18 @@ def register(app): model=cfg["llm"].get("model", "gpt-oss-120b"), base_url=cfg["llm"].get("base_url", "https://api.aillm.ru/v1"), ) - user_prompt = _build_diagnosis_prompt(parsed) - answer = diagnoser.diagnose(SYSTEM_PROMPT, user_prompt) + answer = diagnoser.diagnose(SYSTEM_PROMPT, _build_diagnosis_prompt(parsed)) return jsonify({ "diagnosis": answer, - "parsed": { - "vin": parsed["vin"], - "dtc_stored": parsed["dtc_stored"], - "dtc_pending": parsed["dtc_pending"], - "parameters": parsed["parameters"], - }, + "parsed": _summary(parsed), }) -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) +def _summary(p: dict) -> dict: + return { + "vin": p["vin"], + "dtc_stored": p["dtc_stored"], + "dtc_pending": p["dtc_pending"], + "parameters": p["parameters"], + } diff --git a/web/script_parser.py b/web/script_parser.py new file mode 100644 index 0000000..f92ec09 --- /dev/null +++ b/web/script_parser.py @@ -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)