Spaces:
Runtime error
Runtime error
File size: 2,861 Bytes
f7b331f 4dccba4 f7b331f 4dccba4 f7b331f 4dccba4 f7b331f 4dccba4 f7b331f 4dccba4 f7b331f | 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 | import sqlite3
import os
from datetime import datetime
from huggingface_hub import HfApi, hf_hub_download
REPO_ID = "IDKHowToCodeFr/tinyml-logs"
DB_NAME = 'patient_history.db'
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), DB_NAME)
HF_TOKEN = os.getenv("HF_TOKEN")
api = HfApi()
def sync_from_hub():
if not HF_TOKEN:
print("No HF_TOKEN found. Skipping sync.")
return
try:
print(f"Downloading {DB_NAME} from Hub...")
path = hf_hub_download(repo_id=REPO_ID, filename=DB_NAME, repo_type="dataset", token=HF_TOKEN)
import shutil
shutil.copy(path, DB_PATH)
print("Sync from Hub complete.")
except Exception as e:
print(f"Sync from Hub failed (maybe first run?): {e}")
def sync_to_hub():
if not HF_TOKEN:
return
try:
api.upload_file(
path_or_fileobj=DB_PATH,
path_in_repo=DB_NAME,
repo_id=REPO_ID,
repo_type="dataset",
token=HF_TOKEN
)
except Exception as e:
print(f"Sync to Hub failed: {e}")
def init_db():
sync_from_hub()
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.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,
fall_detection TEXT,
prediction_label TEXT,
confidence REAL
)
''')
conn.commit()
conn.close()
def log_prediction(data, prediction_label, confidence):
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute('''
INSERT INTO predictions (timestamp, heart_rate, spo2, sys_bp, dia_bp, temp, fall_detection, prediction_label, confidence)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
data.Heart_Rate,
data.SpO2_Level,
data.Systolic_BP,
data.Diastolic_BP,
data.Body_Temp,
data.Fall_Detection,
prediction_label,
confidence
))
conn.commit()
conn.close()
sync_to_hub()
def get_history():
if not os.path.exists(DB_PATH):
return []
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute('SELECT * FROM predictions ORDER BY timestamp DESC LIMIT 100')
rows = cursor.fetchall()
col_names = [description[0] for description in cursor.description]
conn.close()
history = []
for row in rows:
history.append(dict(zip(col_names, row)))
return history
# Initialize db when this module is loaded
init_db()
|