Nail/elmer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

project structure: elm.py, db.py, prompts.py, diagnose.py, web UI (Flask), config

Browse Source
This commit is contained in:
“Naeel”
2026-05-25 07:46:04 +04:00
parent 200a27c26e
commit 6cc42bc340
13 changed files with 1089 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
elmer/__init__.py
+3
View File
@@ -0,0 +1,3 @@
"""Elmer — анализ ошибок электроники автомобиля через ELM327 + LLM."""
__version__ = "0.1.0"
elmer/config.py
+39
View File
@@ -0,0 +1,39 @@
"""Загрузка конфигурации из config.yaml."""
import os
from pathlib import Path
import yaml
CONFIG_PATH = Path(os.environ.get("ELMER_CONFIG", Path(__file__).parent.parent / "config.yaml"))
def load() -> dict:
"""Читает config.yaml, подставляет переменные окружения в значения."""
if not CONFIG_PATH.exists():
raise FileNotFoundError(f"Конфиг не найден: {CONFIG_PATH}")
with open(CONFIG_PATH) as f:
config = yaml.safe_load(f)
# Подстановка ${VAR} из переменных окружения
_resolve_env(config)
return config
def _resolve_env(obj):
"""Рекурсивно заменяет ${VAR} на os.environ['VAR']."""
if isinstance(obj, dict):
for k, v in obj.items():
if isinstance(v, str) and v.startswith("${") and v.endswith("}"):
env_var = v[2:-1]
obj[k] = os.environ.get(env_var, "")
else:
_resolve_env(v)
elif isinstance(obj, list):
for i, v in enumerate(obj):
if isinstance(v, str) and v.startswith("${") and v.endswith("}"):
env_var = v[2:-1]
obj[i] = os.environ.get(env_var, "")
else:
_resolve_env(v)
elmer/db.py
+153
View File
@@ -0,0 +1,153 @@
"""SQLite — сохранение сессий диагностики.
Схема:
cars — VIN, марка, модель, год, двигатель
diagnostic_tokens — id (PK), car_id (FK), created_at
llm_messages — token_id (FK), role, content, timestamp
ecu_parameters — token_id (FK), pid_code, value, unit, timestamp
dtc_codes — token_id (FK), code, description, status
"""
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
class Database:
def __init__(self, path: str | Path = "elmer.db"):
self.path = Path(path)
self.conn = sqlite3.connect(str(self.path))
self.conn.row_factory = sqlite3.Row
self._init_schema()
def _init_schema(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS cars (
id INTEGER PRIMARY KEY AUTOINCREMENT,
vin TEXT NOT NULL UNIQUE,
make TEXT,
model TEXT,
year INTEGER,
engine TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS diagnostic_tokens (
id INTEGER PRIMARY KEY AUTOINCREMENT,
car_id INTEGER NOT NULL REFERENCES cars(id),
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS llm_messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
token_id INTEGER NOT NULL REFERENCES diagnostic_tokens(id),
role TEXT NOT NULL, -- 'system' | 'user' | 'assistant'
content TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS ecu_parameters (
id INTEGER PRIMARY KEY AUTOINCREMENT,
token_id INTEGER NOT NULL REFERENCES diagnostic_tokens(id),
pid_code TEXT NOT NULL, -- напр. '0105', '010C'
name TEXT, -- напр. 'coolant_temp', 'rpm'
value REAL,
unit TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS dtc_codes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
token_id INTEGER NOT NULL REFERENCES diagnostic_tokens(id),
code TEXT NOT NULL, -- напр. 'P0301'
description TEXT,
status TEXT, -- 'stored' | 'pending'
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
""")
self.conn.commit()
# ── cars ──────────────────────────────────────────────
def get_or_create_car(self, vin: str) -> int:
"""Возвращает car_id по VIN, создаёт запись если нет."""
row = self.conn.execute("SELECT id FROM cars WHERE vin = ?", (vin,)).fetchone()
if row:
return row["id"]
cur = self.conn.execute("INSERT INTO cars (vin) VALUES (?)", (vin,))
self.conn.commit()
return cur.lastrowid
def update_car_info(self, car_id: int, make: str, model: str, year: int, engine: str):
self.conn.execute(
"UPDATE cars SET make=?, model=?, year=?, engine=? WHERE id=?",
(make, model, year, engine, car_id),
)
self.conn.commit()
# ── tokens ────────────────────────────────────────────
def create_token(self, car_id: int) -> int:
"""Создаёт новую сессию диагностики, возвращает token_id."""
cur = self.conn.execute(
"INSERT INTO diagnostic_tokens (car_id) VALUES (?)", (car_id,)
)
self.conn.commit()
return cur.lastrowid
def last_token_for_car(self, car_id: int) -> int | None:
"""Последняя сессия для VIN (для продолжения диалога), или None."""
row = self.conn.execute(
"SELECT id FROM diagnostic_tokens WHERE car_id=? ORDER BY created_at DESC LIMIT 1",
(car_id,),
).fetchone()
return row["id"] if row else None
# ── llm_messages ──────────────────────────────────────
def add_llm_message(self, token_id: int, role: str, content: str):
self.conn.execute(
"INSERT INTO llm_messages (token_id, role, content) VALUES (?, ?, ?)",
(token_id, role, content),
)
self.conn.commit()
def get_llm_messages(self, token_id: int) -> list[dict]:
"""Возвращает историю диалога для токена."""
rows = self.conn.execute(
"SELECT role, content FROM llm_messages WHERE token_id=? ORDER BY id",
(token_id,),
).fetchall()
return [{"role": r["role"], "content": r["content"]} for r in rows]
# ── ecu_parameters ────────────────────────────────────
def add_parameter(self, token_id: int, pid_code: str, name: str, value: float, unit: str):
self.conn.execute(
"INSERT INTO ecu_parameters (token_id, pid_code, name, value, unit) VALUES (?, ?, ?, ?, ?)",
(token_id, pid_code, name, value, unit),
)
self.conn.commit()
def get_parameters(self, token_id: int) -> list[dict]:
rows = self.conn.execute(
"SELECT pid_code, name, value, unit FROM ecu_parameters WHERE token_id=? ORDER BY id",
(token_id,),
).fetchall()
return [dict(r) for r in rows]
# ── dtc_codes ─────────────────────────────────────────
def add_dtc(self, token_id: int, code: str, description: str = "", status: str = "stored"):
self.conn.execute(
"INSERT INTO dtc_codes (token_id, code, description, status) VALUES (?, ?, ?, ?)",
(token_id, code, description, status),
)
self.conn.commit()
def get_dtcs(self, token_id: int) -> list[dict]:
rows = self.conn.execute(
"SELECT code, description, status FROM dtc_codes WHERE token_id=? ORDER BY id",
(token_id,),
).fetchall()
return [dict(r) for r in rows]
elmer/diagnose.py
+48
View File
@@ -0,0 +1,48 @@
"""Вызов DeepSeek API для диагностики."""
import requests
DEEPSEEK_BASE = "https://api.deepseek.com/v1"
DEFAULT_MODEL = "deepseek-chat"
class Diagnoser:
"""Отправляет данные в DeepSeek и возвращает диагноз."""
def __init__(self, api_key: str, model: str = DEFAULT_MODEL, base_url: str = DEEPSEEK_BASE):
self.api_key = api_key
self.model = model
self.base_url = base_url.rstrip("/")
def ask(self, messages: list[dict]) -> str:
"""Отправляет сообщения в DeepSeek, возвращает текст ответа."""
resp = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
},
json={
"model": self.model,
"messages": messages,
"temperature": 0.3, # пониже — меньше фантазий
"max_tokens": 4096,
},
timeout=60,
)
resp.raise_for_status()
data = resp.json()
return data["choices"][0]["message"]["content"]
def diagnose(
self,
system: str,
user_prompt: str,
history: list[dict] | None = None,
) -> str:
"""Полный цикл: system + история + user_prompt → ответ."""
messages = [{"role": "system", "content": system}]
if history:
messages.extend(history)
messages.append({"role": "user", "content": user_prompt})
return self.ask(messages)
elmer/elm.py
+208
View File
@@ -0,0 +1,208 @@
"""Связь с ELM327 через Bluetooth SPP (pyserial)."""
import re
import time
import serial
class ELM327:
"""Работа с ELM327 по последовательному порту (Bluetooth SPP)."""
# Стандартные PID'ы для чтения
DEFAULT_PIDS = {
"0105": ("coolant_temp", "°C"), # температура ОЖ
"010C": ("rpm", "об/мин"), # обороты двигателя
"010D": ("speed", "км/ч"), # скорость
"0111": ("throttle_pos", "%"), # положение дросселя
"010B": ("map", "кПа"), # давление впуска (MAP)
"010F": ("iat", "°C"), # температура впуска
"011F": ("runtime_since_start", "с"), # время с запуска
"0104": ("engine_load", "%"), # нагрузка двигателя
"0106": ("stft_b1", "%"), # краткосрочный fuel trim bank 1
"0107": ("ltft_b1", "%"), # долгосрочный fuel trim bank 1
}
def __init__(self, port: str, baudrate: int = 38400, timeout: float = 5.0):
self.port = port
self.ser = serial.Serial(
port=port,
baudrate=baudrate,
timeout=timeout,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
)
# ── низкоуровневые команды ────────────────────────────
def _cmd(self, cmd: str, wait: float = 0.2) -> str:
"""Отправляет AT/OBD команду, возвращает сырой ответ."""
self.ser.reset_input_buffer()
self.ser.write((cmd + "\r").encode())
time.sleep(wait)
lines = []
while True:
line = self.ser.readline().decode("utf-8", errors="ignore").strip()
if not line or line == ">":
break
lines.append(line)
return "\n".join(lines)
# ── инициализация ─────────────────────────────────────
def init(self) -> bool:
"""Сброс и настройка ELM327. Возвращает True если OK."""
resp = self._cmd("ATZ", wait=1.0) # сброс
if "ELM" not in resp:
return False
self._cmd("ATE0") # выкл эхо
self._cmd("ATL0") # выкл перевод строки
self._cmd("ATSP0") # авто-протокол
self._cmd("ATH1") # вкл заголовки
return True
# ── чтение VIN ────────────────────────────────────────
def read_vin(self) -> str | None:
"""Читает VIN (режим 09 PID 02). Возвращает VIN или None."""
resp = self._cmd("0902", wait=1.5)
# Формат: 014 0: 49 02 01 57 56 57 ...
# Ищем строку с байтами после 49 02
match = re.search(r"49\s*02\s*(.+)", resp.replace("\n", " ").replace(":", ""))
if not match:
return None
# Собираем HEX байты, переводим в ASCII
hex_bytes = match.group(1).strip().split()
vin = ""
for h in hex_bytes:
h = h.strip()
if len(h) == 2:
try:
vin += chr(int(h, 16))
except ValueError:
pass
return vin if len(vin) == 17 else None
# ── чтение ошибок ─────────────────────────────────────
def read_dtc_codes(self, mode: str = "03") -> list[dict]:
"""Читает коды ошибок.
mode: '03' — сохранённые, '07' — ожидающие.
Возвращает [{"code": "P0301", "description": "", "status": "stored"}, ...].
"""
resp = self._cmd(mode, wait=1.0)
codes = []
# Пример ответа: 43 01 33 00 00 00 00
for line in resp.split("\n"):
line = line.strip()
if not line or "NO DATA" in line.upper():
continue
# Ищем HEX-байты после 43 (mode 03 response) или 47 (mode 07)
match = re.search(r"4[37]\s*(.+)", line.replace(":", ""))
if not match:
continue
hex_bytes = match.group(1).strip().split()
# Парсим по 2 байта на код (первые два байта — количество кодов)
i = 1 # пропускаем байт количества
while i + 1 < len(hex_bytes):
dtc_raw = _decode_dtc(hex_bytes[i], hex_bytes[i + 1])
if dtc_raw and dtc_raw != "P0000":
codes.append({
"code": dtc_raw,
"description": "",
"status": "stored" if mode == "03" else "pending",
})
i += 2
return codes
# ── чтение параметров ─────────────────────────────────
def read_pid(self, pid: str) -> float | None:
"""Читает один PID, возвращает числовое значение или None."""
resp = self._cmd(pid, wait=0.3)
# Ищем строку ответа: 41 XX YY ZZ ...
match = re.search(r"4[12]\s*" + pid[2:4] + r"\s*(.+)", resp.replace(":", ""))
if not match:
return None
hex_bytes = match.group(1).strip().split()
if not hex_bytes:
return None
# Формулы для стандартных PID (SAE J1979)
formulas = {
"05": lambda b: int(b[0], 16) - 40, # coolant °C
"0C": lambda b: (int(b[0], 16) * 256 + int(b[1], 16)) / 4, # RPM
"0D": lambda b: int(b[0], 16), # speed km/h
"11": lambda b: int(b[0], 16) * 100 / 255, # throttle %
"0B": lambda b: int(b[0], 16), # MAP kPa
"0F": lambda b: int(b[0], 16) - 40, # IAT °C
"1F": lambda b: int(b[0], 16) * 256 + int(b[1], 16), # runtime sec
"04": lambda b: int(b[0], 16) * 100 / 255, # load %
"06": lambda b: (int(b[0], 16) - 128) * 100 / 128, # STFT %
"07": lambda b: (int(b[0], 16) - 128) * 100 / 128, # LTFT %
}
pid_short = pid[2:4]
if pid_short in formulas:
try:
return round(formulas[pid_short](hex_bytes), 1)
except (ValueError, IndexError):
return None
# Generic: первый байт как raw
try:
return int(hex_bytes[0], 16)
except (ValueError, IndexError):
return None
def read_all_pids(self, pids: dict | None = None) -> list[dict]:
"""Читает все PID'ы из словаря {pid: (name, unit)}.
Возвращает [{"pid_code": "...", "name": "...", "value": ..., "unit": "..."}, ...].
"""
if pids is None:
pids = self.DEFAULT_PIDS
results = []
for pid_code, (name, unit) in pids.items():
try:
value = self.read_pid(pid_code)
if value is not None:
results.append({
"pid_code": pid_code,
"name": name,
"value": value,
"unit": unit,
})
except Exception:
continue
return results
def close(self):
self._cmd("ATZ", wait=0.5)
self.ser.close()
def _decode_dtc(b1: str, b2: str) -> str | None:
"""Декодирует два HEX-байта в код ошибки вида P0301."""
try:
a, b = int(b1, 16), int(b2, 16)
except ValueError:
return None
# Первые 2 бита первого байта — тип:
types = {0: "P", 1: "C", 2: "B", 3: "U"}
prefix = types.get(a >> 6, "?")
# Оставшиеся биты
d1 = str((a >> 4) & 0x03) # вторая цифра
d2 = str(a & 0x0F) # третья цифра
d3 = f"{(b >> 4) & 0x0F:X}" # четвёртая цифра (hex!)
d4 = f"{b & 0x0F:X}" # пятая цифра (hex!)
return f"{prefix}{d1}{d2}{d3}{d4}"
elmer/prompts.py
+64
View File
@@ -0,0 +1,64 @@
"""Шаблоны промптов для LLM."""
SYSTEM_PROMPT = """Ты — эксперт по диагностике автомобилей. Ты анализируешь коды ошибок OBD2 и параметры ЭБУ.
ПРАВИЛА ОТВЕТА:
1. Начинай с КРАТКОГО резюме (1-2 предложения): что скорее всего не так.
2. Затем — подробный анализ: почему ты так думаешь, какие параметры указывают на это.
3. Указывай степень уверенности в процентах для каждого вывода.
4. Если данных недостаточно — честно скажи, какие параметры нужно ещё считать.
5. Предлагай конкретные действия: что проверить, что заменить, в каком порядке.
6. Предупреждай о рисках: "если не поможет X, тогда проверь Y".
7. НИКОГДА не давай категоричных команд "меняй деталь X", если не уверен на 100%.
8. Пиши на русском языке, простым языком, без излишнего жаргона.
ФОРМАТ ОТВЕТА:
## Краткий диагноз
...
## Подробный анализ
...
## Степень уверенности
- Проблема A: ~XX%
- Проблема B: ~XX%
## Что делать
1. ...
2. ...
## Если не поможет
- Проверь ...
"""
def build_user_prompt(
vin: str,
dtc_codes: list[dict],
parameters: list[dict],
car_info: dict | None = None,
) -> str:
"""Собирает промпт пользователя из данных ЭБУ."""
parts = [f"## Данные диагностики\n"]
parts.append(f"**VIN:** {vin}")
if car_info:
parts.append(f"**Автомобиль:** {car_info.get('make', '?')} {car_info.get('model', '?')} "
f"({car_info.get('year', '?')}), двигатель: {car_info.get('engine', '?')}")
if dtc_codes:
parts.append("\n### Коды ошибок")
for dtc in dtc_codes:
parts.append(f"- **{dtc['code']}** ({dtc.get('status', 'stored')}): {dtc.get('description', '')}")
if parameters:
parts.append("\n### Параметры ЭБУ")
for p in parameters:
parts.append(f"- {p['name']} ({p['pid_code']}): {p['value']} {p['unit']}")
parts.append("\n## Запрос")
parts.append("Дай диагноз на основе этих данных. Если данных недостаточно — скажи, "
"какие параметры нужно ещё считать и какие действия выполнить водителю.")
return "\n".join(parts)