Claude as Your Match Commentator

camera -> YOLO -> tracker -> rules (lesson 31) -> FACTS (JSON) -> Claude -> words

"""Turn detections.json into a short list of FACTS. Claude never sees video; it sees these."""
import json
import math
 
from events import EventBus
from goal_detector import GoalDetector
 
 
def nearest_player(frame):
    """Which player id is closest to the ball in this frame (None if there is no ball or no players)."""
    b = frame["ball"]
    if not b or not frame["players"]:
        return None
    return min(frame["players"], key=lambda p: math.hypot(p["x"] - b["x"], p["y"] - b["y"]))["id"]
 
 
def build_facts(path="detections.json"):
    data = json.load(open(path))
    frames = data["frames"]
    goals = []
    bus = EventBus()
    by_time = {round(f["t"], 2): f for f in frames}
 
    def on_goal(zone, t):
        # who was nearest the ball a second before the goal? (the ball is in the net at t, so look back)
        before = min(frames, key=lambda f: abs(f["t"] - (t - 1.0)))
        goals.append({"time_seconds": round(t, 1), "nearest_player": nearest_player(before)})
 
    bus.on("goal", on_goal)
    det = GoalDetector(data["goal_zone"], bus)
    for fr in frames:
        det.update(fr["ball"], fr["t"])
 
    ids = sorted({p["id"] for f in frames for p in f["players"]})
    return {
        "clip_length_seconds": round(frames[-1]["t"], 1),
        "players_tracked": len(ids),
        "goals_detected": len(goals),
        "goals": goals,
        "note": "Positions come from a camera system and may contain mistakes. Player numbers are tracker ids, not shirt numbers.",
    }
 
 
if __name__ == "__main__":
    print(json.dumps(build_facts(), indent=2))

{
  "clip_length_seconds": 20.0,
  "players_tracked": 8,
  "goals_detected": 3,
  "goals": [
    {
      "time_seconds": 5.7,
      "nearest_player": 1
    },
    {
      "time_seconds": 12.4,
      "nearest_player": 7
    },
    {
      "time_seconds": 19.1,
      "nearest_player": 1
    }
  ],
  "note": "Positions come from a camera system and may contain mistakes. Player numbers are tracker ids, not shirt numbers."
}

"""Ask Claude to commentate from facts. The model name is one constant, so you can swap it."""
import json
import os
 
MODEL = "claude-haiku-5-5"            # fast and cheap: good for short live lines. Try "claude-sonnet-5-5" for a full match report.
 
SYSTEM = """You are a lively but honest football commentator for a school match.
Rules:
- Use ONLY the facts in the JSON the user sends. Never invent names, scores, times, shots, fouls or reasons.
- Player numbers are tracker ids, not shirt numbers or names. Say "player 4", never a name.
- If the facts do not cover something, say so briefly rather than guess.
- Keep it under 80 words. Plain text, no markdown."""
 
 
def build_user_message(facts, style="excited radio commentator"):
    return f"Style: {style}.\nFacts (JSON):\n{json.dumps(facts, indent=2)}\n\nWrite the commentary."
 
 
def get_client():
    """Real client. Needs `pip install anthropic` and the ANTHROPIC_API_KEY environment variable."""
    import anthropic
    if not os.environ.get("ANTHROPIC_API_KEY"):
        raise RuntimeError("Set ANTHROPIC_API_KEY first (never write the key in your code or commit it).")
    return anthropic.Anthropic()
 
 
def commentate(facts, client=None, model=MODEL, style="excited radio commentator", max_tokens=300):
    client = client or get_client()
    msg = client.messages.create(
        model=model,
        max_tokens=max_tokens,
        system=SYSTEM,
        messages=[{"role": "user", "content": build_user_message(facts, style)}],
    )
    return "".join(block.text for block in msg.content if block.type == "text")

"""A pretend Claude for tests: it returns whatever text you give it, in the same shape as the real SDK."""
from types import SimpleNamespace
 
 
class FakeClient:
    def __init__(self, reply="Goal!", fail_times=0):
        self.reply, self.fail_times, self.calls = reply, fail_times, []
        self.messages = SimpleNamespace(create=self._create)
 
    def _create(self, **kwargs):
        self.calls.append(kwargs)
        if self.fail_times > 0:
            self.fail_times -= 1
            raise ConnectionError("network down")
        return SimpleNamespace(content=[SimpleNamespace(type="text", text=self.reply)])

"""Catch invented numbers. Every number in the commentary must be traceable to the facts."""
import re
 
 
def numbers_in(obj):
    """All numbers that appear anywhere in the facts, as strings in a few spellings (5.7, 5, 6...)."""
    out = set()
 
    def walk(x):
        if isinstance(x, dict):
            for v in x.values():
                walk(v)
        elif isinstance(x, list):
            for v in x:
                walk(v)
        elif isinstance(x, (int, float)) and not isinstance(x, bool):
            out.add(f"{x:g}")
            out.add(str(int(round(x))))                       # a commentator may round 5.7 s to "6 seconds"
            out.add(f"{x:.1f}")
        elif isinstance(x, str):
            for m in re.findall(r"\d+(?:\.\d+)?", x):
                out.add(m)
 
    walk(obj)
    return out
 
 
