analytics_engine/fatigue_tracker.py

"""
Fatigue Tracker - Monitors player fatigue indicators over time.

Tracks speed decline and reaction delay increase to identify fatigue patterns.
"""
from typing import Dict, Any, List
import numpy as np
from .base import BaseAnalyticsModule


class FatigueTracker(BaseAnalyticsModule):
    """Tracks player fatigue indicators using speed and reaction metrics."""
    
    def __init__(
        self,
        time_window_minutes: float = 2.0,
        baseline_window_minutes: float = 3.0,
        low_fatigue_threshold: float = 5.0,
        medium_fatigue_threshold: float = 15.0,
    ):
        """
        Initialize fatigue tracker.
        
        Args:
            time_window_minutes: Size of rolling time window for analysis
            baseline_window_minutes: Duration of early-game baseline period
            low_fatigue_threshold: Percentage decline for low fatigue
            medium_fatigue_threshold: Percentage decline for medium fatigue
        """
        super().__init__("fatigue_tracker")
        self.time_window_minutes = time_window_minutes
        self.baseline_window_minutes = baseline_window_minutes
        self.low_threshold = low_fatigue_threshold
        self.medium_threshold = medium_fatigue_threshold
    
    def process(
        self,
        video_frames: List[Any],
        player_tracks: List[Dict],
        ball_tracks: List[Dict],
        tactical_positions: List[Dict],
        player_assignment: List[Dict],
        ball_possession: List[int],
        events: List[Dict],
        shots: List[Dict],
        court_keypoints: List[Dict],
        speeds: List[Dict],
        video_path: str,
        fps: float,
        **kwargs
    ) -> Dict[str, Any]:
        """
        Track fatigue indicators for all players over time.
        
        Returns:
            Dictionary with fatigue metrics per player per time window
        """
        defensive_reactions = kwargs.get("defensive_reactions", [])
        
        fatigue_indices = []
        
        # Get all unique players
        all_players = set()
        for assignment in player_assignment:
            all_players.update(assignment.keys())
        
        # Calculate baseline metrics (first N minutes)
        baseline_frames = int(self.baseline_window_minutes * 60 * fps)
        baseline_metrics = self._calculate_baseline_metrics(
            all_players,
            speeds[:baseline_frames],
            defensive_reactions,
            baseline_frames
        )
        
        # Analyze in time windows
        window_frames = int(self.time_window_minutes * 60 * fps)
        total_frames = len(speeds)
        
        for window_start in range(0, total_frames, window_frames):
            window_end = min(window_start + window_frames, total_frames)
            window_minute = int((window_start / fps) / 60)
            
            # Skip if this is the baseline window
            if window_start < baseline_frames:
                continue
            
            window_speeds = speeds[window_start:window_end]
            
            for player_id in all_players:
                # Calculate current window metrics
                player_speeds = []
                for frame_speeds in window_speeds:
                    if player_id in frame_speeds:
                        player_speeds.append(frame_speeds[player_id])
                
                if not player_speeds:
                    continue
                
                current_avg_speed = np.mean(player_speeds)
                
                # Get baseline for this player
                baseline_speed = baseline_metrics.get(player_id, {}).get("avg_speed", 0)
                baseline_reaction = baseline_metrics.get(player_id, {}).get("avg_reaction_ms")
                
                if baseline_speed == 0:
                    continue
                
                # Calculate speed decline
                speed_drop_pct = ((baseline_speed - current_avg_speed) / baseline_speed) * 100
                
                # Calculate reaction delay increase (if available)
                window_reactions = [
                    r for r in defensive_reactions
                    if r.get("defender_track_id") == player_id
                    and window_start <= r.get("event_frame", 0) < window_end
                    and r.get("reaction_delay_ms") is not None
                ]
                
                if window_reactions and baseline_reaction:
                    current_reaction = np.mean([r["reaction_delay_ms"] for r in window_reactions])
                    reaction_increase_pct = ((current_reaction - baseline_reaction) / baseline_reaction) * 100
                else:
                    current_reaction = None
                    reaction_increase_pct = None
                
                # Classify fatigue level
                if speed_drop_pct < self.low_threshold:
                    fatigue_level = "low"
                elif speed_drop_pct < self.medium_threshold:
                    fatigue_level = "medium"
                else:
                    fatigue_level = "high"
                
                fatigue_indices.append({
                    "player_track_id": player_id,
                    "time_window_start": window_start / fps if fps > 0 else 0,
                    "time_window_end": window_end / fps if fps > 0 else 0,
                    "minute": window_minute,
                    "baseline_speed_mps": float(baseline_speed),
                    "current_speed_mps": float(current_avg_speed),
                    "speed_drop_percentage": float(speed_drop_pct),
                    "baseline_reaction_ms": float(baseline_reaction) if baseline_reaction else None,
                    "current_reaction_ms": float(current_reaction) if current_reaction else None,
                    "reaction_delay_increase_percentage": float(reaction_increase_pct) if reaction_increase_pct else None,
                    "fatigue_level": fatigue_level
                })
        
        # Calculate summary statistics
        if fatigue_indices:
            high_fatigue_count = sum(1 for f in fatigue_indices if f["fatigue_level"] == "high")
            medium_fatigue_count = sum(1 for f in fatigue_indices if f["fatigue_level"] == "medium")
            
            summary = {
                "total_measurements": len(fatigue_indices),
                "high_fatigue_instances": high_fatigue_count,
                "medium_fatigue_instances": medium_fatigue_count,
                "avg_speed_drop_pct": float(np.mean([f["speed_drop_percentage"] for f in fatigue_indices])),
                "max_speed_drop_pct": float(max(f["speed_drop_percentage"] for f in fatigue_indices)),
            }
        else:
            summary = {
                "total_measurements": 0,
                "high_fatigue_instances": 0,
                "medium_fatigue_instances": 0,
                "avg_speed_drop_pct": 0,
                "max_speed_drop_pct": 0,
            }
        
        return {
            "fatigue_indices": fatigue_indices,
            "summary": summary,
            "status": "success"
        }
    
    def _calculate_baseline_metrics(
        self,
        players: set,
        baseline_speeds: List[Dict],
        defensive_reactions: List[Dict],
        baseline_frames: int
    ) -> Dict[int, Dict[str, float]]:
        """
        Calculate baseline metrics for each player from early game.
        
        Args:
            players: Set of player IDs
            baseline_speeds: Speed data from baseline period
            defensive_reactions: All defensive reaction data
            baseline_frames: Number of frames in baseline
        
        Returns:
            Dictionary mapping player_id to baseline metrics
        """
        baseline_metrics = {}
        
        for player_id in players:
            # Collect speeds
            player_speeds = []
            for frame_speeds in baseline_speeds:
                if player_id in frame_speeds:
                    player_speeds.append(frame_speeds[player_id])
            
            if player_speeds:
                avg_speed = np.mean(player_speeds)
            else:
                avg_speed = 0
            
            # Collect reaction times
            player_reactions = [
                r["reaction_delay_ms"]
                for r in defensive_reactions
                if r.get("defender_track_id") == player_id
                and r.get("event_frame", 0) < baseline_frames
                and r.get("reaction_delay_ms") is not None
            ]
            
            if player_reactions:
                avg_reaction = np.mean(player_reactions)
            else:
                avg_reaction = None
            
            baseline_metrics[player_id] = {
                "avg_speed": avg_speed,
                "avg_reaction_ms": avg_reaction
            }
        
        return baseline_metrics