| 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() |
|
|
