Simplify rating script args; minor game log fixes

Author: john-b-yang

codeclash/games/battlecode/battlecode.py

@@ -29,7 +29,7 @@ def copy_logs_from_env(self, round_num):
         copy_from_container(
             container=self.environment,
             src_path="/testbed/sim.log",
-            dest_path=self.log_local / "rounds" / f"sim_{round_num}.log",
+            dest_path=self.log_local / "rounds" / str(round_num) / f"sim_{round_num}.log",
         )
 
     def get_stats(self, agents: list[Any]) -> RoundStats:

codeclash/games/battlesnake/battlesnake.py

@@ -49,7 +49,7 @@ def copy_logs_from_env(self, round_num):
     def get_stats(self, agents: list[Player]) -> RoundStats:
         scores = {}
         for idx in range(self.game_config["sims_per_round"]):
-            ro = self.environment.execute(f"cat game/logs/sim_{idx}.txt")["output"]
+            ro = self.environment.execute(f"cat game/logs/sim_{idx}.jsonl")["output"]
             lines = ro.strip().split("\n")
             results = json.loads(lines[-1]) if lines else {}  # Get the last line which contains the game result
             winner = RESULT_TIE if results["isDraw"] else results["winnerName"]
@@ -99,7 +99,7 @@ def _run_single_simulation(self, cmd: str, idx: int) -> tuple[str, str]:
         """Run a single battlesnake simulation and return log and result outputs."""
         assert_zero_exit_code(
             self.environment.execute(
-                cmd + f" -o logs/sim_{idx}.txt",
+                cmd + f" -o logs/sim_{idx}.jsonl",
                 cwd=f"{self.environment.config.cwd}/game",
             )
         )

codeclash/ratings/elo.py

@@ -3,6 +3,8 @@
 from dataclasses import dataclass
 from pathlib import Path
 
+from codeclash.constants import DIR_LOGS
+
 K_FACTOR = 32  # ELO constant, changeable
 
 
@@ -19,44 +21,51 @@ def expected_score(rating_a, rating_b):
 
 
 def main(log_dir: Path):
+    # Assuming directory structure is:
+    # logs/<user_id>/<game_id>
+    # - players/
+    # - rounds/
+    # - game.log
+    # - metadata.json
     player_profiles = {}
-    for game_log_folder in log_dir.iterdir():
-        if game_log_folder.is_dir():
-            print(f"Processing game log folder: {game_log_folder}")
+    for user_folder in log_dir.iterdir():
+        for game_log_folder in user_folder.iterdir():
+            if game_log_folder.is_dir():
+                print(f"Processing game log folder: {game_log_folder}")
 
-        game_id = game_log_folder.name.split(".")[1]
-        player_ids = [x.name for x in (game_log_folder / "players").iterdir() if x.is_dir()]
-        # Initialize profiles
-        for player in player_ids:
-            key = f"{game_id}.{player}"
-            if key not in player_profiles:
-                player_profiles[key] = PlayerEloProfile(player_id=player, game_id=game_id)
+            game_id = game_log_folder.name.split(".")[1]
+            player_ids = [x.name for x in (game_log_folder / "players").iterdir() if x.is_dir()]
+            # Initialize profiles
+            for player in player_ids:
+                key = f"{game_id}.{player}"
+                if key not in player_profiles:
+                    player_profiles[key] = PlayerEloProfile(player_id=player, game_id=game_id)
 
-        for round_folder in (game_log_folder / "rounds").iterdir():
-            round_results = json.load(open(round_folder / "results.json"))
-            winner = round_results.get("winner")
-            players = round_results.get("players", player_ids)
-            # Only process if there are exactly 2 players
-            if len(players) == 2:
-                p1_key = f"{game_id}.{players[0]}"
-                p2_key = f"{game_id}.{players[1]}"
-                p1 = player_profiles[p1_key]
-                p2 = player_profiles[p2_key]
-                p1.games_played += 1
-                p2.games_played += 1
-                # Determine scores
-                if winner == players[0]:
-                    s1, s2 = 1, 0
-                elif winner == players[1]:
-                    s1, s2 = 0, 1
-                else:
-                    s1, s2 = 0.5, 0.5  # Tie
-                # Calculate expected scores
-                e1 = expected_score(p1.rating, p2.rating)
-                e2 = expected_score(p2.rating, p1.rating)
-                # Update ratings
-                p1.rating += K_FACTOR * (s1 - e1)
-                p2.rating += K_FACTOR * (s2 - e2)
+            for round_folder in (game_log_folder / "rounds").iterdir():
+                round_results = json.load(open(round_folder / "results.json"))
+                winner = round_results.get("winner")
+                players = round_results.get("players", player_ids)
+                # Only process if there are exactly 2 players
+                if len(players) == 2:
+                    p1_key = f"{game_id}.{players[0]}"
+                    p2_key = f"{game_id}.{players[1]}"
+                    p1 = player_profiles[p1_key]
+                    p2 = player_profiles[p2_key]
+                    p1.games_played += 1
+                    p2.games_played += 1
+                    # Determine scores
+                    if winner == players[0]:
+                        s1, s2 = 1, 0
+                    elif winner == players[1]:
+                        s1, s2 = 0, 1
+                    else:
+                        s1, s2 = 0.5, 0.5  # Tie
+                    # Calculate expected scores
+                    e1 = expected_score(p1.rating, p2.rating)
+                    e2 = expected_score(p2.rating, p1.rating)
+                    # Update ratings
+                    p1.rating += K_FACTOR * (s1 - e1)
+                    p2.rating += K_FACTOR * (s2 - e2)
 
     print("Player ELO profiles:")
     for profile in player_profiles.values():
