Add sims_per_round flag

Author: john-b-yang

codeclash/agents/minisweagent.py

@@ -17,7 +17,7 @@
 
 from codeclash.agents.abstract import Player
 from codeclash.agents.utils import GameContext, resolve_api_key
-from codeclash.utils.environment import copy_file_to_container
+from codeclash.utils.environment import copy_to_container
 
 
 class ClashAgent(DefaultAgent):
@@ -107,7 +107,7 @@ def run(self):
                 result=result,
                 print_fct=self.logger.debug,
             )
-            copy_file_to_container(
+            copy_to_container(
                 self.environment,
                 traj_path,
                 self.game_context.log_env / traj_path.name,

codeclash/constants.py

@@ -3,4 +3,6 @@
 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

@@ -7,7 +7,13 @@
 from minisweagent.environments.docker import DockerEnvironment
 
 from codeclash.agents.abstract import Player
-from codeclash.constants import DIR_LOGS, DIR_WORK, GH_ORG
+from codeclash.constants import (
+    DIR_LOGS,
+    DIR_WORK,
+    GH_ORG,
+    OUTPUTS_LOGS,
+    OUTPUTS_RESULTS,
+)
 from codeclash.utils.environment import assert_zero_exit_code, copy_between_containers
 from codeclash.utils.log import get_logger
 
@@ -139,12 +145,12 @@ def _pre_round_setup(self, agents: list[Player]):
 
     @abstractmethod
     def determine_winner(
-        self, result_output: str, agents: list[Player]
+        self, result_outputs: list[str], agents: list[Player]
     ) -> dict[str, str]:
         """Determine the winner of the game based on the result output.
 
         Args:
-            result_output: The specific output containing winning information
+            result_outputs: The specific output(s) containing winning information
             agents: List of agents participating in the round
 
         Returns:
@@ -153,13 +159,13 @@ def determine_winner(
         pass
 
     @abstractmethod
-    def execute_round(self, agents: list[Player]) -> dict[str, str]:
+    def execute_round(self, agents: list[Player]) -> dict[str, list[str]]:
         """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_output" and "result_output"
+            Dictionary with keys "log_outputs" and "result_outputs"
         """
         pass
 
@@ -172,14 +178,14 @@ def run_round(self, agents: list[Player]) -> dict[str, str]:
         """
         self._pre_round_setup(agents)
         result = self.execute_round(agents)
-        log_output = result["log_output"]
-        result_output = result["result_output"]
+        log_outputs = result[OUTPUTS_LOGS]
+        result_outputs = result[OUTPUTS_RESULTS]
 
-        winner_result = self.determine_winner(result_output, agents)
+        winner_result = self.determine_winner(result_outputs, agents)
         winner_name = winner_result["winner"]
 
         return {
-            "log_output": log_output,
-            "result_output": result_output,
+            OUTPUTS_LOGS: log_outputs,
+            OUTPUTS_RESULTS: result_outputs,
             "winner": winner_name,
         }

codeclash/games/battlecode/main.py

@@ -1,9 +1,10 @@
 import re
-import shlex
 from pathlib import Path
 from typing import Any
 
-from codeclash.constants import DIR_WORK, RESULT_TIE
+from tqdm.auto import tqdm
+
+from codeclash.constants import DIR_WORK, OUTPUTS_LOGS, OUTPUTS_RESULTS, RESULT_TIE
 from codeclash.games.abstract import CodeGame
 
 
@@ -23,27 +24,30 @@ def __init__(self, config, *, tournament_id: str, local_output_dir: Path):
             else:
                 self.run_cmd_round += f" --{arg} {val}"
 
-    def determine_winner(self, result_output: str, agents: list[Any]) -> dict[str, str]:
-        self.logger.debug(f"Determining winner from result output: {result_output}")
-        lines = result_output.strip().split("\n")
-        # Get the third-to-last line which contains the winner info
-        winner_line = lines[-3] if len(lines) >= 3 else ""
-        self.logger.debug(f"Winner line: {winner_line}")
-        match = re.search(r"\s\((.*)\)\swins\s\(", winner_line)
-        if match:
-            winner_key = match.group(1)
-            self.logger.debug(f"Winner key from match: {winner_key}")
-            # Map A/B to actual agent names (much closer to original code)
-            winner = {"A": agents[0].name, "B": agents[1].name}.get(
-                winner_key, RESULT_TIE
-            )
-            self.logger.debug(f"Concluding winner: {winner}")
-            return {"winner": winner}
-        else:
-            self.logger.debug("No winner match found, returning tie")
-            return {"winner": RESULT_TIE}
+    def determine_winner(
+        self, result_outputs: list[str], agents: list[Any]
+    ) -> dict[str, str]:
+        winners = []
+        for ro in result_outputs:
+            lines = ro.strip().split("\n")
+            # Get the third-to-last line which contains the winner info
+            winner_line = lines[-3] if len(lines) >= 3 else ""
+            self.logger.debug(f"Winner line: {winner_line}")
+            match = re.search(r"\s\((.*)\)\swins\s\(", winner_line)
+            if match:
+                winner_key = match.group(1)
+                self.logger.debug(f"Winner key from match: {winner_key}")
+                # Map A/B to actual agent names (much closer to original code)
+                winner = {"A": agents[0].name, "B": agents[1].name}.get(
+                    winner_key, RESULT_TIE
+                )
+                winners.append(winner)
+            else:
+                winners.append(RESULT_TIE)
+        winner = max(set(winners), key=winners.count)
+        return {"winner": winner}
 
-    def execute_round(self, agents: list[Any]) -> dict[str, str]:
+    def execute_round(self, agents: list[Any]) -> dict[str, list[str]]:
         for agent in agents:
             src, dest = f"/{agent.name}/src/mysubmission/", str(
                 DIR_WORK / "src" / agent.name
@@ -53,10 +57,12 @@ def execute_round(self, agents: list[Any]) -> dict[str, str]:
             f"--p{idx+1}-dir src --p{idx+1} {agent.name}"
             for idx, agent in enumerate(agents)
         ]
-        cmd = f"{self.run_cmd_round} {shlex.join(args)}"
+        cmd = f"{self.run_cmd_round} {' '.join(args)}"
         self.logger.info(f"Running command: {cmd}")
-        response = self.environment.execute(cmd)
-        assert response["returncode"] == 0, response
-        # For BattleCode, log_output and result_output are the same
-        output = response["output"]
-        return {"log_output": output, "result_output": output}
+        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}

codeclash/games/battlesnake/main.py

@@ -3,6 +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.utils.environment import assert_zero_exit_code
 
@@ -23,18 +24,20 @@ def __init__(self, config, *, tournament_id: str, local_output_dir: Path):
                 self.run_cmd_round += f" --{arg} {val}"
 
     def determine_winner(
-        self, result_output: str, agents: list[Player]
+        self, result_outputs: list[str], agents: list[Player]
     ) -> dict[str, str]:
-        self.logger.debug(f"Determining winner from result output: {result_output}")
-        lines = result_output.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}")
-        winner = json.loads(last_line)["winnerName"]
-        self.logger.debug(f"Concluding winner: {winner}")
+        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}")
+            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, str]:
+    def execute_round(self, agents: list[Player]) -> dict[str, list[str]]:
         cmd = []
         for idx, agent in enumerate(agents):
             port = 8001 + idx
@@ -46,27 +49,33 @@ def execute_round(self, agents: list[Player]) -> dict[str, str]:
 
         time.sleep(3)  # Give servers time to start
 
-        # Create temporary output file for results
-        output_file = f"battlesnake_output_{int(time.time())}.json"
-        cmd_str = " ".join(cmd) + f" -o {output_file}"
-        self.logger.info(f"Running command: {self.run_cmd_round} {cmd_str}")
-
         try:
-            response = assert_zero_exit_code(
-                self.environment.execute(
-                    f"{self.run_cmd_round} {cmd_str}",
-                    cwd=f"{self.environment.config.cwd}/game",
+            log_outputs, result_outputs = [], []
+            for idx in 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}")
+
+                response = assert_zero_exit_code(
+                    self.environment.execute(
+                        f"{self.run_cmd_round} {cmd_str}",
+                        cwd=f"{self.environment.config.cwd}/game",
+                    )
                 )
-            )
 
-            # Read the output file for result information
-            result_response = self.environment.execute(f"cat game/{output_file}")
-            result_output = result_response["output"]
+                # Read the output file for result information
+                result_response = self.environment.execute(f"cat game/{output_file}")
+                result_output = result_response["output"]
+                log_outputs.append(response["output"])
+                result_outputs.append(result_output)
+
+                # Clean up the output file
+                self.environment.execute(f"rm -f game/{output_file}")
 
