#!/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()