@@ -83,6 +92,6 @@ def main(log_dir: Path):
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser()
-    parser.add_argument("log_dir", type=Path, help="Path to `logs/<user>` folder containing game logs")
+    parser.add_argument("-d", "--log_dir", type=Path, help="Path to game logs (Default: logs/)", default=DIR_LOGS)
     args = parser.parse_args()
     main(args.log_dir)

codeclash/ratings/win_rate.py

@@ -3,7 +3,7 @@
 from dataclasses import dataclass
 from pathlib import Path
 
-from codeclash.constants import RESULT_TIE
+from codeclash.constants import DIR_LOGS, RESULT_TIE
 
 
 @dataclass
@@ -19,38 +19,61 @@ def win_rate(self) -> float:
 
 
 def main(log_dir: Path):
+    # Assuming directory structure is:
+    # logs/<user_id>/<game_id>
+    # - players/
+    # - rounds/
+    # - game.log
+    # - metadata.json
     player_profiles = {}
-    for game_log_folder in log_dir.iterdir():
-        if game_log_folder.is_dir():
-            print(f"Processing game log folder: {game_log_folder}")
-
-        game_id = game_log_folder.name.split(".")[1]
-        player_ids = [x.name for x in (game_log_folder / "players").iterdir() if x.is_dir()]
-        num_rounds = len(list((game_log_folder / "rounds").iterdir()))
-
-        for player in player_ids:
-            if f"{game_id}.{player}" in player_profiles:
-                player_profiles[f"{game_id}.{player}"].count += num_rounds
-            else:
-                player_profiles[f"{game_id}.{player}"] = PlayerGameProfile(
-                    player_id=player, game_id=game_id, count=num_rounds
-                )
-
-        for round_folder in (game_log_folder / "rounds").iterdir():
-            round_results = json.load(open(round_folder / "results.json"))
-            winner = round_results.get("winner")
-            if winner != RESULT_TIE:
-                player_profiles[f"{game_id}.{winner}"].wins += 1
+    for user_folder in log_dir.iterdir():
+        for game_log_folder in user_folder.iterdir():
+            if game_log_folder.is_dir():
+                print(f"Processing game log folder: {game_log_folder}")
+
+            game_id = game_log_folder.name.split(".")[1]
+            player_ids = [x.name for x in (game_log_folder / "players").iterdir() if x.is_dir()]
+            num_rounds = len(list((game_log_folder / "rounds").iterdir()))
+
+            for player in player_ids:
+                if f"{game_id}.{player}" in player_profiles:
+                    player_profiles[f"{game_id}.{player}"].count += num_rounds
+                else:
+                    player_profiles[f"{game_id}.{player}"] = PlayerGameProfile(
+                        player_id=player, game_id=game_id, count=num_rounds
+                    )
+
+            for round_folder in (game_log_folder / "rounds").iterdir():
+                round_results = json.load(open(round_folder / "results.json"))
+                winner = round_results.get("winner")
+                if winner != RESULT_TIE:
+                    player_profiles[f"{game_id}.{winner}"].wins += 1
 
     print("Player profiles:")
     for profile in player_profiles.values():
         print(
             f" - {profile.player_id} (Game: {profile.game_id}) - Win Rate: {profile.win_rate:.2%} ({profile.wins}/{profile.count})"
         )
 
+    # Player-specific (game-agnostic) win rates (micro average)
+    total_wins = {}
+    total_games = {}
+    for profile in player_profiles.values():
+        pid = profile.player_id
+        total_wins[pid] = total_wins.get(pid, 0) + profile.wins
+        total_games[pid] = total_games.get(pid, 0) + profile.count
+
+    print("\nPlayer-specific win rates (game-agnostic, micro average):")
+    for pid in total_wins:
+        if total_games[pid] > 0:
+            win_rate = total_wins[pid] / total_games[pid]
+        else:
+            win_rate = 0.0
+        print(f" - {pid}: Win Rate {win_rate:.2%} ({total_wins[pid]}/{total_games[pid]})")
+
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser()
-    parser.add_argument("log_dir", type=Path, help="Path to `logs/<user>` folder containing game logs")
+    parser.add_argument("-d", "--log_dir", type=Path, help="Path to game logs (Default: logs/)", default=DIR_LOGS)
     args = parser.parse_args()
     main(args.log_dir)