227 lines
7.4 KiB
Python
227 lines
7.4 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Simulate a monop game on IRC for testing the board viewer bot.
|
||
|
|
|
||
|
|
Runs monop in a subprocess, connects to IRC, and plays the game
|
||
|
|
with scripted AI players that make simple decisions.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import re
|
||
|
|
import select
|
||
|
|
import ssl
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import time
|
||
|
|
import irc.client
|
||
|
|
import irc.connection
|
||
|
|
|
||
|
|
CONFIG = {
|
||
|
|
"server": "irc.darkscience.net",
|
||
|
|
"port": 6697,
|
||
|
|
"tls": True,
|
||
|
|
"nick": "monop",
|
||
|
|
"channel": "#monop-dev",
|
||
|
|
"num_players": 3,
|
||
|
|
"player_names": ["Alice", "Bob", "Charlie"],
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
class MonopSimulator:
|
||
|
|
def __init__(self, config):
|
||
|
|
self.config = config
|
||
|
|
self.connection = None
|
||
|
|
self.channel = config["channel"]
|
||
|
|
self.proc = None
|
||
|
|
self.game_phase = "init" # init, setup, playing
|
||
|
|
self.setup_step = 0
|
||
|
|
self.current_prompt = ""
|
||
|
|
self.output_buffer = ""
|
||
|
|
self.turn_count = 0
|
||
|
|
self.max_turns = 60 # stop after this many turns
|
||
|
|
|
||
|
|
def start_monop(self):
|
||
|
|
"""Start the monop game process."""
|
||
|
|
self.proc = subprocess.Popen(
|
||
|
|
["/usr/games/monop"],
|
||
|
|
stdin=subprocess.PIPE,
|
||
|
|
stdout=subprocess.PIPE,
|
||
|
|
stderr=subprocess.PIPE,
|
||
|
|
text=True,
|
||
|
|
bufsize=1,
|
||
|
|
)
|
||
|
|
# Set stdout to non-blocking
|
||
|
|
import fcntl
|
||
|
|
fd = self.proc.stdout.fileno()
|
||
|
|
fl = fcntl.fcntl(fd, fcntl.F_GETFL)
|
||
|
|
fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
|
||
|
|
print("monop process started", flush=True)
|
||
|
|
|
||
|
|
def read_monop_output(self):
|
||
|
|
"""Read available output from monop."""
|
||
|
|
lines = []
|
||
|
|
try:
|
||
|
|
while True:
|
||
|
|
data = self.proc.stdout.read(4096)
|
||
|
|
if not data:
|
||
|
|
break
|
||
|
|
self.output_buffer += data
|
||
|
|
except (BlockingIOError, IOError):
|
||
|
|
pass
|
||
|
|
|
||
|
|
while "\n" in self.output_buffer:
|
||
|
|
line, self.output_buffer = self.output_buffer.split("\n", 1)
|
||
|
|
lines.append(line)
|
||
|
|
|
||
|
|
# Check for prompts (lines without newline at end)
|
||
|
|
if self.output_buffer:
|
||
|
|
self.current_prompt = self.output_buffer
|
||
|
|
|
||
|
|
return lines
|
||
|
|
|
||
|
|
def send_to_monop(self, text):
|
||
|
|
"""Send input to monop process."""
|
||
|
|
if self.proc and self.proc.poll() is None:
|
||
|
|
print(f" -> monop input: {text.strip()}", flush=True)
|
||
|
|
self.proc.stdin.write(text + "\n")
|
||
|
|
self.proc.stdin.flush()
|
||
|
|
time.sleep(0.3) # give monop time to process
|
||
|
|
|
||
|
|
def send_to_irc(self, text):
|
||
|
|
"""Send a message to the IRC channel."""
|
||
|
|
if self.connection and text.strip():
|
||
|
|
# Split long lines
|
||
|
|
for line in text.split("\n"):
|
||
|
|
line = line.rstrip()
|
||
|
|
if line:
|
||
|
|
self.connection.privmsg(self.channel, line)
|
||
|
|
time.sleep(0.15) # rate limit
|
||
|
|
|
||
|
|
def decide_action(self, prompt, output_lines):
|
||
|
|
"""AI decision making for the scripted players."""
|
||
|
|
full_text = "\n".join(output_lines) + "\n" + prompt
|
||
|
|
|
||
|
|
# "How many players?"
|
||
|
|
if "How many players?" in full_text:
|
||
|
|
return str(self.config["num_players"])
|
||
|
|
|
||
|
|
# Player name prompts
|
||
|
|
for i, name in enumerate(self.config["player_names"]):
|
||
|
|
if f"Player {i+1}'s name:" in full_text:
|
||
|
|
return name
|
||
|
|
|
||
|
|
# "Do you want to buy?"
|
||
|
|
if "Do you want to buy?" in prompt or "Do you want to buy?" in full_text:
|
||
|
|
return "yes"
|
||
|
|
|
||
|
|
# yes/no questions - default yes
|
||
|
|
if prompt.strip().endswith("?") and ("yes" in prompt.lower() or "no" in prompt.lower()):
|
||
|
|
return "yes"
|
||
|
|
|
||
|
|
# Command prompt (the game's main input) - just roll
|
||
|
|
if "-- Loss of $" in full_text:
|
||
|
|
return "" # just continue
|
||
|
|
|
||
|
|
# If we see a command prompt, roll
|
||
|
|
return "" # empty = roll (default action)
|
||
|
|
|
||
|
|
def on_connect(self, connection, event):
|
||
|
|
print(f"Connected to IRC. Joining {self.channel}", flush=True)
|
||
|
|
self.connection = connection
|
||
|
|
connection.join(self.channel)
|
||
|
|
|
||
|
|
def on_join(self, connection, event):
|
||
|
|
if event.source.nick == self.config["nick"]:
|
||
|
|
print(f"Joined {self.channel}. Starting game in 3s...", flush=True)
|
||
|
|
time.sleep(3)
|
||
|
|
self.start_monop()
|
||
|
|
self.game_loop()
|
||
|
|
|
||
|
|
def game_loop(self):
|
||
|
|
"""Main game loop - read monop output, send to IRC, make decisions."""
|
||
|
|
print("Starting game loop...", flush=True)
|
||
|
|
idle_count = 0
|
||
|
|
|
||
|
|
while self.proc and self.proc.poll() is None and self.turn_count < self.max_turns:
|
||
|
|
lines = self.read_monop_output()
|
||
|
|
|
||
|
|
if lines:
|
||
|
|
idle_count = 0
|
||
|
|
for line in lines:
|
||
|
|
print(f" monop: {line}", flush=True)
|
||
|
|
self.send_to_irc(line)
|
||
|
|
|
||
|
|
# Track turns
|
||
|
|
if "rolled doubles" in line or "'s turn" in line:
|
||
|
|
self.turn_count += 1
|
||
|
|
|
||
|
|
# Check if monop is waiting for input
|
||
|
|
time.sleep(0.5)
|
||
|
|
more_lines = self.read_monop_output()
|
||
|
|
for line in more_lines:
|
||
|
|
lines.append(line)
|
||
|
|
print(f" monop: {line}", flush=True)
|
||
|
|
self.send_to_irc(line)
|
||
|
|
|
||
|
|
# Make a decision if there's a prompt
|
||
|
|
if self.current_prompt or any("?" in l for l in lines):
|
||
|
|
decision = self.decide_action(self.current_prompt, lines)
|
||
|
|
if decision is not None:
|
||
|
|
time.sleep(0.5)
|
||
|
|
self.send_to_monop(decision)
|
||
|
|
self.current_prompt = ""
|
||
|
|
else:
|
||
|
|
# Game might be waiting for default input (roll)
|
||
|
|
time.sleep(1)
|
||
|
|
self.send_to_monop("")
|
||
|
|
else:
|
||
|
|
idle_count += 1
|
||
|
|
if idle_count > 10:
|
||
|
|
# Monop might be waiting for input
|
||
|
|
time.sleep(0.5)
|
||
|
|
self.send_to_monop("")
|
||
|
|
idle_count = 0
|
||
|
|
time.sleep(0.3)
|
||
|
|
|
||
|
|
# Process IRC events
|
||
|
|
try:
|
||
|
|
self.reactor.process_once(timeout=0.1)
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
|
||
|
|
print(f"Game loop ended. Turns: {self.turn_count}", flush=True)
|
||
|
|
self.send_to_irc(f"--- Game simulation ended after {self.turn_count} turns ---")
|
||
|
|
if self.proc:
|
||
|
|
self.proc.terminate()
|
||
|
|
|
||
|
|
def run(self):
|
||
|
|
config = self.config
|
||
|
|
self.reactor = irc.client.Reactor()
|
||
|
|
|
||
|
|
connect_params = {}
|
||
|
|
if config.get("tls", False):
|
||
|
|
ssl_ctx = ssl.create_default_context()
|
||
|
|
hostname = config["server"]
|
||
|
|
ssl_factory = irc.connection.Factory(
|
||
|
|
wrapper=lambda s: ssl_ctx.wrap_socket(s, server_hostname=hostname))
|
||
|
|
connect_params["connect_factory"] = ssl_factory
|
||
|
|
|
||
|
|
server = self.reactor.server()
|
||
|
|
server.connect(
|
||
|
|
config["server"],
|
||
|
|
config["port"],
|
||
|
|
config["nick"],
|
||
|
|
**connect_params,
|
||
|
|
)
|
||
|
|
|
||
|
|
server.add_global_handler("welcome", self.on_connect)
|
||
|
|
server.add_global_handler("join", self.on_join)
|
||
|
|
|
||
|
|
print(f"Connecting to {config['server']}:{config['port']}...", flush=True)
|
||
|
|
self.reactor.process_forever()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
sim = MonopSimulator(CONFIG)
|
||
|
|
sim.run()
|