154 lines
6.7 KiB
Python
154 lines
6.7 KiB
Python
"""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]
|