From d72cb3ce93d444b61428000f75488826968c790e Mon Sep 17 00:00:00 2001 From: Openverse Builder Date: Mon, 1 Jan 2001 00:00:00 +0000 Subject: [PATCH] Add env.py from Openverse builder --- env.py | 237 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 237 insertions(+) create mode 100644 env.py diff --git a/env.py b/env.py new file mode 100644 index 0000000..a290cf2 --- /dev/null +++ b/env.py @@ -0,0 +1,237 @@ +```python +import re +import random +from typing import Any, Dict, List, Optional, Tuple + +import textarena as ta + + +class RunestoneClashEnv(ta.Env): + """ + Environment for "Runestone Clash": a deterministic turn-based two-player grid alignment battle. + Players alternate imprinting magical runes ("⚙" for A, "✶" for B) on a 3×3 Stone Circle. + The first to align three runes in a straight line wins. + """ + + def __init__(self, max_turns: int = 9): + self.max_turns = max_turns + # Compile regexes for quick validation + self.imprint_pattern = re.compile(r"^\[Imprint:(1|2|3),(1|2|3)\]$") + self.pass_pattern = re.compile(r"^\[Pass\]$") + + # ===================================================== + # Core Helpers + # ===================================================== + def _extract_answer_content(self, action: str) -> str: + """ + Extract boxed content from \boxed{...} for machine parsing. + Falls back to returning the raw trimmed string on parse failure. + """ + match = re.search(r"\\boxed\{([^}]*)\}", action, re.DOTALL) + if match: + return match.group(1).strip() + return action.strip() + + # ===================================================== + # Initialization + # ===================================================== + def reset(self, num_players: int, seed: Optional[int] = None): + """ + Resets the environment to an initial Runestone Clash state. + + Args: + num_players: Number of players (must be 2). + seed: Deterministic seed for reproducible starts. + + Returns: + None + """ + if num_players != 2: + raise ValueError("Runestone Clash requires exactly two players.") + + self.state = ta.TwoPlayerState(num_players=num_players, seed=seed, max_turns=self.max_turns) + rng = random.Random(seed) + starting_player = rng.choice([0, 1]) + + game_state: Dict[str, Any] = { + "turn_number": 1, + "active_player": "PlayerA" if starting_player == 0 else "PlayerB", + "rune_grid": [["" for _ in range(3)] for _ in range(3)], + "players": { + "PlayerA": {"symbol": "⚙", "imprints": 0, "skips": 0, "status": "active"}, + "PlayerB": {"symbol": "✶", "imprints": 0, "skips": 0, "status": "active"}, + }, + "winner": None, + "draw": False, + "transcript": [], + "seed": seed, + } + + # Set manually active player according to chosen start + self.state.reset( + game_state=game_state, + player_prompt_function=self._generate_player_prompt, + role_mapping={0: "PlayerA", 1: "PlayerB"} + ) + self.state.manually_set_current_player_id(starting_player) + + self.state.add_observation( + message="The Stone Circle hums with latent power. Runemages, prepare to begin.", + observation_type=ta.ObservationType.GAME_MESSAGE + ) + + return None + + # ===================================================== + # Player Prompt + # ===================================================== + def _generate_player_prompt(self, player_id: int, game_state: Dict[str, Any]) -> str: + """ + Compose the role prompt shown to the current Runemage. + """ + role_name = "Runemage A" if player_id == 0 else "Runemage B" + player_key = "PlayerA" if player_id == 0 else "PlayerB" + opponent_key = "PlayerB" if player_id == 0 else "PlayerA" + player_symbol = game_state["players"][player_key]["symbol"] + opponent_symbol = game_state["players"][opponent_key]["symbol"] + + grid = game_state["rune_grid"] + display_grid = "\n".join( + [ + " ".join(f"[{(cell if cell else ' ')}]" for cell in row) + for row in grid + ] + ) + open_cells = sum(1 for row in grid for c in row if c == "") + turn_number = game_state["turn_number"] + + prompt = ( + f"You are {role_name}, facing your rival in Runestone Clash.\n" + f"The current Stone Circle (3×3) state:\n{display_grid}\n" + f"Your sigil: {player_symbol}\nOpponent's sigil: {opponent_symbol}\n" + f"Turn {turn_number}, open cells remaining: {open_cells}\n\n" + f"Allowed actions:\n" + f" - [Imprint:x,y] : Imprint your rune at coordinates x,y (1–3) if empty.\n" + f" - [Pass] : Skip your turn, only if cells remain.\n" + f"Ensure syntax matches exactly (e.g., [Imprint:2,3]).\n\n" + "Put your final answer within \\boxed{} at the end of your response.\n\n" + "Example valid response:\n" + "I will secure the center of the Stone Circle.\n" + "\\boxed{[Imprint:2,2]}\n\n" + "Example valid response:\n" + "The board is tight; I will bide my time.\n" + "\\boxed{[Pass]}" + ) + return prompt + + # ===================================================== + # Step Logic + # ===================================================== + def step(self, action: str) -> Tuple[bool, ta.Info]: + """ + Perform a single environment step for the current player. + + Args: + action: Raw text from player action. + + Returns: + Tuple (done, info) + """ + # Log the raw message + self.state.add_observation( + message=action, + observation_type=ta.ObservationType.PLAYER_ACTION, + from_id=self.state.current_player_id, + to_id=-1 + ) + current_id = self.state.current_player_id + current_player_key = "PlayerA" if current_id == 0 else "PlayerB" + opponent_player_key = "PlayerB" if current_id == 0 else "PlayerA" + game_state = self.state.game_state + + # Extract boxed content and validate + parsed_action = self._extract_answer_content(action) + + match_imprint = self.imprint_pattern.match(parsed_action) + match_pass = self.pass_pattern.match(parsed_action) + grid = game_state["rune_grid"] + + def check_full_grid(g): + return all(c != "" for row in g for c in row) + + # -------------------- VALIDATION -------------------- + if not (match_imprint or match_pass): + self.state.set_invalid_move("Invalid syntax: does not match required action pattern.") + return self.state.step() + + if match_imprint: + x, y = int(match_imprint.group(1)), int(match_imprint.group(2)) + if not (1 <= x <= 3 and 1 <= y <= 3): + self.state.set_invalid_move(f"Invalid coordinates: cell ({x},{y}) is outside grid boundaries.") + return self.state.step() + if grid[x - 1][y - 1] != "": + self.state.set_invalid_move("Cell already claimed by another rune.") + return self.state.step() + + # Perform imprint + symbol = game_state["players"][current_player_key]["symbol"] + grid[x - 1][y - 1] = symbol + game_state["players"][current_player_key]["imprints"] += 1 + + game_state["transcript"].append({"player": current_player_key, "action": f"[Imprint:{x},{y}]"}) + self.state.add_observation( + message=f"{current_player_key} imprinted a rune at ({x},{y}).", + observation_type=ta.ObservationType.GAME_MESSAGE + ) + + elif match_pass: + if check_full_grid(grid): + self.state.set_invalid_move("Cannot pass: grid fully imprinted.") + return self.state.step() + game_state["players"][current_player_key]["skips"] += 1 + game_state["transcript"].append({"player": current_player_key, "action": "[Pass]"}) + self.state.add_observation( + message=f"{current_player_key} chose to pass this turn.", + observation_type=ta.ObservationType.GAME_MESSAGE + ) + + # -------------------- GAME STATE UPDATE -------------------- + game_state["turn_number"] += 1 + + # -------------------- WIN CHECK -------------------- + def check_win(symbol: str) -> bool: + g = grid + # Rows, columns + for i in range(3): + if g[i][0] == g[i][1] == g[i][2] == symbol and symbol != "": + return True + if g[0][i] == g[1][i] == g[2][i] == symbol and symbol != "": + return True + # Diagonals + if g[0][0] == g[1][1] == g[2][2] == symbol and symbol != "": + return True + if g[0][2] == g[1][1] == g[2][0] == symbol and symbol != "": + return True + return False + + current_symbol = game_state["players"][current_player_key]["symbol"] + if check_win(current_symbol): + game_state["winner"] = current_player_key + game_state["players"][current_player_key]["status"] = "won" + game_state["players"][opponent_player_key]["status"] = "lost" + self.state.set_winner(player_id=current_id, reason=f"{current_player_key} aligned three runes and harnessed the Stone Circle!") + return self.state.step() + + # -------------------- DRAW CHECK -------------------- + if check_full_grid(grid) or game_state["turn_number"] > 9: + game_state["draw"] = True + self.state.set_draw(reason="The Stone Circle is filled; no alignment achieved.") + return self.state.step() + + # -------------------- NEXT TURN -------------------- + next_player = (current_id + 1) % 2 + game_state["active_player"] = "PlayerA" if next_player == 0 else "PlayerB" + self.state.manually_set_current_player_id(next_player) + + return self.state.step() +``` \ No newline at end of file