scripts/test_fast_full_video.py

#!/usr/bin/env python3
"""
FAST full video evaluation using template-based play clock reading.

This script uses PlayDetector with pre-configured fixed coordinates, which:
1. Uses the known scorebug region (skips region discovery/searching)
2. Uses template matching for play clock reading
3. Includes all detection logic (state machine + clock reset detection)

Template capture modes:
- DYNAMIC (default): Captures templates dynamically using OCR on first ~400 frames
- STATIC: Uses pre-existing templates from disk (for faster startup)

Usage:
    cd /Users/andytaylor/Documents/Personal/cfb40
    source .venv/bin/activate

    # Run with DYNAMIC template capture (default - builds templates from video):
    python tests/test_digit_templates/test_fast_full_video.py

    # Run with STATIC templates (uses pre-existing templates from disk):
    python tests/test_digit_templates/test_fast_full_video.py --use-static-templates
"""

import argparse
import json
import logging
import sys
import time
from pathlib import Path
from typing import List, Dict, Any, Optional

from pipeline.play_detector import PlayDetector, DetectionConfig

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# Configuration - all paths are known from previous sessions
VIDEO_PATH = "full_videos/OSU vs Tenn 12.21.24.mkv"
DIGIT_TEMPLATE_PATH = "output/debug/digit_templates"
V3_BASELINE_PATH = "output/benchmarks/v3_special_plays_baseline.json"
PLAYCLOCK_CONFIG_PATH = "output/OSU_vs_Tenn_12_21_24_playclock_config.json"

# Known fixed regions from previous calibration
SCOREBUG_REGION = (128, 975, 1669, 46)  # (x, y, w, h)

# Frame sampling
FRAME_INTERVAL = 0.5  # seconds between samples

# Template collection settings
TEMPLATE_COLLECTION_FRAMES = 400  # Number of frames to use for building templates via OCR


def load_playclock_config():
    """Load play clock region offset from config."""
    with open(PLAYCLOCK_CONFIG_PATH, "r", encoding="utf-8") as f:
        config = json.load(f)
    return (config["x_offset"], config["y_offset"], config["width"], config["height"])


def get_absolute_playclock_coords():
    """Calculate absolute play clock coordinates from scorebug + offset."""
    sb_x, sb_y, _, _ = SCOREBUG_REGION
    pc_x_off, pc_y_off, pc_w, pc_h = load_playclock_config()

    # Absolute coordinates
    abs_x = sb_x + pc_x_off
    abs_y = sb_y + pc_y_off

    return (abs_x, abs_y, pc_w, pc_h)


def load_v3_baseline() -> List[Dict]:
    """Load v3 baseline plays for comparison."""
    with open(V3_BASELINE_PATH, "r", encoding="utf-8") as f:
        data = json.load(f)
    return data.get("plays", [])


def find_matching_play(detected_play: dict, baseline_plays: list, tolerance: float = 5.0) -> Optional[dict]:
    """Find a matching play in the baseline within tolerance."""
    detected_start = detected_play.get("start_time", 0)
    for baseline in baseline_plays:
        baseline_start = baseline.get("start_time", 0)
        if abs(detected_start - baseline_start) <= tolerance:
            return baseline
    return None


def compare_results(detected_plays: list, baseline_plays: list) -> Dict[str, Any]:
    """Compare detected plays against baseline."""
    matched_baseline = set()
    true_positives = []
    false_positives = []

    for detected in detected_plays:
        match = find_matching_play(detected, baseline_plays)
        if match:
            baseline_idx = baseline_plays.index(match)
            if baseline_idx not in matched_baseline:
                matched_baseline.add(baseline_idx)
                true_positives.append({"detected": detected, "baseline": match})
            else:
                false_positives.append(detected)
        else:
            false_positives.append(detected)

    false_negatives = [bp for i, bp in enumerate(baseline_plays) if i not in matched_baseline]

    tp = len(true_positives)
    fp = len(false_positives)
    fn = len(false_negatives)

    precision = tp / (tp + fp) if (tp + fp) > 0 else 0
    recall = tp / (tp + fn) if (tp + fn) > 0 else 0
    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0

    return {
        "true_positives": true_positives,
        "false_positives": false_positives,
        "false_negatives": false_negatives,
        "counts": {"tp": tp, "fp": fp, "fn": fn},
        "metrics": {"precision": precision, "recall": recall, "f1": f1},
    }


