Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |