move to basemodel

src/detection/models.py

@@ -1,32 +1,31 @@
 """
-Dataclass models for per-frame detection.
+Pydantic models for per-frame detection.
 
 These models represent the results of detecting various elements in video frames:
 scorebug presence and timeout indicators.
 """
 
-from dataclasses import dataclass
 from typing import Optional, Tuple, List
 
+from pydantic import BaseModel, Field
 
-@dataclass
-class ScorebugDetection:
+
+class ScorebugDetection(BaseModel):
     """Results from scorebug detection."""
 
-    detected: bool  # Whether scorebug was detected
-    confidence: float  # Confidence score (0.0 to 1.0)
-    bbox: Optional[Tuple[int, int, int, int]] = None  # Bounding box (x, y, width, height)
-    method: str = "unknown"  # Detection method used
-    left_confidence: Optional[float] = None  # Left half confidence (when split detection enabled)
-    right_confidence: Optional[float] = None  # Right half confidence (when split detection enabled)
+    detected: bool = Field(..., description="Whether scorebug was detected")
+    confidence: float = Field(..., description="Confidence score (0.0 to 1.0)")
+    bbox: Optional[Tuple[int, int, int, int]] = Field(None, description="Bounding box (x, y, width, height)")
+    method: str = Field("unknown", description="Detection method used")
+    left_confidence: Optional[float] = Field(None, description="Left half confidence (when split detection enabled)")
+    right_confidence: Optional[float] = Field(None, description="Right half confidence (when split detection enabled)")
 
 
-@dataclass
-class TimeoutRegionConfig:
+class TimeoutRegionConfig(BaseModel):
     """Configuration for a team's timeout indicator region."""
 
-    team_name: str  # "home" or "away"
-    bbox: Tuple[int, int, int, int]  # x, y, width, height for the 3-oval group
+    team_name: str = Field(..., description="'home' or 'away'")
+    bbox: Tuple[int, int, int, int] = Field(..., description="x, y, width, height for the 3-oval group")
 
     def to_dict(self) -> dict:
         """Convert to dictionary for JSON serialization."""
@@ -42,12 +41,11 @@ class TimeoutRegionConfig:
         return cls(team_name=data["team_name"], bbox=bbox)
 
 
-@dataclass
-class TimeoutReading:
+class TimeoutReading(BaseModel):
     """Results from timeout indicator reading."""
 
-    home_timeouts: int  # 0-3 timeouts remaining
-    away_timeouts: int  # 0-3 timeouts remaining
-    confidence: float  # Overall confidence in reading
-    home_oval_states: Optional[List[bool]] = None  # True = white (available), False = dark (used)
-    away_oval_states: Optional[List[bool]] = None  # True = white (available), False = dark (used)
+    home_timeouts: int = Field(..., description="0-3 timeouts remaining")
+    away_timeouts: int = Field(..., description="0-3 timeouts remaining")
+    confidence: float = Field(..., description="Overall confidence in reading")
+    home_oval_states: Optional[List[bool]] = Field(None, description="True = white (available), False = dark (used)")
+    away_oval_states: Optional[List[bool]] = Field(None, description="True = white (available), False = dark (used)")

src/pipeline/models.py

@@ -1,5 +1,5 @@
 """
-Dataclass models for the pipeline module.
+Pydantic models for the pipeline module.
 
 This module contains all the data structures used by the pipeline components
 for configuration, intermediate results, and final output.
@@ -9,10 +9,9 @@ Streaming processing is used for optimal performance (read frame -> process imme
 See docs/ocr_to_template_migration.md for details.
 """
 
-from dataclasses import dataclass, field
 from typing import Optional, List, Dict, Any, Tuple
 
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict, Field
 
 
 # =============================================================================
@@ -20,8 +19,7 @@ from pydantic import BaseModel
 # =============================================================================
 
 
-@dataclass
-class DetectionConfig:
+class DetectionConfig(BaseModel):
     """Configuration for play detection pipeline.
 
     Uses template matching for play clock reading (~34x faster than OCR).
@@ -29,19 +27,19 @@ class DetectionConfig:
     then streaming detection processes each frame immediately via template matching.
     """
 
-    video_path: str  # Path to video file
-    template_path: str  # Path to scorebug template
-    clock_region_config_path: str  # Path to play clock region config
-    start_time: float = 0.0  # Start time in seconds
-    end_time: Optional[float] = None  # End time in seconds (None = full video)
-    frame_interval: float = 0.5  # Interval between frame samples (seconds)
-    use_split_detection: bool = True  # Enable split-half scorebug detection for robustness to partial overlays
+    video_path: str = Field(..., description="Path to video file")
+    template_path: str = Field(..., description="Path to scorebug template")
+    clock_region_config_path: str = Field(..., description="Path to play clock region config")
+    start_time: float = Field(0.0, description="Start time in seconds")
+    end_time: Optional[float] = Field(None, description="End time in seconds (None = full video)")
+    frame_interval: float = Field(0.5, description="Interval between frame samples (seconds)")
+    use_split_detection: bool = Field(True, description="Enable split-half scorebug detection for robustness to partial overlays")
     # Template matching configuration
-    template_collection_frames: int = 400  # Number of frames to use for building digit templates via OCR
-    digit_template_path: Optional[str] = None  # Path to pre-built digit templates (skip collection phase if provided)
+    template_collection_frames: int = Field(400, description="Number of frames to use for building digit templates via OCR")
+    digit_template_path: Optional[str] = Field(None, description="Path to pre-built digit templates (skip collection phase if provided)")
     # Fixed coordinates mode - skip scorebug detection entirely for maximum speed
-    fixed_playclock_coords: Optional[Tuple[int, int, int, int]] = None  # (x, y, w, h) absolute play clock coords
-    fixed_scorebug_coords: Optional[Tuple[int, int, int, int]] = None  # (x, y, w, h) scorebug region (for metadata)
+    fixed_playclock_coords: Optional[Tuple[int, int, int, int]] = Field(None, description="(x, y, w, h) absolute play clock coords")
+    fixed_scorebug_coords: Optional[Tuple[int, int, int, int]] = Field(None, description="(x, y, w, h) scorebug region (for metadata)")
 
 
 # =============================================================================
@@ -49,19 +47,20 @@ class DetectionConfig:
 # =============================================================================
 
 
-@dataclass
-class VideoContext:
+class VideoContext(BaseModel):
     """Container for video properties and processing state."""
 
-    cap: Any  # cv2.VideoCapture (using Any to avoid cv2 typing issues)
-    fps: float  # Frames per second
-    total_frames: int  # Total frame count
-    duration: float  # Video duration in seconds
-    start_time: float  # Segment start time
-    end_time: float  # Segment end time
-    frame_skip: int  # Frames to skip between samples
-    start_frame: int  # First frame to process
-    end_frame: int  # Last frame to process
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    cap: Any = Field(..., description="cv2.VideoCapture (using Any to avoid cv2 typing issues)")
+    fps: float = Field(..., description="Frames per second")
+    total_frames: int = Field(..., description="Total frame count")
+    duration: float = Field(..., description="Video duration in seconds")
+    start_time: float = Field(..., description="Segment start time")
+    end_time: float = Field(..., description="Segment end time")
+    frame_skip: int = Field(..., description="Frames to skip between samples")
+    start_frame: int = Field(..., description="First frame to process")
+    end_frame: int = Field(..., description="Last frame to process")
 
 
 # =============================================================================
@@ -69,20 +68,19 @@ class VideoContext:
 # =============================================================================
 
 
-@dataclass
-class DetectionResult:
+class DetectionResult(BaseModel):
     """Results from play detection pipeline."""
 
