#!/usr/bin/env python3 """ Monop IRC game server: runs monop in a subprocess, bridges stdin/stdout to IRC. Players prefix commands with a dot (.) in the channel. Also includes scripted AI players for testing. """ import os import re import ssl import sys import time import pexpect import irc.client import irc.connection SERVER = "irc.darkscience.net" PORT = 6697 CHANNEL = "#monop-dev" NICK = "monopoly" MONOP = "/usr/games/monop" NUM_PLAYERS = 3 PLAYER_NAMES = ["Alice", "Bob", "Charlie"] class MonopServer: def __init__(self): self.connection = None self.child = None self.joined = False self.game_started = False self.setup_phase = True self.setup_step = 0 # 0=waiting, 1=num_players sent, 2+=names self.output_queue = [] self.last_output_time = 0 self.auto_play = True self.turn_count = 0 def start_monop(self): """Start the monop process.""" self.child = pexpect.spawn(MONOP, encoding='utf-8', timeout=2) print("monop process started", flush=True) def read_monop(self): """Read any available output from monop.""" lines = [] try: while True: # Read one line at a time self.child.expect('\r?\n', timeout=0.3) text = self.child.before.strip() if text: lines.append(text) except pexpect.TIMEOUT: # Also grab any partial prompt if self.child.before: partial = self.child.before.strip() if partial and partial not in lines: lines.append(partial) except pexpect.EOF: pass return lines def send_to_monop(self, text): """Send a line to monop stdin.""" if self.child and self.child.isalive(): print(f" >> monop: {text!r}", flush=True) self.child.sendline(text) time.sleep(0.2) def send_to_irc(self, text): """Send a line to IRC channel.""" if self.connection and text.strip(): # Truncate long lines for line in text.split('\n'): line = line.strip()[:450] if line: try: self.connection.privmsg(CHANNEL, line) time.sleep(0.15) except Exception as e: print(f"IRC send error: {e}", flush=True) def on_connect(self, conn, event): print(f"Connected to IRC", flush=True) self.connection = conn time.sleep(2) # wait before joining conn.join(CHANNEL) def on_join(self, conn, event): if event.source.nick == NICK: print(f"Joined {CHANNEL}", flush=True) self.joined = True self.connection.privmsg(CHANNEL, "🎲 Monopoly game starting! Setting up with AI players...") time.sleep(1) self.start_monop() self.run_setup() def on_pubmsg(self, conn, event): """Handle player commands (dot-prefixed).""" nick = event.source.nick if event.source else "" msg = event.arguments[0] if event.arguments else "" if nick == NICK: return # ignore own messages if msg.startswith("."): cmd = msg[1:].strip() print(f" Player command from {nick}: {cmd!r}", flush=True) self.send_to_monop(cmd) time.sleep(0.5) self.flush_and_send() def run_setup(self): """Automated game setup: set player count and names.""" time.sleep(1) lines = self.read_monop() for l in lines: print(f" monop: {l}", flush=True) self.send_to_irc(l) # Send number of players self.send_to_monop(str(NUM_PLAYERS)) time.sleep(0.5) self.flush_and_send() # Send player names for name in PLAYER_NAMES: time.sleep(0.5) self.send_to_monop(name) time.sleep(0.5) self.flush_and_send() # Read initial rolls time.sleep(1) self.flush_and_send() time.sleep(1) self.flush_and_send() self.setup_phase = False self.game_started = True self.connection.privmsg(CHANNEL, "Game is set up! Auto-playing turns now...") def flush_and_send(self): """Read monop output and relay to IRC.""" lines = self.read_monop() for l in lines: print(f" monop: {l}", flush=True) self.send_to_irc(l) return lines def auto_turn(self): """Play a turn automatically.""" if not self.game_started or not self.child or not self.child.isalive(): return # Send empty line (= roll / default action) self.send_to_monop("") time.sleep(0.8) lines = self.flush_and_send() # Handle prompts full = "\n".join(lines) if "Do you want to buy?" in full: self.send_to_monop("yes") time.sleep(0.5) self.flush_and_send() elif "lose 10%" in full or "10%% of your total" in full: # Income tax - choose $200 flat self.send_to_monop("$200") time.sleep(0.5) self.flush_and_send() elif "mortgage?" in full.lower() or "do you wish to" in full.lower(): self.send_to_monop("yes") time.sleep(0.5) self.flush_and_send() elif "Bid for" in full: self.send_to_monop("0") time.sleep(0.5) self.flush_and_send() elif "How much" in full: self.send_to_monop("0") time.sleep(0.5) self.flush_and_send() elif "Illegal response" in full: # Try different responses self.send_to_monop("$200") time.sleep(0.5) self.flush_and_send() self.turn_count += 1 def run(self): reactor = irc.client.Reactor() ssl_ctx = ssl.create_default_context() factory = irc.connection.Factory( wrapper=lambda s: ssl_ctx.wrap_socket(s, server_hostname=SERVER)) server = reactor.server() server.connect(SERVER, PORT, NICK, connect_factory=factory) server.add_global_handler("welcome", self.on_connect) server.add_global_handler("join", self.on_join) server.add_global_handler("pubmsg", self.on_pubmsg) print(f"Connecting to {SERVER}:{PORT} as {NICK}...", flush=True) start = time.time() last_auto = time.time() max_turns = 50 while time.time() - start < 300: # 5 min timeout reactor.process_once(timeout=0.5) # Auto-play turns every 3 seconds if self.auto_play and self.game_started and time.time() - last_auto > 3: if self.turn_count < max_turns: self.auto_turn() last_auto = time.time() elif self.turn_count == max_turns: self.connection.privmsg(CHANNEL, f"--- Auto-play complete ({max_turns} turns). Game state saved. ---") self.turn_count += 1 # prevent repeating print("Server shutting down", flush=True) if self.connection: self.connection.quit("Game over!") if __name__ == "__main__": ms = MonopServer() ms.run()