231 lines
7.3 KiB
Python
231 lines
7.3 KiB
Python
|
#!/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()
|