thenuke02/cs2-analyzer
0
1#!/usr/bin/env python32"""3Verify the full parse → analyze pipeline works end-to-end.4 5This script tests the complete OpenSight analysis pipeline:61. DemoParser parses the demo file72. DemoAnalyzer computes metrics83. CachedAnalyzer orchestrates everything94. Final result has valid, non-zero data10 11Run with: python scripts/verify_pipeline.py [path/to/demo.dem]12 13If no demo is provided, it will look for test demos in test_demos/ directory.14 15Exit codes:16 0 - Pipeline working correctly17 1 - Pipeline has issues (check output for details)18 2 - Critical failure (missing dependencies, no demo file)19"""20 21import sys22from pathlib import Path23 24# Add src to path for imports25sys.path.insert(0, str(Path(__file__).parent.parent / "src"))26 27 28def find_test_demo() -> Path | None:29 """Find a test demo file."""30 project_root = Path(__file__).parent.parent31 32 # Check common locations33 locations = [34 project_root / "test_demos",35 project_root / "tests" / "fixtures",36 project_root / "demos",37 Path.home() / ".opensight" / "demos",38 ]39 40 for loc in locations:41 if loc.exists():42 demos = list(loc.glob("*.dem")) + list(loc.glob("*.dem.gz"))43 if demos:44 return demos[0]45 46 return None47 48 49def verify_parser(demo_path: Path) -> dict | None:50 """Test the DemoParser directly."""51 print("\n=== Step 1: DemoParser ===")52 53 try:54 from opensight.core.parser import DemoParser55 56 parser = DemoParser(demo_path)57 demo_data = parser.parse()58 59 print(f" Map: {demo_data.map_name}")60 print(f" Rounds: {demo_data.num_rounds}")61 print(f" Players: {len(demo_data.player_names)}")62 print(f" Kills: {len(demo_data.kills)}")63 print(f" Duration: {demo_data.duration_seconds:.0f}s")64 65 # Validate critical fields66 issues = []67 if demo_data.num_rounds == 0:68 issues.append("num_rounds is 0")69 if len(demo_data.kills) == 0:70 issues.append("No kills parsed")71 if len(demo_data.player_names) == 0:72 issues.append("No players found")73 74 if issues:75 print(f" ISSUES: {', '.join(issues)}")76 return None77 78 print(" OK: Parser working correctly")79 return {80 "demo_data": demo_data,81 "map": demo_data.map_name,82 "rounds": demo_data.num_rounds,83 "players": len(demo_data.player_names),84 "kills": len(demo_data.kills),85 }86 except Exception as e:87 print(f" FAIL: Parser error: {e}")88 import traceback89 90 traceback.print_exc()91 return None92 93 94def verify_analyzer(demo_data) -> dict | None:95 """Test the DemoAnalyzer."""96 print("\n=== Step 2: DemoAnalyzer ===")97 98 try:99 from opensight.analysis.analytics import DemoAnalyzer100 101 analyzer = DemoAnalyzer(demo_data)102 analysis = analyzer.analyze()103 104 print(f" Players analyzed: {len(analysis.players)}")105 106 # Check player stats107 zero_stat_players = []108 valid_players = 0109 110 for _steam_id, player in analysis.players.items():111 if player.kills == 0 and player.deaths == 0:112 zero_stat_players.append(player.name)113 else:114 valid_players += 1115 if valid_players <= 3:116 print(117 f" {player.name}: {player.kills}K/{player.deaths}D, ADR: {player.adr:.1f}, Rating: {player.hltv_rating:.2f}"118 )119 120 if zero_stat_players:121 print(122 f" WARN: {len(zero_stat_players)} players with 0K/0D: {zero_stat_players[:3]}..."123 )124 125 if valid_players == 0:126 print(" FAIL: All players have 0 kills AND 0 deaths")127 return None128 129 print(f" OK: {valid_players} players with valid stats")130 return {131 "analysis": analysis,132 "valid_players": valid_players,133 }134 except Exception as e:135 print(f" FAIL: Analyzer error: {e}")136 import traceback137 138 traceback.print_exc()139 return None140 141 142def verify_cached_analyzer(demo_path: Path) -> dict | None:143 """Test the full CachedAnalyzer pipeline."""144 print("\n=== Step 3: CachedAnalyzer (Full Pipeline) ===")145 146 try:147 from opensight.infra.cache import CachedAnalyzer148 149 cached = CachedAnalyzer()150 result = cached.analyze(demo_path, force=True) # Force fresh analysis151 152 if not result:153 print(" FAIL: CachedAnalyzer returned None/empty")154 return None155 156 # Check critical fields in result157 players = result.get("players", {})158 round_timeline = result.get("round_timeline", [])159 match_info = result.get("match_info", {})160 161 print(f" Players in result: {len(players)}")162 print(f" Rounds in timeline: {len(round_timeline)}")163 print(f" Map: {match_info.get('map', 'unknown')}")164 165 # Validate player data166 issues = []167 zero_stat_count = 0168 169 for _steam_id, player in players.items():170 stats = player.get("stats", {})171 kills = stats.get("kills", 0)172 deaths = stats.get("deaths", 0)173 174 if kills == 0 and deaths == 0:175 zero_stat_count += 1176 177 if zero_stat_count > 0:178 issues.append(f"{zero_stat_count} players with 0K/0D")179 if len(round_timeline) == 0:180 issues.append("No round timeline")181 182 if issues:183 print(f" ISSUES: {', '.join(issues)}")184 185 # Print sample player stats186 print("\n Sample player stats:")187 for i, (_steam_id, player) in enumerate(players.items()):188 if i >= 5:189 break190 stats = player.get("stats", {})191 rating = player.get("rating", {})192 print(193 f" {player.get('name', 'Unknown')}: {stats.get('kills', 0)}K/{stats.get('deaths', 0)}D, ADR: {stats.get('adr', 0)}, Rating: {rating.get('hltv_rating', 0)}"194 )195 196 # Check for zero-stat issues (the main bug we're looking for)197 valid_player_count = len(players) - zero_stat_count198 if valid_player_count == 0:199 print(" FAIL: All players have 0 kills AND 0 deaths - data is broken!")200 return None201 202 print(f"\n OK: CachedAnalyzer pipeline working ({valid_player_count} valid players)")203 return {204 "result": result,205 "players": len(players),206 "rounds": len(round_timeline),207 "valid_players": valid_player_count,208 }209 except Exception as e:210 print(f" FAIL: CachedAnalyzer error: {e}")211 import traceback212 213 traceback.print_exc()214 return None215 216 217def main():218 print("=" * 60)219 print("OpenSight Pipeline Verification")220 print("=" * 60)221 222 # Find demo file223 if len(sys.argv) > 1:224 demo_path = Path(sys.argv[1])225 else:226 demo_path = find_test_demo()227 if demo_path is None:228 print("\nERROR: No demo file specified and no test demos found.")229 print("\nUsage: python scripts/verify_pipeline.py <path/to/demo.dem>")230 print("\nOr place demo files in:")231 print(" - test_demos/")232 print(" - tests/fixtures/")233 sys.exit(2)234 235 if not demo_path.exists():236 print(f"\nERROR: Demo file not found: {demo_path}")237 sys.exit(2)238 239 print(f"\nDemo file: {demo_path}")240 print(f"File size: {demo_path.stat().st_size / 1024 / 1024:.1f} MB")241 242 # Run verification steps243 results = {}244 245 # Step 1: Parser246 parser_result = verify_parser(demo_path)247 results["parser"] = parser_result is not None248 249 # Step 2: Analyzer (if parser worked)250 if parser_result:251 analyzer_result = verify_analyzer(parser_result["demo_data"])252 results["analyzer"] = analyzer_result is not None253 else:254 results["analyzer"] = False255 256 # Step 3: CachedAnalyzer (full pipeline)257 cached_result = verify_cached_analyzer(demo_path)258 results["cached_analyzer"] = cached_result is not None259 260 # Summary261 print("\n" + "=" * 60)262 print("VERIFICATION SUMMARY")263 print("=" * 60)264 265 all_passed = all(results.values())266 267 for step, passed in results.items():268 status = "PASS" if passed else "FAIL"269 print(f" {step}: {status}")270 271 if all_passed:272 print("\n SUCCESS: Full pipeline is working correctly!")273 print("Demo parsing and analysis are producing valid data.")274 sys.exit(0)275 else:276 print("\n ISSUES DETECTED: See above for details.")277 print("Some parts of the pipeline are not working correctly.")278 sys.exit(1)279 280 281if __name__ == "__main__":282 main()283