From 6ce682645f6e28e4ffe03debe6d810d401b587b8 Mon Sep 17 00:00:00 2001 From: Jarvis Date: Fri, 20 Feb 2026 21:10:03 +0000 Subject: [PATCH] Add game simulation script for testing --- test/simulate_game.py | 226 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 226 insertions(+) create mode 100644 test/simulate_game.py diff --git a/test/simulate_game.py b/test/simulate_game.py new file mode 100644 index 0000000..288e8ac --- /dev/null +++ b/test/simulate_game.py @@ -0,0 +1,226 @@ +#!/usr/bin/env python3 +"""Simulate a monop game on IRC for testing the board viewer bot. + +Runs monop in a subprocess, connects to IRC, and plays the game +with scripted AI players that make simple decisions. +""" + +import json +import os +import re +import select +import ssl +import subprocess +import sys +import time +import irc.client +import irc.connection + +CONFIG = { + "server": "irc.darkscience.net", + "port": 6697, + "tls": True, + "nick": "monop", + "channel": "#monop-dev", + "num_players": 3, + "player_names": ["Alice", "Bob", "Charlie"], +} + + +class MonopSimulator: + def __init__(self, config): + self.config = config + self.connection = None + self.channel = config["channel"] + self.proc = None + self.game_phase = "init" # init, setup, playing + self.setup_step = 0 + self.current_prompt = "" + self.output_buffer = "" + self.turn_count = 0 + self.max_turns = 60 # stop after this many turns + + def start_monop(self): + """Start the monop game process.""" + self.proc = subprocess.Popen( + ["/usr/games/monop"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + ) + # Set stdout to non-blocking + import fcntl + fd = self.proc.stdout.fileno() + fl = fcntl.fcntl(fd, fcntl.F_GETFL) + fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK) + print("monop process started", flush=True) + + def read_monop_output(self): + """Read available output from monop.""" + lines = [] + try: + while True: + data = self.proc.stdout.read(4096) + if not data: + break + self.output_buffer += data + except (BlockingIOError, IOError): + pass + + while "\n" in self.output_buffer: + line, self.output_buffer = self.output_buffer.split("\n", 1) + lines.append(line) + + # Check for prompts (lines without newline at end) + if self.output_buffer: + self.current_prompt = self.output_buffer + + return lines + + def send_to_monop(self, text): + """Send input to monop process.""" + if self.proc and self.proc.poll() is None: + print(f" -> monop input: {text.strip()}", flush=True) + self.proc.stdin.write(text + "\n") + self.proc.stdin.flush() + time.sleep(0.3) # give monop time to process + + def send_to_irc(self, text): + """Send a message to the IRC channel.""" + if self.connection and text.strip(): + # Split long lines + for line in text.split("\n"): + line = line.rstrip() + if line: + self.connection.privmsg(self.channel, line) + time.sleep(0.15) # rate limit + + def decide_action(self, prompt, output_lines): + """AI decision making for the scripted players.""" + full_text = "\n".join(output_lines) + "\n" + prompt + + # "How many players?" + if "How many players?" in full_text: + return str(self.config["num_players"]) + + # Player name prompts + for i, name in enumerate(self.config["player_names"]): + if f"Player {i+1}'s name:" in full_text: + return name + + # "Do you want to buy?" + if "Do you want to buy?" in prompt or "Do you want to buy?" in full_text: + return "yes" + + # yes/no questions - default yes + if prompt.strip().endswith("?") and ("yes" in prompt.lower() or "no" in prompt.lower()): + return "yes" + + # Command prompt (the game's main input) - just roll + if "-- Loss of $" in full_text: + return "" # just continue + + # If we see a command prompt, roll + return "" # empty = roll (default action) + + def on_connect(self, connection, event): + print(f"Connected to IRC. Joining {self.channel}", flush=True) + self.connection = connection + connection.join(self.channel) + + def on_join(self, connection, event): + if event.source.nick == self.config["nick"]: + print(f"Joined {self.channel}. Starting game in 3s...", flush=True) + time.sleep(3) + self.start_monop() + self.game_loop() + + def game_loop(self): + """Main game loop - read monop output, send to IRC, make decisions.""" + print("Starting game loop...", flush=True) + idle_count = 0 + + while self.proc and self.proc.poll() is None and self.turn_count < self.max_turns: + lines = self.read_monop_output() + + if lines: + idle_count = 0 + for line in lines: + print(f" monop: {line}", flush=True) + self.send_to_irc(line) + + # Track turns + if "rolled doubles" in line or "'s turn" in line: + self.turn_count += 1 + + # Check if monop is waiting for input + time.sleep(0.5) + more_lines = self.read_monop_output() + for line in more_lines: + lines.append(line) + print(f" monop: {line}", flush=True) + self.send_to_irc(line) + + # Make a decision if there's a prompt + if self.current_prompt or any("?" in l for l in lines): + decision = self.decide_action(self.current_prompt, lines) + if decision is not None: + time.sleep(0.5) + self.send_to_monop(decision) + self.current_prompt = "" + else: + # Game might be waiting for default input (roll) + time.sleep(1) + self.send_to_monop("") + else: + idle_count += 1 + if idle_count > 10: + # Monop might be waiting for input + time.sleep(0.5) + self.send_to_monop("") + idle_count = 0 + time.sleep(0.3) + + # Process IRC events + try: + self.reactor.process_once(timeout=0.1) + except Exception: + pass + + print(f"Game loop ended. Turns: {self.turn_count}", flush=True) + self.send_to_irc(f"--- Game simulation ended after {self.turn_count} turns ---") + if self.proc: + self.proc.terminate() + + def run(self): + config = self.config + self.reactor = irc.client.Reactor() + + connect_params = {} + if config.get("tls", False): + ssl_ctx = ssl.create_default_context() + hostname = config["server"] + ssl_factory = irc.connection.Factory( + wrapper=lambda s: ssl_ctx.wrap_socket(s, server_hostname=hostname)) + connect_params["connect_factory"] = ssl_factory + + server = self.reactor.server() + server.connect( + config["server"], + config["port"], + config["nick"], + **connect_params, + ) + + server.add_global_handler("welcome", self.on_connect) + server.add_global_handler("join", self.on_join) + + print(f"Connecting to {config['server']}:{config['port']}...", flush=True) + self.reactor.process_forever() + + +if __name__ == "__main__": + sim = MonopSimulator(CONFIG) + sim.run()