ADVANCED_FEATURES.md

#!/usr/bin/env python3
"""
RSK World
Founder: Molla Samser
Designer & Tester: Rima Khatun
Email: help@rskworld.in
Phone: +91 93305 39277
Address: Nutanhat, Mongolkote, Purba Burdwan, West Bengal, India, 713147
Website: https://rskworld.in/

Advanced Statistics Calculator for Sports Analysis Dataset
Calculates comprehensive statistics from tracking and performance data
"""

import json
import csv
from pathlib import Path
from collections import defaultdict
from statistics import mean, median, stdev
from typing import Dict, List, Any

class SportsStatisticsCalculator:
    """
    Advanced statistics calculator for sports analytics
    """
    
    def __init__(self, data_dir: str = '../data'):
        """
        Initialize the calculator with data directory
        
        Args:
            data_dir: Path to data directory
        """
        self.data_dir = Path(data_dir)
        self.tracking_data = {}
        self.events_data = {}
        self.performance_data = {}
    
    def load_tracking_data(self, game_id: str) -> Dict:
        """
        Load tracking data for a game
        
        Args:
            game_id: Game identifier
            
        Returns:
            Dictionary containing tracking data
        """
        file_path = self.data_dir / 'annotations' / f'tracking_{game_id.split("_")[1]}.json'
        
        if file_path.exists():
            with open(file_path, 'r') as f:
                data = json.load(f)
                self.tracking_data[game_id] = data
                return data
        return {}
    
    def load_events_data(self, game_id: str) -> Dict:
        """
        Load events data for a game
        
        Args:
            game_id: Game identifier
            
        Returns:
            Dictionary containing events data
        """
        file_path = self.data_dir / 'annotations' / f'events_{game_id.split("_")[1]}.json'
        
        if file_path.exists():
            with open(file_path, 'r') as f:
                data = json.load(f)
                self.events_data[game_id] = data
                return data
        return {}
    
    def load_performance_data(self) -> List[Dict]:
        """
        Load performance metrics CSV file
        
        Returns:
            List of dictionaries containing performance data
        """
        file_path = self.data_dir / 'annotations' / 'performance_metrics_001.csv'
        
        if file_path.exists():
            with open(file_path, 'r') as f:
                reader = csv.DictReader(f)
                self.performance_data = list(reader)
                return self.performance_data
        return []
    
    def calculate_player_distances(self, game_id: str) -> Dict[str, float]:
        """
        Calculate total distance covered by each player
        
        Args:
            game_id: Game identifier
            
        Returns:
            Dictionary mapping player_id to total distance
        """
        tracking = self.tracking_data.get(game_id, {})
        players = tracking.get('players', [])
        
        distances = defaultdict(float)
        
        for i in range(len(players) - 1):
            if players[i]['player_id'] == players[i + 1]['player_id']:
                p1 = players[i]['position']
                p2 = players[i + 1]['position']
                
                # Calculate Euclidean distance
                distance = ((p2['x'] - p1['x'])**2 + (p2['y'] - p1['y'])**2)**0.5
                
                # Convert pixels to meters (assuming scale)
                distance_meters = distance * 0.1  # Adjust scale as needed
                
                distances[players[i]['player_id']] += distance_meters
        
        return dict(distances)
    
    def calculate_average_speed(self, game_id: str) -> Dict[str, float]:
        """
        Calculate average speed for each player
        
        Args:
            game_id: Game identifier
            
        Returns:
            Dictionary mapping player_id to average speed (m/s)
        """
        tracking = self.tracking_data.get(game_id, {})
        players = tracking.get('players', [])
        
        speeds = defaultdict(list)
        
        for i in range(len(players) - 1):
            if players[i]['player_id'] == players[i + 1]['player_id']:
                p1 = players[i]
                p2 = players[i + 1]
                
                time_diff = p2['timestamp'] - p1['timestamp']
                if time_diff > 0:
                    p1_pos = p1['position']
                    p2_pos = p2['position']
                    
                    distance = ((p2_pos['x'] - p1_pos['x'])**2 + 
                               (p2_pos['y'] - p1_pos['y'])**2)**0.5 * 0.1
                    
                    speed = distance / time_diff  # m/s
                    speeds[p1['player_id']].append(speed)
        
        # Calculate average speeds
        avg_speeds = {pid: mean(speed_list) for pid, speed_list in speeds.items()}
        return avg_speeds
    
    def calculate_event_statistics(self, game_id: str) -> Dict[str, Any]:
        """
        Calculate statistics about events in a game
        
        Args:
            game_id: Game identifier
            
        Returns:
            Dictionary containing event statistics
        """
        events_data = self.events_data.get(game_id, {})
        events = events_data.get('events', [])
        
        stats = {
            'total_events': len(events),
            'events_by_type': defaultdict(int),
            'events_by_player': defaultdict(int),
            'events_by_team': defaultdict(int),
            'time_intervals': []
        }
        
        for event in events:
            stats['events_by_type'][event['event_type']] += 1
            
            if 'player_id' in event and event['player_id']:
                stats['events_by_player'][event['player_id']] += 1
            
            if 'team' in event and event['team']:
                stats['events_by_team'][event['team']] += 1
            
            if 'timestamp' in event:
                stats['time_intervals'].append(event['timestamp'])
        
        stats['events_by_type'] = dict(stats['events_by_type'])
        stats['events_by_player'] = dict(stats['events_by_player'])
        stats['events_by_team'] = dict(stats['events_by_team'])
        
        if stats['time_intervals']:
            stats['average_time_between_events'] = (
                (max(stats['time_intervals']) - min(stats['time_intervals'])) / 
                len(stats['time_intervals']) if len(stats['time_intervals']) > 1 else 0
            )
        else:
            stats['average_time_between_events'] = 0
        
        return stats
    
    def calculate_performance_statistics(self) -> Dict[str, Any]:
        """
        Calculate statistics from performance metrics
        
        Returns:
            Dictionary containing performance statistics
        """
        if not self.performance_data:
            return {}
        
        stats = {
            'total_players': len(self.performance_data),
            'average_distance': 0,
            'average_speed': 0,
            'max_distance': 0,
            'max_speed': 0,
            'team_statistics': defaultdict(lambda: {
                'players': 0,
                'total_distance': 0,
                'average_speed': []
            })
        }
        
        distances = []
        speeds = []
        
        for player in self.performance_data:
            distance = float(player.get('distance_covered_km', 0))
            speed = float(player.get('speed_avg_kmh', 0))
            
            distances.append(distance)
            speeds.append(speed)
            
            stats['max_distance'] = max(stats['max_distance'], distance)
            stats['max_speed'] = max(stats['max_speed'], speed)
            
            team = player.get('team', 'Unknown')
            stats['team_statistics'][team]['players'] += 1
            stats['team_statistics'][team]['total_distance'] += distance
            stats['team_statistics'][team]['average_speed'].append(speed)
        
        stats['average_distance'] = mean(distances) if distances else 0
        stats['average_speed'] = mean(speeds) if speeds else 0
        stats['team_statistics'] = {
            team: {
                'players': data['players'],
                'total_distance': data['total_distance'],
                'average_speed': mean(data['average_speed']) if data['average_speed'] else 0
            }
            for team, data in stats['team_statistics'].items()
        }
        
        return stats
    
    def generate_comprehensive_report(self, game_id: str) -> Dict[str, Any]:
        """
        Generate a comprehensive statistics report for a game
        
        Args:
            game_id: Game identifier
            
        Returns:
            Dictionary containing comprehensive statistics
        """
        self.load_tracking_data(game_id)
        self.load_events_data(game_id)
        self.load_performance_data()
        
        report = {
            'game_id': game_id,
            'tracking_statistics': {
                'player_distances': self.calculate_player_distances(game_id),
                'average_speeds': self.calculate_average_speed(game_id)
            },
            'event_statistics': self.calculate_event_statistics(game_id),
            'performance_statistics': self.calculate_performance_statistics()
        }
        
        return report
    
    def export_statistics(self, report: Dict[str, Any], output_file: str):
        """
        Export statistics report to JSON file
        
        Args:
            report: Statistics report dictionary
            output_file: Output file path
        """
        with open(output_file, 'w') as f:
            json.dump(report, f, indent=2)
        
        print(f"Statistics report exported to {output_file}")


