Spaces:
Sleeping
Sleeping
| """ | |
| Pydantic models for play tracking. | |
| These models represent detected plays, their temporal boundaries, | |
| and the state machine configuration/state for play detection. | |
| Architecture: | |
| - PlayTracker (parent): Coordinates sub-trackers, holds all plays | |
| - NormalPlayTracker: Handles standard plays (clock reset to 40, countdown) | |
| - SpecialPlayTracker: Handles 40→25 transitions (timeouts, punts, FGs, XPs) | |
| """ | |
| from enum import Enum | |
| from typing import Optional, List, Tuple | |
| from pydantic import BaseModel, Field | |
| # ============================================================================= | |
| # Utility Functions | |
| # ============================================================================= | |
| def determine_timeout_team(home_change: int, away_change: int) -> Optional[str]: | |
| """ | |
| Determine which team called a timeout based on timeout count changes. | |
| Validates that EXACTLY one team's count decreased by 1 while the other | |
| team's count stayed the same. | |
| Args: | |
| home_change: Change in home team timeouts (before - after, positive = decrease) | |
| away_change: Change in away team timeouts (before - after, positive = decrease) | |
| Returns: | |
| "home" or "away" if a valid timeout pattern detected, None otherwise | |
| """ | |
| if home_change == 1 and away_change == 0: | |
| return "home" | |
| if away_change == 1 and home_change == 0: | |
| return "away" | |
| return None | |
| # ============================================================================= | |
| # Enums | |
| # ============================================================================= | |
| class TrackerMode(Enum): | |
| """Which sub-tracker is currently active.""" | |
| NORMAL = "normal" # Normal play tracker is active | |
| SPECIAL = "special" # Special play tracker is handling 40→25 transition | |
| class PlayState(Enum): | |
| """Current state of play detection (used by NormalPlayTracker).""" | |
| IDLE = "idle" # No scorebug detected, waiting | |
| PRE_SNAP = "pre_snap" # Scorebug visible, clock ticking down before snap | |
| PLAY_IN_PROGRESS = "play_in_progress" # Ball snapped, play is live | |
| POST_PLAY = "post_play" # Play ended, waiting for next play setup | |
| NO_SCOREBUG = "no_scorebug" # Scorebug lost during/after play (e.g., replay) | |
| class SpecialPlayPhase(Enum): | |
| """ | |
| State machine phases for SpecialPlayTracker when handling 40→25 transitions. | |
| Flow: | |
| 1. CHECKING_COUNTDOWN: Check if 25 immediately counts down (weird clock) | |
| 2. MONITORING_TIMEOUT: Wait 4-7s for timeout indicator change | |
| 3. WAITING_FOR_END: Track as special play until 7s or scorebug disappears | |
| 4. RESOLVED: Classification complete, ready to hand back control | |
| """ | |
| CHECKING_COUNTDOWN = "checking_countdown" # First 2s: checking if 25 counts down | |
| MONITORING_TIMEOUT = "monitoring_timeout" # 4-7s: watching for timeout indicator change | |
| WAITING_FOR_END = "waiting_for_end" # Tracking special play until end condition | |
| RESOLVED = "resolved" # Classification complete | |
| class PlayEvent(BaseModel): | |
| """Represents a detected play with start and end times.""" | |
| 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)") | |
| has_flag: bool = Field(False, description="Whether a penalty FLAG was detected during this play") | |
| class TrackPlayStateConfig(BaseModel): | |
| """Configuration settings for play state tracking. | |
| These values control the detection thresholds and timing parameters | |
| used by the state machine to identify play boundaries. | |
| """ | |
| clock_stable_frames: int = Field(3, description="Frames with same clock value to consider it 'stable'") | |
| max_play_duration: float = Field(15.0, description="Maximum expected play duration in seconds") | |
| scorebug_lost_timeout: float = Field(30.0, description="Seconds before resetting state when scorebug lost") | |
| required_countdown_ticks: int = Field(3, description="Number of consecutive descending ticks required to confirm play end") | |
| min_clock_jump_for_reset: int = Field(5, description="Minimum jump in clock value to consider it a valid reset (40 from X where X <= 40 - this value)") | |
| # FLAG detection settings (penalty flag indicator) | |
| flag_extension_timeout: float = Field(15.0, description="Max seconds to extend play capture after FLAG first detected") | |
| capture_flag_plays: bool = Field(True, description="Whether to capture plays with FLAGS (bypasses quiet time filter)") | |
| # Opening kickoff detection settings | |
| opening_kickoff_min_consecutive_frames: int = Field(3, description="Required consecutive frames with valid clock readings to confirm kickoff start") | |
| max_opening_kickoff_duration: float = Field(90.0, description="Maximum valid kickoff duration (seconds). If exceeded, reject and keep searching.") | |
| class TimeoutInfo(BaseModel): | |
| """Timeout information for a frame.""" | |
| home_timeouts: Optional[int] = Field(None, description="Number of home team timeouts remaining") | |
| away_timeouts: Optional[int] = Field(None, description="Number of away team timeouts remaining") | |
| confidence: float = Field(0.0, description="Confidence of the timeout reading") | |
| class FlagInfo(BaseModel): | |
| """FLAG indicator information for a frame.""" | |
| detected: bool = Field(False, description="Whether FLAG is detected (yellow present AND valid hue)") | |
| yellow_ratio: float = Field(0.0, description="Ratio of yellow pixels in FLAG region") | |
| mean_hue: float = Field(0.0, description="Mean hue of yellow pixels (helps distinguish orange)") | |
| scorebug_verified: bool = Field(True, description="Whether scorebug was verified present via template matching") | |
| is_valid_yellow: bool = Field(False, description="True if mean_hue >= threshold (not orange)") | |
| class ClockResetStats(BaseModel): | |
| """Statistics about clock reset classifications.""" | |
| total: int = Field(0, description="Total 40→25 resets detected") | |
| weird_clock: int = Field(0, description="Class A: 25 counts down immediately (rejected)") | |
| timeout: int = Field(0, description="Class B: Timeout indicator changed") | |
| special: int = Field(0, description="Class C: Special play (injury/punt/FG/XP)") | |
| # ============================================================================= | |
| # State models for restructured tracker architecture | |
| # ============================================================================= | |
| class SpecialPlayHandoff(BaseModel): | |
| """ | |
| Data passed from NormalPlayTracker to SpecialPlayTracker when a 40→25 transition is detected. | |
| Contains all context needed for the SpecialPlayTracker to classify and handle the transition. | |
| """ | |
| transition_timestamp: float = Field(..., description="When the 40→25 transition occurred") | |
| home_timeouts_at_40: Optional[int] = Field(None, description="Home team timeouts before transition") | |
| away_timeouts_at_40: Optional[int] = Field(None, description="Away team timeouts before transition") | |
| timeout_confidence_at_40: float = Field(0.0, description="Confidence of timeout reading at 40") | |
| was_in_play: bool = Field(False, description="Whether a play was in progress when transition detected") | |
| play_start_time: Optional[float] = Field(None, description="Start time of play in progress (if any)") | |
| time_at_40: float = Field(0.0, description="How long clock was at 40 before transitioning to 25") | |
| from_freeze_detection: bool = Field(False, description="Whether this handoff came from clock freeze→25 detection (always treat as special play)") | |
| class NormalTrackerState(BaseModel): | |
| """ | |
| State for NormalPlayTracker. | |
| Tracks clock behavior and play boundaries for normal plays. | |
| """ | |
| # Current detection state | |
| state: PlayState = Field(PlayState.IDLE, description="Current state of the normal play tracker") | |
| # Clock tracking | |
| last_clock_value: Optional[int] = Field(None, description="Last observed play clock value") | |
| last_clock_timestamp: Optional[float] = Field(None, description="Timestamp of last clock reading") | |
| clock_stable_count: int = Field(0, description="Number of consecutive frames with same clock value") | |
| last_scorebug_timestamp: Optional[float] = Field(None, description="Timestamp of last scorebug detection") | |
| # Current play tracking | |
| current_play_start_time: Optional[float] = Field(None, description="Start time of the current play") | |
| current_play_start_method: Optional[str] = Field(None, description="Method used to detect play start") | |
| current_play_start_clock: Optional[int] = Field(None, description="Clock value when play started") | |
| current_play_clock_base: int = Field(40, description="Clock base for current play (40 or 25)") | |
| current_play_type: str = Field("normal", description="Type of current play: 'normal' or 'special'") | |
| first_40_timestamp: Optional[float] = Field(None, description="When we first saw 40 in current play") | |
| countdown_history: List[Tuple[float, int]] = Field(default_factory=list, description="(timestamp, clock_value) pairs") | |
| direct_end_time: Optional[float] = Field(None, description="Direct end time observation") | |
| # Timeout tracking | |
| last_home_timeouts: Optional[int] = Field(None, description="Last observed home team timeouts") | |
| last_away_timeouts: Optional[int] = Field(None, description="Last observed away team timeouts") | |
| last_timeout_confidence: float = Field(0.0, description="Confidence of last timeout reading") | |
| # FLAG tracking (penalty flag indicator) | |
| flag_detected_at: Optional[float] = Field(None, description="When FLAG first appeared in current FLAG event") | |
| flag_last_seen_at: Optional[float] = Field(None, description="Last time FLAG was visible") | |
| flag_active: bool = Field(False, description="Whether FLAG is currently active (extends play capture)") | |
| current_play_has_flag: bool = Field(False, description="Whether FLAG was detected during the current play") | |
| # Clock freeze tracking (for special play detection) | |
| clock_freeze_start_timestamp: Optional[float] = Field(None, description="When clock first reached current frozen value") | |
| clock_freeze_value: Optional[int] = Field(None, description="The clock value that has been frozen/stable") | |
| # Last play info (for continuation detection) | |
| last_play_end_time: Optional[float] = Field(None, description="End time of the last completed play") | |
| # Handoff signaling | |
| request_special_handoff: bool = Field(False, description="Flag to request handoff to SpecialPlayTracker") | |
| pending_handoff: Optional[SpecialPlayHandoff] = Field(None, description="Data for pending handoff to special tracker") | |
| # Opening kickoff tracking | |
| # The opening kickoff is special because the scorebug appears mid-play | |
| # We detect it when scorebug first appears and end it on first clock reset to 40 | |
| # Requires k consecutive valid clock readings to confirm kickoff (filters out pre-game noise) | |
| opening_kickoff_active: bool = Field(False, description="Whether we're currently tracking the opening kickoff") | |
| opening_kickoff_complete: bool = Field(False, description="Whether opening kickoff has been recorded (only happens once)") | |
| first_scorebug_timestamp: Optional[float] = Field(None, description="When scorebug first appeared in the video (verified by template matching)") | |
| first_clock_reading_timestamp: Optional[float] = Field(None, description="When we first got a valid clock reading") | |
| opening_kickoff_consecutive_readings: int = Field(0, description="Count of consecutive frames with valid clock readings for kickoff detection") | |
| opening_kickoff_candidate_timestamp: Optional[float] = Field(None, description="Timestamp when consecutive clock readings started (candidate kickoff start)") | |
| class SpecialTrackerState(BaseModel): | |
| """ | |
| State for SpecialPlayTracker. | |
| Manages the 40→25 transition classification state machine. | |
| """ | |
| # Current phase of special play handling | |
| phase: SpecialPlayPhase = Field(SpecialPlayPhase.RESOLVED, description="Current phase of special play handling") | |
| # Transition context (from handoff) | |
| transition_timestamp: Optional[float] = Field(None, description="When the 40→25 transition occurred") | |
| home_timeouts_at_40: Optional[int] = Field(None, description="Home team timeouts before transition") | |
| away_timeouts_at_40: Optional[int] = Field(None, description="Away team timeouts before transition") | |
| timeout_confidence_at_40: float = Field(0.0, description="Confidence of timeout reading at 40") | |
| was_in_play: bool = Field(False, description="Whether a play was in progress when transition detected") | |
| play_start_time: Optional[float] = Field(None, description="Start time of play that led to this transition") | |
| time_at_40: float = Field(0.0, description="How long clock was at 40 before transitioning") | |
| from_freeze_detection: bool = Field(False, description="Whether this came from clock freeze→25 detection (always special play)") | |
| # Classification tracking | |
| lowest_clock_seen: int = Field(25, description="Lowest clock value seen since 25 (for countdown detection)") | |
| last_scorebug_timestamp: Optional[float] = Field(None, description="Last time scorebug was visible") | |
| # Scorebug disappearance tracking (debounce for noisy detection) | |
| consecutive_scorebug_absent: int = Field(0, description="Number of consecutive frames without scorebug detection") | |
| # Timeout confirmation tracking (require k consecutive confirmations) | |
| pending_timeout_team: Optional[str] = Field(None, description="Team with pending timeout detection") | |
| consecutive_timeout_confirmations: int = Field(0, description="Number of consecutive frames confirming timeout") | |
| saw_timeout_increase: bool = Field(False, description="Whether we saw any timeout count INCREASE (invalidates timeout detection)") | |
| # Resolution | |
| resolution_complete: bool = Field(False, description="Whether classification is complete") | |
| classification: Optional[str] = Field(None, description="Final classification: 'weird_clock', 'timeout', 'special'") | |
| timeout_team: Optional[str] = Field(None, description="Which team called timeout: 'home' or 'away'") | |