-    video: str  # Video filename
-    segment_start: float  # Segment start time
-    segment_end: float  # Segment end time
-    total_frames_processed: int  # Number of frames analyzed
-    frames_with_scorebug: int  # Frames where scorebug was detected
-    frames_with_clock: int  # Frames where clock was read successfully
-    plays: List[Dict[str, Any]] = field(default_factory=list)  # Detected plays as dicts
-    stats: Dict[str, Any] = field(default_factory=dict)  # Summary statistics
-    timing: Dict[str, float] = field(default_factory=dict)  # Timing breakdown by section
-    config: Dict[str, Any] = field(default_factory=dict)  # Configuration used for this run (regions, thresholds, etc.)
+    video: str = Field(..., description="Video filename")
+    segment_start: float = Field(..., description="Segment start time")
+    segment_end: float = Field(..., description="Segment end time")
+    total_frames_processed: int = Field(..., description="Number of frames analyzed")
+    frames_with_scorebug: int = Field(..., description="Frames where scorebug was detected")
+    frames_with_clock: int = Field(..., description="Frames where clock was read successfully")
+    plays: List[Dict[str, Any]] = Field(default_factory=list, description="Detected plays as dicts")
+    stats: Dict[str, Any] = Field(default_factory=dict, description="Summary statistics")
+    timing: Dict[str, float] = Field(default_factory=dict, description="Timing breakdown by section")
+    config: Dict[str, Any] = Field(default_factory=dict, description="Configuration used for this run (regions, thresholds, etc.)")
 
 
 # =============================================================================
@@ -97,12 +95,12 @@ class ChunkResult(BaseModel):
     when passing data between worker processes.
     """
 
-    chunk_id: int  # Identifier for this chunk
-    start_time: float  # Chunk start time in seconds
-    end_time: float  # Chunk end time in seconds
-    frames_processed: int  # Total frames processed in this chunk
-    frames_with_scorebug: int  # Frames where scorebug was detected
-    frames_with_clock: int  # Frames where clock was successfully read
-    frame_data: List[Dict[str, Any]]  # Per-frame detection results
-    io_time: float  # Time spent on video I/O operations
-    processing_time: float  # Total processing time for this chunk
+    chunk_id: int = Field(..., description="Identifier for this chunk")
+    start_time: float = Field(..., description="Chunk start time in seconds")
+    end_time: float = Field(..., description="Chunk end time in seconds")
+    frames_processed: int = Field(..., description="Total frames processed in this chunk")
+    frames_with_scorebug: int = Field(..., description="Frames where scorebug was detected")
+    frames_with_clock: int = Field(..., description="Frames where clock was successfully read")
+    frame_data: List[Dict[str, Any]] = Field(..., description="Per-frame detection results")
+    io_time: float = Field(..., description="Time spent on video I/O operations")
+    processing_time: float = Field(..., description="Total processing time for this chunk")

src/readers/models.py

@@ -1,40 +1,38 @@
 """
-Dataclass models for value reading results.
+Pydantic models for value reading results.
 
 These models represent the outputs of reading values from detected regions:
 play clock readings, template match results, etc.
 """
 
-from dataclasses import dataclass
 from typing import Optional
 
+from pydantic import BaseModel, Field
 
-@dataclass
-class PlayClockReading:
+
+class PlayClockReading(BaseModel):
     """Result from play clock OCR reading."""
 
-    detected: bool  # Whether a valid play clock value was detected
-    value: Optional[int]  # Play clock value (0-40 seconds), None if unreadable
-    confidence: float  # Confidence score (0.0 to 1.0)
-    raw_text: str  # Raw OCR output for debugging
+    detected: bool = Field(..., description="Whether a valid play clock value was detected")
+    value: Optional[int] = Field(..., description="Play clock value (0-40 seconds), None if unreadable")
+    confidence: float = Field(..., description="Confidence score (0.0 to 1.0)")
+    raw_text: str = Field(..., description="Raw OCR output for debugging")
 
 
-@dataclass
-class TemplateMatchResult:
+class TemplateMatchResult(BaseModel):
     """Result from template matching for a single digit."""
 
-    digit_value: int  # Matched digit value (-1 for blank, 0-9 for digits)
-    confidence: float  # Match confidence (0.0 to 1.0)
-    is_valid: bool  # Whether match confidence exceeds threshold
+    digit_value: int = Field(..., description="Matched digit value (-1 for blank, 0-9 for digits)")
+    confidence: float = Field(..., description="Match confidence (0.0 to 1.0)")
+    is_valid: bool = Field(..., description="Whether match confidence exceeds threshold")
 
 
-@dataclass
-class TemplatePlayClockReading:
+class TemplatePlayClockReading(BaseModel):
     """Result from template-based play clock reading."""
 
-    detected: bool  # Whether a valid clock value was read
-    value: Optional[int]  # Clock value (0-40), None if unreadable
-    confidence: float  # Overall confidence score
-    tens_match: Optional[TemplateMatchResult]  # Tens digit match result
-    ones_match: Optional[TemplateMatchResult]  # Ones digit match result
-    method: str  # "template_single", "template_double", or "template"
+    detected: bool = Field(..., description="Whether a valid clock value was read")
+    value: Optional[int] = Field(..., description="Clock value (0-40), None if unreadable")
+    confidence: float = Field(..., description="Overall confidence score")
+    tens_match: Optional[TemplateMatchResult] = Field(..., description="Tens digit match result")
+    ones_match: Optional[TemplateMatchResult] = Field(..., description="Ones digit match result")
+    method: str = Field(..., description="'template_single', 'template_double', or 'template'")

src/setup/models.py

@@ -1,49 +1,49 @@
 """