def main():
    """
    Main function to demonstrate statistics calculator usage
    """
    calculator = SportsStatisticsCalculator()
    
    print("Sports Analysis Dataset - Statistics Calculator")
    print("=" * 50)
    
    # Load data
    print("\nLoading data...")
    calculator.load_tracking_data('game_001')
    calculator.load_events_data('game_001')
    calculator.load_performance_data()
    
    # Calculate statistics
    print("\nCalculating statistics...")
    report = calculator.generate_comprehensive_report('game_001')
    
    # Display results
    print("\n=== Statistics Report ===")
    print(f"\nGame ID: {report['game_id']}")
    
    print("\n--- Event Statistics ---")
    event_stats = report['event_statistics']
    print(f"Total Events: {event_stats.get('total_events', 0)}")
    print(f"Events by Type: {event_stats.get('events_by_type', {})}")
    
    print("\n--- Performance Statistics ---")
    perf_stats = report['performance_statistics']
    print(f"Total Players: {perf_stats.get('total_players', 0)}")
    print(f"Average Distance: {perf_stats.get('average_distance', 0):.2f} km")
    print(f"Average Speed: {perf_stats.get('average_speed', 0):.2f} km/h")
    
    # Export report
    calculator.export_statistics(report, 'statistics_report.json')
    print("\nReport generation complete!")


if __name__ == '__main__':
    main()