From 5aa96d2163fee4d8d8103f2b2047734510bf6244 Mon Sep 17 00:00:00 2001 From: Jarvis Date: Sat, 21 Feb 2026 08:57:10 +0000 Subject: [PATCH] Add standalone autopilot players script --- monop_players.py | 570 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 570 insertions(+) create mode 100644 monop_players.py diff --git a/monop_players.py b/monop_players.py new file mode 100644 index 0000000..31032b9 --- /dev/null +++ b/monop_players.py @@ -0,0 +1,570 @@ +#!/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 + + 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 + deadline = time.time() + 15 + while time.time() < deadline: + data = self.sock.recv(4096) + 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 " 001 " in self.buffer: + break + + self._send(f"JOIN {self.channel}") + time.sleep(0.3) + self.buffer = "" + self.log(f"Connected and joined {self.channel}") + + 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): + """Send after a short delay (avoids flooding, more natural).""" + d = delay if delay is not None else RESPONSE_DELAY + def _do(): + time.sleep(d) + self.say(msg) + 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.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: "{name}, is the trade ok?" + m = re.match(r'^(.+?), is the trade ok\?$', msg) + if m: + name = m.group(1) + if name.lower() == self.nick.lower(): + self.say_delayed("yes") + return + + # 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 + + # ============================================================ + # 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?": + # monop will present options via getinp; it takes the property + # name. We'll wait for the valid inputs prompt. + self.awaiting_prompt = "mortgage_choice" + return + + if msg == "Which property do you want to unmortgage?": + self.awaiting_prompt = "unmortgage_choice" + 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("percentage") + 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:": + 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, just end turn (empty/done) + self.say_delayed("roll") + 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 + + # ============================================================ + # 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)) + return True + + if msg.startswith("Sorry. Number must range from"): + if self.player_index == 0: + self.say_delayed(str(self.num_players)) + 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) + 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()