WORDS = {"one": "1", "two": "2", "three": "3", "four": "4", "five": "5", "six": "6", "seven": "7",
         "eight": "8", "nine": "9", "ten": "10"}
 
 
def unsupported_numbers(text, facts):
    allowed = numbers_in(facts)
    found = re.findall(r"\d+(?:\.\d+)?", text)
    for w in re.findall(r"[a-z]+", text.lower()):
        if w in WORDS:
            found.append(WORDS[w])
    return sorted({n for n in found if n not in allowed}, key=float)

from facts import build_facts
from commentary import commentate
from check_numbers import unsupported_numbers
 
facts = build_facts()
text = commentate(facts)                         # real call
print(text)
print("numbers not in the facts:", unsupported_numbers(text, facts))

"""Live lines: the detector must NEVER wait for the internet. Events go in a queue; a worker thread talks to Claude."""
import queue
import threading
import time
 
from commentary import commentate
 
 
class CommentaryWorker:
    def __init__(self, client=None, model=None, retries=2, on_line=print):
        self.q = queue.Queue(maxsize=10)                   # if Claude is slow, drop old news instead of piling up
        self.client, self.model, self.retries, self.on_line = client, model, retries, on_line
        self.lines, self.errors, self.dropped = [], [], 0
        self.thread = threading.Thread(target=self._run, daemon=True)
        self.thread.start()
 
    def submit(self, facts, style="excited radio commentator"):
        """Called from the match loop. Returns immediately."""
        try:
            self.q.put_nowait((facts, style))
        except queue.Full:
            self.dropped += 1
 
    def _run(self):
        while True:
            item = self.q.get()
            if item is None:
                return
            facts, style = item
            for attempt in range(self.retries + 1):
                try:
                    kwargs = {"client": self.client, "style": style}
                    if self.model:
                        kwargs["model"] = self.model
                    text = commentate(facts, **kwargs)
                    self.lines.append(text)
                    self.on_line(text)
                    break
                except Exception as e:                     # network, rate limit, bad key: log and carry on
                    self.errors.append(f"attempt {attempt + 1}: {type(e).__name__}: {e}")
                    time.sleep(0.01 * (attempt + 1))       # tiny wait here; use seconds with a real API
            self.q.task_done()
 
    def close(self):
        self.q.put(None)
        self.thread.join(timeout=5)

import time
from check_numbers import unsupported_numbers
from commentary import SYSTEM, build_user_message, commentate
from facts import build_facts
from fake_client import FakeClient
from live_commentary import CommentaryWorker
 
bad = 0
def check(name, ok, extra=""):
    global bad; bad += not ok; print("PASS" if ok else "FAIL", name, extra)
 
facts = build_facts()
check("3 goals in the facts", facts["goals_detected"] == 3)
 
fc = FakeClient("And it is in! Player 1 scores at 5.7 seconds, and that is goal number 1 of 3 in this clip.")
text = commentate(facts, client=fc)
call = fc.calls[0]
check("system prompt forbids inventing", "ONLY the facts" in call["system"])
check("facts are in the user message", '"goals_detected": 3' in call["messages"][0]["content"])
check("model constant is used", call["model"] == "claude-haiku-5-5")
check("text returned", text.startswith("And it is in"))
 
good = "Three goals: player 1 at 5.7 seconds, player 7 at 12.4 seconds and player 1 again at 19.1 seconds."
check("true numbers pass", unsupported_numbers(good, facts) == [], unsupported_numbers(good, facts))
made_up = "Player 9 smashes it in at 7.2 seconds and it is 4-0 after 90 minutes."
print("   invented numbers found:", unsupported_numbers(made_up, facts))
check("invented numbers are caught", {"9", "7.2", "4", "0", "90"} <= set(unsupported_numbers(made_up, facts)))
check("number words are caught", "4" in unsupported_numbers("That is four goals.", facts))
check("rounding is allowed", unsupported_numbers("Around 6 seconds in.", facts) == [])
 
# worker: success
got = []
w = CommentaryWorker(client=FakeClient("Goal!"), on_line=got.append)
t0 = time.time(); w.submit(facts); submit_time = time.time() - t0
w.q.join(); w.close()
check("submit returns at once", submit_time < 0.05, f"{submit_time*1000:.1f} ms")
check("line delivered", got == ["Goal!"])
 
# worker: network fails once, then works
w = CommentaryWorker(client=FakeClient("Back online!", fail_times=1), retries=2, on_line=got.append)
w.submit(facts); w.q.join(); w.close()
check("retry works", w.lines == ["Back online!"] and len(w.errors) == 1, w.errors)
 
# worker: always fails -> never raises, match goes on
w = CommentaryWorker(client=FakeClient("x", fail_times=99), retries=1, on_line=got.append)
w.submit(facts); w.q.join(); w.close()
check("permanent failure is logged, not raised", w.lines == [] and len(w.errors) == 2, w.errors)
 
# worker: flood -> drop
class Slow(FakeClient):
    def _create(self, **kw):
        time.sleep(0.2); return super()._create(**kw)
w = CommentaryWorker(client=Slow("ok"), on_line=lambda s: None)
for _ in range(30): w.submit(facts)
check("a flood is dropped, not queued forever", w.dropped > 0, f"dropped {w.dropped}")
print("failures:", bad)