Spaces:
Running
Running
File size: 2,350 Bytes
eaa7aa4 74480c7 eaa7aa4 |
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 |
import sqlite3
from constants import DATABASE_PATH
class MetricsStorage:
def __init__(self, db_path=DATABASE_PATH):
self.db_path = db_path
self.setup_database()
def setup_database(self):
"""Initialize the SQLite database and create tables if they don't exist"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS inference_metrics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
inference_time REAL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
"""
)
conn.commit()
def add_metric(self, inference_time):
"""Add a new inference time measurement to the database"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute(
"INSERT INTO inference_metrics (inference_time) VALUES (?)",
(inference_time,),
)
conn.commit()
def get_recent_metrics(self, limit=80):
"""Get the most recent metrics from the database"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute(
"SELECT inference_time FROM inference_metrics ORDER BY timestamp DESC LIMIT ?",
(limit,),
)
results = cursor.fetchall()
return [r[0] for r in reversed(results)]
def get_total_inferences(self):
"""Get the total number of inferences recorded"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM inference_metrics")
return cursor.fetchone()[0]
def get_average_time(self, limit=80):
"""Get the average inference time from the most recent entries"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute(
"SELECT AVG(inference_time) FROM (SELECT inference_time FROM inference_metrics ORDER BY timestamp DESC LIMIT ?)",
(limit,),
)
result = cursor.fetchone()[0]
return result if result is not None else 0
|