Make get_stats operate on local logs

Author: john-b-yang

codeclash/constants.py

@@ -2,5 +2,6 @@
 
 DIR_LOGS = Path("logs")
 DIR_WORK = Path("/testbed")
+FILE_RESULTS = "results.json"
 GH_ORG = "emagedoc"
 RESULT_TIE = "Tie"

codeclash/games/battlecode/battlecode.py

@@ -29,13 +29,13 @@ 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" / str(round_num) / f"sim_{round_num}.log",
+            dest_path=self.log_round(round_num) / f"sim_{round_num}.log",
         )
 
-    def get_stats(self, agents: list[Any]) -> RoundStats:
+    def get_stats(self, agents: list[Any], round_num: int) -> RoundStats:
         winners = []
-        ro = self.environment.execute(f"cat {BATTLECODE_LOG}")["output"]
-        lines = ro.strip().split("\n")
+        with open(self.log_round(round_num) / f"sim_{round_num}.log") as f:
+            lines = f.read().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}")

codeclash/games/battlesnake/battlesnake.py

@@ -43,17 +43,17 @@ def copy_logs_from_env(self, round_num):
         copy_from_container(
             container=self.environment,
             src_path=f"{self.environment.config.cwd}/game/logs",
-            dest_path=self.log_local / "rounds" / str(round_num),
+            dest_path=self.log_round(round_num),
         )
 
-    def get_stats(self, agents: list[Player]) -> RoundStats:
+    def get_stats(self, agents: list[Player], round_num: int) -> RoundStats:
         scores = {}
         for idx in range(self.game_config["sims_per_round"]):
-            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"]
-            scores[winner] = scores.get(winner, 0) + 1
+            with open(self.log_round(round_num) / f"logs/sim_{idx}.jsonl") as f:
+                lines = f.read().strip().split("\n")
+                results = json.loads(lines[-1])  # Get the last line which contains the game result
+                winner = RESULT_TIE if results["isDraw"] else results["winnerName"]
+                scores[winner] = scores.get(winner, 0) + 1
 
         winner = max(scores, key=scores.get)
         winner = RESULT_TIE if list(scores.values()).count(scores[winner]) > 1 else winner

codeclash/games/corewar/corewar.py

@@ -6,6 +6,8 @@
 from codeclash.games.game import CodeGame, RoundStats
 from codeclash.utils.environment import copy_from_container
 
+COREWAR_LOG = "sim.log"
+
 
 class CoreWarGame(CodeGame):
     name: str = "CoreWar"
@@ -24,12 +26,13 @@ def copy_logs_from_env(self, round_num: int) -> None:
         super().copy_logs_from_env(round_num)
         copy_from_container(
             container=self.environment,
-            src_path="/testbed/output.log",
-            dest_path=self.log_local / "rounds" / str(round_num) / "output.log",
+            src_path=f"/testbed/{COREWAR_LOG}",
+            dest_path=self.log_round(round_num) / COREWAR_LOG,
         )
 
-    def get_stats(self, agents: list[Player]) -> RoundStats:
-        result_output = self.environment.execute("cat output.log")["output"]
+    def get_stats(self, agents: list[Player], round_num: int) -> RoundStats:
+        with open(self.log_round(round_num) / COREWAR_LOG) as f:
+            result_output = f.read()
         self.logger.debug(f"Determining winner from result output: {result_output}")
         scores = []
         n = len(agents) * 2
@@ -61,7 +64,7 @@ def get_stats(self, agents: list[Player]) -> RoundStats:
 
     def execute_round(self, agents: list[Player]):
         args = [f"/{agent.name}/warriors/warrior.red" for agent in agents]
-        cmd = f"{self.run_cmd_round} {shlex.join(args)} -r {self.game_config['sims_per_round']} > output.log;"
+        cmd = f"{self.run_cmd_round} {shlex.join(args)} -r {self.game_config['sims_per_round']} > {COREWAR_LOG};"
         self.logger.info(f"Running game: {cmd}")
         response = self.environment.execute(cmd)
         assert response["returncode"] == 0, response

codeclash/games/dummy/dummy_game.py

@@ -13,12 +13,13 @@ def copy_logs_from_env(self, round_num):
         copy_from_container(
             container=self.environment,
             src_path="/testbed/result.log",
-            dest_path=self.log_local / "rounds" / str(round_num) / "result.log",
+            dest_path=self.log_round(round_num) / "result.log",
         )
 
-    def get_stats(self, agents: list[Player]) -> RoundStats:
-        result_output = self.environment.execute("cat result.log")["output"]
-        lines = result_output.split("FINAL_RESULTS")[-1].splitlines()
+    def get_stats(self, agents: list[Player], round_num: int) -> RoundStats:
+        with open(self.log_round(round_num) / "result.log") as f:
+            round_log = f.read()
+        lines = round_log.split("FINAL_RESULTS")[-1].splitlines()
 
         scores = {}
         for line in lines:

codeclash/games/dummy/main.py

@@ -99,11 +99,9 @@ def build_image(self):
             self.logger.error(f"❌ Failed to build Docker image: {result.stderr}\n{result.stdout}{result.stderr}")
             raise RuntimeError(f"Failed to build Docker image: {result.stderr}")
 
-    def get_metadata(self) -> dict:
-        """This is what we write to metadata.json.
-        You can subclass extend this to add more details for specific games.
-        """
-        return self._metadata
+    def copy_logs_from_env(self, round_num: int) -> None:
+        """Copy logs from the game's environment to the local machine."""
+        (self.log_local / "rounds" / str(round_num)).mkdir(parents=True, exist_ok=True)
 
     def end(self, cleanup: bool = False):
         if cleanup:
