Spaces:
Runtime error
Runtime error
File size: 2,427 Bytes
4cea933 0b2f8bd 604d53b 4cea933 0b2f8bd 4cea933 0b2f8bd 4cea933 604d53b 4cea933 0b2f8bd 4cea933 604d53b 4cea933 604d53b 4cea933 604d53b 4cea933 0b2f8bd 4cea933 0b2f8bd 604d53b 4cea933 604d53b | 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 | import aiosqlite
import os
from datetime import datetime
import zoneinfo
from typing import List, Dict, Any
from config import settings
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), settings.db_name)
async def init_db() -> None:
async with aiosqlite.connect(DB_PATH, timeout=20.0) as db:
await db.execute("PRAGMA busy_timeout = 10000;")
try:
await db.execute("PRAGMA journal_mode=WAL;")
except Exception:
pass
await db.execute("PRAGMA synchronous=NORMAL;")
await db.execute('''
CREATE TABLE IF NOT EXISTS predictions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp DATETIME,
heart_rate REAL,
spo2 REAL,
sys_bp REAL,
dia_bp REAL,
temp REAL,
prediction_label TEXT,
confidence REAL
)
''')
await db.commit()
async def log_prediction(data: Any, prediction_label: str, confidence: float) -> None:
async with aiosqlite.connect(DB_PATH, timeout=20.0) as db:
await db.execute("PRAGMA busy_timeout = 10000;")
try:
ist = zoneinfo.ZoneInfo("Asia/Kolkata")
ts = datetime.now(ist).strftime("%Y-%m-%d %H:%M:%S")
except Exception:
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
await db.execute('''
INSERT INTO predictions (timestamp, heart_rate, spo2, sys_bp, dia_bp, temp, prediction_label, confidence)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
''', (
ts,
getattr(data, "Heart_Rate", 0.0),
getattr(data, "SpO2_Level", 0.0),
getattr(data, "Systolic_BP", 0.0),
getattr(data, "Diastolic_BP", 0.0),
getattr(data, "Body_Temp", 0.0),
prediction_label,
confidence
))
await db.commit()
async def get_history() -> List[Dict[str, Any]]:
if not os.path.exists(DB_PATH):
return []
async with aiosqlite.connect(DB_PATH, timeout=20.0) as db:
await db.execute("PRAGMA busy_timeout = 10000;")
db.row_factory = aiosqlite.Row
async with db.execute('SELECT * FROM predictions ORDER BY timestamp DESC LIMIT 100') as cursor:
rows = await cursor.fetchall()
return [dict(row) for row in rows]
|