"""Эндпоинты для толстого клиента: скрипты и батчевая загрузка. GET /api/v1/script — выдача скрипта диагностики POST /api/v1/session/upload — приём батча, LLM-анализ, возврат диагноза """ import json import logging from datetime import datetime, timezone 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"] if data["vin"]: parts.append(f"**VIN:** {data['vin']}") if data["dtc_stored"]: parts.append(f"\n**Сохранённые ошибки:** {', '.join(data['dtc_stored'])}") if data["dtc_pending"]: parts.append(f"**Ожидающие ошибки:** {', '.join(data['dtc_pending'])}") if data["parameters"]: parts.append("\n**Параметры:**") for p in data["parameters"]: parts.append(f"- {p['name']}: {p['value']}") 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) return "\n".join(parts) def register(app): """Регистрирует эндпоинты скриптов на Flask-приложении.""" @app.route("/api/v1/script", methods=["GET"]) def get_script(): """Возвращает скрипт диагностики для клиента.""" from flask import jsonify # В будущем — выбирать скрипт в зависимости от VIN / истории script = build_default_script() return jsonify(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 from elmer.prompts import SYSTEM_PROMPT data = request.get_json(silent=True) if not data or "responses" not in data: return jsonify({"error": "missing 'responses'"}), 400 responses = data["responses"] logger.info(f"Upload: {len(responses)} responses") parsed = _parse_batch(responses) # LLM cfg = load() api_key = cfg["deepseek"]["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"], }, }) diagnoser = Diagnoser( api_key=api_key, model=cfg["deepseek"].get("model", "deepseek-chat"), ) user_prompt = _build_diagnosis_prompt(parsed) answer = diagnoser.diagnose(SYSTEM_PROMPT, user_prompt) return jsonify({ "diagnosis": answer, "parsed": { "vin": parsed["vin"], "dtc_stored": parsed["dtc_stored"], "dtc_pending": parsed["dtc_pending"], "parameters": parsed["parameters"], }, }) 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)