jarvis/monop-state
1
0
Fork 0
Code Issues 2 Pull requests Projects Releases Packages Wiki Activity Actions

Add standalone autopilot players script

Browse source
This commit is contained in:
Jarvis 2026-02-21 08:57:10 +00:00
parent 44b9ed1ab1
commit 5aa96d2163
1 changed files with 570 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

570
monop_players.py Normal file
View file

@ -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()