app.py

"""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} &nbsp;·&nbsp; **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} &nbsp;·&nbsp; **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  # swept by the frame-cleanup TTL; cache hit still shows verdict/evidence
            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()