From 0cb1cbceed59a2a72697a022e41627fecc3f7994 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Thu, 22 Jan 2026 13:18:22 +0000 Subject: [PATCH] Review code, fix missing UtilityWasteDetector logic, and ensure all tests pass. - Deleted duplicate/incorrect `src/main.py` - Implemented `UtilityWasteDetector.detect` in `src/mistakes/detectors.py` to identify wasted flashes. - Updated `src/parser/demo_parser.py` to extract player name and team for grenade events. - Added unit tests for `UtilityWasteDetector` including handling of missing round columns. - Verified all 177 tests pass. --- src/main.py | 78 -------------------------------- src/mistakes/detectors.py | 88 ++++++++++++++++++++++++++++++++++-- src/parser/demo_parser.py | 6 +-- tests/test_mistakes.py | 95 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 182 insertions(+), 85 deletions(-) delete mode 100644 src/main.py diff --git a/src/main.py b/src/main.py deleted file mode 100644 index 7dca3e3..0000000 --- a/src/main.py +++ /dev/null @@ -1,78 +0,0 @@ -# Copyright (c) 2026 Pl4yer-ONE -# This file is part of FragAudit. -# Licensed under GPLv3 or commercial license. - -import argparse -from pathlib import Path -import sys -from concurrent.futures import ThreadPoolExecutor - -# Add project root to path -project_root = Path(__file__).parent.parent -sys.path.append(str(project_root)) - -from src.pipeline.analyzer import MatchAnalyzer - -def main(): - parser = argparse.ArgumentParser(description="CS2 AI Coach - Production Analytics Engine") - parser.add_argument("input_path", type=str, help="Path to .dem file or directory containing .dem files") - parser.add_argument("--output", type=str, default="outputs", help="Output directory") - parser.add_argument("--analyze-movement", action="store_true", help="Enable computationally expensive movement analysis") - parser.add_argument("--workers", type=int, default=1, help="Number of parallel workers for batch processing") - - args = parser.parse_args() - - input_path = Path(args.input_path) - if not input_path.exists(): - print(f"Error: Input path '{input_path}' does not exist.") - sys.exit(1) - - analyzer = MatchAnalyzer(output_dir=args.output) - - demos_to_process = [] - if input_path.is_file(): - if input_path.suffix == ".dem": - demos_to_process.append(input_path) - else: - print("Error: Input file must be a .dem file") - sys.exit(1) - elif input_path.is_dir(): - print(f"Scanning directory: {input_path}") - demos_to_process = list(input_path.glob("*.dem")) - print(f"Found {len(demos_to_process)} demo files.") - - if not demos_to_process: - print("No demo files found to process.") - sys.exit(0) - - # Process - print(f"Starting batch analysis of {len(demos_to_process)} demos with {args.workers} workers...") - - successful = 0 - failed = 0 - - # Use ThreadPoolExecutor for batch processing (CPU bound mostly, but I/O heavy parser) - # ProcessPool might be better but serialization is tricky with pandas objects. - # Sequential for safety first, or ThreadPool. - - if args.workers > 1 and len(demos_to_process) > 1: - with ThreadPoolExecutor(max_workers=args.workers) as executor: - futures = {executor.submit(analyzer.analyze_match, demo, args.analyze_movement): demo for demo in demos_to_process} - for future in futures: - res = future.result() - if res["status"] == "success": - successful += 1 - else: - failed += 1 - else: - for demo in demos_to_process: - res = analyzer.analyze_match(demo, args.analyze_movement) - if res["status"] == "success": - successful += 1 - else: - failed += 1 - - print(f"\n Batch Complete. Success: {successful}, Failed: {failed}") - -if __name__ == "__main__": - main() diff --git a/src/mistakes/detectors.py b/src/mistakes/detectors.py index 72b5167..485f611 100644 --- a/src/mistakes/detectors.py +++ b/src/mistakes/detectors.py @@ -19,6 +19,7 @@ from typing import List, Dict, Any, Optional from enum import Enum import math +import pandas as pd # Import contextual WPA for weighted calculations try: @@ -338,14 +339,93 @@ def detect(self, parsed_demo, round_num: int) -> List[DetectedMistake]: # Check for flash assists or thrown flashes # Simplified: if flash thrown and no kill within window, flag as waste grenades = getattr(parsed_demo, 'grenades', None) - if grenades is None: + if grenades is None or grenades.empty: return mistakes # Filter flashes in this round - # Implementation depends on demo parser structure - # Placeholder logic: + round_col = 'total_rounds_played' if 'total_rounds_played' in grenades.columns else 'round_num' + if round_col in grenades.columns: + round_grenades = grenades[grenades[round_col] == round_num] + else: + # Fallback: if no round info in grenades, assume they are global and filter by tick? + # Or assume we can't map them. + # But the parser usually adds round info if available or we can use round start/end ticks. + # For now, let's check ticks if rounds dataframe exists + if hasattr(parsed_demo, 'rounds') and not parsed_demo.rounds.empty: + round_info = parsed_demo.rounds[parsed_demo.rounds['round_num'] == round_num if 'round_num' in parsed_demo.rounds.columns else parsed_demo.rounds.index == round_num-1] + if not round_info.empty: + start_tick = round_info.iloc[0].get('round_start_tick', 0) if 'round_start_tick' in round_info.columns else round_info.iloc[0].get('start_tick', 0) + end_tick = round_info.iloc[0].get('round_end_tick', 99999999) if 'round_end_tick' in round_info.columns else round_info.iloc[0].get('end_tick', 99999999) + + round_grenades = grenades[(grenades['tick'] >= start_tick) & (grenades['tick'] <= end_tick)] + else: + return mistakes + else: + return mistakes + + flashes = round_grenades[round_grenades['grenade_type'] == 'flashbang'] + if flashes.empty: + return mistakes + + kills = parsed_demo.kills + if kills is None or kills.empty: + # If no kills but flashes thrown, maybe wasted? + # But we only check for entry/trade support. + pass + + # Get kills for this round + round_kills = pd.DataFrame() + if kills is not None and not kills.empty: + k_round_col = 'total_rounds_played' if 'total_rounds_played' in kills.columns else 'round_num' + if k_round_col in kills.columns: + round_kills = kills[kills[k_round_col] == round_num] + else: + # Fallback: filter by tick if round info is available in parsed_demo.rounds + if hasattr(parsed_demo, 'rounds') and not parsed_demo.rounds.empty: + # Use the same logic as for grenades above + round_info = parsed_demo.rounds[parsed_demo.rounds['round_num'] == round_num if 'round_num' in parsed_demo.rounds.columns else parsed_demo.rounds.index == round_num-1] + if not round_info.empty: + start_tick = round_info.iloc[0].get('round_start_tick', 0) if 'round_start_tick' in round_info.columns else round_info.iloc[0].get('start_tick', 0) + end_tick = round_info.iloc[0].get('round_end_tick', 99999999) if 'round_end_tick' in round_info.columns else round_info.iloc[0].get('end_tick', 99999999) + + round_kills = kills[(kills['tick'] >= start_tick) & (kills['tick'] <= end_tick)] + + for _, flash in flashes.iterrows(): + flash_tick = int(flash.get('tick', 0)) + thrower = str(flash.get('name', flash.get('user_name', ''))) + team = str(flash.get('team_name', '')) + + if not thrower: + continue + + # Check for ANY kill by ANYONE on thrower's team within window + # Window: flash_tick to flash_tick + window + window_ticks = self.FLASH_FOLLOWUP_WINDOW_MS / 15.625 + + success = False + if not round_kills.empty: + for _, kill in round_kills.iterrows(): + kill_tick = int(kill.get('tick', 0)) + attacker_team = str(kill.get('attacker_team_name', '')) + + if attacker_team == team: + if 0 <= (kill_tick - flash_tick) <= window_ticks: + success = True + break + + if not success: + mistakes.append(DetectedMistake( + round=round_num, + timestamp_ms=int(flash_tick * 15.625), + player=thrower, + error_type=self.error_type.value, + severity=Severity.LOW.value, + wpa_loss=0.02, + map_pos={"x": float(flash.get('x', flash.get('X', 0))), "y": float(flash.get('y', flash.get('Y', 0)))}, + details=f"Flash with no entry/kill follow-up within 2s" + )) - return mistakes # Requires grenade tracking data + return mistakes class PostplantMisplayDetector(MistakeDetector): diff --git a/src/parser/demo_parser.py b/src/parser/demo_parser.py index 6c14046..49e9401 100644 --- a/src/parser/demo_parser.py +++ b/src/parser/demo_parser.py @@ -199,9 +199,9 @@ def _parse_with_demoparser2(self) -> ParsedDemo: # Parse grenade detonations try: - he_df = parser.parse_event("hegrenade_detonate", player=["X", "Y", "Z"]) - smoke_df = parser.parse_event("smokegrenade_detonate", player=["X", "Y", "Z"]) - flash_det_df = parser.parse_event("flashbang_detonate", player=["X", "Y", "Z"]) + he_df = parser.parse_event("hegrenade_detonate", player=["X", "Y", "Z", "name", "team_name"]) + smoke_df = parser.parse_event("smokegrenade_detonate", player=["X", "Y", "Z", "name", "team_name"]) + flash_det_df = parser.parse_event("flashbang_detonate", player=["X", "Y", "Z", "name", "team_name"]) grenades_list = [] if he_df is not None and len(he_df) > 0: diff --git a/tests/test_mistakes.py b/tests/test_mistakes.py index fe9c92b..28d7c50 100644 --- a/tests/test_mistakes.py +++ b/tests/test_mistakes.py @@ -311,6 +311,101 @@ def test_flash_followup_window(self): detector = UtilityWasteDetector() assert detector.FLASH_FOLLOWUP_WINDOW_MS == 2000 + def test_detects_wasted_flash(self, mock_demo_with_kills): + """Detect flash with no follow-up kill.""" + # Create a demo with a flash that has no follow-up + demo = mock_demo_with_kills + + # Add grenade data + # Flash at tick 1500 (approx 96s) + # Kill at tick 2000 (approx 128s) -> >2s later + demo.grenades = pd.DataFrame({ + 'grenade_type': ['flashbang'], + 'tick': [1500], + 'name': ['PlayerA'], + 'team_name': ['CT'], + 'x': [0], 'y': [0], 'z': [0], + 'round_num': [1] # Ensure round number is set + }) + + # The existing kills in mock_demo_with_kills are at ticks 1000, 1100, 1300, 2000... + # Flash at 1500. Next kill is 2000. + # Difference = 500 ticks. 500 * 0.015625 = 7.8s > 2s. + # So this flash should be wasted. + + detector = UtilityWasteDetector() + mistakes = detector.detect(demo, 1) # Round 1 + + # Should detect 1 mistake + assert len(mistakes) == 1 + assert mistakes[0].error_type == "UTILITY_WASTE" + assert mistakes[0].player == "PlayerA" + + def test_ignores_good_flash(self, mock_demo_with_kills): + """Ignore flash that has follow-up kill.""" + demo = mock_demo_with_kills + + # Flash at tick 1050 + # Kill at tick 1100 (PlayerB kills PlayerA) - within 50 ticks (<1s) + demo.grenades = pd.DataFrame({ + 'grenade_type': ['flashbang'], + 'tick': [1050], + 'name': ['PlayerB'], + 'team_name': ['Terrorist'], + 'x': [0], 'y': [0], 'z': [0], + 'round_num': [1] # Ensure round number is set + }) + + detector = UtilityWasteDetector() + mistakes = detector.detect(demo, 1) + + assert len(mistakes) == 0 + + def test_detects_with_missing_round_columns(self, mock_demo_with_kills): + """Detect mistakes when round columns are missing from dataframes.""" + demo = mock_demo_with_kills + + # Remove round_num from kills to trigger tick-based filtering + if 'round_num' in demo.kills.columns: + demo.kills = demo.kills.drop(columns=['round_num']) + + # Ensure rounds info exists for mapping + demo.rounds = pd.DataFrame({ + 'round_num': [1], + 'start_tick': [0], + 'end_tick': [2000], + }) + + # Add flash and kill within window + # Flash at 1050, Kill at 1100. Should be NO mistake. + demo.grenades = pd.DataFrame({ + 'grenade_type': ['flashbang'], + 'tick': [1050], + 'name': ['PlayerB'], + 'team_name': ['Terrorist'], + 'x': [0], 'y': [0], 'z': [0] + # No round_num here either + }) + + detector = UtilityWasteDetector() + mistakes = detector.detect(demo, 1) + + assert len(mistakes) == 0 + + # Add wasted flash + # Flash at 1500, no kill after (next kill in mock_demo is 2000 > 2s later) + demo.grenades = pd.DataFrame({ + 'grenade_type': ['flashbang'], + 'tick': [1500], + 'name': ['PlayerA'], + 'team_name': ['CT'], + 'x': [0], 'y': [0], 'z': [0] + }) + + mistakes = detector.detect(demo, 1) + assert len(mistakes) == 1 + assert mistakes[0].player == "PlayerA" + # ============================================================================ # Postplant Misplay Detector Tests