-Dataclass models for the setup/template building phase.
+Pydantic models for the setup/template building phase.
 
 These models are used during the initial template collection phase where
 OCR-labeled samples are collected and averaged into digit templates.
 Also includes region configuration models for setup.
 """
 
-from dataclasses import dataclass
-
 import numpy as np
+from pydantic import BaseModel, ConfigDict, Field
 
 
-@dataclass
-class DigitSample:
+class DigitSample(BaseModel):
     """A single digit sample extracted from a play clock region."""
 
-    digit_value: int  # 0-9 for ones digit, 0-4 for tens digit, -1 for blank
-    is_tens_digit: bool  # True if this is the tens place digit
-    position: str  # "left", "center", or "right" - where digit appears in region
-    image: np.ndarray  # The digit image (grayscale, preprocessed)
-    source_clock_value: int  # The full clock value this was extracted from
-    timestamp: float  # Video timestamp where this was captured
-    confidence: float  # OCR confidence for this sample
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    digit_value: int = Field(..., description="0-9 for ones digit, 0-4 for tens digit, -1 for blank")
+    is_tens_digit: bool = Field(..., description="True if this is the tens place digit")
+    position: str = Field(..., description="'left', 'center', or 'right' - where digit appears in region")
+    image: np.ndarray = Field(..., description="The digit image (grayscale, preprocessed)")
+    source_clock_value: int = Field(..., description="The full clock value this was extracted from")
+    timestamp: float = Field(..., description="Video timestamp where this was captured")
+    confidence: float = Field(..., description="OCR confidence for this sample")
 
 
-@dataclass
-class DigitTemplate:
+class DigitTemplate(BaseModel):
     """A template for matching a specific digit."""
 
-    digit_value: int  # 0-9 for ones, 0-4 for tens, -1 for blank
-    is_tens_digit: bool  # True if this is a tens place template
-    position: str  # "left", "center", or "right" - where digit appears in region
-    template: np.ndarray  # The template image (grayscale)
-    sample_count: int  # Number of samples used to build this template
-    avg_confidence: float  # Average OCR confidence of source samples
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    digit_value: int = Field(..., description="0-9 for ones, 0-4 for tens, -1 for blank")
+    is_tens_digit: bool = Field(..., description="True if this is a tens place template")
+    position: str = Field(..., description="'left', 'center', or 'right' - where digit appears in region")
+    template: np.ndarray = Field(..., description="The template image (grayscale)")
+    sample_count: int = Field(..., description="Number of samples used to build this template")
+    avg_confidence: float = Field(..., description="Average OCR confidence of source samples")
 
 
-@dataclass
-class PlayClockRegionConfig:
+class PlayClockRegionConfig(BaseModel):
     """Configuration for the play clock region relative to the scorebug bounding box."""
 
-    x_offset: int  # X offset from scorebug left edge
-    y_offset: int  # Y offset from scorebug top edge
-    width: int  # Width of play clock region
-    height: int  # Height of play clock region
-    source_video: str  # Video used to identify region
-    scorebug_template: str  # Template used for scorebug detection
-    samples_used: int  # Number of frames used to verify region
+    x_offset: int = Field(..., description="X offset from scorebug left edge")
+    y_offset: int = Field(..., description="Y offset from scorebug top edge")
+    width: int = Field(..., description="Width of play clock region")
+    height: int = Field(..., description="Height of play clock region")
+    source_video: str = Field(..., description="Video used to identify region")
+    scorebug_template: str = Field(..., description="Template used for scorebug detection")
+    samples_used: int = Field(..., description="Number of frames used to verify region")

src/tracking/models.py

@@ -1,24 +1,24 @@
 """
