|
| 1 | +import re |
| 2 | +import shlex |
| 3 | +import subprocess |
| 4 | +from collections import Counter |
| 5 | +from concurrent.futures import ThreadPoolExecutor, as_completed |
| 6 | + |
| 7 | +from tqdm.auto import tqdm |
| 8 | + |
| 9 | +from codeclash.agents.player import Player |
| 10 | +from codeclash.constants import RESULT_TIE |
| 11 | +from codeclash.games.game import CodeGame, RoundStats |
| 12 | + |
| 13 | +HALITE_LOG = "sim_{idx}.log" |
| 14 | + |
| 15 | + |
| 16 | +class HaliteGame(CodeGame): |
| 17 | + name: str = "Halite" |
| 18 | + description: str = """Halite is a strategic programming game where players write bots to control ships that gather resources, build structures, and compete for dominance on a grid-based map. |
| 19 | +Victory is achieved by outmaneuvering opponents, optimizing resource collection, and strategically expanding your territory""" |
| 20 | + default_args: dict = {} |
| 21 | + |
| 22 | + def __init__(self, config, **kwargs): |
| 23 | + super().__init__(config, **kwargs) |
| 24 | + self.run_cmd_round: str = f"./environment/halite --replaydirectory {self.log_env}" |
| 25 | + for arg, val in self.game_config.get("args", self.default_args).items(): |
| 26 | + if isinstance(val, bool): |
| 27 | + if val: |
| 28 | + self.run_cmd_round += f" --{arg}" |
| 29 | + else: |
| 30 | + self.run_cmd_round += f" --{arg} {val}" |
| 31 | + |
| 32 | + def _run_single_simulation(self, agents: list[Player], idx: int, cmd: str): |
| 33 | + """Run a single halite simulation and return the output.""" |
| 34 | + cmd = f"{cmd} > {self.log_env / HALITE_LOG.format(idx=idx)}" |
| 35 | + |
| 36 | + # Run the simulation and return the output |
| 37 | + try: |
| 38 | + response = self.environment.execute(cmd, timeout=120) |
| 39 | + except subprocess.TimeoutExpired: |
| 40 | + self.logger.warning(f"Halite simulation {idx} timed out: {cmd}") |
| 41 | + return |
| 42 | + if response["returncode"] != 0: |
| 43 | + self.logger.warning( |
| 44 | + f"Halite simulation {idx} failed with exit code {response['returncode']}:\n{response['output']}" |
| 45 | + ) |
| 46 | + |
| 47 | + def execute_round(self, agents: list[Player]): |
| 48 | + entries = [] |
| 49 | + for agent in agents: |
| 50 | + entries.append(f"python /{agent.name}/airesources/Python/RandomBot.py") |
| 51 | + cmd = f"{self.run_cmd_round} {shlex.join(entries)}" |
| 52 | + self.logger.info(f"Running game: {cmd}") |
| 53 | + with ThreadPoolExecutor(5) as executor: |
| 54 | + futures = [ |
| 55 | + executor.submit(self._run_single_simulation, agents, idx, cmd) |
| 56 | + for idx in range(self.game_config["sims_per_round"]) |
| 57 | + ] |
| 58 | + for future in tqdm(as_completed(futures), total=len(futures)): |
| 59 | + future.result() |
| 60 | + |
| 61 | + def get_results(self, agents: list[Player], round_num: int, stats: RoundStats): |
| 62 | + winners = [] |
| 63 | + pattern = r"Player\s#(\d+),\s(.*),\scame\sin\srank\s#(\d+)" |
| 64 | + for idx in range(self.game_config["sims_per_round"]): |
| 65 | + log_file = self.log_round(round_num) / HALITE_LOG.format(idx=idx) |
| 66 | + with open(log_file) as f: |
| 67 | + lines = f.readlines()[-len(agents) - 1 :] |
| 68 | + for line in lines: |
| 69 | + match = re.search(pattern, line) |
| 70 | + if match: |
| 71 | + player_idx = int(match.group(1)) - 1 |
| 72 | + rank = int(match.group(3)) |
| 73 | + if rank == 1: |
| 74 | + winners.append(agents[player_idx].name) |
| 75 | + |
| 76 | + # Count wins |
| 77 | + win_counts = Counter(winners) |
| 78 | + |
| 79 | + # Find all winners with the maximum count |
| 80 | + max_wins = max(win_counts.values(), default=0) |
| 81 | + overall_winners = [name for name, count in win_counts.items() if count == max_wins] |
| 82 | + |
| 83 | + # Update stats |
| 84 | + stats.winner = RESULT_TIE if len(overall_winners) > 1 else overall_winners[0] |
| 85 | + stats.scores = dict(win_counts) |
| 86 | + for player, score in win_counts.items(): |
| 87 | + if player != RESULT_TIE: |
| 88 | + stats.player_stats[player].score = score |
| 89 | + |
| 90 | + def validate_code(self, agent: Player) -> tuple[bool, str | None]: |
| 91 | + return True, None |
0 commit comments