-            # Clean up the output file
-            self.environment.execute(f"rm -f game/{output_file}")
+                time.sleep(0.1)
 
-            return {"log_output": response["output"], "result_output": result_output}
+            return {OUTPUTS_LOGS: log_outputs, OUTPUTS_RESULTS: result_outputs}
         finally:
             # Kill all python servers when done
             self.environment.execute("pkill -f 'python main.py' || true")

codeclash/games/corewar/main.py

@@ -3,6 +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
 
 
@@ -22,8 +23,9 @@ def __init__(self, config, *, tournament_id: str, local_output_dir: Path):
                 self.run_cmd_round += f" -{arg} {val}"
 
     def determine_winner(
-        self, result_output: str, agents: list[Player]
+        self, result_outputs: list[str], agents: list[Player]
     ) -> dict[str, str]:
+        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
@@ -51,12 +53,13 @@ def determine_winner(
             self.logger.debug("No scores found, returning unknown")
             return {"winner": "unknown"}
 
-    def execute_round(self, agents: list[Player]) -> dict[str, str]:
+    def execute_round(self, agents: list[Player]) -> dict[str, list[str]]:
         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}")
         response = self.environment.execute(cmd)
         assert response["returncode"] == 0, response
-        # For CoreWar, log_output and result_output are the same
-        output = response["output"]
-        return {"log_output": output, "result_output": output}
+        # For CoreWar, log_outputs and result_outputs are the same
+        output = [response["output"]]
+        return {OUTPUTS_LOGS: output, OUTPUTS_RESULTS: output}

codeclash/games/robocode/main.py

@@ -3,8 +3,9 @@
 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.utils.environment import copy_file_to_container
+from codeclash.utils.environment import copy_to_container
 
 
 class RoboCodeGame(CodeGame):
@@ -25,7 +26,7 @@ def __init__(self, config, *, tournament_id: str, local_output_dir: Path):
     def _get_battle_config(self) -> str:
         default_battle_config = {
             "battle": {
-                "numRounds": 10,
+                "numRounds": self.game_config.get("sims_per_round", 100),
                 "gunCoolingRate": 0.1,
                 "rules": {"inactivityTime": 450, "hideEnemyNames": True},
             },
@@ -56,12 +57,13 @@ def dict_to_lines(d, prefix=""):
         return "\n".join(battle_lines)
 
     def determine_winner(
-        self, result_output: str, agents: list[Player]
+        self, result_outputs: list[str], agents: list[Player]
     ) -> dict[str, str]:
+        result_output = result_outputs[0]  # Get the first (and only) element
         self.logger.debug(f"Determining winner from result output: {result_output}")
         lines = result_output.strip().split("\n")
         # Get the second line which contains the winner info (closer to original)
-        winner_line = lines[1] if len(lines) >= 2 else ""
+        winner_line = lines[2] if len(lines) >= 3 else ""
         self.logger.debug(f"Winner line: {winner_line}")
         if winner_line:
             winner = winner_line.split()[1].rsplit(".", 1)[0]
@@ -71,7 +73,7 @@ def determine_winner(
             self.logger.debug("No winner line found, returning unknown")
             return {"winner": "unknown"}
 
-    def execute_round(self, agents: list[Player]) -> dict[str, str]:
+    def execute_round(self, agents: list[Player]) -> dict[str, list[str]]:
         for agent in agents:
             # Copy the agent codebase into the game codebase and compile it
             for cmd in [
@@ -93,7 +95,7 @@ def execute_round(self, agents: list[Player]) -> dict[str, str]:
 robocode.battle.selectedRobots={selected_robots}
 """
             )
-        copy_file_to_container(self.environment, battle_file, f"battles/{battle_file}")
+        copy_to_container(self.environment, battle_file, f"battles/{battle_file}")
         subprocess.run(f"rm -f {battle_file}", shell=True)
 
         # Run battle with results output to file
@@ -110,4 +112,4 @@ def execute_round(self, agents: list[Player]) -> dict[str, str]:
         # Clean up the results file
         self.environment.execute(f"rm -f {results_file}")
 
-        return {"log_output": response["output"], "result_output": result_output}
+        return {OUTPUTS_LOGS: [response["output"]], OUTPUTS_RESULTS: [result_output]}