""" Timeout tracker detector module. This module provides functions to detect timeout indicator changes on the scorebug. Each team has 3 timeout indicators (white ovals when available, dark when used). Detecting when an oval changes from white to dark indicates a timeout was called. Two detection modes are supported: 1. Legacy mode (DetectTimeouts): Divides region into 3 equal parts 2. Calibrated mode (CalibratedTimeoutDetector): Uses blob-detected oval positions """ import json import logging from pathlib import Path from typing import Any, Optional, Tuple, List import cv2 import numpy as np from .models import CalibratedTimeoutRegion, OvalLocation, TimeoutRegionConfig, TimeoutReading logger = logging.getLogger(__name__) class DetectTimeouts: """ Tracks timeout indicators on the scorebug. Each team has 3 timeout indicators displayed as ovals: - White oval = timeout available - Dark oval = timeout used The tracker monitors these indicators and detects when a timeout is called (white oval becomes dark). """ # Threshold for determining if a pixel is "bright" (part of white oval) BRIGHT_PIXEL_THRESHOLD = 200 # On 0-255 scale # Minimum percentage of bright pixels for an oval to be considered "white" (available) # Based on analysis: available ovals have 0.15-0.19 ratio, used ovals have ~0.00 ratio BRIGHT_PIXEL_RATIO_THRESHOLD = 0.10 # 10% of pixels must be bright for available timeout # Minimum confidence for a valid reading MIN_CONFIDENCE = 0.5 def __init__( self, home_region: Optional[TimeoutRegionConfig] = None, away_region: Optional[TimeoutRegionConfig] = None, config_path: Optional[str] = None, ): """ Initialize the timeout tracker. Args: home_region: Configuration for home team's timeout indicators away_region: Configuration for away team's timeout indicators config_path: Path to JSON config file with regions (alternative to direct config) """ self.home_region = home_region self.away_region = away_region self._configured = home_region is not None and away_region is not None # Previous reading for change detection self._prev_reading: Optional[TimeoutReading] = None # Load from config if provided if config_path and not self._configured: self._load_config(config_path) if self._configured: logger.info("DetectTimeouts initialized with regions") logger.info(" Home region: %s", self.home_region.bbox if self.home_region else None) logger.info(" Away region: %s", self.away_region.bbox if self.away_region else None) else: logger.info("DetectTimeouts initialized (not configured - call configure_regions first)") def _load_config(self, config_path: str) -> None: """Load timeout regions from a JSON config file.""" path = Path(config_path) if not path.exists(): logger.warning("Timeout tracker config not found: %s", config_path) return with open(path, "r", encoding="utf-8") as f: data = json.load(f) if "home_timeout_region" in data: self.home_region = TimeoutRegionConfig.from_dict(data["home_timeout_region"]) if "away_timeout_region" in data: self.away_region = TimeoutRegionConfig.from_dict(data["away_timeout_region"]) self._configured = self.home_region is not None and self.away_region is not None if self._configured: logger.info("Loaded timeout tracker config from: %s", config_path) def save_config(self, config_path: str) -> None: """Save timeout regions to a JSON config file.""" if not self._configured: logger.warning("Cannot save config - tracker not configured") return data = {} if self.home_region: data["home_timeout_region"] = self.home_region.to_dict() if self.away_region: data["away_timeout_region"] = self.away_region.to_dict() path = Path(config_path) path.parent.mkdir(parents=True, exist_ok=True) with open(path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) logger.info("Saved timeout tracker config to: %s", config_path) def is_configured(self) -> bool: """Check if the tracker is configured with regions.""" return self._configured def set_regions(self, home_region: TimeoutRegionConfig, away_region: TimeoutRegionConfig) -> None: """ Set the timeout indicator regions. Args: home_region: Configuration for home team's timeout indicators away_region: Configuration for away team's timeout indicators """ self.home_region = home_region self.away_region = away_region self._configured = True logger.info("Timeout regions set: home=%s, away=%s", home_region.bbox, away_region.bbox) def _extract_oval_bright_ratios(self, frame: np.ndarray[Any, Any], region: TimeoutRegionConfig) -> List[float]: """ Extract the ratio of bright pixels for each oval in a region. Divides the region into 3 equal VERTICAL parts (one per oval) since timeout indicators are stacked vertically (3 horizontal bars). Args: frame: Input frame (BGR format) region: Region configuration Returns: List of 3 bright pixel ratios (0.0-1.0), one per oval """ x, y, w, h = region.bbox # Validate bounds frame_h, frame_w = frame.shape[:2] if x < 0 or y < 0 or x + w > frame_w or y + h > frame_h: logger.warning("Timeout region out of bounds: %s", region.bbox) return [0.0, 0.0, 0.0] # Extract the region roi = frame[y : y + h, x : x + w] # Convert to grayscale for brightness analysis gray = cv2.cvtColor(roi, cv2.COLOR_BGR2GRAY) # Divide into 3 equal VERTICAL sections (one per oval) - they're stacked oval_height = h // 3 bright_ratios = [] for i in range(3): start_y = i * oval_height end_y = start_y + oval_height if i < 2 else h # Last oval gets remaining height oval_region = gray[start_y:end_y, :] # Count pixels above brightness threshold total_pixels = oval_region.size bright_pixels = np.sum(oval_region >= self.BRIGHT_PIXEL_THRESHOLD) bright_ratio = bright_pixels / total_pixels if total_pixels > 0 else 0.0 bright_ratios.append(float(bright_ratio)) return bright_ratios def _classify_ovals(self, bright_ratios: List[float]) -> List[bool]: """ Classify each oval as white (available) or dark (used). An oval is considered "white" (timeout available) if it has enough bright pixels above the threshold. Args: bright_ratios: List of bright pixel ratios for each oval Returns: List of booleans: True = white/available, False = dark/used """ return [ratio >= self.BRIGHT_PIXEL_RATIO_THRESHOLD for ratio in bright_ratios] def _count_available_timeouts(self, oval_states: List[bool]) -> int: """Count how many timeouts are available (white ovals).""" return sum(1 for state in oval_states if state) def read_timeouts(self, frame: np.ndarray[Any, Any]) -> TimeoutReading: """ Read the current timeout count for each team. Args: frame: Input frame (BGR format) Returns: TimeoutReading with current timeout counts """ if not self._configured: logger.warning("Timeout tracker not configured") return TimeoutReading(home_timeouts=3, away_timeouts=3, confidence=0.0) # Asserts: _configured guarantees regions are set assert self.home_region is not None assert self.away_region is not None # Read home team timeouts using bright pixel ratio home_bright_ratios = self._extract_oval_bright_ratios(frame, self.home_region) home_states = self._classify_ovals(home_bright_ratios) home_count = self._count_available_timeouts(home_states) # Read away team timeouts using bright pixel ratio away_bright_ratios = self._extract_oval_bright_ratios(frame, self.away_region) away_states = self._classify_ovals(away_bright_ratios) away_count = self._count_available_timeouts(away_states) # Calculate confidence based on how distinct the readings are all_ratios = home_bright_ratios + away_bright_ratios confidence = self._calculate_confidence(all_ratios) reading = TimeoutReading( home_timeouts=home_count, away_timeouts=away_count, confidence=confidence, home_oval_states=home_states, away_oval_states=away_states, ) logger.debug( "Timeout reading: home=%d (states=%s, ratios=%s), away=%d (states=%s, ratios=%s), conf=%.2f", home_count, home_states, [f"{r:.2f}" for r in home_bright_ratios], away_count, away_states, [f"{r:.2f}" for r in away_bright_ratios], confidence, ) return reading def _calculate_confidence(self, bright_ratios: List[float]) -> float: """ Calculate confidence based on how distinct the bright pixel ratios are. High confidence when ratios are clearly above or below threshold. Low confidence when ratios are near the threshold. """ if not bright_ratios: return 0.0 # Calculate distance from threshold for each ratio distances = [abs(r - self.BRIGHT_PIXEL_RATIO_THRESHOLD) for r in bright_ratios] # Average distance, normalized (threshold is 0.15, so max meaningful distance is ~0.85) avg_distance = sum(distances) / len(distances) confidence = min(1.0, avg_distance / 0.1) # 0.1 ratio distance = full confidence return confidence def detect_timeout_change(self, curr_reading: TimeoutReading) -> Optional[str]: """ Detect if a timeout was just called by comparing with previous reading. Args: curr_reading: Current timeout reading Returns: "home" if home team called timeout, "away" if away team did, None otherwise """ if self._prev_reading is None: self._prev_reading = curr_reading return None # Check for timeout decrement home_change = self._prev_reading.home_timeouts - curr_reading.home_timeouts away_change = self._prev_reading.away_timeouts - curr_reading.away_timeouts result = None if home_change > 0: logger.info("HOME team timeout detected: %d -> %d", self._prev_reading.home_timeouts, curr_reading.home_timeouts) result = "home" elif away_change > 0: logger.info("AWAY team timeout detected: %d -> %d", self._prev_reading.away_timeouts, curr_reading.away_timeouts) result = "away" self._prev_reading = curr_reading return result def update(self, frame: np.ndarray[Any, Any]) -> Tuple[TimeoutReading, Optional[str]]: """ Read timeouts and detect any change in one call. Args: frame: Input frame Returns: Tuple of (current reading, team that called timeout or None) """ reading = self.read_timeouts(frame) change = self.detect_timeout_change(reading) return reading, change def reset_tracking(self) -> None: """Reset the previous reading for fresh tracking.""" self._prev_reading = None logger.debug("Timeout tracking reset") def visualize(self, frame: np.ndarray[Any, Any], reading: Optional[TimeoutReading] = None) -> np.ndarray[Any, Any]: """ Draw timeout regions and states on frame for visualization. Args: frame: Input frame reading: Optional reading to display (if None, will read from frame) Returns: Frame with visualization overlay """ vis_frame = frame.copy() if not self._configured: cv2.putText(vis_frame, "Timeout tracker not configured", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 255), 2) return vis_frame if reading is None: reading = self.read_timeouts(frame) # Draw home team region if self.home_region: x, y, w, h = self.home_region.bbox cv2.rectangle(vis_frame, (x, y), (x + w, y + h), (255, 0, 0), 2) # Blue cv2.putText(vis_frame, f"HOME: {reading.home_timeouts}", (x, y - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 0, 0), 1) # Draw individual oval states if reading.home_oval_states: oval_width = w // 3 for i, state in enumerate(reading.home_oval_states): color = (0, 255, 0) if state else (0, 0, 255) # Green if available, red if used cx = x + i * oval_width + oval_width // 2 cy = y + h + 10 cv2.circle(vis_frame, (cx, cy), 5, color, -1) # Draw away team region if self.away_region: x, y, w, h = self.away_region.bbox cv2.rectangle(vis_frame, (x, y), (x + w, y + h), (0, 165, 255), 2) # Orange cv2.putText(vis_frame, f"AWAY: {reading.away_timeouts}", (x, y - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 165, 255), 1) # Draw individual oval states if reading.away_oval_states: oval_width = w // 3 for i, state in enumerate(reading.away_oval_states): color = (0, 255, 0) if state else (0, 0, 255) cx = x + i * oval_width + oval_width // 2 cy = y + h + 10 cv2.circle(vis_frame, (cx, cy), 5, color, -1) # Add confidence cv2.putText(vis_frame, f"Conf: {reading.confidence:.2f}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2) return vis_frame class CalibratedTimeoutDetector: """ Timeout detector using calibrated oval positions. This detector uses precise oval locations discovered during calibration, rather than dividing the region into equal parts. This provides more accurate timeout detection by checking brightness at the exact oval locations. """ # Brightness threshold ratio - oval is "dark" if below this fraction of baseline BRIGHTNESS_THRESHOLD_RATIO = 0.5 # Minimum brightness for an oval to be considered valid (scorebug visible) # If the brightest oval is below this, we consider the scorebug not visible # Calibrated ovals have baseline brightness ~185-195, so we need at least 120 # to confidently say the scorebug is visible with readable ovals MIN_VALID_BRIGHTNESS = 120 def __init__( self, home_region: Optional[CalibratedTimeoutRegion] = None, away_region: Optional[CalibratedTimeoutRegion] = None, config_path: Optional[str] = None, ): """ Initialize the calibrated timeout detector. Args: home_region: Calibrated region for home team's timeout indicators away_region: Calibrated region for away team's timeout indicators config_path: Path to JSON config file with calibrated regions """ self.home_region = home_region self.away_region = away_region self._configured = home_region is not None and away_region is not None # Previous reading for change detection self._prev_reading: Optional[TimeoutReading] = None # Load from config if provided if config_path and not self._configured: self._load_config(config_path) if self._configured: home_ovals = len(self.home_region.ovals) if self.home_region else 0 away_ovals = len(self.away_region.ovals) if self.away_region else 0 logger.info("CalibratedTimeoutDetector initialized: home=%d ovals, away=%d ovals", home_ovals, away_ovals) else: logger.info("CalibratedTimeoutDetector initialized (not configured - call calibrate first)") def _load_config(self, config_path: str) -> None: """Load calibrated timeout regions from a JSON config file.""" path = Path(config_path) if not path.exists(): logger.warning("Calibrated timeout config not found: %s", config_path) return with open(path, "r", encoding="utf-8") as f: data = json.load(f) # Check for calibrated regions (with ovals) if "home_timeout_region" in data and "ovals" in data["home_timeout_region"]: self.home_region = CalibratedTimeoutRegion.from_dict(data["home_timeout_region"]) if "away_timeout_region" in data and "ovals" in data["away_timeout_region"]: self.away_region = CalibratedTimeoutRegion.from_dict(data["away_timeout_region"]) self._configured = self.home_region is not None and self.away_region is not None if self._configured: logger.info("Loaded calibrated timeout config from: %s", config_path) def save_config(self, config_path: str) -> None: """Save calibrated timeout regions to a JSON config file.""" if not self._configured: logger.warning("Cannot save config - detector not calibrated") return data = {} if self.home_region: data["home_timeout_region"] = self.home_region.to_dict() if self.away_region: data["away_timeout_region"] = self.away_region.to_dict() path = Path(config_path) path.parent.mkdir(parents=True, exist_ok=True) with open(path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) logger.info("Saved calibrated timeout config to: %s", config_path) def is_configured(self) -> bool: """Check if the detector is configured with calibrated regions.""" return self._configured def set_regions(self, home_region: CalibratedTimeoutRegion, away_region: CalibratedTimeoutRegion) -> None: """ Set the calibrated timeout regions. Args: home_region: Calibrated region for home team away_region: Calibrated region for away team """ self.home_region = home_region self.away_region = away_region self._configured = True logger.info( "Calibrated regions set: home=%d ovals, away=%d ovals", len(home_region.ovals), len(away_region.ovals), ) def _check_oval_brightness(self, frame: np.ndarray[Any, Any], region: CalibratedTimeoutRegion, oval: OvalLocation) -> Tuple[bool, float]: """ Check if a specific oval is still bright (timeout available). Args: frame: Full video frame (BGR format) region: Calibrated region containing the oval oval: OvalLocation to check Returns: Tuple of (is_bright, current_brightness) """ rx, ry, _, _ = region.bbox # Calculate absolute position in frame abs_x = rx + oval.x abs_y = ry + oval.y # Validate bounds frame_h, frame_w = frame.shape[:2] if abs_x < 0 or abs_y < 0 or abs_x + oval.width > frame_w or abs_y + oval.height > frame_h: logger.warning("Oval position out of bounds: (%d, %d)", abs_x, abs_y) return False, 0.0 # Extract the oval region oval_roi = frame[abs_y : abs_y + oval.height, abs_x : abs_x + oval.width] # Convert to grayscale and calculate mean brightness gray = cv2.cvtColor(oval_roi, cv2.COLOR_BGR2GRAY) current_brightness = float(np.mean(np.asarray(gray))) # Compare to baseline threshold = oval.baseline_brightness * self.BRIGHTNESS_THRESHOLD_RATIO is_bright = current_brightness >= threshold return is_bright, current_brightness def _read_team_timeouts(self, frame: np.ndarray[Any, Any], region: CalibratedTimeoutRegion) -> Tuple[int, List[bool], float]: """ Read timeout count for a single team. Args: frame: Full video frame (BGR format) region: Calibrated region for the team Returns: Tuple of (timeout_count, oval_states, max_brightness) - timeout_count: Number of available timeouts (bright ovals) - oval_states: List of booleans for each oval - max_brightness: Maximum brightness across all ovals (for validity check) """ oval_states = [] max_brightness = 0.0 for oval in region.ovals: is_bright, brightness = self._check_oval_brightness(frame, region, oval) oval_states.append(is_bright) max_brightness = max(max_brightness, brightness) timeout_count = sum(1 for state in oval_states if state) return timeout_count, oval_states, max_brightness def read_timeouts(self, frame: np.ndarray[Any, Any]) -> TimeoutReading: """ Read the current timeout count for each team. Args: frame: Input frame (BGR format) Returns: TimeoutReading with current timeout counts Note: confidence=0.0 if scorebug appears not visible (all ovals too dark) """ if not self._configured: logger.warning("Calibrated timeout detector not configured") return TimeoutReading(home_timeouts=3, away_timeouts=3, confidence=0.0) assert self.home_region is not None assert self.away_region is not None # Read home team timeouts home_count, home_states, home_max_brightness = self._read_team_timeouts(frame, self.home_region) # Read away team timeouts away_count, away_states, away_max_brightness = self._read_team_timeouts(frame, self.away_region) # Check if scorebug is likely not visible # At least ONE region must have valid brightness to indicate scorebug is visible # A region with 0 timeouts will have all dark ovals (low brightness), which is valid # but a region showing end zone/players will have different patterns home_visible = home_max_brightness >= self.MIN_VALID_BRIGHTNESS away_visible = away_max_brightness >= self.MIN_VALID_BRIGHTNESS # Scorebug is visible if at least one team has bright ovals # (a team with 0 timeouts will have dark ovals but other team should be bright) scorebug_visible = home_visible or away_visible # Additional check: if BOTH regions are dark, scorebug is definitely not visible # But if only ONE is dark and the other is bright, the dark one likely has 0 timeouts if not scorebug_visible: logger.debug( "Scorebug appears not visible (home_max=%.1f, away_max=%.1f, threshold=%.1f)", home_max_brightness, away_max_brightness, self.MIN_VALID_BRIGHTNESS, ) confidence = 0.0 else: confidence = 1.0 if (len(home_states) == 3 and len(away_states) == 3) else 0.8 reading = TimeoutReading( home_timeouts=home_count, away_timeouts=away_count, confidence=confidence, home_oval_states=home_states, away_oval_states=away_states, ) logger.debug( "Calibrated timeout reading: home=%d (states=%s), away=%d (states=%s)", home_count, home_states, away_count, away_states, ) return reading def detect_timeout_change(self, curr_reading: TimeoutReading) -> Optional[str]: """ Detect if a timeout was just called by comparing with previous reading. Args: curr_reading: Current timeout reading Returns: "home" if home team called timeout, "away" if away team did, None otherwise """ # Skip low-confidence readings (scorebug not visible) if curr_reading.confidence < 0.5: logger.debug("Skipping timeout change detection - low confidence reading (%.2f)", curr_reading.confidence) return None if self._prev_reading is None: self._prev_reading = curr_reading return None # Skip if previous reading was low confidence if self._prev_reading.confidence < 0.5: self._prev_reading = curr_reading return None # Check for timeout decrement - must be exactly 1 decrease for one team home_change = self._prev_reading.home_timeouts - curr_reading.home_timeouts away_change = self._prev_reading.away_timeouts - curr_reading.away_timeouts result = None # Valid timeout: exactly one team decreases by exactly 1, other stays same if home_change == 1 and away_change == 0: logger.info("HOME team timeout detected: %d -> %d", self._prev_reading.home_timeouts, curr_reading.home_timeouts) result = "home" elif away_change == 1 and home_change == 0: logger.info("AWAY team timeout detected: %d -> %d", self._prev_reading.away_timeouts, curr_reading.away_timeouts) result = "away" elif home_change != 0 or away_change != 0: # Invalid pattern - both changed or changed by more than 1 logger.debug( "Invalid timeout pattern: home %d->%d (Δ%d), away %d->%d (Δ%d)", self._prev_reading.home_timeouts, curr_reading.home_timeouts, home_change, self._prev_reading.away_timeouts, curr_reading.away_timeouts, away_change, ) self._prev_reading = curr_reading return result def update(self, frame: np.ndarray[Any, Any]) -> Tuple[TimeoutReading, Optional[str]]: """ Read timeouts and detect any change in one call. Args: frame: Input frame Returns: Tuple of (current reading, team that called timeout or None) """ reading = self.read_timeouts(frame) change = self.detect_timeout_change(reading) return reading, change def reset_tracking(self) -> None: """Reset the previous reading for fresh tracking.""" self._prev_reading = None logger.debug("Calibrated timeout tracking reset") def visualize(self, frame: np.ndarray[Any, Any], reading: Optional[TimeoutReading] = None) -> np.ndarray[Any, Any]: """ Draw calibrated oval positions and states on frame for visualization. Args: frame: Input frame reading: Optional reading to display (if None, will read from frame) Returns: Frame with visualization overlay """ vis_frame = frame.copy() if not self._configured: cv2.putText(vis_frame, "Calibrated timeout detector not configured", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 255), 2) return vis_frame if reading is None: reading = self.read_timeouts(frame) # Draw home team region and ovals if self.home_region: rx, ry, rw, rh = self.home_region.bbox cv2.rectangle(vis_frame, (rx, ry), (rx + rw, ry + rh), (255, 0, 0), 2) cv2.putText(vis_frame, f"HOME: {reading.home_timeouts}", (rx, ry - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 0, 0), 1) # Draw each calibrated oval for i, oval in enumerate(self.home_region.ovals): abs_x = rx + oval.x abs_y = ry + oval.y state = reading.home_oval_states[i] if reading.home_oval_states and i < len(reading.home_oval_states) else True color = (0, 255, 0) if state else (0, 0, 255) cv2.rectangle(vis_frame, (abs_x, abs_y), (abs_x + oval.width, abs_y + oval.height), color, 2) # Draw away team region and ovals if self.away_region: rx, ry, rw, rh = self.away_region.bbox cv2.rectangle(vis_frame, (rx, ry), (rx + rw, ry + rh), (0, 165, 255), 2) cv2.putText(vis_frame, f"AWAY: {reading.away_timeouts}", (rx, ry - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 165, 255), 1) # Draw each calibrated oval for i, oval in enumerate(self.away_region.ovals): abs_x = rx + oval.x abs_y = ry + oval.y state = reading.away_oval_states[i] if reading.away_oval_states and i < len(reading.away_oval_states) else True color = (0, 255, 0) if state else (0, 0, 255) cv2.rectangle(vis_frame, (abs_x, abs_y), (abs_x + oval.width, abs_y + oval.height), color, 2) cv2.putText(vis_frame, f"Conf: {reading.confidence:.2f}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2) return vis_frame