-Dataclass models for play tracking.
+Pydantic models for play tracking.
 
 These models represent detected plays and their temporal boundaries.
 """
 
-from dataclasses import dataclass
 from typing import Optional
 
+from pydantic import BaseModel, Field
 
-@dataclass
-class PlayEvent:
+
+class PlayEvent(BaseModel):
     """Represents a detected play with start and end times."""
 
-    play_number: int  # Sequential play number
-    start_time: float  # Video timestamp (seconds) when play started
-    end_time: float  # Video timestamp (seconds) when play ended - from backward counting
-    confidence: float  # Overall confidence score
-    start_method: str  # How start was detected: "clock_reset", "clock_reset_25", "clock_freeze"
-    end_method: str  # How end was detected: "backward_calc" (primary), "direct_detect" (secondary)
-    direct_end_time: Optional[float] = None  # End time from direct detection (for comparison)
-    start_clock_value: Optional[int] = None  # Clock value at start detection
-    end_clock_value: Optional[int] = None  # Clock value used for backward calculation
-    play_type: str = "normal"  # Type of play: "normal", "special" (punt/fg/xp after 25-second reset)
+    play_number: int = Field(..., description="Sequential play number")
+    start_time: float = Field(..., description="Video timestamp (seconds) when play started")
+    end_time: float = Field(..., description="Video timestamp (seconds) when play ended - from backward counting")
+    confidence: float = Field(..., description="Overall confidence score")
+    start_method: str = Field(..., description="How start was detected: 'clock_reset', 'clock_reset_25', 'clock_freeze'")
+    end_method: str = Field(..., description="How end was detected: 'backward_calc' (primary), 'direct_detect' (secondary)")
+    direct_end_time: Optional[float] = Field(None, description="End time from direct detection (for comparison)")
+    start_clock_value: Optional[int] = Field(None, description="Clock value at start detection")
+    end_clock_value: Optional[int] = Field(None, description="Clock value used for backward calculation")
+    play_type: str = Field("normal", description="Type of play: 'normal', 'special' (punt/fg/xp after 25-second reset)")

src/video/ffmpeg_ops.py

@@ -17,11 +17,12 @@ import subprocess
 import tempfile
 import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
-from dataclasses import dataclass
 from enum import Enum
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 
+from pydantic import BaseModel, Field
+
 logger = logging.getLogger(__name__)
 
 
@@ -33,14 +34,13 @@ class ClipMethod(Enum):
     ULTRAFAST = "ultrafast"  # Frame-accurate, faster encoding
 
 
-@dataclass
-class ClipInfo:
+class ClipInfo(BaseModel):
     """Information about a clip to extract."""
 
-    start_time: float  # Start time in seconds
-    end_time: float  # End time in seconds
-    output_path: Path  # Path for output file
-    play_number: Optional[int] = None  # Optional play number for logging
+    start_time: float = Field(..., description="Start time in seconds")
+    end_time: float = Field(..., description="End time in seconds")
+    output_path: Path = Field(..., description="Path for output file")
+    play_number: Optional[int] = Field(None, description="Optional play number for logging")
 
 
 def extract_clip_stream_copy(