Field Trip 2: Deploying at a Live Sports Event

"""Run this at home the night before, and again at the venue before kickoff."""
import importlib
import os
import shutil
import socket
import sys
from dataclasses import dataclass
 
 
@dataclass
class Result:
    name: str
    status: str            # "PASS", "WARN" (go, but know it) or "FAIL" (do not start)
    detail: str
 
 
def check_python(minimum=(3, 9)):
    ok = sys.version_info >= minimum
    return Result("python version", "PASS" if ok else "FAIL", sys.version.split()[0])
 
 
def check_packages(names=("cv2", "numpy"), optional=("ultralytics", "anthropic", "serial")):
    out = []
    for n in names:
        try:
            importlib.import_module(n)
            out.append(Result(f"package {n}", "PASS", "installed"))
        except ImportError:
            out.append(Result(f"package {n}", "FAIL", "pip install it before you leave"))
    for n in optional:
        try:
            importlib.import_module(n)
            out.append(Result(f"package {n}", "PASS", "installed"))
        except ImportError:
            out.append(Result(f"package {n}", "WARN", "missing: that feature will be off"))
    return out
 
 
def check_files(paths):
    return [Result(f"file {p}", "PASS" if os.path.isfile(p) else "FAIL", "found" if os.path.isfile(p) else "missing") for p in paths]
 
 
def check_disk(folder=".", need_gb=5.0):
    free = shutil.disk_usage(folder).free / 1e9
    return Result("disk space", "PASS" if free >= need_gb else "FAIL", f"{free:.1f} GB free, need {need_gb}")
 
 
def check_camera(open_camera):
    """open_camera: a function returning (ok, width, height). Injected, so we can test without a camera."""
    try:
        ok, w, h = open_camera()
    except Exception as e:
        return Result("camera", "FAIL", f"{type(e).__name__}: {e}")
    return Result("camera", "PASS" if ok else "FAIL", f"{w}x{h}" if ok else "could not read a frame")
 
 
def check_api_key():
    has = bool(os.environ.get("ANTHROPIC_API_KEY"))
    return Result("Claude API key", "PASS" if has else "WARN", "set" if has else "not set: commentary will be off")
 
 
def check_internet(host="api.anthropic.com", port=443, timeout=2.0, connect=socket.create_connection):
    try:
        connect((host, port), timeout=timeout).close()
        return Result("internet", "PASS", f"reached {host}")
    except OSError:
        return Result("internet", "WARN", "no connection: Claude features will be off (the detector works offline)")
 
 
def run(results):
    flat = []
    for r in results:
        flat.extend(r if isinstance(r, list) else [r])
    for r in flat:
        print(f"{r.status:<5} {r.name:<28} {r.detail}")
    fails = [r for r in flat if r.status == "FAIL"]
    warns = [r for r in flat if r.status == "WARN"]
    print(f"\n{len(flat) - len(fails) - len(warns)} pass, {len(warns)} warnings, {len(fails)} failures")
    print("GO" if not fails else "NO GO: fix the failures first")
    return not fails
 
 
def real_camera(index=0):
    import cv2
    cap = cv2.VideoCapture(index)
    ok, frame = cap.read()
    cap.release()
    return (ok, frame.shape[1], frame.shape[0]) if ok else (False, 0, 0)
 
 
if __name__ == "__main__":
    go = run([
        check_python(),
        check_packages(),
        check_files(["best.pt", "calibration.json", "zones.json"]),
        check_disk(".", need_gb=5.0),
        check_camera(real_camera),
        check_api_key(),
        check_internet(),
    ])
    sys.exit(0 if go else 1)

PASS  python version               3.13.16
PASS  file best.pt                 found
FAIL  file zones.json              missing
PASS  camera                       1280x720
WARN  internet                     no connection: Claude features will be off (the detector works offline)
 
3 pass, 1 warnings, 1 failures
NO GO: fix the failures first

"""One CSV for the whole day. Every line is flushed at once, so a crash or a flat battery loses nothing."""
import csv
import os
import time
 
 
class EventLog:
    def __init__(self, path="event_log.csv", clock=time.time):
        self.path, self.clock = path, clock
        new = not os.path.exists(path) or os.path.getsize(path) == 0
        self.f = open(path, "a", newline="")
        self.w = csv.writer(self.f)
        self.t0 = clock()
        if new:
            self.w.writerow(["wall_time", "match_seconds", "kind", "text"])
            self.f.flush()
 
    def add(self, kind, text="", match_seconds=None):
        now = self.clock()
        if match_seconds is None:
            match_seconds = now - self.t0
        self.w.writerow([time.strftime("%H:%M:%S", time.localtime(now)), round(match_seconds, 1), kind, text])
        self.f.flush()                                  # to the operating system now, not "when the program ends"
        os.fsync(self.f.fileno())                       # and onto the disk
 
    def note(self, text):                               # a human typed it: "kickoff", "camera knocked", "rain started"
        self.add("note", text)
 
    def close(self):
        self.f.close()

