File size: 6,586 Bytes
2d7db98 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 | """
SQLite database setup using Python's built-in sqlite3.
Database file: backend/evolve.db
No extra packages needed — sqlite3 is part of Python stdlib.
"""
import sqlite3
import os
import json
from datetime import datetime
from pathlib import Path
# DB path: use DB_PATH env var if set (Hugging Face /data/), else local
import os
_env_path = os.environ.get("DB_PATH")
DB_PATH = Path(_env_path) if _env_path else Path(__file__).parent.parent / "evolve.db"
def get_conn() -> sqlite3.Connection:
"""Return a new SQLite connection with row_factory for dict-like access."""
conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row # rows behave like dicts
conn.execute("PRAGMA journal_mode=WAL") # better concurrent read performance
conn.execute("PRAGMA foreign_keys=ON")
return conn
def init_db():
"""Create all tables if they don't exist. Called once on startup."""
conn = get_conn()
cur = conn.cursor()
# ── Uploaded data tables ────────────────────────────────────────
cur.execute("""
CREATE TABLE IF NOT EXISTS bms_data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
vehicle_id TEXT NOT NULL,
timestamp TEXT NOT NULL,
soc REAL,
voltage REAL,
current REAL,
temperature REAL,
cycle_count INTEGER,
odometer_km REAL,
uploaded_at TEXT NOT NULL DEFAULT (datetime('now'))
)
""")
cur.execute("""
CREATE TABLE IF NOT EXISTS fleet_data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
asset_id TEXT NOT NULL UNIQUE,
category TEXT,
route_distance_km REAL,
payload_tons REAL,
duty_cycle_hrs REAL,
dwell_time_hrs REAL,
terrain_score REAL,
diesel_l_per_100km REAL,
age_years REAL,
uploaded_at TEXT NOT NULL DEFAULT (datetime('now'))
)
""")
cur.execute("""
CREATE TABLE IF NOT EXISTS supplier_data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
supplier_id TEXT NOT NULL UNIQUE,
name TEXT,
tier INTEGER,
material TEXT,
country TEXT,
geopolitical_risk REAL,
esg_score REAL,
quality_score REAL,
lead_time_days INTEGER,
concentration_pct REAL,
uploaded_at TEXT NOT NULL DEFAULT (datetime('now'))
)
""")
cur.execute("""
CREATE TABLE IF NOT EXISTS inspection_data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
lot_id TEXT NOT NULL,
supplier_id TEXT NOT NULL,
material TEXT,
defect_rate REAL,
inspection_date TEXT,
root_cause_hint TEXT,
uploaded_at TEXT NOT NULL DEFAULT (datetime('now'))
)
""")
# ── Activity / audit log ────────────────────────────────────────
cur.execute("""
CREATE TABLE IF NOT EXISTS activity_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL DEFAULT (datetime('now')),
user TEXT NOT NULL DEFAULT 'Admin',
action TEXT NOT NULL,
endpoint TEXT,
method TEXT,
status_code INTEGER,
details TEXT -- JSON string for extra metadata
)
""")
# ── Upload history ──────────────────────────────────────────────
cur.execute("""
CREATE TABLE IF NOT EXISTS upload_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL DEFAULT (datetime('now')),
user TEXT NOT NULL DEFAULT 'Admin',
data_type TEXT NOT NULL, -- bms / fleet / suppliers / inspections
filename TEXT,
row_count INTEGER,
status TEXT NOT NULL DEFAULT 'success',
error_msg TEXT
)
""")
conn.commit()
conn.close()
print(f"[DB] SQLite database ready at: {DB_PATH}")
# ── Helpers ─────────────────────────────────────────────────────────
def log_activity(action: str, endpoint: str = None, method: str = None,
status_code: int = 200, details: dict = None, user: str = "Admin"):
"""Insert one row into activity_log."""
conn = get_conn()
conn.execute(
"""INSERT INTO activity_log (timestamp, user, action, endpoint, method, status_code, details)
VALUES (datetime('now'), ?, ?, ?, ?, ?, ?)""",
(user, action, endpoint, method, status_code, json.dumps(details) if details else None)
)
conn.commit()
conn.close()
def log_upload(data_type: str, filename: str, row_count: int,
status: str = "success", error_msg: str = None, user: str = "Admin"):
"""Insert one row into upload_history."""
conn = get_conn()
conn.execute(
"""INSERT INTO upload_history (timestamp, user, data_type, filename, row_count, status, error_msg)
VALUES (datetime('now'), ?, ?, ?, ?, ?, ?)""",
(user, data_type, filename, row_count, status, error_msg)
)
conn.commit()
conn.close()
def get_activity_log(limit: int = 100) -> list:
conn = get_conn()
rows = conn.execute(
"SELECT * FROM activity_log ORDER BY timestamp DESC LIMIT ?", (limit,)
).fetchall()
conn.close()
return [dict(r) for r in rows]
def get_upload_history(limit: int = 50) -> list:
conn = get_conn()
rows = conn.execute(
"SELECT * FROM upload_history ORDER BY timestamp DESC LIMIT ?", (limit,)
).fetchall()
conn.close()
return [dict(r) for r in rows]
|