analytics_engine/lineup_impact.py

"""
Lineup Impact Engine - Analyzes performance of specific 5-player combinations.

Segments game by unique lineups and calculates offensive/defensive ratings,
spacing quality, and error rates for each combination.
"""
from typing import Dict, Any, List
import numpy as np
from .base import BaseAnalyticsModule


class LineupImpactEngine(BaseAnalyticsModule):
    """Analyzes performance metrics for specific player combinations."""
    
    def __init__(self):
        """Initialize lineup impact engine."""
        super().__init__("lineup_impact")
    
    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 lineup performance across the game.
        
        Returns:
            Dictionary with lineup metrics
        """
        # Get spacing metrics and defensive reactions from kwargs if available
        spacing_metrics = kwargs.get("spacing_metrics", [])
        defensive_reactions = kwargs.get("defensive_reactions", [])
        
        # Detect lineup segments
        lineup_segments = self._detect_lineup_segments(
            player_assignment,
            ball_possession,
            fps
        )
        
        lineup_metrics = []
        
        for segment in lineup_segments:
            team_id = segment["team_id"]
            lineup_hash = segment["lineup_hash"]
            player_ids = segment["player_ids"]
            start_frame = segment["start_frame"]
            end_frame = segment["end_frame"]
            
            # Calculate metrics for this lineup segment
            segment_shots = [
                s for s in shots
                if start_frame <= s.get("start_frame", s.get("frame", 0)) < end_frame
                and s.get("team") == team_id
            ]
            
            segment_shots_allowed = [
                s for s in shots
                if start_frame <= s.get("start_frame", s.get("frame", 0)) < end_frame
                and s.get("team") != team_id
            ]
            
            # Points scored (simplified: 2 points per made shot)
            points_scored = sum(2 for s in segment_shots if s.get("outcome") == "made")
            points_allowed = sum(2 for s in segment_shots_allowed if s.get("outcome") == "made")
            
            # Possessions (estimate from shots + turnovers)
            possessions = len(segment_shots)  # Simplified
            
            # Offensive/defensive ratings (per 100 possessions)
            if possessions > 0:
                offensive_rating = (points_scored / possessions) * 100
                defensive_rating = (points_allowed / possessions) * 100
                net_rating = offensive_rating - defensive_rating
            else:
                offensive_rating = 0
                defensive_rating = 0
                net_rating = 0
            
            # Average spacing score for this lineup
            segment_spacing = [
                m for m in spacing_metrics
                if start_frame <= m.get("frame", 0) < end_frame
            ]
            
            if segment_spacing:
                # Convert quality to numeric score
                quality_scores = []
                for m in segment_spacing:
                    if m["spacing_quality"] == "good":
                        quality_scores.append(3.0)
                    elif m["spacing_quality"] == "average":
                        quality_scores.append(2.0)
                    else:
                        quality_scores.append(1.0)
                avg_spacing_score = float(np.mean(quality_scores))
            else:
                avg_spacing_score = 0.0
            
            # Defensive error rate (late closeouts / total defensive events)
            segment_def_reactions = [
                r for r in defensive_reactions
                if start_frame <= r.get("event_frame", 0) < end_frame
            ]
            
            if segment_def_reactions:
                late_closeouts = sum(1 for r in segment_def_reactions if r.get("late_closeout", False))
                defensive_error_rate = late_closeouts / len(segment_def_reactions)
            else:
                defensive_error_rate = 0.0
            
            # Turnovers (from interceptions)
            segment_interceptions = [
                e for e in events
                if e.get("event_type") == "interception"
                and start_frame <= e.get("frame", 0) < end_frame
            ]
            turnovers = len(segment_interceptions)
            
            # Duration
            duration_seconds = (end_frame - start_frame) / fps if fps > 0 else 0
            duration_minutes = duration_seconds / 60
            
            lineup_metrics.append({
                "team_id": team_id,
                "lineup_hash": lineup_hash,
                "player_track_ids": player_ids,
                "possessions_count": possessions,
                "points_scored": points_scored,
                "points_allowed": points_allowed,
                "offensive_rating": float(offensive_rating),
                "defensive_rating": float(defensive_rating),
                "net_rating": float(net_rating),
                "avg_spacing_score": avg_spacing_score,
                "turnovers": turnovers,
                "defensive_error_rate": float(defensive_error_rate),
                "total_minutes": float(duration_minutes),
                "start_frame": start_frame,
                "end_frame": end_frame
            })
        
        # Calculate summary statistics
        if lineup_metrics:
            best_lineup = max(lineup_metrics, key=lambda x: x["net_rating"])
            worst_lineup = min(lineup_metrics, key=lambda x: x["net_rating"])
            
            summary = {
                "total_lineups": len(lineup_metrics),
                "best_lineup_hash": best_lineup["lineup_hash"],
                "best_lineup_net_rating": best_lineup["net_rating"],
                "worst_lineup_hash": worst_lineup["lineup_hash"],
                "worst_lineup_net_rating": worst_lineup["net_rating"],
                "avg_net_rating": float(np.mean([m["net_rating"] for m in lineup_metrics])),
            }
        else:
            summary = {
                "total_lineups": 0,
                "best_lineup_hash": None,
                "best_lineup_net_rating": 0,
                "worst_lineup_hash": None,
                "worst_lineup_net_rating": 0,
                "avg_net_rating": 0,
            }
        
        return {
            "lineup_metrics": lineup_metrics,
            "summary": summary,
            "status": "success"
        }
    
    def _detect_lineup_segments(
        self,
        player_assignment: List[Dict],
        ball_possession: List[int],
        fps: float,
        min_duration_seconds: float = 30.0
    ) -> List[Dict]:
        """
        Detect continuous segments with the same 5-player lineup.
        
        Args:
            player_assignment: Per-frame team assignments
            ball_possession: Per-frame possession
            fps: Frames per second
            min_duration_seconds: Minimum segment duration to track
        
        Returns:
            List of lineup segments
        """
        segments = []
        current_lineups = {1: None, 2: None}  # Track lineup for each team
        segment_starts = {1: 0, 2: 0}
        
        min_frames = int(min_duration_seconds * fps)
        
        for frame_idx in range(len(player_assignment)):
            assignment = player_assignment[frame_idx]
            
            # Get players for each team
            team_1_players = sorted([pid for pid, team in assignment.items() if team == 1])
            team_2_players = sorted([pid for pid, team in assignment.items() if team == 2])
            
            # Create lineup hashes
            lineup_1_hash = "_".join(map(str, team_1_players)) if len(team_1_players) >= 3 else None
            lineup_2_hash = "_".join(map(str, team_2_players)) if len(team_2_players) >= 3 else None
            
            # Check for lineup changes
            for team_id, lineup_hash, players in [
                (1, lineup_1_hash, team_1_players),
                (2, lineup_2_hash, team_2_players)
            ]:
                if lineup_hash is None:
                    continue
                
                if current_lineups[team_id] != lineup_hash:
                    # Lineup changed
                    if current_lineups[team_id] is not None:
                        # Save previous segment if long enough
                        duration = frame_idx - segment_starts[team_id]
                        if duration >= min_frames:
                            segments.append({
                                "team_id": team_id,
                                "lineup_hash": current_lineups[team_id],
                                "player_ids": self._parse_lineup_hash(current_lineups[team_id]),
                                "start_frame": segment_starts[team_id],
                                "end_frame": frame_idx
                            })
                    
                    # Start new segment
                    current_lineups[team_id] = lineup_hash
                    segment_starts[team_id] = frame_idx
        
        # Close final segments
        for team_id in [1, 2]:
            if current_lineups[team_id] is not None:
                duration = len(player_assignment) - segment_starts[team_id]
                if duration >= min_frames:
                    segments.append({
                        "team_id": team_id,
                        "lineup_hash": current_lineups[team_id],
                        "player_ids": self._parse_lineup_hash(current_lineups[team_id]),
                        "start_frame": segment_starts[team_id],
                        "end_frame": len(player_assignment)
                    })
        
        return segments
    
    def _parse_lineup_hash(self, lineup_hash: str) -> List[int]:
        """Parse lineup hash back to list of player IDs."""
        return [int(x) for x in lineup_hash.split("_")]