#!/usr/bin/env python3 """ Autopilot players for monop-irc. Spawns one IRC connection per player. Each player watches monop bot output and responds appropriately to keep the game moving. Players are not sophisticated — they just roll, buy when affordable, and handle prompts. Usage: python3 monop_players.py [--host HOST] [--port PORT] [--channel CHAN] [--players alice,bob] """ import socket import threading import time import re import random import argparse import sys HOST = "127.0.0.1" PORT = 6667 CHANNEL = "#monop" PREFIX = "." BOT_NICK = "monop" # How long to wait before responding (avoids flooding) RESPONSE_DELAY = 0.8 class PlayerBot: """A single IRC-connected monop player.""" def __init__(self, nick, channel, host, port, player_names, player_index): self.nick = nick self.channel = channel self.host = host self.port = port self.player_names = player_names self.player_index = player_index # which player number we are (0-based) self.num_players = len(player_names) self.sock = None self.buffer = "" self.lock = threading.Lock() # Game state tracking (minimal — just enough to respond correctly) self.setup_phase = True self.setup_registrations_seen = 0 self.current_player = None # whose turn it is self.my_money = 1500 self.in_jail = False self.jail_turns = 0 self.in_debt = False self.in_auction = False self.auction_bid = 0 self.awaiting_prompt = None # what kind of prompt we're waiting to answer self.game_started = False self.game_over = False self.rolled_this_turn = False # Properties we own (just names, for mortgage decisions) self.my_properties = [] self.mortgaged = set() # Track if we already responded to current prompt self._prompt_answered = False # Seen "goes first" yet self._first_player_announced = False # Trading state self.in_trade = False self.trade_props_offered = 0 self.turns_played = 0 def log(self, msg): print(f"[{self.nick}] {msg}", flush=True) def connect(self): """Connect to IRC and join channel.""" self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.connect((self.host, self.port)) self._send(f"NICK {self.nick}") self._send(f"USER {self.nick} 0 * :{self.nick}") # Wait for registration self.sock.settimeout(10) deadline = time.time() + 15 while time.time() < deadline: try: data = self.sock.recv(4096) except socket.timeout: self.log("Timeout waiting for registration") break if not data: raise ConnectionError("Connection closed during registration") self.buffer += data.decode("utf-8", errors="replace") # Handle pings during registration while "\r\n" in self.buffer: line, self.buffer = self.buffer.split("\r\n", 1) if line.startswith("PING"): self._send("PONG" + line[4:]) if " 433 " in line: # Nick in use — try with underscore self.nick = self.nick + "_" self.log(f"Nick in use, trying {self.nick}") self._send(f"NICK {self.nick}") if " 001 " in self.buffer: break self.sock.settimeout(None) self._send(f"JOIN {self.channel}") time.sleep(0.5) self.buffer = "" self.log(f"Connected and joined {self.channel}") # If we're the first player, send the player count immediately # in case "How many players?" was sent before we joined if self.player_index == 0: time.sleep(1.0) self.log("Sending player count (in case we missed the prompt)") self.say(str(self.num_players)) def _send(self, line): with self.lock: self.sock.sendall((line + "\r\n").encode()) def say(self, msg): """Send a prefixed message to the game channel.""" self._send(f"PRIVMSG {self.channel} :{PREFIX}{msg}") self.log(f" -> .{msg}") def say_delayed(self, msg, delay=None, force=False): """Send after a short delay. Rechecks is_my_turn() before sending.""" d = delay if delay is not None else RESPONSE_DELAY def _do(): time.sleep(d) if force or self.is_my_turn(): self.say(msg) else: self.log(f" (suppressed .{msg} — not my turn)") threading.Thread(target=_do, daemon=True).start() def is_my_turn(self): return (self.current_player and self.current_player.lower() == self.nick.lower()) def run(self): """Main loop: read IRC messages and respond.""" self.connect() while not self.game_over: try: data = self.sock.recv(4096) if not data: break self.buffer += data.decode("utf-8", errors="replace") while "\r\n" in self.buffer: line, self.buffer = self.buffer.split("\r\n", 1) self._handle_irc_line(line) except Exception as e: self.log(f"Error: {e}") break self.log("Exiting") def _handle_irc_line(self, line): """Handle a raw IRC line.""" if line.startswith("PING"): self._send("PONG" + line[4:]) return # Parse PRIVMSG from monop bot m = re.match(r':(\S+?)!\S+\s+PRIVMSG\s+(\S+)\s+:(.+)', line) if not m: return sender = m.group(1) target = m.group(2) message = m.group(3).strip() if target.lower() != self.channel.lower(): return if sender.lower() != BOT_NICK.lower(): return self._handle_bot_msg(message) def _handle_bot_msg(self, msg): """Process a message from the monop bot.""" if not msg: return # ============================================================ # SETUP PHASE # ============================================================ if self.setup_phase: if self._handle_setup(msg): return # ============================================================ # GAME OVER # ============================================================ if "WINS!!!!!" in msg or msg == "The party is over.": self.game_over = True self.log(f"Game over: {msg}") return # ============================================================ # CHECKPOINT LINE: "{name} ({N}) (cash ${M}) on {square}" # This marks the START of someone's turn. # ============================================================ m = re.match( r'^(.+?)\s+\((\d+)\)\s+\(cash\s+\$(-?\d+)\)\s+on\s+(.+)$', msg ) if m: name = m.group(1) money = int(m.group(3)) self.current_player = name self.in_debt = False self.awaiting_prompt = None self._prompt_answered = False self.rolled_this_turn = False if name.lower() == self.nick.lower(): self.my_money = money self.in_jail = False self.jail_turns = 0 if self.is_my_turn(): self.turns_played += 1 # ~10% chance to initiate a trade (after turn 5, not in debt) if (self.turns_played > 5 and not self.in_debt and random.random() < 0.10): self.in_trade = True self.trade_props_offered = 0 self.log("Initiating a trade!") self.say_delayed("trade") else: self.say_delayed("roll") self.rolled_this_turn = True return # ============================================================ # JAIL DETECTION # ============================================================ m = re.match(r'^\(This is your (\w+) turn in JAIL\)$', msg) if m: if self.is_my_turn(): self.in_jail = True ordinal = m.group(1) if ordinal == "3rd": self.jail_turns = 3 elif ordinal == "2nd": self.jail_turns = 2 else: self.jail_turns = 1 # Try rolling doubles self.say_delayed("roll") return if msg == "Double roll gets you out.": if self.is_my_turn(): self.in_jail = False return if msg.startswith("It's your third turn"): # Forced to pay $50 if self.is_my_turn(): self.in_jail = False return # ============================================================ # DOUBLES — ROLL AGAIN # ============================================================ m = re.match(r'^(.+?) rolled doubles\.\s+Goes again$', msg) if m: name = m.group(1) if name.lower() == self.nick.lower(): self.rolled_this_turn = False self.say_delayed("roll") self.rolled_this_turn = True return # Triple doubles -> jail if msg == "That's 3 doubles. You go to jail": if self.is_my_turn(): self.in_jail = True return # ============================================================ # BUY PROPERTY # ============================================================ if msg == "Do you want to buy?": if self.is_my_turn(): # Buy if we can afford it (we'll always say yes for simplicity) self.say_delayed("yes") return # Track what we bought m = re.match(r'^That would cost \$(\d+)$', msg) if m: # Just informational return # ============================================================ # AUCTION # ============================================================ if msg.startswith("So it goes up for auction"): self.in_auction = True self.auction_bid = 0 return if self.in_auction: # Bid prompt: "name:" (just the name followed by colon) m = re.match(r'^(.+?)\s*:\s*$', msg) if m: name = m.group(1).strip() if name.lower() == self.nick.lower(): if self.auction_bid == 0: # Open with a small bid self.say_delayed("1") elif self.auction_bid < min(200, self.my_money // 2): bid = self.auction_bid + random.randint(5, 20) self.say_delayed(str(bid)) else: # Drop out self.say_delayed("0") return m = re.match(r'^You must bid higher than (\d+)', msg) if m: self.auction_bid = int(m.group(1)) return if msg.startswith("It goes to") or msg.startswith("Nobody seems"): self.in_auction = False return # ============================================================ # YES/NO PROMPTS # ============================================================ if msg in ( "Do you want to mortgage it?", "Do you want to unmortgage it?", "Is that ok?", ): if self.is_my_turn(): self.say_delayed("yes") return # (Trade confirmation handled in TRADING section below) # Resign confirmation if msg == "Do you really want to resign?": if self.is_my_turn(): self.say_delayed("yes") return if msg.startswith("Who do you wish to resign to?"): if self.is_my_turn(): self.say_delayed("bank") return if msg.startswith("Which player do you wish to trade with?"): # We don't trade proactively — shouldn't hit this return # (Trade property prompt handled in TRADING section below) # ============================================================ # DEBT / FORCED MORTGAGE # ============================================================ if msg == "How are you going to fix it up?": self.in_debt = True if self.is_my_turn(): self.say_delayed("mortgage") return if msg == "-- You are now Solvent ---": self.in_debt = False return if msg == "that leaves you broke": return if msg == "You don't have any un-mortgaged property.": if self.is_my_turn() and self.in_debt: self.say_delayed("sell houses") return if msg == "You don't have any houses to sell!!": if self.is_my_turn() and self.in_debt: self.say_delayed("resign") return # Mortgage property selection if msg == "Which property do you want to mortgage?": if self.is_my_turn(): if self.in_debt: self.say_delayed("?") # get list, pick first else: self.say_delayed("done") # not in debt, stop mortgaging return if msg == "Which property do you want to unmortgage?": if self.is_my_turn(): self.say_delayed("done") # don't unmortgage proactively return # ============================================================ # TAX CHOICE # ============================================================ # "Do you wish to lose 10% of your total worth or $200? " if msg.startswith("Do you wish to lose 10"): if self.is_my_turn(): # 10% is usually cheaper early game self.say_delayed("10%") return # ============================================================ # HOUSE BUYING/SELLING # ============================================================ # "How many houses do you wish to buy for" / "sell from" if msg.startswith("How many houses do you wish to"): # We don't buy/sell houses proactively, but if prompted # during debt resolution, respond with 0 or done return # Property prompt during house buy/sell: "{name} ({n}):" or "{name} (H):" m = re.match(r'^(.+?)\s+\((?:\d+|H)\)\s*:\s*$', msg) if m: if self.is_my_turn(): # Don't buy/sell any houses self.say_delayed("0") return # ============================================================ # COMMAND PROMPT # ============================================================ if msg == "-- Command:": self.in_trade = False # trade ended (accepted, rejected, or cancelled) if self.is_my_turn(): if self.in_debt: self.say_delayed("mortgage") elif not self.rolled_this_turn: self.say_delayed("roll") self.rolled_this_turn = True # else: already rolled, waiting for prompts to resolve return # ============================================================ # VALID INPUTS (help/error recovery) # ============================================================ if msg.startswith("Valid inputs are:"): if self.is_my_turn(): options_str = msg[len("Valid inputs are:"):].strip() options = [o.strip().strip("'\"") for o in options_str.split(",")] self.log(f"Valid options: {options}") # Pick first sensible option for opt in options: opt_l = opt.lower().strip() if opt_l in ("quit", "save", "restore", ""): continue if opt_l == "done": self.say_delayed("done") return # For yes/no, pick yes if opt_l == "yes": self.say_delayed("yes") return if opt_l == "no": self.say_delayed("no") return # For property/player names, pick first self.say_delayed(opt) return return # ============================================================ # ILLEGAL ACTION — wrong player tried to act # ============================================================ m = re.match( r"^Illegal action: bad player \((.+?)'s turn, not (.+?)\)$", msg ) if m: correct = m.group(1) wrong = m.group(2) self.current_player = correct self.log(f"Turn correction: {correct}'s turn (not {wrong})") # If it's now our turn, roll if self.is_my_turn() and not self.rolled_this_turn: self.say_delayed("roll", delay=1.5) self.rolled_this_turn = True return # Illegal response if msg.startswith('Illegal response:'): self.log(f"Illegal response error: {msg}") return # ============================================================ # OWNERSHIP TRACKING (for mortgage decisions) # ============================================================ # "That got you $X" (mortgage succeeded) m = re.match(r'^That got you \$(\d+)$', msg) if m: return # ============================================================ # TRADING # ============================================================ # "There ain't no-one around to trade WITH!!" if msg.startswith("There ain't no-one"): self.in_trade = False if self.is_my_turn(): self.say_delayed("roll") self.rolled_this_turn = True return # "player {name} ({N}):" — trade property listing phase m = re.match(r'^player (.+?) \((\d+)\):$', msg) if m and self.in_trade: # This announces whose properties are being listed self.trade_props_offered = 0 return # "Which property do you wish to trade?" — pick a property or done if msg == "Which property do you wish to trade?": if self.in_trade: if self.trade_props_offered == 0 and random.random() < 0.5: # Offer one property then done self.trade_props_offered = 1 self.say_delayed("?", force=True) # get the list else: self.say_delayed("done", force=True) return # "You have $X. How much are you trading?" — offer some cash m = re.match(r'^You have \$(\d+)\.\s+How much are you trading\?', msg) if m: if self.in_trade: cash = int(m.group(1)) # Offer 0-25% of cash randomly offer = random.randint(0, max(1, cash // 4)) self.say_delayed(str(offer), force=True) return # "You have N get-out-of-jail-free cards. How many are you trading?" m = re.match(r'^You have (\d+) get-out-of-jail-free card', msg) if m: if self.in_trade: self.say_delayed("0", force=True) return # "You've already allocated that." if msg == "You've already allocated that.": if self.in_trade: self.say_delayed("done", force=True) return # "{name}, is the trade ok?" — 50/50 accept or reject m = re.match(r'^(.+?), is the trade ok\?$', msg) if m: name = m.group(1) if name.lower() == self.nick.lower(): if random.random() < 0.5: self.log("Accepting trade!") self.say_delayed("yes", force=True) else: self.log("Rejecting trade!") self.say_delayed("no", force=True) return # "Trade is done!" — trade completed if msg == "Trade is done!": self.in_trade = False # Now we need to roll if self.is_my_turn() and not self.rolled_this_turn: self.say_delayed("roll") self.rolled_this_turn = True return # Trade summary lines (just informational) if msg.startswith("Player ") and msg.endswith("gives:"): return # ============================================================ # EVERYTHING ELSE — ignore silently # ============================================================ # Card text, rent notifications, lucky messages, etc. # We don't need to respond to these. def _handle_setup(self, msg): """Handle setup phase. Returns True if message was handled.""" if msg == "How many players?": # Only the first player sends the number if self.player_index == 0: self.say_delayed(str(self.num_players), force=True) return True if msg.startswith("Sorry. Number must range from"): if self.player_index == 0: self.say_delayed(str(self.num_players), force=True) return True # "Player N, say ''me'' please." m = re.match(r"^Player (\d+), say ''me'' please\.$", msg) if m: player_num = int(m.group(1)) # Player N is 1-indexed, our index is 0-indexed if player_num - 1 == self.player_index: self.say_delayed(self.nick, delay=1.0, force=True) return True # Duplicate name error if "the same person" in msg: return True # Roll results during setup m = re.match(r'^(.+?)\s+\((\d+)\)\s+rolls\s+(\d+)$', msg) if m: return True # Re-roll for ties if "rolled the same thing" in msg: return True # Who goes first — end of setup m = re.match(r'^(.+?)\s+\((\d+)\)\s+goes first$', msg) if m: self.setup_phase = False self.game_started = True self.current_player = m.group(1) self.log(f"Game started! {self.current_player} goes first.") return True return False def main(): parser = argparse.ArgumentParser(description="Monop autopilot players") parser.add_argument("--host", default=HOST) parser.add_argument("--port", type=int, default=PORT) parser.add_argument("--channel", default=CHANNEL) parser.add_argument( "--players", default="alice,bob", help="Comma-separated player names" ) args = parser.parse_args() player_names = [n.strip() for n in args.players.split(",")] print(f"Starting {len(player_names)} autopilot players: {player_names}") print(f"Connecting to {args.host}:{args.port} {args.channel}") threads = [] bots = [] for i, name in enumerate(player_names): bot = PlayerBot( nick=name, channel=args.channel, host=args.host, port=args.port, player_names=player_names, player_index=i, ) bots.append(bot) t = threading.Thread(target=bot.run, daemon=True, name=f"player-{name}") threads.append(t) # Stagger connections slightly to avoid race conditions for i, t in enumerate(threads): t.start() time.sleep(1.0) # Wait for all to finish (or ctrl-c) try: while any(t.is_alive() for t in threads): time.sleep(1) except KeyboardInterrupt: print("\nShutting down...") for bot in bots: bot.game_over = True if __name__ == "__main__": main()