@@ -112,6 +110,9 @@ def end(self, cleanup: bool = False):
                     subprocess.run(f"rm -rf {artifact}", shell=True)
             self.logger.info(f"🧼 Cleaned up {self.name} game")
 
+    def log_round(self, round_num: int) -> Path:
+        return self.log_local / "rounds" / str(round_num)
+
     def get_environment(self, branch_name: str | None = None) -> DockerEnvironment:
         """Get docker container ID with the game code installed."""
         self.build_image()
@@ -142,6 +143,12 @@ def get_environment(self, branch_name: str | None = None) -> DockerEnvironment:
             assert_zero_exit_code(environment.execute(cmd), logger=self.logger)
         return environment
 
+    def get_metadata(self) -> dict:
+        """This is what we write to metadata.json.
+        You can subclass extend this to add more details for specific games.
+        """
+        return self._metadata
+
     def _pre_round_setup(self, agents: list[Player]):
         """Copy agent codebases into game's container"""
         for agent in agents:
@@ -158,12 +165,20 @@ def _pre_round_setup(self, agents: list[Player]):
             logger=self.logger,
         )
 
-    def copy_logs_from_env(self, round_num: int) -> None:
-        """Copy logs from the game's environment to the local machine."""
-        (self.log_local / "rounds" / str(round_num)).mkdir(parents=True, exist_ok=True)
+    def run_round(self, agents: list[Player], round_num: int) -> RoundStats:
+        """
+        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)
+        self.execute_round(agents)
+        self.copy_logs_from_env(round_num)
+        return self.get_stats(agents, round_num)
 
     @abstractmethod
-    def get_stats(self, agents: list[Player]) -> RoundStats:
+    def get_stats(self, agents: list[Player], round_num: int) -> RoundStats:
         """Determine the winner of the game based on the result output.
 
         Args:
@@ -181,16 +196,3 @@ def execute_round(self, agents: list[Player]):
         includes the pre-round setup, post-round setup, and winner determination.
         """
         pass
-
-    def run_round(self, agents: list[Player], round_num: int) -> RoundStats:
-        """
-        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)
-        self.execute_round(agents)
-        stats = self.get_stats(agents)
-        self.copy_logs_from_env(round_num)
-        return stats

codeclash/games/game.py

@@ -2,6 +2,9 @@
 
 from codeclash.agents.player import Player
 from codeclash.games.game import CodeGame, RoundStats
+from codeclash.utils.environment import copy_from_container
+
+HB_LOG_DIR = Path("/testbed/engine/logs/")
 
 
 class HuskyBenchGame(CodeGame):
@@ -19,15 +22,27 @@ def __init__(self, config, *, tournament_id: str, local_output_dir: Path):
             else:
                 self.run_cmd_round += f" --{arg} {val}"
 
-    def get_stats(self, result_outputs: list[str], agents: list[Player]) -> RoundStats:
+    def copy_logs_from_env(self, round_num):
+        super().copy_logs_from_env(round_num)
+        log_path = self.log_round(round_num)
+        copy_from_container(
+            container=self.environment,
+            src_path=HB_LOG_DIR,
+            dest_path=log_path,
+        )
+
+    def get_stats(self, agents: list[Player], round_num: int) -> RoundStats:
         return RoundStats(winner="N/A", scores={})
 
     def execute_round(self, agents: list[Player]):
+        self.environment.execute(f"rm -rf {HB_LOG_DIR}; mkdir -p {HB_LOG_DIR}")
         try:
             self.logger.debug("Starting game servers")
-            self.environment.execute(self.run_cmd_round + " > output.log &")
+            self.environment.execute(f"{self.run_cmd_round} > {HB_LOG_DIR / 'engine.log'} &")
             for agent in agents:
-                self.environment.execute("python client/main.py --port 8000 &", cwd=f"/{agent.name}")
+                self.environment.execute(
+                    f"python client/main.py --port 8000 > {HB_LOG_DIR / f'{agent.name}.log'} &", cwd=f"/{agent.name}"
+                )
         finally:
             # Kill all python servers when done
             self.environment.execute("pkill -f 'python client/main.py' || true")

codeclash/games/huskybench/huskybench.py

@@ -58,11 +58,12 @@ def copy_logs_from_env(self, round_num: int) -> None:
         copy_from_container(
             container=self.environment,
             src_path="/testbed/logs",
-            dest_path=self.log_local / "rounds" / str(round_num),
+            dest_path=self.log_round(round_num),
         )
 
-    def get_stats(self, agents: list[Player]) -> RoundStats:
-        result_output = self.environment.execute("cat logs/results.txt")["output"]
+    def get_stats(self, agents: list[Player], round_num: int) -> RoundStats:
+        with open(self.log_round(round_num) / "logs/results.txt") as f:
+            result_output = f.read()
         print(result_output)
         lines = result_output.strip().split("\n")
 

codeclash/games/robocode/robocode.py

@@ -21,14 +21,14 @@ def copy_logs_from_env(self, round_num: int) -> None:
         copy_from_container(
             container=self.environment,
             src_path="/testbed/logs",
-            dest_path=self.log_local / "rounds" / str(round_num),
+            dest_path=self.log_round(round_num),
         )
 
-    def get_stats(self, agents: list[Player]) -> RoundStats:
+    def get_stats(self, agents: list[Player], round_num: int) -> RoundStats:
         winners = []
         for idx in range(self.game_config.get("sims_per_round", 100)):
-            ro = self.environment.execute(f"cat logs/sim_{idx}.txt")["output"]
-            lines = ro.strip().split("\n")
+            with open(self.log_round(round_num) / f"logs/sim_{idx}.txt") as f:
+                lines = f.read().strip().split("\n")
 
             # Get the last 2 lines which contain the game result (same as original)
             relevant_lines = lines[-2:] if len(lines) >= 2 else lines