File size: 10,097 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 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 | """
Defensive Reaction Engine - Analyzes defensive reaction times and closeout speeds.
For every offensive event (shot, pass, drive), measures how quickly the nearest
defender reacts and closes out.
"""
from typing import Dict, Any, List
import numpy as np
from .base import BaseAnalyticsModule
class DefensiveReactionEngine(BaseAnalyticsModule):
"""Analyzes defensive reaction times and closeout quality."""
def __init__(
self,
late_closeout_delay_ms: float = 500,
late_closeout_speed_mps: float = 4.0,
reaction_window_frames: int = 30, # ~1 second at 30fps
):
"""
Initialize defensive reaction engine.
Args:
late_closeout_delay_ms: Reaction delay threshold for "late" classification
late_closeout_speed_mps: Speed threshold for "late" classification
reaction_window_frames: Number of frames to analyze after event
"""
super().__init__("defensive_reaction")
self.late_delay_threshold = late_closeout_delay_ms
self.late_speed_threshold = late_closeout_speed_mps
self.reaction_window = reaction_window_frames
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]:
"""
Analyze defensive reactions for all offensive events.
Returns:
Dictionary with defensive reaction metrics for each event
"""
defensive_reactions = []
# Analyze shots (primary offensive events)
for shot in shots:
frame = shot.get("start_frame", shot.get("frame", 0))
if frame >= len(player_assignment) or frame >= len(tactical_positions):
continue
assignment = player_assignment[frame]
tactical_pos = tactical_positions[frame]
# Get shooter info
shooter_id = shot.get("player_id")
if not shooter_id or shooter_id not in assignment:
continue
offense_team = assignment[shooter_id]
defense_team = 1 if offense_team == 2 else 2
# Find nearest defender
shooter_pos = tactical_pos.get(shooter_id)
if not shooter_pos:
continue
nearest_defender = None
min_distance = float('inf')
for player_id, team_id in assignment.items():
if team_id == defense_team and player_id in tactical_pos:
defender_pos = tactical_pos[player_id]
dist = self._euclidean_distance(shooter_pos, defender_pos)
if dist < min_distance:
min_distance = dist
nearest_defender = player_id
if not nearest_defender:
continue
# Measure reaction (simplified: check if defender moved toward shooter)
reaction_metrics = self._measure_reaction(
frame,
nearest_defender,
shooter_id,
tactical_positions,
speeds,
fps
)
defensive_reactions.append({
"event_id": f"shot_{frame}",
"event_type": "shot",
"event_frame": frame,
"defender_track_id": nearest_defender,
"offensive_player_track_id": shooter_id,
"distance_at_event": float(min_distance),
**reaction_metrics
})
# Analyze passes
for event in events:
if event.get("event_type") != "pass":
continue
frame = event.get("frame", 0)
if frame >= len(player_assignment) or frame >= len(tactical_positions):
continue
assignment = player_assignment[frame]
tactical_pos = tactical_positions[frame]
passer_id = event.get("player_id")
if not passer_id or passer_id not in assignment:
continue
offense_team = assignment[passer_id]
defense_team = 1 if offense_team == 2 else 2
passer_pos = tactical_pos.get(passer_id)
if not passer_pos:
continue
# Find nearest defender
nearest_defender = None
min_distance = float('inf')
for player_id, team_id in assignment.items():
if team_id == defense_team and player_id in tactical_pos:
defender_pos = tactical_pos[player_id]
dist = self._euclidean_distance(passer_pos, defender_pos)
if dist < min_distance:
min_distance = dist
nearest_defender = player_id
if not nearest_defender:
continue
reaction_metrics = self._measure_reaction(
frame,
nearest_defender,
passer_id,
tactical_positions,
speeds,
fps
)
defensive_reactions.append({
"event_id": f"pass_{frame}",
"event_type": "pass",
"event_frame": frame,
"defender_track_id": nearest_defender,
"offensive_player_track_id": passer_id,
"distance_at_event": float(min_distance),
**reaction_metrics
})
# Calculate summary statistics
if defensive_reactions:
late_closeouts = sum(1 for r in defensive_reactions if r.get("late_closeout", False))
avg_reaction_delay = np.mean([
r["reaction_delay_ms"] for r in defensive_reactions
if r.get("reaction_delay_ms") is not None
])
avg_closeout_speed = np.mean([
r["closeout_speed_mps"] for r in defensive_reactions
if r.get("closeout_speed_mps") is not None
])
summary = {
"total_defensive_events": len(defensive_reactions),
"late_closeouts": late_closeouts,
"late_closeout_rate": (late_closeouts / len(defensive_reactions) * 100),
"avg_reaction_delay_ms": float(avg_reaction_delay) if not np.isnan(avg_reaction_delay) else None,
"avg_closeout_speed_mps": float(avg_closeout_speed) if not np.isnan(avg_closeout_speed) else None,
}
else:
summary = {
"total_defensive_events": 0,
"late_closeouts": 0,
"late_closeout_rate": 0,
"avg_reaction_delay_ms": None,
"avg_closeout_speed_mps": None,
}
return {
"defensive_reactions": defensive_reactions,
"summary": summary,
"status": "success"
}
def _measure_reaction(
self,
event_frame: int,
defender_id: int,
offensive_player_id: int,
tactical_positions: List[Dict],
speeds: List[Dict],
fps: float
) -> Dict[str, Any]:
"""
Measure defender's reaction to an offensive event.
Args:
event_frame: Frame when event occurred
defender_id: Track ID of defender
offensive_player_id: Track ID of offensive player
tactical_positions: All tactical positions
speeds: All speed data
fps: Frames per second
Returns:
Dictionary with reaction metrics
"""
# Simplified reaction measurement
# In full implementation, would track defender movement vector toward offensive player
reaction_start_frame = None
max_speed = 0.0
# Look at next N frames after event
for offset in range(1, min(self.reaction_window, len(tactical_positions) - event_frame)):
frame_idx = event_frame + offset
if frame_idx >= len(speeds):
break
# Get defender speed in this frame
frame_speeds = speeds[frame_idx]
if defender_id in frame_speeds:
speed = frame_speeds[defender_id]
if speed > max_speed:
max_speed = speed
# Detect reaction start (speed increase)
if speed > 3.0 and reaction_start_frame is None: # 3 m/s threshold
reaction_start_frame = frame_idx
# Calculate metrics
if reaction_start_frame:
reaction_delay_frames = reaction_start_frame - event_frame
reaction_delay_ms = (reaction_delay_frames / fps) * 1000 if fps > 0 else 0
else:
reaction_delay_ms = None
closeout_speed_mps = max_speed if max_speed > 0 else None
# Determine if late closeout
late_closeout = False
if reaction_delay_ms and reaction_delay_ms > self.late_delay_threshold:
late_closeout = True
if closeout_speed_mps and closeout_speed_mps < self.late_speed_threshold:
late_closeout = True
return {
"reaction_start_frame": reaction_start_frame,
"reaction_delay_ms": reaction_delay_ms,
"closeout_speed_mps": closeout_speed_mps,
"late_closeout": late_closeout
}
|