from __future__ import annotations import time from datetime import datetime, timezone from pathlib import Path from typing import Any import gradio as gr import pandas as pd import predictors from tracker import RunTracker, hash_image, hash_text LOG_PATH = Path(__file__).parent / "runs.jsonl" tracker = RunTracker(LOG_PATH) def _client_meta(request: gr.Request | None) -> dict[str, Any]: if request is None: return {} headers = getattr(request, "headers", {}) or {} user_agent = "" try: user_agent = headers.get("user-agent", "") except AttributeError: user_agent = dict(headers).get("user-agent", "") return {"user_agent": user_agent} def predict_text(text: str, model_version: str, request: gr.Request): if not text or not text.strip(): gr.Warning("Empty text — nothing to predict.") return None, "" input_hash, input_meta = hash_text(text) t0 = time.perf_counter() result = predictors.predict(model_version, "text", text) latency_ms = (time.perf_counter() - t0) * 1000.0 run_id = tracker.record( model_version=model_version, input_kind="text", input_hash=input_hash, input_meta=input_meta, output=result["output"], latency_ms=latency_ms, client_meta=_client_meta(request), ) return result, f"**Run ID:** `{run_id}` · {latency_ms:.1f} ms" def predict_image(image, model_version: str, request: gr.Request): if image is None: gr.Warning("No image uploaded.") return None, "" input_hash, input_meta = hash_image(image) t0 = time.perf_counter() result = predictors.predict(model_version, "image", image) latency_ms = (time.perf_counter() - t0) * 1000.0 run_id = tracker.record( model_version=model_version, input_kind="image", input_hash=input_hash, input_meta=input_meta, output=result["output"], latency_ms=latency_ms, client_meta=_client_meta(request), ) return result, f"**Run ID:** `{run_id}` · {latency_ms:.1f} ms" def _parse_date(value: str) -> datetime | None: if not value or not value.strip(): return None try: return datetime.fromisoformat(value).replace(tzinfo=timezone.utc) except ValueError: return None def load_history(model_filter: str, since_str: str, until_str: str): model = None if (not model_filter or model_filter == "All") else model_filter since = _parse_date(since_str) until = _parse_date(until_str) rows = list( tracker.iter_runs( model_version=model, since=since, until=until, limit=500, ) ) if not rows: return pd.DataFrame( columns=["run_id", "timestamp", "model_version", "input_kind", "latency_ms"] ) df = pd.DataFrame( [ { "run_id": r.get("run_id"), "timestamp": r.get("timestamp"), "model_version": r.get("model_version"), "input_kind": r.get("input_kind"), "latency_ms": round(float(r.get("latency_ms", 0.0)), 2), } for r in rows ] ) return df def show_detail(evt: gr.SelectData, df: pd.DataFrame): if df is None or df.empty or evt is None: return {} try: row_idx = evt.index[0] if isinstance(evt.index, (list, tuple)) else evt.index except Exception: return {} if row_idx is None or row_idx >= len(df): return {} run_id = df.iloc[row_idx]["run_id"] for row in tracker.iter_runs(limit=10_000): if row.get("run_id") == run_id: return row return {} def load_stats(): stats = tracker.stats() per_model_rows = [ { "model_version": model, "count": data["count"], "p50_ms": None if data["p50_ms"] is None else round(data["p50_ms"], 2), "p95_ms": None if data["p95_ms"] is None else round(data["p95_ms"], 2), } for model, data in stats["per_model"].items() ] per_model_df = pd.DataFrame( per_model_rows, columns=["model_version", "count", "p50_ms", "p95_ms"] ) per_day = stats["per_day"] if not per_day: empty_df = pd.DataFrame({"date": [], "runs": []}) return per_model_df, empty_df plot_df = pd.DataFrame( [{"date": day, "runs": count} for day, count in per_day.items()] ) return per_model_df, plot_df def refresh_download(): if LOG_PATH.exists() and LOG_PATH.stat().st_size > 0: return str(LOG_PATH) return None def clear_log(confirm: bool): if not confirm: gr.Warning("Tick the confirm box to clear the log.") return refresh_download(), "Not cleared." dropped = tracker.clear() return refresh_download(), f"Cleared {dropped} run(s)." def build_app() -> gr.Blocks: models = predictors.list_models() model_choices_with_all = ["All", *models] with gr.Blocks(title="Model Prediction Tracker") as demo: gr.Markdown("# Model Prediction Tracker") gr.Markdown( "Every prediction gets a unique run ID and is appended to " "`runs.jsonl`. Hugging Face Spaces have an **ephemeral filesystem** — " "use the Download tab to save the log before the Space rebuilds." ) with gr.Tabs(): with gr.Tab("Predict"): with gr.Row(): model_dd = gr.Dropdown( choices=models, value=models[0], label="Model version", ) with gr.Tabs(): with gr.Tab("Text"): text_in = gr.Textbox(label="Text input", lines=3) text_btn = gr.Button("Predict", variant="primary") text_run = gr.Markdown() text_out = gr.JSON(label="Prediction") text_btn.click( predict_text, inputs=[text_in, model_dd], outputs=[text_out, text_run], ) with gr.Tab("Image"): img_in = gr.Image(type="pil", label="Image input") img_btn = gr.Button("Predict", variant="primary") img_run = gr.Markdown() img_out = gr.JSON(label="Prediction") img_btn.click( predict_image, inputs=[img_in, model_dd], outputs=[img_out, img_run], ) with gr.Tab("History") as history_tab: with gr.Row(): hist_model = gr.Dropdown( choices=model_choices_with_all, value="All", label="Filter by model", ) hist_since = gr.Textbox( label="Since (YYYY-MM-DD)", placeholder="2026-01-01" ) hist_until = gr.Textbox( label="Until (YYYY-MM-DD)", placeholder="2026-12-31" ) hist_refresh = gr.Button("Refresh") hist_df = gr.Dataframe( headers=[ "run_id", "timestamp", "model_version", "input_kind", "latency_ms", ], interactive=False, wrap=True, label="Runs (newest first)", ) hist_detail = gr.JSON(label="Selected run detail") hist_refresh.click( load_history, inputs=[hist_model, hist_since, hist_until], outputs=[hist_df], ) hist_model.change( load_history, inputs=[hist_model, hist_since, hist_until], outputs=[hist_df], ) hist_df.select(show_detail, inputs=[hist_df], outputs=[hist_detail]) history_tab.select( load_history, inputs=[hist_model, hist_since, hist_until], outputs=[hist_df], ) with gr.Tab("Stats") as stats_tab: stats_refresh = gr.Button("Refresh") stats_per_model = gr.Dataframe( headers=["model_version", "count", "p50_ms", "p95_ms"], interactive=False, label="Per-model latency", ) stats_per_day = gr.BarPlot( x="date", y="runs", title="Runs per day", label="Runs per day", ) stats_refresh.click( load_stats, outputs=[stats_per_model, stats_per_day] ) stats_tab.select(load_stats, outputs=[stats_per_model, stats_per_day]) with gr.Tab("Download") as download_tab: gr.Markdown( "Download the raw JSONL log. Clearing the log is " "**irreversible** — tick the confirm box first." ) dl_file = gr.File(label="runs.jsonl", value=refresh_download()) dl_refresh = gr.Button("Refresh file") clear_confirm = gr.Checkbox( label="I want to clear the log", value=False ) clear_btn = gr.Button("Clear log", variant="stop") clear_status = gr.Markdown() dl_refresh.click(refresh_download, outputs=[dl_file]) download_tab.select(refresh_download, outputs=[dl_file]) clear_btn.click( clear_log, inputs=[clear_confirm], outputs=[dl_file, clear_status], ) return demo demo = build_app() if __name__ == "__main__": demo.launch()