Added dataclasses for return type; added score tracking for games

Author: john-b-yang

codeclash/agents/abstract.py

@@ -8,7 +8,7 @@
 from codeclash.agents.utils import GameContext
 from codeclash.constants import GH_ORG
 from codeclash.tournaments.utils.git_utils import filter_git_diff
-from codeclash.utils.environment import assert_zero_exit_code, create_file_on_container
+from codeclash.utils.environment import assert_zero_exit_code, create_file_in_container
 from codeclash.utils.log import get_logger
 
 load_dotenv()
@@ -101,7 +101,7 @@ def reset_and_apply_patch(
             self.logger.debug("No patch to apply, skipping")
             return
 
-        create_file_on_container(
+        create_file_in_container(
             container=self.environment,  # type: ignore
             content=patch,
             dest_path="tmp_patch.txt",

codeclash/constants.py

@@ -3,6 +3,4 @@
 DIR_LOGS = Path("logs")
 DIR_WORK = Path("/testbed")
 GH_ORG = "emagedoc"
-OUTPUTS_LOGS = "log_outputs"
-OUTPUTS_RESULTS = "result_outputs"
 RESULT_TIE = "Tie"

codeclash/games/abstract.py

@@ -2,22 +2,42 @@
 import os
 import subprocess
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
 from pathlib import Path
+from typing import Any
 
 from minisweagent.environments.docker import DockerEnvironment
 
 from codeclash.agents.abstract import Player
-from codeclash.constants import (
-    DIR_LOGS,
-    DIR_WORK,
-    GH_ORG,
-    OUTPUTS_LOGS,
-    OUTPUTS_RESULTS,
-)
+from codeclash.constants import DIR_LOGS, DIR_WORK, GH_ORG
 from codeclash.utils.environment import assert_zero_exit_code, copy_between_containers
 from codeclash.utils.log import get_logger
 
 
+@dataclass
+class RoundStats:
+    winner: str
+    scores: dict[
+        str, float
+    ]  # Map of player to game metric (e.g. # of wins, assets accumulated)
+    details: dict[str, Any] = None  # Optional, for game-specific info
+
+    def __str__(self) -> str:
+        return "\n".join([f"- Winner: {self.winner}", f"- Scores: {self.scores}"])
+
+
+@dataclass
+class RoundData:
+    logs: list[str]
+    results: list[str]
+
+
+@dataclass
+class RoundRecord:
+    data: RoundData
+    stats: RoundStats
+
+
 class CodeGame(ABC):
     name: str
 
@@ -144,48 +164,37 @@ def _pre_round_setup(self, agents: list[Player]):
         )
 
     @abstractmethod
-    def determine_winner(
-        self, result_outputs: list[str], agents: list[Player]
-    ) -> dict[str, str]:
+    def get_stats(self, result_outputs: list[str], agents: list[Player]) -> RoundStats:
         """Determine the winner of the game based on the result output.
 
         Args:
             result_outputs: The specific output(s) containing winning information
             agents: List of agents participating in the round
 
         Returns:
-            Dictionary with key "winner" containing the winner's name
+            RoundStats object
         """
         pass
 
     @abstractmethod
-    def execute_round(self, agents: list[Player]) -> dict[str, list[str]]:
+    def execute_round(self, agents: list[Player]) -> RoundData:
         """Subclasses implement their game-specific logic here.
         This is the low level implementation, you probably want to use run_round instead, which
         includes the pre-round setup, post-round setup, and winner determination.
 
         Returns:
-            Dictionary with keys "log_outputs" and "result_outputs"
+            RoundData object
         """
         pass
 
-    def run_round(self, agents: list[Player]) -> dict[str, str]:
+    def run_round(self, agents: list[Player]) -> RoundRecord:
         """
         Run a single round of the game with the given agents.
 
         Returns the log output, result output, and winner name. All bookkeeping should be
         handled by the tournament class.
         """
         self._pre_round_setup(agents)
-        result = self.execute_round(agents)
-        log_outputs = result[OUTPUTS_LOGS]
-        result_outputs = result[OUTPUTS_RESULTS]
-
-        winner_result = self.determine_winner(result_outputs, agents)
-        winner_name = winner_result["winner"]
-
-        return {
-            OUTPUTS_LOGS: log_outputs,
-            OUTPUTS_RESULTS: result_outputs,
-            "winner": winner_name,
-        }
+        data = self.execute_round(agents)
+        stats = self.get_stats(data.results, agents)
+        return RoundRecord(data=data, stats=stats)

codeclash/games/battlecode/main.py

@@ -4,8 +4,8 @@
 
 from tqdm.auto import tqdm
 
-from codeclash.constants import DIR_WORK, OUTPUTS_LOGS, OUTPUTS_RESULTS, RESULT_TIE
-from codeclash.games.abstract import CodeGame
+from codeclash.constants import DIR_WORK, RESULT_TIE
+from codeclash.games.abstract import CodeGame, RoundData, RoundStats
 
 
 class BattleCodeGame(CodeGame):
@@ -24,9 +24,7 @@ def __init__(self, config, *, tournament_id: str, local_output_dir: Path):
             else:
                 self.run_cmd_round += f" --{arg} {val}"
 
-    def determine_winner(
-        self, result_outputs: list[str], agents: list[Any]
-    ) -> dict[str, str]:
+    def get_stats(self, result_outputs: list[str], agents: list[Any]) -> RoundStats:
         winners = []
         for ro in result_outputs:
             lines = ro.strip().split("\n")
@@ -44,10 +42,12 @@ def determine_winner(
                 winners.append(winner)
             else:
                 winners.append(RESULT_TIE)
-        winner = max(set(winners), key=winners.count)
-        return {"winner": winner}
+        return RoundStats(
+            winner=max(set(winners), key=winners.count),
+            scores={agent.name: winners.count(agent.name) for agent in agents},
+        )
 
-    def execute_round(self, agents: list[Any]) -> dict[str, list[str]]:
+    def execute_round(self, agents: list[Any]) -> RoundData:
         for agent in agents:
             src, dest = f"/{agent.name}/src/mysubmission/", str(
                 DIR_WORK / "src" / agent.name
@@ -58,11 +58,11 @@ def execute_round(self, agents: list[Any]) -> dict[str, list[str]]:
             for idx, agent in enumerate(agents)
         ]
         cmd = f"{self.run_cmd_round} {' '.join(args)}"
-        self.logger.info(f"Running command: {cmd}")
+        self.logger.info(f"Running game: {cmd}")
         outputs = []
         for _ in tqdm(range(self.game_config["sims_per_round"])):
             response = self.environment.execute(cmd)
             assert response["returncode"] == 0, response
             # For BattleCode, log_outputs and result_outputs are the same
             outputs.append(response["output"])
-        return {OUTPUTS_LOGS: outputs, OUTPUTS_RESULTS: outputs}
+        return RoundData(logs=outputs, results=outputs)

codeclash/games/battlesnake/main.py

@@ -2,9 +2,11 @@
 import time
 from pathlib import Path
 
+from tqdm.auto import tqdm
+
 from codeclash.agents.abstract import Player
-from codeclash.constants import OUTPUTS_LOGS, OUTPUTS_RESULTS
-from codeclash.games.abstract import CodeGame
+from codeclash.constants import RESULT_TIE
+from codeclash.games.abstract import CodeGame, RoundData, RoundStats
 from codeclash.utils.environment import assert_zero_exit_code
 
 
@@ -23,21 +25,25 @@ def __init__(self, config, *, tournament_id: str, local_output_dir: Path):
             else:
                 self.run_cmd_round += f" --{arg} {val}"
 
-    def determine_winner(
-        self, result_outputs: list[str], agents: list[Player]
-    ) -> dict[str, str]:
+    def get_stats(self, result_outputs: list[str], agents: list[Player]) -> RoundStats:
         winners = []
         for ro in result_outputs:
             lines = ro.strip().split("\n")
-            # Get the last line which contains the game result
-            last_line = lines[-1] if lines else ""
-            self.logger.debug(f"Last line: {last_line}")
+            last_line = (
+                lines[-1] if lines else ""
+            )  # Get the last line which contains the game result
             winner = json.loads(last_line)["winnerName"]
             winners.append(winner)
-        winner = max(set(winners), key=winners.count)
-        return {"winner": winner}
 
-    def execute_round(self, agents: list[Player]) -> dict[str, list[str]]:
+        win_counts = {agent.name: winners.count(agent.name) for agent in agents}
+        max_wins = max(win_counts.values())
+        winners = [name for name, wins in win_counts.items() if wins == max_wins]
+        return RoundStats(
+            winner=RESULT_TIE if len(winners) > 1 else winners[0],
+            scores=win_counts,
+        )
+
+    def execute_round(self, agents: list[Player]) -> RoundData:
         cmd = []
         for idx, agent in enumerate(agents):
             port = 8001 + idx
@@ -51,15 +57,16 @@ def execute_round(self, agents: list[Player]) -> dict[str, list[str]]:
 
         try:
             log_outputs, result_outputs = [], []
-            for idx in range(self.game_config["sims_per_round"]):
+            cmd = self.run_cmd_round + " " + " ".join(cmd)
+            self.logger.info(f"Running game: {cmd}")
+            for idx in tqdm(range(self.game_config["sims_per_round"])):
                 # Create temporary output file for results
                 output_file = f"battlesnake_output_{idx}_{int(time.time())}.json"
-                cmd_str = " ".join(cmd) + f" -o {output_file}"
-                self.logger.info(f"Running command: {self.run_cmd_round} {cmd_str}")
 
+                # Run game
                 response = assert_zero_exit_code(
                     self.environment.execute(
-                        f"{self.run_cmd_round} {cmd_str}",
+                        cmd + f" -o {output_file}",
                         cwd=f"{self.environment.config.cwd}/game",
                     )
                 )
@@ -72,10 +79,9 @@ def execute_round(self, agents: list[Player]) -> dict[str, list[str]]:
 
                 # Clean up the output file
                 self.environment.execute(f"rm -f game/{output_file}")
+                time.sleep(0.05)
 
-                time.sleep(0.1)
-
-            return {OUTPUTS_LOGS: log_outputs, OUTPUTS_RESULTS: result_outputs}
+            return RoundData(log_outputs, result_outputs)
         finally:
             # Kill all python servers when done
             self.environment.execute("pkill -f 'python main.py' || true")

codeclash/games/corewar/main.py

@@ -3,8 +3,7 @@
 from pathlib import Path
 
 from codeclash.agents.abstract import Player
-from codeclash.constants import OUTPUTS_LOGS, OUTPUTS_RESULTS
-from codeclash.games.abstract import CodeGame
+from codeclash.games.abstract import CodeGame, RoundData, RoundStats
 
 
 class CoreWarGame(CodeGame):
@@ -22,44 +21,43 @@ def __init__(self, config, *, tournament_id: str, local_output_dir: Path):
             else:
                 self.run_cmd_round += f" -{arg} {val}"
 
-    def determine_winner(
-        self, result_outputs: list[str], agents: list[Player]
-    ) -> dict[str, str]:
+    def get_stats(self, result_outputs: list[str], agents: list[Player]) -> RoundStats:
         result_output = result_outputs[0]  # Get the first (and only) element
         self.logger.debug(f"Determining winner from result output: {result_output}")
         scores = []
         n = len(agents) * 2
         lines = result_output.strip().split("\n")
+
         # Get the last n lines which contain the scores (closer to original)
         relevant_lines = lines[-n:] if len(lines) >= n else lines
+        relevant_lines = [l for l in relevant_lines if len(l.strip()) > 0]
         self.logger.debug(f"Relevant lines for scoring: {relevant_lines}")
 
+        # Go through each line; we assume score position is correlated with agent index
         for line in relevant_lines:
             match = re.search(r".*\sby\s.*\sscores\s(\d+)", line)
             if match:
                 score = int(match.group(1))
                 scores.append(score)
-                self.logger.debug(f"Found score: {score} from line: {line}")
 
-        self.logger.debug(f"All scores: {scores}")
         if scores:
-            max_score_index = scores.index(max(scores))
-            winner = agents[max_score_index].name
-            self.logger.debug(
-                f"Concluding winner: {winner} with index {max_score_index}"
+            if len(scores) != len(agents):
+                self.logger.error(f"Have {len(scores)} scores but {len(agents)} agents")
+            return RoundStats(
+                winner=agents[scores.index(max(scores))].name,
+                scores={agent.name: score for agent, score in zip(agents, scores)},
+                details={"stdout": "\n".join(relevant_lines)},
             )
-            return {"winner": winner}
         else:
             self.logger.debug("No scores found, returning unknown")
-            return {"winner": "unknown"}
+            return RoundStats(
+                winner="unknown", scores={agent.name: 0 for agent in agents}
+            )
 
-    def execute_round(self, agents: list[Player]) -> dict[str, list[str]]:
+    def execute_round(self, agents: list[Player]) -> RoundData:
         args = [f"/{agent.name}/warriors/warrior.red" for agent in agents]
-        cmd = f"{self.run_cmd_round} {shlex.join(args)}"
-        cmd += f" -r {self.game_config['sims_per_round']}"
-        self.logger.info(f"Running command: {cmd}")
+        cmd = f"{self.run_cmd_round} {shlex.join(args)} -r {self.game_config['sims_per_round']}"
+        self.logger.info(f"Running game: {cmd}")
         response = self.environment.execute(cmd)
         assert response["returncode"] == 0, response
-        # For CoreWar, log_outputs and result_outputs are the same
-        output = [response["output"]]
-        return {OUTPUTS_LOGS: output, OUTPUTS_RESULTS: output}
+        return RoundData([response["output"]], [response["output"]])