silvanus930
CMS: overview, score column, filtering/sorting, charts, CQ health, calibration UI, library browser, alerts, auth
b84c410 | """SQLite schema and access helpers for the compression CMS. | |
| Single source of truth for every processed item's outcome. Populated by | |
| ingest.py tailing the compression service's compression_outcomes.jsonl | |
| (see miner/compression/app.py:_log_compression_outcome) -- the CMS never | |
| talks to the compression service directly, it only reads what it already | |
| logs. | |
| """ | |
| from __future__ import annotations | |
| import sqlite3 | |
| from contextlib import contextmanager | |
| from pathlib import Path | |
| from scoring import compression_score | |
| DB_PATH = Path("/root/vidaio-cms/cms.db") | |
| SCHEMA = """ | |
| CREATE TABLE IF NOT EXISTS items ( | |
| task_id TEXT PRIMARY KEY, | |
| ts TEXT NOT NULL, | |
| codec TEXT, | |
| codec_mode TEXT, | |
| vmaf_threshold REAL, | |
| target_bitrate REAL, | |
| compression_type TEXT, | |
| cq INTEGER, | |
| vmaf REAL, | |
| original_size INTEGER, | |
| compressed_size INTEGER, | |
| ratio REAL, | |
| score REAL, | |
| elapsed_seconds REAL, | |
| searched INTEGER, | |
| has_reference_sample INTEGER DEFAULT 0, | |
| has_compressed_sample INTEGER DEFAULT 0 | |
| ); | |
| CREATE INDEX IF NOT EXISTS idx_items_ts ON items(ts); | |
| CREATE INDEX IF NOT EXISTS idx_items_codec_thr ON items(codec, vmaf_threshold); | |
| CREATE TABLE IF NOT EXISTS ingest_state ( | |
| source TEXT PRIMARY KEY, | |
| byte_offset INTEGER NOT NULL DEFAULT 0 | |
| ); | |
| CREATE TABLE IF NOT EXISTS alerts ( | |
| id INTEGER PRIMARY KEY AUTOINCREMENT, | |
| ts TEXT NOT NULL, | |
| kind TEXT NOT NULL, | |
| severity TEXT NOT NULL, | |
| message TEXT NOT NULL, | |
| dedupe_key TEXT | |
| ); | |
| CREATE INDEX IF NOT EXISTS idx_alerts_ts ON alerts(ts); | |
| CREATE UNIQUE INDEX IF NOT EXISTS idx_alerts_dedupe ON alerts(dedupe_key); | |
| CREATE TABLE IF NOT EXISTS calibration_runs ( | |
| id INTEGER PRIMARY KEY AUTOINCREMENT, | |
| started_at TEXT NOT NULL, | |
| finished_at TEXT, | |
| codec TEXT NOT NULL, | |
| vmaf_threshold REAL NOT NULL, | |
| cq_grid TEXT NOT NULL, | |
| limit_clips INTEGER NOT NULL, | |
| status TEXT NOT NULL DEFAULT 'running', | |
| pid INTEGER, | |
| log_tail TEXT | |
| ); | |
| """ | |
| def connect(): | |
| DB_PATH.parent.mkdir(parents=True, exist_ok=True) | |
| conn = sqlite3.connect(DB_PATH) | |
| conn.row_factory = sqlite3.Row | |
| try: | |
| yield conn | |
| conn.commit() | |
| finally: | |
| conn.close() | |
| def init_db() -> None: | |
| with connect() as conn: | |
| conn.executescript(SCHEMA) | |
| # CREATE TABLE IF NOT EXISTS doesn't alter an already-existing table, | |
| # so columns added after the first deploy need an explicit | |
| # migration -- idempotent via the duplicate-column error, not a | |
| # tracked migration system, since this schema is still young enough | |
| # that isn't worth the overhead yet. | |
| existing_cols = {row["name"] for row in conn.execute("PRAGMA table_info(items)")} | |
| if "score" not in existing_cols: | |
| conn.execute("ALTER TABLE items ADD COLUMN score REAL") | |
| def get_ingest_offset(source: str) -> int: | |
| with connect() as conn: | |
| row = conn.execute( | |
| "SELECT byte_offset FROM ingest_state WHERE source = ?", (source,) | |
| ).fetchone() | |
| return row["byte_offset"] if row else 0 | |
| def set_ingest_offset(source: str, offset: int) -> None: | |
| with connect() as conn: | |
| conn.execute( | |
| "INSERT INTO ingest_state (source, byte_offset) VALUES (?, ?) " | |
| "ON CONFLICT(source) DO UPDATE SET byte_offset = excluded.byte_offset", | |
| (source, offset), | |
| ) | |
| def upsert_item(record: dict, has_reference: bool, has_compressed: bool) -> None: | |
| score = compression_score(record.get("ratio"), record.get("vmaf"), record.get("vmaf_threshold")) | |
| with connect() as conn: | |
| conn.execute( | |
| """ | |
| INSERT INTO items ( | |
| task_id, ts, codec, codec_mode, vmaf_threshold, target_bitrate, | |
| compression_type, cq, vmaf, original_size, compressed_size, ratio, score, | |
| elapsed_seconds, searched, has_reference_sample, has_compressed_sample | |
| ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) | |
| ON CONFLICT(task_id) DO UPDATE SET | |
| ts=excluded.ts, codec=excluded.codec, codec_mode=excluded.codec_mode, | |
| vmaf_threshold=excluded.vmaf_threshold, target_bitrate=excluded.target_bitrate, | |
| compression_type=excluded.compression_type, cq=excluded.cq, vmaf=excluded.vmaf, | |
| original_size=excluded.original_size, compressed_size=excluded.compressed_size, | |
| ratio=excluded.ratio, score=excluded.score, elapsed_seconds=excluded.elapsed_seconds, | |
| searched=excluded.searched, | |
| has_reference_sample=excluded.has_reference_sample, | |
| has_compressed_sample=excluded.has_compressed_sample | |
| """, | |
| ( | |
| record["task_id"], record["ts"], record.get("codec"), | |
| record.get("codec_mode"), record.get("vmaf_threshold"), | |
| record.get("target_bitrate"), record.get("compression_type"), | |
| record.get("cq"), record.get("vmaf"), record.get("original_size"), | |
| record.get("compressed_size"), record.get("ratio"), score, | |
| record.get("elapsed_seconds"), int(bool(record.get("searched"))), | |
| int(has_reference), int(has_compressed), | |
| ), | |
| ) | |
| _SORT_COLUMNS = {"ts", "score", "vmaf", "ratio", "cq", "elapsed_seconds"} | |
| def list_items( | |
| codec: str | None = None, | |
| vmaf_threshold: float | None = None, | |
| limit: int = 100, | |
| offset: int = 0, | |
| sort: str = "ts", | |
| order: str = "desc", | |
| ) -> list[dict]: | |
| sort_col = sort if sort in _SORT_COLUMNS else "ts" | |
| order_sql = "ASC" if order.lower() == "asc" else "DESC" | |
| query = "SELECT * FROM items WHERE 1=1" | |
| params: list = [] | |
| if codec: | |
| query += " AND codec = ?" | |
| params.append(codec) | |
| if vmaf_threshold is not None: | |
| query += " AND vmaf_threshold = ?" | |
| params.append(vmaf_threshold) | |
| query += f" ORDER BY {sort_col} {order_sql} LIMIT ? OFFSET ?" | |
| params.extend([limit, offset]) | |
| with connect() as conn: | |
| return [dict(r) for r in conn.execute(query, params).fetchall()] | |
| def get_item(task_id: str) -> dict | None: | |
| with connect() as conn: | |
| row = conn.execute("SELECT * FROM items WHERE task_id = ?", (task_id,)).fetchone() | |
| return dict(row) if row else None | |
| def distinct_categories() -> list[dict]: | |
| with connect() as conn: | |
| rows = conn.execute( | |
| "SELECT DISTINCT codec, vmaf_threshold FROM items ORDER BY codec, vmaf_threshold" | |
| ).fetchall() | |
| return [dict(r) for r in rows] | |
| def stats_summary() -> dict: | |
| with connect() as conn: | |
| total = conn.execute("SELECT COUNT(*) c FROM items").fetchone()["c"] | |
| by_category = conn.execute( | |
| """ | |
| SELECT codec, vmaf_threshold, COUNT(*) n, | |
| AVG(vmaf) avg_vmaf, AVG(ratio) avg_ratio, AVG(score) avg_score, | |
| MIN(ts) oldest, MAX(ts) newest | |
| FROM items GROUP BY codec, vmaf_threshold ORDER BY codec, vmaf_threshold | |
| """ | |
| ).fetchall() | |
| return { | |
| "total_items": total, | |
| "by_category": [dict(r) for r in by_category], | |
| } | |
| def score_trend(days: int = 14) -> list[dict]: | |
| """Daily mean/median-ish (avg + min) score, most recent `days` days.""" | |
| with connect() as conn: | |
| rows = conn.execute( | |
| """ | |
| SELECT substr(ts, 1, 10) as day, COUNT(*) n, | |
| AVG(score) avg_score, MIN(score) min_score, MAX(score) max_score | |
| FROM items | |
| WHERE score IS NOT NULL | |
| GROUP BY day | |
| ORDER BY day DESC | |
| LIMIT ? | |
| """, | |
| (days,), | |
| ).fetchall() | |
| return list(reversed([dict(r) for r in rows])) | |
| def rate_distortion_points(limit: int = 500) -> list[dict]: | |
| with connect() as conn: | |
| rows = conn.execute( | |
| """ | |
| SELECT task_id, codec, vmaf_threshold, vmaf, ratio, score | |
| FROM items | |
| WHERE vmaf IS NOT NULL AND ratio IS NOT NULL | |
| ORDER BY ts DESC LIMIT ? | |
| """, | |
| (limit,), | |
| ).fetchall() | |
| return [dict(r) for r in rows] | |
| # --- Alerts --------------------------------------------------------------- | |
| def add_alert(kind: str, severity: str, message: str, dedupe_key: str, ts: str) -> None: | |
| """dedupe_key makes repeated identical conditions (e.g. the same stalled | |
| restart count) collapse to one row instead of spamming -- INSERT OR | |
| IGNORE against the unique index on dedupe_key. | |
| """ | |
| with connect() as conn: | |
| conn.execute( | |
| "INSERT OR IGNORE INTO alerts (ts, kind, severity, message, dedupe_key) VALUES (?, ?, ?, ?, ?)", | |
| (ts, kind, severity, message, dedupe_key), | |
| ) | |
| def recent_alerts(limit: int = 50) -> list[dict]: | |
| with connect() as conn: | |
| rows = conn.execute( | |
| "SELECT * FROM alerts ORDER BY ts DESC LIMIT ?", (limit,) | |
| ).fetchall() | |
| return [dict(r) for r in rows] | |
| # --- Calibration runs ------------------------------------------------------- | |
| def start_calibration_run(codec: str, vmaf_threshold: float, cq_grid: str, limit_clips: int, pid: int, started_at: str) -> int: | |
| with connect() as conn: | |
| cur = conn.execute( | |
| """ | |
| INSERT INTO calibration_runs (started_at, codec, vmaf_threshold, cq_grid, limit_clips, status, pid) | |
| VALUES (?, ?, ?, ?, ?, 'running', ?) | |
| """, | |
| (started_at, codec, vmaf_threshold, cq_grid, limit_clips, pid), | |
| ) | |
| return cur.lastrowid | |
| def update_calibration_run(run_id: int, status: str, log_tail: str, finished_at: str | None = None) -> None: | |
| with connect() as conn: | |
| conn.execute( | |
| "UPDATE calibration_runs SET status=?, log_tail=?, finished_at=COALESCE(?, finished_at) WHERE id=?", | |
| (status, log_tail, finished_at, run_id), | |
| ) | |
| def list_calibration_runs(limit: int = 20) -> list[dict]: | |
| with connect() as conn: | |
| rows = conn.execute( | |
| "SELECT * FROM calibration_runs ORDER BY started_at DESC LIMIT ?", (limit,) | |
| ).fetchall() | |
| return [dict(r) for r in rows] | |
| def get_calibration_run(run_id: int) -> dict | None: | |
| with connect() as conn: | |
| row = conn.execute("SELECT * FROM calibration_runs WHERE id = ?", (run_id,)).fetchone() | |
| return dict(row) if row else None | |