""" Play tracker parent module. Coordinates the NormalPlayTracker, SpecialPlayTracker, and FlagTracker sub-trackers, managing handoffs between them and maintaining the master list of all plays. Architecture: - PlayTracker (this class): Coordinates sub-trackers, holds all plays - NormalPlayTracker: Handles standard plays (clock reset to 40, countdown) - SpecialPlayTracker: Handles 40→25 transitions (timeouts, punts, FGs, XPs) - FlagTracker: Independently tracks FLAG (penalty) events Control flow: 1. Normal mode: NormalPlayTracker processes clock readings 2. When 40→25 detected: NormalPlayTracker signals handoff 3. Special mode: SpecialPlayTracker classifies the transition 4. After resolution: Control returns to NormalPlayTracker FLAG tracking runs in parallel: - FlagTracker is updated on EVERY frame regardless of mode - FLAG plays are tracked independently and merged at the end - FLAG plays have absolute priority - they are ALWAYS included """ import logging from typing import Any, Dict, List, Optional from detection import ScorebugDetection from readers import PlayClockReading from .models import ( ClockResetStats, FlagInfo, PlayEvent, PlayState, SpecialPlayHandoff, TimeoutInfo, TrackerMode, TrackPlayStateConfig, ) from .flag_tracker import FlagTracker from .normal_play_tracker import NormalPlayTracker from .special_play_tracker import SpecialPlayTracker logger = logging.getLogger(__name__) class PlayTracker: """ Parent tracker that coordinates NormalPlayTracker, SpecialPlayTracker, and FlagTracker. Responsibilities: - Maintain the master list of all detected plays - Route updates to the appropriate sub-tracker based on current mode - Handle handoffs between sub-trackers when 40→25 transitions occur - Independently track FLAG events (always included in output) - Synchronize play counts between sub-trackers """ def __init__(self, config: Optional[TrackPlayStateConfig] = None): """ Initialize the play tracker. Args: config: Configuration settings. Uses defaults if not provided. """ self.config = config or TrackPlayStateConfig() # Master list of all plays (normal + special, FLAGS handled separately) self._plays: List[PlayEvent] = [] # FLAG plays tracked independently self._flag_plays: List[PlayEvent] = [] # Current active tracker mode self._active_mode: TrackerMode = TrackerMode.NORMAL # Sub-trackers self._normal_tracker = NormalPlayTracker(self.config) self._special_tracker = SpecialPlayTracker() self._flag_tracker = FlagTracker() # ========================================================================= # Public properties # ========================================================================= @property def plays(self) -> List[PlayEvent]: """List of all detected plays (normal + special, not FLAG).""" return self._plays @property def flag_plays(self) -> List[PlayEvent]: """List of all FLAG plays (tracked independently).""" return self._flag_plays @property def state(self) -> PlayState: """Current state of the normal play tracker (for backward compatibility).""" return self._normal_tracker.state @property def active_mode(self) -> TrackerMode: """Which sub-tracker is currently active.""" return self._active_mode @property def clock_reset_stats(self) -> ClockResetStats: """Statistics about 40→25 clock reset classifications.""" return self._special_tracker.stats @property def flag_tracker_stats(self) -> Dict[str, Any]: """Statistics about FLAG tracking.""" return self._flag_tracker.get_stats() # ========================================================================= # Main update method # ========================================================================= def update( self, timestamp: float, scorebug: ScorebugDetection, clock: PlayClockReading, timeout_info: Optional[TimeoutInfo] = None, flag_info: Optional[FlagInfo] = None, ) -> Optional[PlayEvent]: """ Update the tracker with new frame data. Routes the update to the appropriate sub-tracker based on current mode, handles handoffs, and maintains the master play list. FLAG tracking runs in parallel on EVERY frame, regardless of mode. Args: timestamp: Current video timestamp in seconds scorebug: Scorebug detection result clock: Play clock reading result timeout_info: Optional timeout indicator information flag_info: Optional FLAG indicator information for penalty detection Returns: PlayEvent if a play just ended, None otherwise """ completed_play = None # Determine actual scorebug visibility (for disappearance detection and flag filtering) # In fixed coords mode: detected=True (assumed), template_matched=actual visibility # In standard mode: template_matched is None, use detected scorebug_actually_visible = scorebug.template_matched if scorebug.template_matched is not None else scorebug.detected # FLAG tracking runs in parallel, regardless of normal/special mode # Uses actual visibility (template_matched) to filter false positives during commercials flag_play = self._flag_tracker.update( timestamp=timestamp, scorebug_detected=scorebug_actually_visible, flag_info=flag_info, ) if flag_play is not None: # Add FLAG play to separate list (will be merged later) self._flag_plays.append(flag_play) logger.debug("FLAG Play #%d added (%.1fs - %.1fs)", len(self._flag_plays), flag_play.start_time, flag_play.end_time) # Normal/Special play tracking if self._active_mode == TrackerMode.NORMAL: completed_play = self._update_normal_mode(timestamp, scorebug, clock, timeout_info, flag_info) else: # TrackerMode.SPECIAL completed_play = self._update_special_mode(timestamp, scorebug, clock, timeout_info, scorebug_actually_visible) # Add completed play to master list if completed_play is not None: # Check if opening kickoff is still active - if so, end it first # This handles cases where special plays are created before a normal play if self._normal_tracker._state.opening_kickoff_active: kickoff_play = self._end_opening_kickoff_from_parent(completed_play.start_time) if kickoff_play is not None: kickoff_play = kickoff_play.model_copy(update={"play_number": len(self._plays) + 1}) self._plays.append(kickoff_play) logger.debug("Opening kickoff Play #%d added before first detected play", kickoff_play.play_number) # Update play number to be sequential in master list completed_play = completed_play.model_copy(update={"play_number": len(self._plays) + 1}) self._plays.append(completed_play) logger.debug("Play #%d added to master list (mode=%s)", completed_play.play_number, self._active_mode.value) return completed_play def _update_normal_mode( self, timestamp: float, scorebug: ScorebugDetection, clock: PlayClockReading, timeout_info: Optional[TimeoutInfo], flag_info: Optional[FlagInfo] = None, ) -> Optional[PlayEvent]: """Process update in normal mode.""" # Update the normal tracker completed_play = self._normal_tracker.update( timestamp=timestamp, scorebug_detected=scorebug.detected, clock_value=clock.value if clock.detected else None, timeout_info=timeout_info, flag_info=flag_info, ) # Check if normal tracker is requesting handoff to special tracker if self._normal_tracker.request_special_handoff: handoff = self._normal_tracker.pending_handoff if handoff is not None: self._activate_special_mode(handoff) return completed_play def _update_special_mode( self, timestamp: float, scorebug: ScorebugDetection, clock: PlayClockReading, timeout_info: Optional[TimeoutInfo], scorebug_actually_visible: bool = True, ) -> Optional[PlayEvent]: """Process update in special mode.""" # Update the special tracker with actual scorebug visibility # (for disappearance detection in special play end logic) completed_play = self._special_tracker.update( timestamp=timestamp, scorebug_detected=scorebug_actually_visible, clock_value=clock.value if clock.detected else None, timeout_info=timeout_info, ) # Check if special tracker has completed resolution if self._special_tracker.resolution_complete: self._return_to_normal_mode() return completed_play # ========================================================================= # Mode transitions # ========================================================================= def _activate_special_mode(self, handoff: SpecialPlayHandoff) -> None: """ Activate special mode to handle a 40→25 transition. Args: handoff: SpecialPlayHandoff data from NormalPlayTracker """ logger.info( "Activating SPECIAL mode at %.1fs (was_in_play=%s)", handoff.transition_timestamp, handoff.was_in_play, ) # Clear the handoff request from normal tracker self._normal_tracker.clear_handoff_request() # Synchronize play count to special tracker self._special_tracker.set_play_count(len(self._plays)) # Activate special tracker with handoff data self._special_tracker.activate(handoff) # Switch to special mode self._active_mode = TrackerMode.SPECIAL def _return_to_normal_mode(self) -> None: """Return to normal mode after special tracker completes.""" classification = self._special_tracker.classification last_clock = self._special_tracker.get_last_clock_value() logger.info( "Returning to NORMAL mode (classification=%s, last_clock=%d)", classification, last_clock, ) # Reset special tracker self._special_tracker.reset() # Resume normal tracker with last known clock value self._normal_tracker.resume_after_special(clock_value=last_clock) # Switch back to normal mode self._active_mode = TrackerMode.NORMAL def _end_opening_kickoff_from_parent(self, first_play_start: float) -> Optional[PlayEvent]: """ End the opening kickoff when the first play is detected. This is called from the parent PlayTracker when any play (normal or special) is completed, ensuring the opening kickoff is captured even if special plays are created before normal play detection kicks in. Note: Kickoff plays that are too long will be filtered out by PlayMerger using max_kickoff_duration filter. Args: first_play_start: Start time of the first detected play Returns: PlayEvent for the opening kickoff, or None if not applicable """ state = self._normal_tracker._state # Get the first clock reading timestamp as kickoff start start_time = state.first_clock_reading_timestamp if start_time is None: return None # Kickoff ends just before the first play starts end_time = first_play_start - 0.5 if end_time <= start_time: end_time = start_time + 2.0 # Minimum 2 second duration # Create the kickoff play play = PlayEvent( play_number=0, # Will be updated by caller start_time=start_time, end_time=end_time, confidence=0.75, start_method="opening_kickoff", end_method="first_play_detected", direct_end_time=first_play_start, start_clock_value=None, end_clock_value=None, play_type="kickoff", ) # Mark opening kickoff as complete in normal tracker state.opening_kickoff_active = False state.opening_kickoff_complete = True logger.info( "Opening kickoff: %.1fs - %.1fs (duration: %.1fs)", play.start_time, play.end_time, play.end_time - play.start_time, ) return play # ========================================================================= # Public API methods # ========================================================================= def get_plays(self) -> List[PlayEvent]: """Get all tracked plays (normal + special, not FLAG).""" return self._plays.copy() def get_flag_plays(self) -> List[PlayEvent]: """Get all FLAG plays (tracked independently).""" return self._flag_plays.copy() def get_state(self) -> PlayState: """Get current state of normal tracker.""" return self._normal_tracker.state def get_stats(self) -> Dict[str, Any]: """Get statistics about tracked plays.""" if not self._plays: return { "total_plays": 0, "clock_reset_events": self._special_tracker.stats.model_dump(), "flag_stats": self._flag_tracker.get_stats(), } durations = [p.end_time - p.start_time for p in self._plays] return { "total_plays": len(self._plays), "avg_duration": sum(durations) / len(durations), "min_duration": min(durations), "max_duration": max(durations), "start_methods": {m: sum(1 for p in self._plays if p.start_method == m) for m in set(p.start_method for p in self._plays)}, "end_methods": {m: sum(1 for p in self._plays if p.end_method == m) for m in set(p.end_method for p in self._plays)}, "play_types": {t: sum(1 for p in self._plays if p.play_type == t) for t in set(p.play_type for p in self._plays)}, "clock_reset_events": self._special_tracker.stats.model_dump(), "flag_stats": self._flag_tracker.get_stats(), } def finalize(self, timestamp: float) -> None: """ Finalize tracking at end of video. This ensures any active FLAG event is properly closed. Args: timestamp: Final video timestamp """ flag_play = self._flag_tracker.finalize(timestamp) if flag_play is not None: self._flag_plays.append(flag_play) logger.info("Finalized active FLAG event at end of video")