| """AI VAR — Gradio UI (HF Spaces entrypoint). |
| |
| Upload one or more camera angles (videos ≤2 min and/or images); each file is |
| treated as a separate view and fused SoccerNet-VARS style. |
| """ |
| import traceback |
| from pathlib import Path |
|
|
| import gradio as gr |
| import spaces |
|
|
| from aivar import cleanup |
| from aivar.ingest import IngestError |
| from aivar.llm import BudgetExceeded |
| from aivar.pipeline import analyze |
| from aivar.ratelimit import RateLimited |
| from aivar.schemas import AnalysisResult, Decision, EvidenceStatus |
|
|
| cleanup.start_sweeper() |
|
|
|
|
| @spaces.GPU(duration=90) |
| def gpu_vision(views): |
| """Runs the YOLO+ByteTrack stage on ZeroGPU-allocated hardware.""" |
| from aivar import vision |
| for v in views: |
| vision.analyze_view(v) |
| return views |
|
|
| STATUS_ICON = {EvidenceStatus.CONFIRMED: "✅", |
| EvidenceStatus.CONTRADICTED: "❌", |
| EvidenceStatus.NOT_VISIBLE: "🕶️"} |
|
|
| DECISION_COLOR = {Decision.RED_CARD: "#c62828", Decision.YELLOW_CARD: "#f9a825", |
| Decision.PENALTY: "#6a1b9a", Decision.INSUFFICIENT_EVIDENCE: "#546e7a"} |
|
|
|
|
| def _verdict_md(r: AnalysisResult) -> str: |
| v = r.verdict |
| color = DECISION_COLOR.get(v.decision, "#2e7d32") |
| cached = " · ⚡ from cache (0 Gemini calls)" if r.from_cache else f" · {r.gemini_calls} Gemini calls" |
| lines = [ |
| f"## <span style='color:{color}'>{v.decision.value}</span>", |
| f"**Incident:** {v.incident.value} · **Confidence:** {v.confidence}%{cached}", |
| ] |
| inc = r.incident |
| if any([inc.offending_team, inc.offending_player, inc.fouled_team, inc.fouled_player]): |
| foul_by = " ".join(p for p in [inc.offending_team, inc.offending_player] if p) or "unknown" |
| on = " ".join(p for p in [inc.fouled_team, inc.fouled_player] if p) or "unknown" |
| lines.append(f"**Foul by:** {foul_by} · **On:** {on}") |
| lines += ["", f"**Why:** {v.why}"] |
| if v.why_not: |
| lines += ["", f"**Why not:** {v.why_not}"] |
| if v.rule_citations: |
| lines += ["", "### 📖 Rule citations (IFAB Laws of the Game 2025/26)"] |
| for c in v.rule_citations: |
| lines.append(f"> **{c.law} — {c.section}**: “{c.quote}”") |
| if v.missing_evidence: |
| lines += ["", "### 🔍 Missing evidence"] |
| lines += [f"- {m}" for m in v.missing_evidence] |
| if v.recommendation: |
| lines += ["", f"**Recommendation:** {v.recommendation}"] |
| return "\n".join(lines) |
|
|
|
|
| def _evidence_md(r: AnalysisResult) -> str: |
| if not r.evidence: |
| return "_No checklist evaluated._" |
| lines = ["### Evidence checklist (fused across angles)"] |
| for e in r.evidence: |
| icon = STATUS_ICON[e.status] |
| src = f" — via Angle {e.source_angle}" if e.source_angle else "" |
| crit = " **[critical]**" if e.critical else "" |
| conf = f" ({e.confidence}%)" if e.confidence else "" |
| lines.append(f"- {icon} **{e.question}**{crit}{src}{conf} \n {e.detail}") |
| if e.conflict: |
| lines.append(" ⚠️ *Angles disagree on this item.*") |
| return "\n".join(lines) |
|
|
|
|
| def preview_files(files): |
| if not files: |
| return [] |
| paths = [f.name if hasattr(f, "name") else f for f in files] |
| return [(p, f"Angle {i}") for i, p in enumerate(paths, start=1)] |
|
|
|
|
| def run(files, user_key, progress=gr.Progress()): |
| if not files: |
| raise gr.Error("Upload at least one video (≤2 min) or image.") |
| paths = [f.name if hasattr(f, "name") else f for f in files] |
| api_key = (user_key or "").strip() or None |
| try: |
| result = analyze(paths, progress=lambda m: progress(0, desc=m), api_key=api_key, |
| vision_fn=gpu_vision) |
| except (IngestError, BudgetExceeded, RateLimited) as e: |
| raise gr.Error(str(e)) |
| except Exception as e: |
| traceback.print_exc() |
| raise gr.Error(f"Analysis failed: {e}") |
|
|
| gallery = [] |
| for view in result.views: |
| for kf in view.keyframes: |
| if not Path(kf.path).exists(): |
| continue |
| tag = f"Angle {view.angle_id} @ {kf.timestamp:.2f}s" |
| if kf.is_replay: |
| tag += " (replay)" |
| gallery.append((kf.path, tag)) |
| return _verdict_md(result), _evidence_md(result), gallery |
|
|
|
|
| with gr.Blocks(title="AI VAR — Football Referee Assistant") as demo: |
| gr.Markdown( |
| "# ⚽ AI VAR — Intelligent Referee Decision Assistant\n" |
| "Upload **multiple camera angles** — videos (≤2 min) and/or photos of the same " |
| "incident. Decisions are grounded in the **IFAB Laws of the Game 2025/26** and " |
| "the system refuses to guess when evidence is insufficient.") |
| with gr.Row(): |
| with gr.Column(scale=1): |
| files = gr.File(label="Camera angles (videos / images)", |
| file_count="multiple", |
| file_types=[".mp4", ".mov", ".avi", ".mkv", ".webm", |
| ".jpg", ".jpeg", ".png", ".webp"]) |
| preview_out = gr.Gallery(label="Preview — uploaded angles", columns=3, height=240) |
| user_key = gr.Textbox( |
| label="Your Gemini API key (only needed after the daily free limit)", |
| type="password", placeholder="AIza…") |
| gr.Markdown("*Free tier: 25 analyses/day globally. After that, paste your " |
| "own key — it is used only for your request and never stored.*") |
| btn = gr.Button("🔍 Analyze incident", variant="primary") |
| gr.Markdown("*Repeat uploads of the same footage — even re-encoded or " |
| "trimmed — are served instantly from the perceptual cache. " |
| "Extracted keyframe images are auto-deleted 10 minutes after " |
| "creation for privacy/disk hygiene — cached verdicts stay " |
| "instant, but keyframe thumbnails may no longer display.*") |
| with gr.Column(scale=2): |
| verdict_out = gr.Markdown(label="Verdict") |
| evidence_out = gr.Markdown(label="Evidence") |
| gallery_out = gr.Gallery(label="Annotated keyframes by angle", columns=6, height=260) |
|
|
| files.change(preview_files, inputs=[files], outputs=[preview_out]) |
| btn.click(run, inputs=[files, user_key], outputs=[verdict_out, evidence_out, gallery_out]) |
|
|
| if __name__ == "__main__": |
| demo.launch() |
|
|
