diff --git a/backend/README.md b/backend/README.md
new file mode 100644
index 00000000..9275ee49
--- /dev/null
+++ b/backend/README.md
@@ -0,0 +1,6 @@
+## Camera identification
+Camera selection follows: explicit config → auto-identify → filename hint → detected homography.
+Automatic matching uses median NCC of normalized 96×54 grayscale thumbnails across five startup frames: at least 0.75 correlation and a 0.12 lead over the runner-up, configured in `PipelineConfig`.
+Detector-vs-detector keypoints (at least 10 shared points) are a recorded cross-check only. Only courts 2, 4, and 6 have fingerprints; both court-1 setups require explicit config or a recognized filename hint.
+Add a court with `python3 -m backend.tools.calibrate_court --image --camera-id `, save its reference frame under `calibration_frames/`, and add its camera ID and frame filename to `camera_fingerprints.json`.
+Validate locally with `python3 -m backend.tools.identify_camera [more...] --frames 5 --json` (CPU; requires model weights).
diff --git a/backend/calibration_frames/camera_fingerprints.json b/backend/calibration_frames/camera_fingerprints.json
new file mode 100644
index 00000000..b7f5aa57
--- /dev/null
+++ b/backend/calibration_frames/camera_fingerprints.json
@@ -0,0 +1,5 @@
+{
+ "uc_davis_court2": { "frame": "court2.png" },
+ "uc_davis_court4": { "frame": "court4.png" },
+ "uc_davis_court6": { "frame": "court6.png" }
+}
diff --git a/backend/models/__init__.py b/backend/models/__init__.py
index aad443d0..13377630 100644
--- a/backend/models/__init__.py
+++ b/backend/models/__init__.py
@@ -1,8 +1,22 @@
-from backend.models.ball_tracker import BallDetector
-from backend.models.court_line_detector import CourtLineDetector
-from backend.models.player_tracker import PlayerTracker
-from backend.models.bounce_detector import BounceDetector
-from backend.models.stroke_detector import ActionRecognition
-from backend.models.stroke_classifier_tcn import PoseStrokeClassifier
-from backend.models.trajectory_rectifier import TrajectoryRectifier
+"""Import each model only when requested, including the standalone court CLI."""
+from importlib import import_module
+
+_EXPORTS = {
+ "BallDetector": "ball_tracker",
+ "CourtLineDetector": "court_line_detector",
+ "PlayerTracker": "player_tracker",
+ "BounceDetector": "bounce_detector",
+ "ActionRecognition": "stroke_detector",
+ "PoseStrokeClassifier": "stroke_classifier_tcn",
+ "TrajectoryRectifier": "trajectory_rectifier",
+}
+__all__ = list(_EXPORTS)
+
+
+def __getattr__(name):
+ if name not in _EXPORTS:
+ raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+ value = getattr(import_module(f"{__name__}.{_EXPORTS[name]}"), name)
+ globals()[name] = value
+ return value
diff --git a/backend/models/court_line_detector.py b/backend/models/court_line_detector.py
index f583c44c..35193168 100644
--- a/backend/models/court_line_detector.py
+++ b/backend/models/court_line_detector.py
@@ -1,12 +1,13 @@
-import torch
-import torchvision.transforms as transforms
import cv2
-from torchvision import models
import numpy as np
class CourtLineDetector:
def __init__(self, model_path, device='cuda'):
+ import torch
+ import torchvision.transforms as transforms
+ from torchvision import models
+
self.device = device
self.model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V1)
self.model.fc = torch.nn.Linear(self.model.fc.in_features, 14*2)
@@ -21,6 +22,8 @@ def __init__(self, model_path, device='cuda'):
])
def predict(self, image):
+ import torch
+
image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
image_tensor = self.transform(image_rgb).unsqueeze(0).to(self.device)
with torch.no_grad():
diff --git a/backend/pipeline/config.py b/backend/pipeline/config.py
index 4a907ed0..731812b8 100644
--- a/backend/pipeline/config.py
+++ b/backend/pipeline/config.py
@@ -1,14 +1,27 @@
# Pipeline Config class to replace colab global flags
import os
-from dataclasses import dataclass
+from dataclasses import dataclass, field
from typing import Optional, Tuple
-import torch
# Absolute path to calibration_frames/ regardless of working directory
_CALIBRATION_DIR = os.path.join(os.path.dirname(__file__), '..', 'calibration_frames')
_DEFAULT_CALIBRATION_PATH = os.path.normpath(
os.path.join(_CALIBRATION_DIR, 'court_calibration.json')
)
+_DEFAULT_FINGERPRINTS_PATH = os.path.normpath(
+ os.path.join(_CALIBRATION_DIR, 'camera_fingerprints.json')
+)
+
+
+def _default_device() -> str:
+ # Importing config and explicitly choosing CPU must work without torch.
+ try:
+ import torch
+ except ModuleNotFoundError as exc:
+ if exc.name != "torch":
+ raise
+ return "cpu"
+ return "cuda" if torch.cuda.is_available() else "cpu"
@dataclass
@@ -16,7 +29,7 @@ class PipelineConfig:
"""Central configuration for the tennis analysis pipeline"""
# ========== Device ==========
- device: str = "cuda" if torch.cuda.is_available() else "cpu"
+ device: str = field(default_factory=_default_device)
# ========== Video Processing ==========
enforce_720p: bool = False
@@ -91,7 +104,14 @@ class PipelineConfig:
# Path to court_calibration.json produced by backend.tools.calibrate_court.
# When set together with camera_id, per-frame court detection is skipped.
calibration_path: Optional[str] = _DEFAULT_CALIBRATION_PATH
- camera_id: Optional[str] = 'uc_davis_court1_zoomed'
+ camera_fingerprints_path: str = _DEFAULT_FINGERPRINTS_PATH
+ camera_id: Optional[str] = None # An explicit ID always overrides identification.
+ camera_auto_identify: bool = True # Match startup backgrounds to reference frames.
+ camera_match_min_ncc: float = 0.75 # Minimum median background correlation.
+ camera_match_min_ncc_margin: float = 0.12 # Required lead over the runner-up.
+ camera_match_min_points: int = 10 # Overlapping points for the diagnostic cross-check.
+ # Sample evenly across the first court_detection_startup_frames * 6 frames.
+ camera_identify_frames: int = 5
# ========== Bounce Detection ==========
# CatBoost regressor confidence threshold (0-1). Higher = fewer but more certain bounces.
diff --git a/backend/pipeline/run.py b/backend/pipeline/run.py
index 3a3770c5..ca9411f9 100644
--- a/backend/pipeline/run.py
+++ b/backend/pipeline/run.py
@@ -5,14 +5,14 @@
from tqdm import tqdm
from pathlib import Path
-from backend.models import BallDetector, CourtLineDetector, PlayerTracker, BounceDetector, ActionRecognition, PoseStrokeClassifier, TrajectoryRectifier
+from backend.models import CourtLineDetector
from backend.vision import HomographyEstimator, CourtReference, draw_ball_trace, draw_court_keypoints_and_lines, draw_minimap_ball_and_bounces, draw_minimap_players, draw_player_bboxes, draw_stroke_labels
from backend.vision import SwingDetector, extract_pose_sequence
from backend.vision.calibration import load_calibration
+from backend.vision.camera_identify import CameraMatch, filename_camera_hint, identify_from_frames, load_fingerprints
from backend.vision.heatmaps import generate_minimap_heatmaps, generate_player_shot_dot_map
from backend.vision.postprocess import detect_shot_frames
-from backend.pipeline.storage import upload_processed_video, upload_heatmap_png, get_supabase, make_streamable_mp4, upload_results_parallel, ProcessedVideoUploadError, PROCESSED_VIDEO_UPLOAD_ERROR
from backend.pipeline.config import PipelineConfig
from backend.pipeline.rallies import (
RALLY_GAP_SECONDS,
@@ -1744,25 +1744,69 @@ def err_rate(key: str) -> str:
return None
-_OPPONENT_COURT_MAP = {
- "2": "uc_davis_court2",
- "4": "uc_davis_court4",
- "6": "uc_davis_court6",
-}
-
-
def _resolve_camera_id(video_path: str) -> str | None:
"""
- Infer camera_id from the video filename.
+ Infer an advisory camera hint from the video filename; never override detection.
Matches filenames containing 'Court' followed by a court number (e.g. StMarys_Court2.mp4).
- Returns None if no match — pipeline falls back to per-frame court detection.
+ Returns None if no match. Used only when automatic identification finds nothing.
"""
- import re
- stem = Path(video_path).stem
- match = re.search(r'[Cc]ourt(\d+)', stem)
- if match:
- return _OPPONENT_COURT_MAP.get(match.group(1))
- return None
+ return filename_camera_hint(video_path)
+
+
+_CAMERA_STARTUP_SPREAD = 6
+
+
+def _identify_camera_for_video(video_path: str, config: PipelineConfig, court_detector) -> CameraMatch:
+ """Sample spread-out startup frames and identify a camera without changing config."""
+ fingerprints = load_fingerprints(Path(config.camera_fingerprints_path).parent, config.camera_fingerprints_path)
+ if config.camera_identify_frames < 1 or config.court_detection_startup_frames < 1:
+ raise ValueError("Camera identification frame counts must be positive")
+ frames = []
+ cap = cv2.VideoCapture(video_path)
+ try:
+ if not cap.isOpened():
+ raise ValueError(f"Cannot open video for camera identification: {video_path}")
+ limit = config.court_detection_startup_frames * _CAMERA_STARTUP_SPREAD
+ total = cap.get(cv2.CAP_PROP_FRAME_COUNT)
+ if np.isfinite(total) and total > 0:
+ limit = min(limit, int(total))
+ indices = np.linspace(0, limit - 1, min(config.camera_identify_frames, limit), dtype=int)
+ for index in indices:
+ cap.set(cv2.CAP_PROP_POS_FRAMES, int(index))
+ ok, frame = cap.read()
+ if ok:
+ frames.append(frame)
+ else:
+ print(f"[Camera] Could not read sampled frame {index}", flush=True)
+ finally:
+ cap.release()
+ return identify_from_frames(
+ frames, court_detector, fingerprints,
+ min_ncc=config.camera_match_min_ncc,
+ min_margin=config.camera_match_min_ncc_margin,
+ min_points=config.camera_match_min_points,
+ )
+
+
+def _record_camera_match(supabase, match_id: str, camera_id: str | None, match: CameraMatch, hint: str | None):
+ """Write optional camera metadata; pre-migration failures must not fail a run."""
+ def finite_or_none(value):
+ return float(value) if value is not None and np.isfinite(value) else None
+
+ metadata = {
+ "reason": match.reason,
+ "best": finite_or_none(match.best),
+ "runner_up": finite_or_none(match.runner_up),
+ "hint": hint,
+ "scores": {camera: finite_or_none(distance) for camera, distance in match.scores.items()},
+ "keypoint_camera_id": match.keypoint_camera_id,
+ "keypoint_scores": {camera: finite_or_none(distance) for camera, distance in match.keypoint_scores.items()},
+ "keypoint_agrees": match.keypoint_agrees,
+ }
+ try:
+ supabase.table("matches").update({"camera_id": camera_id, "camera_match": metadata}).eq("id", match_id).execute()
+ except Exception as e:
+ print(f"[Camera] Decision write failed (camera columns may be missing): {e}", flush=True)
def run_pipeline(video_path: str, match_id: str, local_mode: bool = False, config: PipelineConfig = None):
@@ -1770,6 +1814,8 @@ def run_pipeline(video_path: str, match_id: str, local_mode: bool = False, confi
Main pipeline entry point (Modal-compatible).
"""
import sys
+ from backend.models import BallDetector, PlayerTracker, BounceDetector, ActionRecognition, PoseStrokeClassifier, TrajectoryRectifier
+ from backend.pipeline.storage import upload_processed_video, upload_heatmap_png, get_supabase, make_streamable_mp4, upload_results_parallel, ProcessedVideoUploadError, PROCESSED_VIDEO_UPLOAD_ERROR
# Belt-and-suspenders in case the Modal Image env var doesn't take.
try:
sys.stdout.reconfigure(line_buffering=True)
@@ -1786,10 +1832,11 @@ def run_pipeline(video_path: str, match_id: str, local_mode: bool = False, confi
# step muxes audio from this so annotated playback isn't silent.
original_video_path = video_path
- # Auto-resolve camera_id from filename unless explicitly set by caller
- if config.camera_id == PipelineConfig.camera_id:
- resolved = _resolve_camera_id(video_path)
- config.camera_id = resolved
+ # Retain the filename logs for continuity; this is only an advisory hint.
+ filename_hint = _resolve_camera_id(video_path)
+ print(f"[Camera] filename hint={filename_hint}", flush=True)
+ if config.camera_id is None:
+ resolved = filename_hint
if resolved:
print(f"[Pipeline] Camera ID resolved from filename: {resolved}", flush=True)
else:
@@ -1965,22 +2012,58 @@ def update_progress(value: float, stage: "str | None" = None):
calibrated_H_ref = None
calibrated_H_frame = None
calibrated_keypoints = None
+ cal_H_ref, cal_H_frame, cal_kps = None, None, None
+ camera_match = CameraMatch(config.camera_id, {}, None, None, 0, "explicit" if config.camera_id else "disabled")
if config.calibration_path and config.camera_id:
cal_H_ref, cal_H_frame, cal_kps = load_calibration(config.calibration_path, config.camera_id)
- if cal_H_ref is not None:
- calibrated_H_ref = cal_H_ref
- calibrated_H_frame = cal_H_frame
- calibrated_keypoints = cal_kps
- print(f"[Pipeline] Calibration: {config.camera_id} loaded")
+ elif config.camera_id is None and config.camera_auto_identify:
+ try:
+ court_detector = CourtLineDetector(
+ model_path=os.path.join(WEIGHTS_DIR, "keypoints_model.pth"),
+ device=device,
+ )
+ camera_match = _identify_camera_for_video(video_path, config, court_detector)
+ if camera_match.camera_id:
+ cal_H_ref, cal_H_frame, cal_kps = load_calibration(config.calibration_path, camera_match.camera_id)
+ if cal_H_ref is not None:
+ config.camera_id = camera_match.camera_id
+ except Exception as e:
+ print(f"[Camera] Identification failed; continuing with fallback: {e}", flush=True)
+ camera_match = CameraMatch(None, {}, None, None, 0, "error")
+ print(
+ f"[Camera] identified={camera_match.camera_id or 'none'} reason={camera_match.reason} "
+ f"best={camera_match.best} runner_up={camera_match.runner_up} hint={filename_hint} "
+ f"keypoint_camera_id={camera_match.keypoint_camera_id} keypoint_agrees={camera_match.keypoint_agrees} "
+ f"keypoint_scores={camera_match.keypoint_scores}",
+ flush=True,
+ )
+ if camera_match.camera_id and filename_hint:
+ agreement = "agrees" if camera_match.camera_id == filename_hint else "disagrees"
+ print(f"[Camera] filename hint {agreement} with detected camera", flush=True)
+ if cal_H_ref is None and filename_hint and config.calibration_path:
+ try:
+ cal_H_ref, cal_H_frame, cal_kps = load_calibration(config.calibration_path, filename_hint)
+ if cal_H_ref is not None:
+ config.camera_id = filename_hint
+ print("[Camera] falling back to filename hint", flush=True)
+ except Exception as e:
+ print(f"[Camera] Filename hint calibration failed; using court detection: {e}", flush=True)
+
+ if cal_H_ref is not None:
+ calibrated_H_ref = cal_H_ref
+ calibrated_H_frame = cal_H_frame
+ calibrated_keypoints = cal_kps
+ print(f"[Pipeline] Calibration: {config.camera_id} loaded")
# If no calibration was loaded, detect the court once on the first N frames.
# This replaces per-frame detection in Pass 2 — the camera is fixed.
if calibrated_H_ref is None:
print(f"[Pipeline] No calibration — running court detection on first {config.court_detection_startup_frames} frames")
- court_detector = CourtLineDetector(
- model_path=os.path.join(WEIGHTS_DIR, "keypoints_model.pth"),
- device=device,
- )
+ if court_detector is None:
+ court_detector = CourtLineDetector(
+ model_path=os.path.join(WEIGHTS_DIR, "keypoints_model.pth"),
+ device=device,
+ )
cap_startup = cv2.VideoCapture(video_path)
startup_frames = []
for _ in range(config.court_detection_startup_frames):
@@ -1996,6 +2079,9 @@ def update_progress(value: float, stage: "str | None" = None):
else:
print("[Pipeline] WARNING: Court detection failed on startup frames")
+ if not local_mode:
+ _record_camera_match(supabase, match_id, config.camera_id, camera_match, filename_hint)
+
# ---------- Pass 1: Ball + Player tracking + pose extraction ----------
cap = cv2.VideoCapture(video_path)
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
diff --git a/backend/tests/test_camera_identify.py b/backend/tests/test_camera_identify.py
new file mode 100644
index 00000000..a0685fe8
--- /dev/null
+++ b/backend/tests/test_camera_identify.py
@@ -0,0 +1,618 @@
+"""Camera matching tests use synthetic signatures and fake detectors, never models."""
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from backend.vision.camera_identify import (
+ CameraMatch,
+ identify_by_background,
+ identify_by_keypoints,
+ identify_from_frames,
+ load_fingerprints,
+ ncc,
+ score_against,
+)
+from backend.vision import camera_identify as matcher
+
+
+CALIBRATION_DIR = Path(__file__).resolve().parents[1] / "calibration_frames"
+CAMERAS = ("uc_davis_court2", "uc_davis_court4", "uc_davis_court6")
+GATES = {"max_distance": 60.0, "min_margin_ratio": 0.8, "min_points": 10}
+NCC_GATES = {"min_ncc": 0.75, "min_margin": 0.12}
+FRAME_SIZE = (1920, 1080)
+
+
+@pytest.fixture
+def calibrations():
+ # Stand-ins for outputs of the same detector on each reference frame.
+ base = np.column_stack((np.linspace(200, 1400, 14), np.linspace(150, 800, 14)))
+ return {camera: (base + [offset, 0]).tolist() for camera, offset in zip(CAMERAS, (0, 60, -60))}
+
+
+def _signature(pattern):
+ centered = pattern - pattern.mean()
+ return centered / centered.std()
+
+
+@pytest.fixture
+def fingerprints(monkeypatch):
+ rng = np.random.default_rng(7)
+ monkeypatch.setattr(matcher, "thumbnail_signature", lambda frame: frame)
+ return {camera: {"signature": _signature(rng.normal(size=(54, 96))), "frame_path": f"{camera}.png"}
+ for camera in CAMERAS}
+
+
+def test_fingerprint_manifest_lists_only_courts_with_reference_frames():
+ import json
+
+ entries = json.loads((CALIBRATION_DIR / "camera_fingerprints.json").read_text())
+ assert entries == {camera: {"frame": f"court{camera[-1]}.png"} for camera in CAMERAS}
+ assert all((CALIBRATION_DIR / entry["frame"]).is_file() for entry in entries.values())
+
+
+@pytest.mark.parametrize("camera", CAMERAS)
+def test_identical_signature_matches_itself(fingerprints, camera):
+ signature = fingerprints[camera]["signature"]
+ assert ncc(signature, signature) == pytest.approx(1.0)
+ match = identify_by_background([signature], fingerprints, **NCC_GATES)
+ assert match.camera_id == camera
+ assert match.reason == "matched"
+ assert match.best == pytest.approx(1.0)
+ assert list(match.scores.values()) == sorted(match.scores.values(), reverse=True)
+
+
+def test_noisy_signature_still_matches(fingerprints):
+ query = _signature(fingerprints[CAMERAS[0]]["signature"] + np.random.default_rng(0).normal(0, 0.5, (54, 96)))
+ match = identify_by_background([query], fingerprints, **NCC_GATES)
+ assert match.camera_id == CAMERAS[0]
+ assert match.best - match.runner_up >= NCC_GATES["min_margin"]
+
+
+def test_similar_backgrounds_are_ambiguous(fingerprints):
+ base = fingerprints[CAMERAS[0]]["signature"]
+ other = fingerprints[CAMERAS[1]]["signature"]
+ orthogonal = _signature(other - ncc(base, other) * base)
+ similar = 0.9 * base + np.sqrt(1 - 0.9 ** 2) * orthogonal
+ references = {"base": {"signature": base}, "similar": {"signature": similar}}
+ match = identify_by_background([base], references, **NCC_GATES)
+ assert match.runner_up == pytest.approx(0.9)
+ assert match.camera_id is None
+ assert match.reason == "ambiguous"
+
+
+def test_uncorrelated_background_is_too_low(fingerprints):
+ query = _signature(np.random.default_rng(8).normal(size=(54, 96)))
+ match = identify_by_background([query], fingerprints, **NCC_GATES)
+ assert match.camera_id is None
+ assert match.reason == "too_low"
+
+
+def test_background_without_references(fingerprints):
+ match = identify_by_background([fingerprints[CAMERAS[0]]["signature"]], {}, **NCC_GATES)
+ assert match.camera_id is None
+ assert match.reason == "no_fingerprints"
+
+
+@pytest.mark.parametrize("constant", [False, True])
+def test_thumbnail_preprocessing_with_numpy_opencv_stub(monkeypatch, constant):
+ import sys
+ from types import SimpleNamespace
+ from unittest.mock import Mock
+
+ thumbnail = np.ones((54, 96)) if constant else np.arange(54 * 96).reshape(54, 96)
+ canonical = np.zeros((1080, 1920, 3), dtype=np.uint8)
+ gray = canonical[:, :, 0]
+ cv2 = SimpleNamespace(COLOR_BGR2GRAY=6, INTER_AREA=3,
+ resize=Mock(side_effect=[canonical, thumbnail]), cvtColor=Mock(return_value=gray))
+ monkeypatch.setitem(sys.modules, "cv2", cv2)
+ result = matcher.thumbnail_signature(np.zeros((720, 1280, 3), dtype=np.uint8))
+ assert cv2.resize.call_args_list[0].args[1] == FRAME_SIZE
+ assert cv2.resize.call_args_list[1].args[1] == (96, 54)
+ assert cv2.resize.call_args_list[1].kwargs == {"interpolation": cv2.INTER_AREA}
+ assert cv2.cvtColor.call_args.args[0] is canonical
+ assert cv2.cvtColor.call_args.args[1] == cv2.COLOR_BGR2GRAY
+ assert result.mean() == pytest.approx(0, abs=1e-12)
+ assert np.isfinite(result).all()
+ assert result.std() == pytest.approx(0 if constant else 1)
+
+
+def test_detector_fingerprints_runs_once_per_reference(monkeypatch, fingerprints, calibrations):
+ calls = []
+
+ def reference_frame(path):
+ calls.append(path)
+ return np.zeros((1080, 1920, 3), dtype=np.uint8)
+
+ monkeypatch.setattr(matcher, "_reference_frame", reference_frame)
+ detector = FakeDetector([calibrations[camera] for camera in CAMERAS])
+ result = matcher.detector_fingerprints(detector, fingerprints)
+ assert result == calibrations
+ assert calls == [fingerprints[camera]["frame_path"] for camera in CAMERAS]
+
+
+def test_keypoint_cross_check_uses_median(monkeypatch, fingerprints, calibrations):
+ monkeypatch.setattr(matcher, "detector_fingerprints", lambda detector, references: calibrations)
+ detector = FakeDetector([calibrations[CAMERAS[0]], [(9999, 9999)] * 14, calibrations[CAMERAS[0]]])
+ frames = [fingerprints[CAMERAS[0]]["signature"]] * 3
+ match = identify_from_frames(frames, detector, fingerprints, frame_size=FRAME_SIZE, min_points=10, **NCC_GATES)
+ assert match.keypoint_agrees is True
+ assert match.keypoint_scores[CAMERAS[0]] == 0
+
+
+@pytest.mark.parametrize("cross_check,agrees", [("uc_davis_court4", True), ("uc_davis_court2", False)])
+def test_background_median_decides_and_keypoints_only_cross_check(monkeypatch, fingerprints, calibrations, cross_check, agrees):
+ frames = [fingerprints["uc_davis_court4"]["signature"], fingerprints["uc_davis_court2"]["signature"],
+ fingerprints["uc_davis_court4"]["signature"]]
+ monkeypatch.setattr(matcher, "detector_fingerprints", lambda detector, references: calibrations)
+ match = identify_from_frames(frames, FakeDetector([calibrations[cross_check]] * 3), fingerprints,
+ frame_size=FRAME_SIZE, min_points=10, **NCC_GATES)
+ assert match.camera_id == "uc_davis_court4"
+ assert match.best == pytest.approx(1.0)
+ assert match.keypoint_camera_id == cross_check
+ assert match.keypoint_agrees is agrees
+ assert match.keypoint_scores[cross_check] == 0
+ assert list(match.keypoint_scores.values()) == sorted(match.keypoint_scores.values())
+
+
+def test_background_rejection_never_uses_keypoints(fingerprints):
+ query = _signature(np.random.default_rng(8).normal(size=(54, 96)))
+ # An empty detector iterator would raise if inference ran.
+ match = identify_from_frames([query], FakeDetector([]), fingerprints, min_points=10, **NCC_GATES)
+ assert match.reason == "too_low"
+ assert match.keypoint_agrees is None
+ assert match.keypoint_scores == {}
+
+
+def test_cross_check_failure_keeps_background_match(monkeypatch, fingerprints, capsys):
+ def fail(*args):
+ raise RuntimeError("Cross-check unavailable")
+
+ monkeypatch.setattr(matcher, "detector_fingerprints", fail)
+ match = identify_from_frames([fingerprints[CAMERAS[0]]["signature"]], FakeDetector([]), fingerprints,
+ min_points=10, **NCC_GATES)
+ assert match.camera_id == CAMERAS[0]
+ assert match.reason == "matched"
+ assert match.keypoint_agrees is None
+ assert "Cross-check unavailable" in capsys.readouterr().out
+
+
+@pytest.mark.parametrize("camera", CAMERAS)
+def test_detector_reference_identifies_itself(calibrations, camera):
+ match = identify_by_keypoints(calibrations[camera], calibrations, FRAME_SIZE, **GATES)
+ assert match.camera_id == camera
+ assert match.best == 0
+ assert match.valid_points == 14
+ assert match.reason == "matched"
+ assert list(match.scores.values()) == sorted(match.scores.values())
+
+
+@pytest.mark.parametrize("camera", CAMERAS)
+def test_noise_preserves_identity(calibrations, camera):
+ detected = np.asarray(calibrations[camera]) + np.random.default_rng(0).normal(0, 4, (14, 2))
+ match = identify_by_keypoints(detected.tolist(), calibrations, FRAME_SIZE, **GATES)
+ assert match.camera_id == camera
+ assert match.reason == "matched"
+
+
+def test_shifted_court2_distance_gate(calibrations):
+ detected = (np.asarray(calibrations["uc_davis_court2"]) + [40, 0]).tolist()
+ # Isolate the optional absolute distance gate for the keypoint path.
+ only_court2 = {"uc_davis_court2": calibrations["uc_davis_court2"]}
+ strict = identify_by_keypoints(detected, only_court2, FRAME_SIZE, **{**GATES, "max_distance": 30})
+ assert strict.camera_id is None
+ assert strict.reason == "too_far"
+ relaxed = identify_by_keypoints(detected, only_court2, FRAME_SIZE, **GATES)
+ assert relaxed.camera_id == "uc_davis_court2"
+
+
+def test_halfway_between_courts_is_ambiguous(calibrations):
+ halfway = (np.asarray(calibrations["uc_davis_court2"]) + calibrations["uc_davis_court6"]) / 2
+ match = identify_by_keypoints(halfway.tolist(), calibrations, FRAME_SIZE, **GATES)
+ assert match.camera_id is None
+ assert match.reason == "ambiguous"
+
+
+def test_too_few_detected_points(calibrations):
+ detected = calibrations["uc_davis_court2"][:8] + [None] * 6
+ match = identify_by_keypoints(detected, calibrations, FRAME_SIZE, **GATES)
+ assert match.camera_id is None
+ assert match.reason == "no_detection"
+ assert match.valid_points == 8
+
+
+@pytest.mark.parametrize("camera", CAMERAS)
+def test_frame_size_scaling(calibrations, camera):
+ detected = (np.asarray(calibrations[camera]) * (2 / 3)).tolist()
+ match = identify_by_keypoints(detected, calibrations, (1280, 720), **GATES)
+ assert match.camera_id == camera
+ assert match.best < 1
+
+
+class FakeDetector:
+ def __init__(self, detections):
+ self.detections = iter(detections)
+
+ def infer_single(self, frame):
+ return next(self.detections)
+
+
+@pytest.mark.parametrize("detected", [None, [], [None] * 14])
+def test_no_detection(calibrations, detected):
+ match = identify_by_keypoints(detected, calibrations, FRAME_SIZE, **GATES)
+ assert match.camera_id is None
+ assert match.reason == "no_detection"
+
+
+def test_no_calibrations(calibrations):
+ match = identify_by_keypoints(calibrations["uc_davis_court4"], {}, FRAME_SIZE, **GATES)
+ assert match.camera_id is None
+ assert match.reason == "no_calibrations"
+
+
+def test_empty_frames(fingerprints):
+ match = identify_from_frames([], FakeDetector([]), fingerprints, min_points=10, **NCC_GATES)
+ assert match.reason == "no_detection"
+
+
+def test_detector_returns_none(monkeypatch, fingerprints, calibrations):
+ monkeypatch.setattr(matcher, "detector_fingerprints", lambda detector, references: calibrations)
+ match = identify_from_frames([fingerprints[CAMERAS[0]]["signature"]], FakeDetector([None]), fingerprints,
+ frame_size=FRAME_SIZE, min_points=10, **NCC_GATES)
+ assert match.reason == "matched"
+ assert match.keypoint_camera_id is None
+ assert match.keypoint_agrees is None
+
+
+def test_score_uses_only_shared_finite_points():
+ detected = [(3, 4), None, (np.nan, 1), (1, 1)]
+ stored = [(0, 0), (0, 0), (0, 0), None]
+ assert score_against(detected, stored, FRAME_SIZE) == (5.0, 1)
+ assert score_against([None], [(0, 0)], FRAME_SIZE) == (float("inf"), 0)
+
+
+def test_sparse_calibration_cannot_win(calibrations):
+ detected = calibrations["uc_davis_court4"]
+ entries = {"sparse": detected[:8] + [None] * 6, **calibrations}
+ match = identify_by_keypoints(detected, entries, FRAME_SIZE, **GATES)
+ assert match.camera_id == "uc_davis_court4"
+
+
+def test_single_calibration_can_match(calibrations):
+ points = calibrations["uc_davis_court4"]
+ match = identify_by_keypoints(points, {"only": points}, FRAME_SIZE, **GATES)
+ assert match.camera_id == "only"
+ assert match.runner_up is None
+
+
+def test_duplicate_calibrations_are_ambiguous(calibrations):
+ points = calibrations["uc_davis_court4"]
+ match = identify_by_keypoints(points, {"first": points, "second": points}, FRAME_SIZE, **GATES)
+ assert match.camera_id is None
+ assert match.reason == "ambiguous"
+
+
+def test_loader_skips_missing_reference_frames(tmp_path, capsys):
+ path = tmp_path / "fingerprints.json"
+ path.write_text('{"missing": {"frame": "missing.png"}}')
+ assert load_fingerprints(tmp_path, path) == {}
+ assert "missing.png" in capsys.readouterr().out
+
+
+def test_loader_skips_unreadable_reference_frames(tmp_path, monkeypatch, capsys):
+ path = tmp_path / "fingerprints.json"
+ path.write_text('{"broken": {"frame": "broken.png"}}')
+ (tmp_path / "broken.png").touch()
+
+ def unreadable(path):
+ raise ValueError(f"Cannot read camera reference frame: {path}")
+
+ monkeypatch.setattr(matcher, "_reference_frame", unreadable)
+ assert load_fingerprints(tmp_path, path) == {}
+ assert "broken.png" in capsys.readouterr().out
+
+
+@pytest.mark.parametrize("contents", ['[]', '{"bad": []}', '{"bad": {"frame": null}}'])
+def test_loader_rejects_malformed_entries(tmp_path, contents):
+ path = tmp_path / "calibration.json"
+ path.write_text(contents)
+ with pytest.raises(ValueError):
+ load_fingerprints(tmp_path, path)
+
+
+def test_numpy_only_imports_and_config_without_torch():
+ import subprocess
+ import sys
+
+ code = '''
+import importlib.abc
+import sys
+class NoOptionalDependencies(importlib.abc.MetaPathFinder):
+ def find_spec(self, fullname, *args):
+ if fullname.split(".")[0] in {"torch", "cv2", "scipy", "pandas"}:
+ raise ModuleNotFoundError(fullname, name=fullname)
+sys.meta_path.insert(0, NoOptionalDependencies())
+from backend.vision.camera_identify import identify_by_background
+from backend.pipeline.config import PipelineConfig
+assert "torch" not in sys.modules
+config = PipelineConfig()
+assert config.device == "cpu"
+assert config.camera_id is None
+assert config.camera_auto_identify is True
+assert config.camera_match_min_ncc == 0.75
+assert config.camera_match_min_ncc_margin == 0.12
+assert config.camera_fingerprints_path.endswith("calibration_frames/camera_fingerprints.json")
+assert config.camera_match_min_points == 10
+assert config.camera_identify_frames == 5
+'''
+ result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True)
+ assert result.returncode == 0, result.stderr
+
+
+def test_identify_camera_for_synthetic_video(tmp_path, monkeypatch, calibrations):
+ cv2 = pytest.importorskip("cv2")
+ import backend.pipeline.run as pipeline
+ from backend.pipeline.config import PipelineConfig
+
+ class FakeCourtDetector:
+ def __init__(self, **kwargs):
+ self.calls = 0
+
+ def infer_single(self, frame):
+ self.calls += 1
+ assert (frame.shape[1], frame.shape[0]) == FRAME_SIZE
+ if self.calls <= len(CAMERAS):
+ return calibrations[CAMERAS[self.calls - 1]]
+ return calibrations["uc_davis_court4"]
+
+ monkeypatch.setattr(pipeline, "CourtLineDetector", FakeCourtDetector)
+ path = tmp_path / "Court2.avi" # A misleading hint cannot override the detector.
+ writer = cv2.VideoWriter(str(path), cv2.VideoWriter_fourcc(*"MJPG"), 10, FRAME_SIZE)
+ assert writer.isOpened()
+ reference_frame = cv2.imread(str(CALIBRATION_DIR / "court4.png"))
+ try:
+ for _ in range(12):
+ writer.write(reference_frame)
+ finally:
+ writer.release()
+ detector = pipeline.CourtLineDetector()
+ config = PipelineConfig(device="cpu")
+ match = pipeline._identify_camera_for_video(str(path), config, detector)
+ assert match.camera_id == "uc_davis_court4"
+ assert match.best > 0.99
+ assert match.keypoint_agrees is True
+ assert detector.calls == config.camera_identify_frames + len(CAMERAS)
+ assert config.camera_id is None
+
+
+@pytest.fixture
+def pipeline_startup(monkeypatch):
+ """Run real calibration wiring, stopping before tracking or external writes."""
+ pytest.importorskip("cv2")
+ import sys
+ from types import ModuleType, SimpleNamespace
+ from unittest.mock import Mock
+ import backend.models as models
+ import backend.pipeline.run as pipeline
+
+ class StopBeforeTracking(BaseException):
+ pass
+
+ state = SimpleNamespace(detectors=[], fallback_calls=[], identify_calls=[], payloads=[])
+ for name in models.__all__:
+ if name != "CourtLineDetector":
+ monkeypatch.setitem(vars(models), name, Mock())
+ storage = ModuleType("backend.pipeline.storage")
+ for name in ("upload_processed_video", "upload_heatmap_png", "get_supabase", "make_streamable_mp4", "upload_results_parallel"):
+ setattr(storage, name, Mock())
+ storage.ProcessedVideoUploadError = RuntimeError
+ storage.PROCESSED_VIDEO_UPLOAD_ERROR = "Upload failed"
+ monkeypatch.setitem(sys.modules, "backend.pipeline.storage", storage)
+ monkeypatch.setattr(pipeline, "CourtLineDetector", lambda **kwargs: state.detectors.append(object()) or state.detectors[-1])
+ monkeypatch.setattr(pipeline, "HomographyEstimator", Mock())
+ monkeypatch.setattr(pipeline, "CourtReference", Mock())
+ monkeypatch.setattr(pipeline, "_detect_court_once", lambda frames, detector, estimator: state.fallback_calls.append(detector) or (np.eye(3), []))
+ monkeypatch.setattr(pipeline, "tqdm", Mock(side_effect=StopBeforeTracking))
+ capture = Mock()
+ capture.read.return_value = (True, np.zeros((2, 2, 3), dtype=np.uint8))
+ capture.get.return_value = 1
+ monkeypatch.setattr(pipeline.cv2, "VideoCapture", Mock(return_value=capture))
+ state.pipeline, state.stop, state.storage = pipeline, StopBeforeTracking, storage
+ return state
+
+
+@pytest.mark.parametrize("explicit,auto,identified,hint,raises,expected,fallback", [
+ ("uc_davis_court1_zoomed", True, "uc_davis_court4", "Court2", False, "uc_davis_court1_zoomed", False),
+ (None, True, "uc_davis_court4", "Court2", False, "uc_davis_court4", False),
+ (None, True, None, "Court2", False, "uc_davis_court2", False),
+ (None, True, None, "unknown", False, None, True),
+ (None, True, None, "Court2", True, "uc_davis_court2", False),
+ (None, True, None, "unknown", True, None, True),
+ (None, False, None, "Court2", False, None, True),
+])
+def test_pipeline_camera_fallback_order(pipeline_startup, monkeypatch, capsys, explicit, auto, identified, hint, raises, expected, fallback):
+ from backend.pipeline.config import PipelineConfig
+
+ state = pipeline_startup
+ config = PipelineConfig(device="cpu", camera_id=explicit, camera_auto_identify=auto)
+
+ def identify(path, cfg, detector):
+ state.identify_calls.append(detector)
+ if raises:
+ raise RuntimeError("Detector failed")
+ return CameraMatch(identified, {}, 1.0 if identified else None, None, 14, "matched" if identified else "too_low")
+
+ monkeypatch.setattr(state.pipeline, "_identify_camera_for_video", identify)
+ with pytest.raises(state.stop):
+ state.pipeline.run_pipeline(f"{hint}.mp4", "local-test", local_mode=True, config=config)
+ assert config.camera_id == expected
+ assert bool(state.fallback_calls) == fallback
+ assert len(state.detectors) == (0 if explicit else 1)
+ assert len(state.identify_calls) == (1 if auto and not explicit else 0)
+ if fallback:
+ assert state.fallback_calls[0] is state.detectors[0]
+ output = capsys.readouterr().out
+ if identified and not explicit:
+ assert "filename hint disagrees" in output
+ if expected == "uc_davis_court2":
+ assert "[Camera] falling back to filename hint" in output
+ state.storage.get_supabase.assert_not_called()
+
+
+def test_pipeline_unavailable_hint_uses_detected_homography(pipeline_startup, monkeypatch, tmp_path):
+ from backend.pipeline.config import PipelineConfig
+
+ state = pipeline_startup
+ config = PipelineConfig(device="cpu", calibration_path=str(tmp_path / "missing.json"))
+ monkeypatch.setattr(state.pipeline, "_identify_camera_for_video", lambda *args: CameraMatch(None, {}, None, None, 0, "no_fingerprints"))
+ with pytest.raises(state.stop):
+ state.pipeline.run_pipeline("Court2.mp4", "local-test", local_mode=True, config=config)
+ assert config.camera_id is None
+ assert state.fallback_calls == state.detectors
+
+
+def test_camera_metadata_is_json_safe_and_missing_columns_only_log(monkeypatch, capsys):
+ pytest.importorskip("cv2")
+ import json
+ from unittest.mock import Mock
+ import backend.pipeline.run as pipeline
+
+ database = Mock()
+ match = CameraMatch(None, {"sparse": float("nan")}, None, float("nan"), 0, "no_detection",
+ keypoint_scores={"sparse": float("inf")})
+ pipeline._record_camera_match(database, "match-id", "uc_davis_court2", match, "uc_davis_court2")
+ payload = database.table.return_value.update.call_args.args[0]
+ json.dumps(payload, allow_nan=False)
+ assert payload == {"camera_id": "uc_davis_court2", "camera_match": {
+ "reason": "no_detection", "best": None, "runner_up": None,
+ "hint": "uc_davis_court2", "scores": {"sparse": None},
+ "keypoint_camera_id": None, "keypoint_scores": {"sparse": None}, "keypoint_agrees": None,
+ }}
+ database.table.return_value.update.return_value.eq.assert_called_with("id", "match-id")
+ database.table.return_value.update.return_value.eq.return_value.execute.side_effect = RuntimeError("column camera_match does not exist")
+ pipeline._record_camera_match(database, "match-id", None, match, None)
+ assert "Decision write failed" in capsys.readouterr().out
+
+
+def test_pipeline_stores_camera_decision_outside_local_mode(pipeline_startup, monkeypatch, capsys):
+ from backend.pipeline.config import PipelineConfig
+
+ state = pipeline_startup
+ match = CameraMatch("uc_davis_court4", {"uc_davis_court4": 0.93}, 0.93, 0.65, 14, "matched",
+ keypoint_camera_id="uc_davis_court2", keypoint_scores={"uc_davis_court2": 18.3}, keypoint_agrees=False)
+ monkeypatch.setattr(state.pipeline, "_identify_camera_for_video", lambda *args: match)
+ database = state.storage.get_supabase.return_value
+ database.table.return_value.select.return_value.eq.return_value.single.return_value.execute.return_value.data = {}
+ with pytest.raises(state.stop):
+ state.pipeline.run_pipeline("Court2.mp4", "saved-match", config=PipelineConfig(device="cpu"))
+ updates = [call.args[0] for call in database.table.return_value.update.call_args_list]
+ decision = next(update for update in updates if "camera_match" in update)
+ assert decision["camera_id"] == "uc_davis_court4"
+ assert decision["camera_match"]["hint"] == "uc_davis_court2"
+ assert decision["camera_match"]["reason"] == "matched"
+ assert decision["camera_match"]["best"] == 0.93
+ assert decision["camera_match"]["runner_up"] == 0.65
+ assert decision["camera_match"]["keypoint_camera_id"] == "uc_davis_court2"
+ assert decision["camera_match"]["keypoint_scores"] == {"uc_davis_court2": 18.3}
+ assert decision["camera_match"]["keypoint_agrees"] is False
+ output = capsys.readouterr().out
+ assert "best=0.93 runner_up=0.65" in output
+ assert "keypoint_agrees=False" in output
+ assert "keypoint_scores={'uc_davis_court2': 18.3}" in output
+
+
+@pytest.mark.parametrize("total,expected", [(120, [0, 14, 29, 44, 59]), (3, [0, 1, 2])])
+def test_pipeline_samples_evenly_and_releases_capture(monkeypatch, calibrations, total, expected):
+ cv2 = pytest.importorskip("cv2")
+ from unittest.mock import Mock
+ import backend.pipeline.run as pipeline
+ from backend.pipeline.config import PipelineConfig
+
+ capture = Mock()
+ capture.get.return_value = total
+ capture.read.return_value = (True, cv2.imread(str(CALIBRATION_DIR / "court4.png")))
+ monkeypatch.setattr(cv2, "VideoCapture", Mock(return_value=capture))
+ detector = FakeDetector([calibrations[camera] for camera in CAMERAS] + [calibrations["uc_davis_court4"]] * len(expected))
+ match = pipeline._identify_camera_for_video("input.mp4", PipelineConfig(device="cpu"), detector)
+ assert match.camera_id == "uc_davis_court4"
+ assert [call.args[1] for call in capture.set.call_args_list] == expected
+ capture.release.assert_called_once()
+
+
+def test_cli_samples_only_first_ten_seconds(tmp_path, monkeypatch):
+ cv2 = pytest.importorskip("cv2")
+ from unittest.mock import Mock
+ from backend.tools import identify_camera as cli
+
+ path = tmp_path / "video.mp4"
+ path.touch()
+ capture = Mock()
+ capture.get.side_effect = lambda prop: 30 if prop == cv2.CAP_PROP_FPS else 600
+ capture.read.return_value = (True, np.zeros((2, 2, 3), dtype=np.uint8))
+ monkeypatch.setattr(cv2, "VideoCapture", Mock(return_value=capture))
+ assert len(cli._load_frames(path, 5)) == 5
+ assert [call.args[1] for call in capture.set.call_args_list] == [0, 74, 149, 224, 299]
+ capture.release.assert_called_once()
+
+
+def test_cli_help_needs_no_model_runtime():
+ import subprocess
+ import sys
+
+ code = '''
+import importlib.abc
+import sys
+class NoModelRuntime(importlib.abc.MetaPathFinder):
+ def find_spec(self, fullname, *args):
+ if fullname.split(".")[0] in {"torch", "torchvision", "cv2", "numpy"}:
+ raise ModuleNotFoundError(fullname, name=fullname)
+sys.meta_path.insert(0, NoModelRuntime())
+from backend.tools.identify_camera import main
+sys.exit(main(["--help"]))
+'''
+ result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True)
+ assert result.returncode == 0, result.stderr
+ assert "--frames" in result.stdout
+
+
+def test_cli_json_is_sorted_and_continues_after_bad_input(monkeypatch, calibrations, fingerprints, capsys):
+ import json
+ from backend.tools import identify_camera as cli
+
+ def load(path, count):
+ if path.name == "bad.mp4":
+ raise ValueError("Unreadable input")
+ return [np.zeros((1080, 1920, 3), dtype=np.uint8)] * count
+
+ monkeypatch.setattr(cli, "_load_frames", load)
+ monkeypatch.setattr(matcher, "load_fingerprints", lambda *args: fingerprints)
+ monkeypatch.setattr(matcher, "thumbnail_signature", lambda frame: fingerprints["uc_davis_court4"]["signature"])
+ monkeypatch.setattr(matcher, "detector_fingerprints", lambda detector, references: calibrations)
+ monkeypatch.setattr(cli, "_create_detector", lambda: FakeDetector([calibrations["uc_davis_court4"]] * 3))
+ assert cli.main(["bad.mp4", "Court2.mp4", "--frames", "3", "--json"]) == 0
+ output = capsys.readouterr()
+ results = json.loads(output.out)
+ assert results[0]["reason"] == "error"
+ assert results[1]["camera_id"] == "uc_davis_court4"
+ assert results[1]["hint"] == "uc_davis_court2"
+ scores = list(results[1]["scores"].values())
+ assert scores == sorted(scores, reverse=True)
+ assert results[1]["best"] == pytest.approx(1.0)
+ assert results[1]["keypoint_camera_id"] == "uc_davis_court4"
+ assert results[1]["keypoint_agrees"] is True
+ distances = list(results[1]["keypoint_scores"].values())
+ assert distances == sorted(distances)
+ assert "Unreadable input" in output.err
+
+
+def test_cli_prints_both_signal_tables(capsys):
+ from backend.tools import identify_camera as cli
+
+ cli._print_table([{"input": "Court2.mp4", "camera_id": "uc_davis_court2", "reason": "matched", "hint": "uc_davis_court2",
+ "scores": {"uc_davis_court2": 0.93}, "keypoint_camera_id": "uc_davis_court2",
+ "keypoint_scores": {"uc_davis_court2": 18.3}, "keypoint_agrees": True}])
+ output = capsys.readouterr().out
+ assert "Background NCC" in output
+ assert "Keypoint cross-check" in output
+ assert "0.93" in output and "18.30" in output
+ assert "matched" in output and "True" in output
diff --git a/backend/tools/identify_camera.py b/backend/tools/identify_camera.py
new file mode 100644
index 00000000..714da87d
--- /dev/null
+++ b/backend/tools/identify_camera.py
@@ -0,0 +1,142 @@
+"""Inspect camera matches locally: python3 -m backend.tools.identify_camera --help."""
+from __future__ import annotations
+
+import argparse
+import contextlib
+import json
+import math
+import sys
+from pathlib import Path
+
+
+BACKEND_DIR = Path(__file__).resolve().parents[1]
+WEIGHTS_PATH = BACKEND_DIR / "weights/keypoints_model.pth"
+VIDEO_SAMPLE_SECONDS = 10
+IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".bmp", ".tif", ".tiff", ".webp"}
+
+
+def _positive_frames(value: str) -> int:
+ frames = int(value)
+ if frames < 1:
+ raise argparse.ArgumentTypeError("--frames must be positive")
+ return frames
+
+
+def _load_frames(path: Path, count: int) -> list:
+ import cv2
+ import numpy as np
+
+ if not path.is_file():
+ raise ValueError(f"Input file does not exist: {path}")
+ if path.suffix.lower() in IMAGE_SUFFIXES:
+ image = cv2.imread(str(path))
+ if image is None:
+ raise ValueError(f"Cannot read image: {path}")
+ return [image]
+ cap = cv2.VideoCapture(str(path))
+ try:
+ if not cap.isOpened():
+ raise ValueError(f"Cannot open video: {path}")
+ fps = cap.get(cv2.CAP_PROP_FPS)
+ if not math.isfinite(fps) or fps <= 0:
+ raise ValueError(f"Cannot determine video frame rate: {path}")
+ limit = max(1, math.ceil(fps * VIDEO_SAMPLE_SECONDS))
+ total = cap.get(cv2.CAP_PROP_FRAME_COUNT)
+ if math.isfinite(total) and total > 0:
+ limit = min(limit, int(total))
+ frames = []
+ for index in np.linspace(0, limit - 1, min(count, limit), dtype=int):
+ cap.set(cv2.CAP_PROP_POS_FRAMES, int(index))
+ ok, frame = cap.read()
+ if ok:
+ frames.append(frame)
+ else:
+ print(f"[Camera] Cannot read frame {index} from {path}", file=sys.stderr)
+ return frames
+ finally:
+ cap.release()
+
+
+def _create_detector():
+ # This import loads torch only when the model is constructed, after argparse.
+ from backend.models.court_line_detector import CourtLineDetector
+
+ return CourtLineDetector(model_path=str(WEIGHTS_PATH), device="cpu")
+
+
+def _inspect_inputs(inputs: list[str], frame_count: int) -> list[dict]:
+ from backend.pipeline.config import PipelineConfig
+ from backend.vision.camera_identify import filename_camera_hint, identify_from_frames, load_fingerprints
+
+ config = PipelineConfig(device="cpu")
+ results = []
+ detector = None
+ for source in inputs:
+ result = {"input": source, "camera_id": None, "reason": "error", "scores": {}, "hint": filename_camera_hint(source),
+ "best": None, "runner_up": None, "keypoint_camera_id": None, "keypoint_scores": {}, "keypoint_agrees": None}
+ try:
+ frames = _load_frames(Path(source), frame_count)
+ fingerprints = load_fingerprints(Path(config.camera_fingerprints_path).parent, config.camera_fingerprints_path)
+ if detector is None:
+ detector = _create_detector()
+ match = identify_from_frames(
+ frames, detector, fingerprints,
+ min_ncc=config.camera_match_min_ncc,
+ min_margin=config.camera_match_min_ncc_margin,
+ min_points=config.camera_match_min_points,
+ )
+ result.update(camera_id=match.camera_id, reason=match.reason, best=match.best, runner_up=match.runner_up,
+ scores=match.scores, keypoint_camera_id=match.keypoint_camera_id, keypoint_agrees=match.keypoint_agrees,
+ keypoint_scores={camera: distance if math.isfinite(distance) else None for camera, distance in match.keypoint_scores.items()})
+ except Exception as exc:
+ result["error"] = str(exc)
+ print(f"[Camera] {source}: {exc}", file=sys.stderr)
+ results.append(result)
+ return results
+
+
+def _print_table(results: list[dict]) -> None:
+ print("Background NCC (higher is better)")
+ print("input\tidentified camera\treason\tper-camera median NCC\tfilename hint")
+ for result in results:
+ scores = ", ".join(f"{camera}={distance:.2f}" if distance is not None else f"{camera}=n/a"
+ for camera, distance in result["scores"].items())
+ print(f"{result['input']}\t{result['camera_id'] or 'none'}\t{result['reason']}\t{scores or '-'}\t{result['hint'] or 'none'}")
+ print("\nKeypoint cross-check (lower is better)")
+ print("input\tkeypoint camera\tagrees\tper-camera median distance (px)")
+ for result in results:
+ scores = ", ".join(f"{camera}={distance:.2f}" if distance is not None else f"{camera}=n/a"
+ for camera, distance in result["keypoint_scores"].items())
+ print(f"{result['input']}\t{result['keypoint_camera_id'] or 'none'}\t{result['keypoint_agrees']}\t{scores or '-'}")
+
+
+def main(argv: list[str] | None = None) -> int:
+ from backend.pipeline.config import PipelineConfig
+
+ parser = argparse.ArgumentParser(description="Identify cameras by background NCC with a CPU keypoint cross-check.")
+ parser.add_argument("inputs", nargs="+", metavar="video-or-image")
+ parser.add_argument("--frames", type=_positive_frames, default=PipelineConfig.camera_identify_frames,
+ help="frames sampled evenly within the first 10 seconds (default: %(default)s)")
+ parser.add_argument("--json", action="store_true", help="print results as JSON instead of a table")
+ try:
+ args = parser.parse_args(argv)
+ except SystemExit:
+ return 0 # Diagnostic-only CLI, including help and argument errors.
+ try:
+ with contextlib.redirect_stdout(sys.stderr):
+ results = _inspect_inputs(args.inputs, args.frames)
+ except Exception as exc:
+ # Missing optional runtime dependencies must still produce a diagnostic.
+ results = [{"input": source, "camera_id": None, "reason": "error", "scores": {}, "hint": None, "error": str(exc),
+ "best": None, "runner_up": None, "keypoint_camera_id": None, "keypoint_scores": {}, "keypoint_agrees": None}
+ for source in args.inputs]
+ print(f"[Camera] {exc}", file=sys.stderr)
+ if args.json:
+ print(json.dumps(results, indent=2, allow_nan=False))
+ else:
+ _print_table(results)
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/backend/vision/__init__.py b/backend/vision/__init__.py
index e268031d..5861b4e2 100644
--- a/backend/vision/__init__.py
+++ b/backend/vision/__init__.py
@@ -1,6 +1,23 @@
-from backend.vision.homography import HomographyEstimator
-from backend.vision.court_reference import CourtReference
-from backend.vision.drawing import draw_ball_trace, draw_court_keypoints_and_lines, draw_minimap_ball_and_bounces, draw_minimap_players, draw_player_bboxes, draw_stroke_labels
-from backend.vision.postprocess import detect_shot_frames
-from backend.vision.swing_detector import SwingDetector, extract_pose_sequence
-from backend.vision.calibration import load_calibration, save_calibration, visualize_keypoints
+"""Load vision exports on demand so keypoint matching only requires numpy."""
+from importlib import import_module
+
+
+_EXPORTS = {
+ "homography": ("HomographyEstimator",),
+ "court_reference": ("CourtReference",),
+ "drawing": ("draw_ball_trace", "draw_court_keypoints_and_lines", "draw_minimap_ball_and_bounces",
+ "draw_minimap_players", "draw_player_bboxes", "draw_stroke_labels"),
+ "postprocess": ("detect_shot_frames",),
+ "swing_detector": ("SwingDetector", "extract_pose_sequence"),
+ "calibration": ("load_calibration", "save_calibration", "visualize_keypoints"),
+}
+__all__ = [name for names in _EXPORTS.values() for name in names]
+
+
+def __getattr__(name):
+ for module, names in _EXPORTS.items():
+ if name in names:
+ value = getattr(import_module(f"{__name__}.{module}"), name)
+ globals()[name] = value
+ return value
+ raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
diff --git a/backend/vision/camera_identify.py b/backend/vision/camera_identify.py
new file mode 100644
index 00000000..33b7a505
--- /dev/null
+++ b/backend/vision/camera_identify.py
@@ -0,0 +1,248 @@
+"""Identify cameras by background correlation, with detector keypoints as a cross-check."""
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import dataclass, field, replace
+from pathlib import Path
+
+import numpy as np
+
+
+CALIBRATION_FRAME_SIZE = (1920, 1080)
+THUMBNAIL_SIZE = (96, 54)
+SIGNATURE_EPSILON = 1e-6
+_OPPONENT_COURT_MAP = {"2": "uc_davis_court2", "4": "uc_davis_court4", "6": "uc_davis_court6"}
+
+
+def filename_camera_hint(video_path: str | Path) -> str | None:
+ """Share the advisory filename convention between the pipeline and CLI."""
+ match = re.search(r'[Cc]ourt(\d+)', Path(video_path).stem)
+ return _OPPONENT_COURT_MAP.get(match.group(1)) if match else None
+
+
+@dataclass(frozen=True)
+class CameraMatch:
+ camera_id: str | None
+ scores: dict[str, float] # Median background NCC, descending (higher is better).
+ best: float | None
+ runner_up: float | None
+ valid_points: int
+ reason: str
+ keypoint_camera_id: str | None = None
+ keypoint_scores: dict[str, float] = field(default_factory=dict)
+ keypoint_agrees: bool | None = None
+
+
+@dataclass(frozen=True)
+class _KeypointMatch:
+ """Keep diagnostic pixel distances separate from CameraMatch's NCC scores."""
+ camera_id: str | None
+ scores: dict[str, float]
+ best: float | None
+ runner_up: float | None
+ valid_points: int
+ reason: str
+
+
+def _reference_frame(frame_path: str | Path) -> np.ndarray:
+ import cv2
+
+ frame = cv2.imread(str(frame_path))
+ if frame is None:
+ raise ValueError(f"Cannot read camera reference frame: {frame_path}")
+ if (frame.shape[1], frame.shape[0]) != CALIBRATION_FRAME_SIZE:
+ frame = cv2.resize(frame, CALIBRATION_FRAME_SIZE)
+ return frame
+
+
+def thumbnail_signature(frame_bgr, size=THUMBNAIL_SIZE) -> np.ndarray:
+ """Normalize a grayscale background thumbnail; import OpenCV only when needed."""
+ import cv2
+
+ if frame_bgr.ndim != 3 or frame_bgr.shape[2] != 3 or not frame_bgr.size:
+ raise ValueError("Camera signature requires a non-empty BGR frame")
+ if len(size) != 2 or any(not isinstance(value, int) or value < 1 for value in size):
+ raise ValueError("Thumbnail size must be a positive (width, height) pair")
+ frame = frame_bgr
+ if (frame.shape[1], frame.shape[0]) != CALIBRATION_FRAME_SIZE:
+ frame = cv2.resize(frame, CALIBRATION_FRAME_SIZE)
+ gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+ thumbnail = cv2.resize(gray, size, interpolation=cv2.INTER_AREA).astype(float)
+ return (thumbnail - thumbnail.mean()) / (thumbnail.std() + SIGNATURE_EPSILON)
+
+
+def ncc(a, b) -> float:
+ """Normalized correlation of two already standardized, equally sized signatures."""
+ a, b = np.asarray(a, dtype=float), np.asarray(b, dtype=float)
+ if a.shape != b.shape or not a.size or not np.isfinite(a).all() or not np.isfinite(b).all():
+ raise ValueError("Signatures must have the same non-empty shape and finite values")
+ return float(np.mean(a * b))
+
+
+def load_fingerprints(calibration_dir, fingerprints_path) -> dict:
+ """Load reference thumbnails, logging and skipping missing/unreadable frames."""
+ with Path(fingerprints_path).open() as source:
+ entries = json.load(source)
+ if not isinstance(entries, dict):
+ raise ValueError("Camera fingerprints must be an object keyed by camera ID")
+ result = {}
+ for camera, entry in entries.items():
+ if not isinstance(entry, dict) or not isinstance(entry.get("frame"), str) or not entry["frame"]:
+ raise ValueError(f"Fingerprint {camera!r} must name a reference frame")
+ frame_path = (Path(calibration_dir) / entry["frame"]).resolve()
+ if not frame_path.is_file():
+ print(f"[Camera] Skipping missing reference frame: {frame_path}", flush=True)
+ continue
+ try:
+ frame = _reference_frame(frame_path)
+ except ValueError as exc:
+ print(f"[Camera] Skipping reference frame: {exc}", flush=True)
+ continue
+ result[camera] = {"signature": thumbnail_signature(frame), "frame_path": str(frame_path)}
+ return result
+
+
+def identify_by_background(frames, fingerprints, *, min_ncc, min_margin) -> CameraMatch:
+ """Choose the highest median NCC when its absolute score and lead pass the gates."""
+ if not np.isfinite(min_ncc) or not -1 <= min_ncc <= 1:
+ raise ValueError("min_ncc must be between -1 and 1")
+ if not np.isfinite(min_margin) or not 0 <= min_margin <= 2:
+ raise ValueError("min_margin must be between 0 and 2")
+ if not frames:
+ return CameraMatch(None, {}, None, None, 0, "no_detection")
+ if not fingerprints:
+ return CameraMatch(None, {}, None, None, 0, "no_fingerprints")
+ signatures = [thumbnail_signature(frame) for frame in frames]
+ scores = {camera: float(np.median([ncc(signature, entry["signature"]) for signature in signatures]))
+ for camera, entry in fingerprints.items()}
+ ordered = dict(sorted(scores.items(), key=lambda item: item[1], reverse=True))
+ candidates = list(ordered)
+ best = ordered[candidates[0]]
+ runner_up = ordered[candidates[1]] if len(candidates) > 1 else None
+ camera_id = None
+ if best < min_ncc:
+ reason = "too_low"
+ elif runner_up is not None and (best == runner_up or best - runner_up < min_margin):
+ reason = "ambiguous"
+ else:
+ camera_id, reason = candidates[0], "matched"
+ return CameraMatch(camera_id, ordered, best, runner_up, 0, reason)
+
+
+def detector_fingerprints(court_detector, fingerprints) -> dict:
+ """Run the same detector once per reference frame, always at 1920x1080."""
+ return {camera: court_detector.infer_single(_reference_frame(entry["frame_path"]))
+ for camera, entry in fingerprints.items()}
+
+
+def _points_array(points) -> np.ndarray:
+ """Keep missing/non-finite points unusable without changing their indices."""
+ if points is None or len(points) == 0:
+ return np.empty((0, 2), dtype=float)
+ array = np.asarray([(np.nan, np.nan) if p is None else p for p in points], dtype=float)
+ if array.ndim != 2 or array.shape[1] != 2:
+ raise ValueError("Court keypoints must be pairs of coordinates or None")
+ return array
+
+
+def score_against(detected: list[tuple[float, float] | None], stored: list[tuple[float, float] | None],
+ frame_size: tuple[int, int]) -> tuple[float, int]:
+ """Mean pixel distance at shared valid indices, scaling stored points first.
+
+ Return (inf, 0) for no overlap. The caller enforces its minimum point count.
+ """
+ size = np.asarray(frame_size, dtype=float)
+ if size.shape != (2,) or not np.isfinite(size).all() or (size <= 0).any():
+ raise ValueError("Frame size must be a positive (width, height) pair")
+ predicted, reference = _points_array(detected), _points_array(stored)
+ overlap = min(len(predicted), len(reference))
+ predicted = predicted[:overlap]
+ reference = reference[:overlap] * (size / CALIBRATION_FRAME_SIZE)
+ valid = np.isfinite(predicted).all(axis=1) & np.isfinite(reference).all(axis=1)
+ count = int(valid.sum())
+ distance = float(np.linalg.norm(predicted[valid] - reference[valid], axis=1).mean()) if count else float("inf")
+ return distance, count
+
+
+def _pick_keypoint_match(scores, valid_points, *, min_points, max_distance=None, min_margin_ratio=None) -> _KeypointMatch:
+ if max_distance is not None and (not np.isfinite(max_distance) or max_distance < 0):
+ raise ValueError("max_distance must be finite and non-negative")
+ if min_margin_ratio is not None and (not np.isfinite(min_margin_ratio) or not 0 < min_margin_ratio <= 1):
+ raise ValueError("min_margin_ratio must be between zero (exclusive) and one")
+ if not isinstance(min_points, int) or min_points < 1:
+ raise ValueError("min_points must be a positive integer")
+ ordered = dict(sorted(scores.items(), key=lambda item: item[1]))
+ candidates = [(camera, distance) for camera, distance in ordered.items() if np.isfinite(distance)]
+ best = candidates[0][1] if candidates else None
+ runner_up = candidates[1][1] if len(candidates) > 1 else None
+ camera_id = None
+ if not scores:
+ reason = "no_calibrations"
+ elif valid_points < min_points or best is None:
+ reason = "no_detection"
+ elif max_distance is not None and best > max_distance:
+ reason = "too_far"
+ elif runner_up is not None and (best == runner_up or (min_margin_ratio is not None and best > min_margin_ratio * runner_up)):
+ reason = "ambiguous"
+ else:
+ camera_id, reason = candidates[0][0], "matched"
+ return _KeypointMatch(camera_id, ordered, best, runner_up, valid_points, reason)
+
+
+def identify_by_keypoints(detected: list[tuple[float, float] | None] | None, calibrations: dict,
+ frame_size: tuple[int, int], *, min_points: int,
+ max_distance=None, min_margin_ratio=None) -> _KeypointMatch:
+ """Rank detector-vs-detector distances; optional distance gates are diagnostic only."""
+ valid_points = int(np.isfinite(_points_array(detected)).all(axis=1).sum())
+ scores = {}
+ for camera_id, stored in calibrations.items():
+ distance, count = score_against(detected, stored, frame_size)
+ scores[camera_id] = distance if count >= min_points else float("inf")
+ match = _pick_keypoint_match(scores, valid_points, max_distance=max_distance,
+ min_margin_ratio=min_margin_ratio, min_points=min_points)
+ if not valid_points:
+ return _KeypointMatch(None, match.scores, None, None, 0, "no_detection")
+ return match
+
+
+def _keypoint_cross_check(frames, court_detector, fingerprints, *, frame_size=None, min_points) -> _KeypointMatch:
+ """Median detector-vs-detector distances across usable query frames.
+
+ valid_points is the median detected point count across usable frames.
+ """
+ calibrations = detector_fingerprints(court_detector, fingerprints)
+ distances = {camera: [] for camera in calibrations}
+ counts = []
+ for frame in frames:
+ detected = court_detector.infer_single(frame)
+ size = frame_size if frame_size is not None else (frame.shape[1], frame.shape[0])
+ match = identify_by_keypoints(detected, calibrations, size, min_points=min_points)
+ if match.valid_points < min_points:
+ continue
+ counts.append(match.valid_points)
+ for camera, distance in match.scores.items():
+ if np.isfinite(distance):
+ distances[camera].append(distance)
+ scores = {camera: float(np.median(values)) if values else float("inf") for camera, values in distances.items()}
+ match = _pick_keypoint_match(scores, int(np.median(counts)) if counts else 0, min_points=min_points)
+ if not counts:
+ return _KeypointMatch(None, match.scores, None, None, 0, "no_detection")
+ return match
+
+
+def identify_from_frames(frames: list[np.ndarray], court_detector, fingerprints: dict, *,
+ min_ncc: float, min_margin: float, min_points: int, frame_size=None) -> CameraMatch:
+ """Background NCC decides; keypoint agreement is recorded without changing it."""
+ match = identify_by_background(frames, fingerprints, min_ncc=min_ncc, min_margin=min_margin)
+ if match.camera_id is None or court_detector is None:
+ return match
+ try:
+ keypoints = _keypoint_cross_check(frames, court_detector, fingerprints,
+ frame_size=frame_size, min_points=min_points)
+ except Exception as exc:
+ print(f"[Camera] Keypoint cross-check failed; keeping background match: {exc}", flush=True)
+ return match
+ return replace(match, valid_points=keypoints.valid_points, keypoint_camera_id=keypoints.camera_id,
+ keypoint_scores=keypoints.scores,
+ keypoint_agrees=keypoints.camera_id == match.camera_id if keypoints.camera_id else None)
diff --git a/frontend/web/app/api/recordings/[id]/route.ts b/frontend/web/app/api/recordings/[id]/route.ts
index 1250d3d1..8b768977 100644
--- a/frontend/web/app/api/recordings/[id]/route.ts
+++ b/frontend/web/app/api/recordings/[id]/route.ts
@@ -51,7 +51,7 @@ export async function GET(
const { id } = await params;
- const FULL_COLS = "id, name, status, progress, processing_stage, error, results_path, input_path, created_at, fps, num_frames, bounce_heatmap_path, player_heatmap_path, player_shot_map_path, bounce_count, shot_count, rally_count, forehand_count, backhand_count, serve_count, in_bounds_bounces, out_bounds_bounces, scouting_report, player_id, favorited, keypoints, notes, shots, coverage_grid, position_summary, net_approach_summary, error_summary, rallies, rally_summary";
+ const FULL_COLS = "id, name, status, progress, processing_stage, error, results_path, input_path, created_at, fps, num_frames, bounce_heatmap_path, player_heatmap_path, player_shot_map_path, bounce_count, shot_count, rally_count, forehand_count, backhand_count, serve_count, in_bounds_bounces, out_bounds_bounces, scouting_report, player_id, favorited, keypoints, notes, shots, coverage_grid, position_summary, net_approach_summary, error_summary, rallies, rally_summary, camera_id, camera_match";
// Pre-migration safety: if a column the API references hasn't been added
// to `matches` yet, Postgres returns "column matches. does not
// exist". Retry up to N times, each time stripping ONLY the column named
@@ -69,6 +69,8 @@ export async function GET(
"error_summary",
"rallies",
"rally_summary",
+ "camera_id",
+ "camera_match",
]);
let { data, error } = await supabaseAdmin
.from("matches")
@@ -163,6 +165,8 @@ export async function GET(
error: data.error || null,
videoUrl,
inputPath: data.input_path ?? null,
+ cameraId: data.camera_id ?? null,
+ cameraMatch: data.camera_match ?? null,
bounceHeatmapUrl,
playerHeatmapUrl,
playerShotMapUrl,
diff --git a/supabase/migrations/20260914_add_camera_id.sql b/supabase/migrations/20260914_add_camera_id.sql
new file mode 100644
index 00000000..939d2149
--- /dev/null
+++ b/supabase/migrations/20260914_add_camera_id.sql
@@ -0,0 +1,3 @@
+-- Brian applies this migration in the Supabase SQL editor.
+ALTER TABLE public.matches ADD COLUMN IF NOT EXISTS camera_id text;
+ALTER TABLE public.matches ADD COLUMN IF NOT EXISTS camera_match jsonb;