268 lines
12 KiB
Python
268 lines
12 KiB
Python
"""Эндпоинты для толстого клиента: скрипты и батчевая загрузка.
|
|
|
|
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**Сохранённые ошибки (mode 03):** {', '.join(data['dtc_stored'])}")
|
|
if data["dtc_pending"]:
|
|
parts.append(f"**Ожидающие ошибки (mode 07):** {', '.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)
|
|
|
|
parts.append("\n---")
|
|
parts.append("## Запрос на анализ")
|
|
parts.append("Дай ГЛУБОКИЙ, РАЗВЁРНУТЫЙ анализ на основе этих данных. "
|
|
"Не ограничивайся кратким резюме — мне нужен полный технический разбор.\n"
|
|
"1. Разбери КАЖДУЮ ошибку: все возможные причины, от частых к редким.\n"
|
|
"2. Проанализируй КАЖДЫЙ параметр: норма/отклонение, на что влияет.\n"
|
|
"3. Найди ВЗАИМОСВЯЗИ между ошибками и параметрами.\n"
|
|
"4. Предложи КОНКРЕТНЫЙ план проверок: что сначала, что потом, как проверять.\n"
|
|
"5. Если данных мало — скажи какие PID'ы досчитать и зачем.\n"
|
|
"6. Дай оценки уверенности в ПРОЦЕНТАХ для каждой версии.")
|
|
|
|
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["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"],
|
|
},
|
|
})
|
|
|
|
diagnoser = Diagnoser(
|
|
api_key=api_key,
|
|
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)
|
|
|
|
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)
|