""" Play state machine module for tracking play start and end times. This module tracks play clock state changes to determine when plays begin and end. The primary method for determining play end time is backward counting from the next observed play clock value after the play. Clock Reset Classification: The state machine also classifies 40→25 clock reset events: - Class A (weird_clock): 25 counts down immediately → rejected - Class B (timeout): Timeout indicator changed → tracked as timeout - Class C (special): Neither A nor B → special play (punt/FG/XP) Architecture (v2 - Restructured): This class now serves as a facade that delegates to the new PlayTracker, which coordinates two specialized sub-trackers: - NormalPlayTracker: Handles standard plays (clock reset to 40, countdown) - SpecialPlayTracker: Handles 40→25 transitions (timeouts, punts, FGs, XPs) The API remains backward compatible with the original implementation. """ import logging from typing import Any, Dict, List, Optional from detection import ScorebugDetection from readers import PlayClockReading from .models import FlagInfo, PlayEvent, PlayState, TrackPlayStateConfig, TimeoutInfo from .play_tracker import PlayTracker logger = logging.getLogger(__name__) class TrackPlayState: """ State machine for tracking play boundaries using play clock behavior. This class maintains backward compatibility with the original API while internally using the new restructured PlayTracker architecture. Identification Strategy: - Play START: Identified when play clock resets to 40 (or potentially freezes - needs validation) - Play END: **Always use backward counting** - calculate from next observed clock value after play Requires K consecutive descending clock ticks to confirm (avoids false positives) Backward Counting: When the play clock reappears showing value X (where X < 40), the play end time is: play_end_time = current_time - (40 - X) This method is reliable even when the broadcast cuts to replays. """ def __init__(self, config: Optional[TrackPlayStateConfig] = None): """Initialize the state machine. Args: config: Configuration settings. Uses defaults if not provided. """ self.config = config or TrackPlayStateConfig() # Use the new PlayTracker internally self._tracker = PlayTracker(self.config) # ========================================================================= # Properties for backward compatibility (access state fields directly) # ========================================================================= @property def state(self) -> PlayState: """Current state of the play state machine.""" return self._tracker.state @state.setter def state(self, _value: PlayState) -> None: # Note: Direct state setting is deprecated but maintained for compatibility # The new architecture manages state internally logger.warning("Direct state setting is deprecated in the new architecture") @property def plays(self) -> List[PlayEvent]: """List of all tracked plays.""" return self._tracker.plays # ========================================================================= # 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 state machine with new frame data. Args: timestamp: Current video timestamp in seconds scorebug: Scorebug detection result clock: Play clock reading result timeout_info: Optional timeout indicator information for clock reset classification flag_info: Optional FLAG indicator information for penalty detection Returns: PlayEvent if a play just ended, None otherwise """ return self._tracker.update(timestamp, scorebug, clock, timeout_info, flag_info) # ========================================================================= # Public API methods # ========================================================================= def get_plays(self) -> List[PlayEvent]: """Get all tracked plays (normal + special, not FLAG).""" return self._tracker.get_plays() def get_flag_plays(self) -> List[PlayEvent]: """Get all FLAG plays (tracked independently).""" return self._tracker.get_flag_plays() def get_state(self) -> PlayState: """Get current state.""" return self._tracker.get_state() def get_stats(self) -> Dict[str, Any]: """Get statistics about tracked plays.""" return self._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 """ self._tracker.finalize(timestamp)