"""A cheap self-check: is the picture good enough to trust? Run it every second or so, show a warning light."""
import cv2
import numpy as np
 
 
def picture_health(frame, dark_below=70, bright_above=200, blurry_below=20.0, noisy_above=3000.0):
    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
    brightness = float(gray.mean())
    sharpness = float(cv2.Laplacian(gray, cv2.CV_64F).var())          # low = blurry
    problems = []
    if brightness < dark_below:
        problems.append("too dark")
    if brightness > bright_above:
        problems.append("too bright")
    if sharpness < blurry_below:
        problems.append("blurry")
    if sharpness > noisy_above:                                       # noise makes edges everywhere, so this number shoots UP
        problems.append("noisy")
    return {"brightness": round(brightness, 1), "sharpness": round(sharpness, 1), "problems": problems}
 
 
class BallWatchdog:
    """Warn if the ball has not been seen for `limit` frames in a row (covered, lost, or camera problem)."""
 
    def __init__(self, limit=90):
        self.limit, self.missed = limit, 0
 
    def update(self, ball):
        self.missed = 0 if ball else self.missed + 1
        return self.missed >= self.limit

import csv, os, tempfile
import numpy as np
from event_log import EventLog
from health import picture_health
from preflight import *
 
bad = 0
def check(name, ok, extra=""):
    global bad; bad += not ok; print("PASS" if ok else "FAIL", name, extra)
 
d = tempfile.mkdtemp()
open(os.path.join(d, "best.pt"), "w").write("x")
files = check_files([os.path.join(d, "best.pt"), os.path.join(d, "zones.json")])
check("file found / missing", [r.status for r in files] == ["PASS", "FAIL"])
check("disk: impossible demand fails", check_disk(d, need_gb=10**9).status == "FAIL")
check("disk: tiny demand passes", check_disk(d, need_gb=0.0001).status == "PASS")
check("camera ok", check_camera(lambda: (True, 1920, 1080)).status == "PASS")
check("camera not ok", check_camera(lambda: (False, 0, 0)).status == "FAIL")
def boom(): raise RuntimeError("busy")
check("camera exception is a FAIL, not a crash", check_camera(boom).status == "FAIL")
def no_net(*a, **k): raise OSError("unreachable")
check("no internet is a WARN", check_internet(connect=no_net).status == "WARN")
class S:
    def close(self): pass
check("internet ok", check_internet(connect=lambda *a, **k: S()).status == "PASS")
os.environ.pop("ANTHROPIC_API_KEY", None)
check("no key is a WARN", check_api_key().status == "WARN")
check("python ok", check_python().status == "PASS")
check("missing optional package is only a WARN", all(r.status != "FAIL" for r in check_packages(names=(), optional=("nonexistent_pkg_xyz",))))
check("missing required package is a FAIL", check_packages(names=("nonexistent_pkg_xyz",), optional=())[0].status == "FAIL")
print("--- sample report")
go = run([check_python(), files, check_camera(lambda: (True, 1280, 720)), check_internet(connect=no_net)])
check("run() says NO GO when a file is missing", go is False)
print("---")
 
# event log: crash safe
p = os.path.join(d, "log.csv"); t = [1000.0]
log = EventLog(p, clock=lambda: t[0])
t[0] = 1003.5; log.add("goal", "north goal", match_seconds=5.7)
t[0] = 1010.0; log.note("camera knocked, recalibrating")
rows = list(csv.reader(open(p)))                      # read while it is still open: nothing waits in a buffer
check("rows visible before close (crash safe)", len(rows) == 3 and rows[1][2] == "goal" and rows[2][3].startswith("camera"), rows)
log.close()
log2 = EventLog(p); log2.add("start", "second run"); log2.close()
rows = list(csv.reader(open(p)))
check("second run appends, header only once", len(rows) == 4 and sum(r[0] == "wall_time" for r in rows) == 1, len(rows))
frame = np.full((360, 640, 3), 30, np.uint8)
check("health check from lesson 36 reused", "too dark" in picture_health(frame)["problems"])
print("failures:", bad)