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

686 lines
24 KiB
Python
Raw Blame History

#!/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
if msg.startswith("Which property do you wish to trade?"):
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?":
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("?") # get the list
else:
self.say_delayed("done")
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))
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")
return
# "You've already allocated that."
if msg == "You've already allocated that.":
if self.in_trade:
self.say_delayed("done")
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")
else:
self.log("Rejecting trade!")
self.say_delayed("no")
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()
Reference in a new issue View git blame Copy permalink