feat: identify the court camera from the video instead of the filename

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 <frame> --camera-id <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 <video-or-image> [more...] --frames 5 --json` (CPU; requires model weights).

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" }
+}

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

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():

backend/pipeline/config.py

@@ -1,22 +1,35 @@
 # 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
 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.

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,32 +1744,78 @@ 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):
     """
     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))