File size: 8,957 Bytes
c6abe34 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 | """
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
|