def run_fast_evaluation(use_static_templates: bool = False):
    """
    Run FAST full video evaluation using PlayDetector with fixed coordinates.

    Args:
        use_static_templates: If True, use pre-existing templates from disk.
                              If False (default), capture templates dynamically via OCR.
    """
    mode = "STATIC TEMPLATES" if use_static_templates else "DYNAMIC TEMPLATE CAPTURE"
    logger.info("=" * 70)
    logger.info("FAST FULL VIDEO EVALUATION")
    logger.info("Mode: %s", mode)
    logger.info("Using PlayDetector with FIXED COORDINATES mode")
    if use_static_templates:
        logger.info("Loading pre-existing digit templates from: %s", DIGIT_TEMPLATE_PATH)
    else:
        logger.info("Will capture templates dynamically using OCR on first %d frames", TEMPLATE_COLLECTION_FRAMES)
    logger.info("=" * 70)

    # Check required files exist
    required_files = [
        (VIDEO_PATH, "Video"),
        (V3_BASELINE_PATH, "V3 baseline"),
        (PLAYCLOCK_CONFIG_PATH, "Play clock config"),
    ]

    # Only require digit templates directory if using static templates
    if use_static_templates:
        required_files.append((DIGIT_TEMPLATE_PATH, "Digit templates"))

    for path, name in required_files:
        if not Path(path).exists():
            logger.error("%s not found: %s", name, path)
            return False

    # Load baseline
    logger.info("\n[Step 1] Loading v3 baseline...")
    baseline_plays = load_v3_baseline()
    logger.info("V3 baseline plays: %d", len(baseline_plays))

    # Get fixed coordinates
    logger.info("\n[Step 2] Loading fixed coordinates...")
    playclock_coords = get_absolute_playclock_coords()
    logger.info("Play clock absolute coords: %s", playclock_coords)
    logger.info("Scorebug region: %s", SCOREBUG_REGION)

    # Create PlayDetector config with fixed coordinates
    logger.info("\n[Step 3] Creating PlayDetector with fixed coordinates mode...")

    # Build config - key difference is whether we provide digit_template_path
    config = DetectionConfig(
        video_path=VIDEO_PATH,
        template_path="",  # Not needed in fixed coords mode
        clock_region_config_path=PLAYCLOCK_CONFIG_PATH,  # Used for region dimensions
        start_time=0.0,
        end_time=None,  # Full video
        frame_interval=FRAME_INTERVAL,
        # STATIC mode: provide path to load pre-existing templates
        # DYNAMIC mode: None means templates will be built during detection
        digit_template_path=DIGIT_TEMPLATE_PATH if use_static_templates else None,
        template_collection_frames=TEMPLATE_COLLECTION_FRAMES,
        # FIXED COORDINATES MODE - skips scorebug detection
        fixed_playclock_coords=playclock_coords,
        fixed_scorebug_coords=SCOREBUG_REGION,
    )

    # Create detector and run
    logger.info("\n[Step 4] Running detection (%s mode)...", mode)
    start_time = time.time()

    detector = PlayDetector(config)
    result = detector.detect()

    elapsed = time.time() - start_time

    # Extract plays from result and apply minimum duration filter
    # Using 1.0s instead of 3.0s because special plays (XP/FG completions) have short durations
    MIN_PLAY_DURATION = 1.0  # seconds - filters only weird clock noise
    all_plays = result.plays
    detected_plays = [p for p in all_plays if p.get("duration", 0) >= MIN_PLAY_DURATION]
    filtered_count = len(all_plays) - len(detected_plays)
    if filtered_count > 0:
        logger.info("Filtered out %d plays with duration < %.1fs", filtered_count, MIN_PLAY_DURATION)
    logger.info("\nDetected plays: %d", len(detected_plays))

    # Compare against baseline
    logger.info("\n[Step 5] Comparing against v3 baseline...")
    comparison = compare_results(detected_plays, baseline_plays)

    counts = comparison["counts"]
    metrics = comparison["metrics"]

    logger.info("\n" + "=" * 70)
    logger.info("EVALUATION RESULTS")
    logger.info("=" * 70)
    logger.info("Mode: %s", mode)
    logger.info("V3 Baseline plays: %d", len(baseline_plays))
    logger.info("Detected plays: %d", len(detected_plays))
    logger.info("")
    logger.info("True Positives: %d", counts["tp"])
    logger.info("False Positives: %d", counts["fp"])
    logger.info("False Negatives: %d", counts["fn"])
    logger.info("")
    logger.info("Precision: %.1f%%", metrics["precision"] * 100)
    logger.info("Recall: %.1f%%", metrics["recall"] * 100)
    logger.info("F1 Score: %.1f%%", metrics["f1"] * 100)
    logger.info("")
    logger.info("Total time: %.1f seconds (%.1f minutes)", elapsed, elapsed / 60)
    logger.info("Frames processed: %d", result.total_frames_processed)
    if result.total_frames_processed > 0:
        logger.info("Speed: %.1f frames/sec, %.2f ms/frame", result.total_frames_processed / elapsed, 1000 * elapsed / result.total_frames_processed)

    # Log timing breakdown
    logger.info("\n--- Timing Breakdown ---")
    for key, val in result.timing.items():
        logger.info("  %s: %.2fs", key, val)

    # Check for the specific play at 1:52:06 (6726s)
    target_time = 6726.0
    found_target = False
    for play in detected_plays:
        if abs(play.get("start_time", 0) - target_time) <= 10:
            found_target = True
            logger.info("\n*** MILESTONE: Play near 1:52:06 (6726s) WAS DETECTED! ***")
            logger.info("  Start time: %.1fs", play.get("start_time", 0))
            break

    if not found_target:
        logger.info("\n*** WARNING: Play near 1:52:06 (6726s) was NOT detected ***")

    # Show missed plays
    if comparison["false_negatives"]:
        logger.info("\n--- MISSED PLAYS (first 10) ---")
        for i, play in enumerate(comparison["false_negatives"][:10]):
            start = play.get("start_time", 0)
            play_type = play.get("play_type", "unknown")
            logger.info("  %d. t=%d:%05.2f (%.1fs) [%s]", i + 1, int(start // 60), start % 60, start, play_type)

    # Show false positives
    if comparison["false_positives"]:
        logger.info("\n--- FALSE POSITIVES (first 10) ---")
        for i, play in enumerate(comparison["false_positives"][:10]):
            start = play.get("start_time", 0)
            duration = play.get("duration", 0)
            play_type = play.get("play_type", "unknown")
            logger.info("  %d. t=%d:%05.2f (%.1fs) duration=%.1fs [%s]", i + 1, int(start // 60), start % 60, start, duration, play_type)

    # Save results
    output_filename = "fast_template_evaluation_dynamic.json" if not use_static_templates else "fast_template_evaluation_static.json"
    output_path = Path("output/benchmarks") / output_filename
    output_path.parent.mkdir(parents=True, exist_ok=True)

    results_data = {
        "video": VIDEO_PATH,
        "method": "dynamic_template_capture" if not use_static_templates else "static_templates",
        "template_mode": "dynamic" if not use_static_templates else "static",
        "baseline_plays": len(baseline_plays),
        "detected_plays": len(detected_plays),
        "counts": counts,
        "metrics": metrics,
        "elapsed_seconds": elapsed,
        "frames_processed": result.total_frames_processed,
        "frames_per_second": result.total_frames_processed / elapsed if elapsed > 0 else 0,
        "ms_per_frame": 1000 * elapsed / result.total_frames_processed if result.total_frames_processed > 0 else 0,
        "timing": result.timing,
        "plays": detected_plays,
    }

    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(results_data, f, indent=2)
    logger.info("\nResults saved to: %s", output_path)

    # Print comparison summary for easy reference
    logger.info("\n" + "=" * 70)
    logger.info("COMPARISON SUMMARY")
    logger.info("=" * 70)
    logger.info("| Metric         | v3 Baseline | Current Run |")
    logger.info("|----------------|-------------|-------------|")
    logger.info("| Plays detected | %11d | %11d |", len(baseline_plays), len(detected_plays))
    logger.info("| Recall         |        100%% | %10.1f%% |", metrics["recall"] * 100)
    logger.info("| Precision      |        100%% | %10.1f%% |", metrics["precision"] * 100)
    logger.info("| F1 Score       |           - | %10.1f%% |", metrics["f1"] * 100)
    logger.info("| Processing     |    9.6 min  | %7.1f min |", elapsed / 60)
    logger.info("=" * 70)

    return metrics["recall"] >= 0.90


def main():
    """Main entry point with argument parsing."""
    parser = argparse.ArgumentParser(
        description="Fast full video evaluation using template-based play clock reading.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
    # Run with DYNAMIC template capture (default):
    python test_fast_full_video.py

    # Run with STATIC templates:
    python test_fast_full_video.py --use-static-templates
        """,
    )
    parser.add_argument(
        "--use-static-templates",
        action="store_true",
        help="Use pre-existing templates from disk instead of capturing dynamically (default: False)",
    )
    args = parser.parse_args()

    success = run_fast_evaluation(use_static_templates=args.use_static_templates)
    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()