138 lines
4.1 KiB
Python
138 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Run monop locally with scripted players via pexpect, feed output to parser."""
|
|
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
import json
|
|
import pexpect
|
|
from datetime import datetime, timezone
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'bot'))
|
|
from parser import MonopParser
|
|
|
|
STATE_FILE = os.path.join(os.path.dirname(__file__), '..', 'site', 'game-state.json')
|
|
MONOP = "/usr/games/monop"
|
|
PLAYERS = ["Alice", "Bob", "Charlie"]
|
|
MAX_TURNS = 80
|
|
|
|
|
|
def write_state(parser):
|
|
state = parser.get_state()
|
|
state["lastUpdated"] = datetime.now(timezone.utc).isoformat()
|
|
tmp = STATE_FILE + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
json.dump(state, f, indent=2)
|
|
os.replace(tmp, STATE_FILE)
|
|
|
|
|
|
def process_output(parser, text):
|
|
"""Feed text lines to parser."""
|
|
for line in text.split("\r\n"):
|
|
line = line.strip()
|
|
if line:
|
|
print(f" monop: {line}", flush=True)
|
|
parser.parse_line(line)
|
|
|
|
|
|
def main():
|
|
parser = MonopParser()
|
|
for i, name in enumerate(PLAYERS):
|
|
parser.add_player(name, i + 1)
|
|
|
|
child = pexpect.spawn(MONOP, encoding='utf-8', timeout=5)
|
|
child.logfile_read = None # we'll handle output ourselves
|
|
|
|
turn_count = 0
|
|
|
|
def expect_and_respond(patterns_responses, default=""):
|
|
"""Wait for patterns and respond."""
|
|
nonlocal turn_count
|
|
patterns = [p for p, _ in patterns_responses]
|
|
try:
|
|
idx = child.expect(patterns, timeout=3)
|
|
output = child.before + child.after
|
|
process_output(parser, output)
|
|
write_state(parser)
|
|
resp = patterns_responses[idx][1]
|
|
if callable(resp):
|
|
resp = resp()
|
|
print(f" -> {resp!r}", flush=True)
|
|
child.sendline(resp)
|
|
return True
|
|
except pexpect.TIMEOUT:
|
|
output = child.before or ""
|
|
if output:
|
|
process_output(parser, output)
|
|
write_state(parser)
|
|
return False
|
|
except pexpect.EOF:
|
|
output = child.before or ""
|
|
if output:
|
|
process_output(parser, output)
|
|
write_state(parser)
|
|
return None
|
|
|
|
# Phase 1: Setup
|
|
print("=== Game Setup ===", flush=True)
|
|
expect_and_respond([("How many players\\?", str(len(PLAYERS)))])
|
|
for i, name in enumerate(PLAYERS):
|
|
expect_and_respond([(f"Player {i+1}'s name:", name)])
|
|
|
|
# Phase 2: Playing
|
|
print("\n=== Game Play ===", flush=True)
|
|
consecutive_timeouts = 0
|
|
|
|
while turn_count < MAX_TURNS:
|
|
result = expect_and_respond([
|
|
("Do you want to buy\\?", "yes"),
|
|
("do you wish to ", "yes"),
|
|
("Bid for ", "0"),
|
|
("How much do you", "0"),
|
|
("mortgage\\?", "yes"),
|
|
("Which file", "/dev/null"),
|
|
("do you wish to sell", "no"),
|
|
("Will you buy", "no"),
|
|
("rolled doubles", ""),
|
|
("'s turn", ""),
|
|
# The main prompt - just hit enter to roll
|
|
("\\? ", ""),
|
|
("\n", ""),
|
|
])
|
|
|
|
if result is None:
|
|
print("Game ended (EOF)", flush=True)
|
|
break
|
|
elif result is False:
|
|
consecutive_timeouts += 1
|
|
if consecutive_timeouts > 5:
|
|
# Try sending empty line to nudge
|
|
child.sendline("")
|
|
consecutive_timeouts = 0
|
|
else:
|
|
consecutive_timeouts = 0
|
|
turn_count += 1
|
|
|
|
# Read any remaining output
|
|
try:
|
|
extra = child.read_nonblocking(4096, timeout=0.3)
|
|
if extra:
|
|
process_output(parser, extra)
|
|
write_state(parser)
|
|
except (pexpect.TIMEOUT, pexpect.EOF):
|
|
pass
|
|
|
|
print(f"\n=== Done: {turn_count} turns ===", flush=True)
|
|
write_state(parser)
|
|
child.close()
|
|
|
|
state = parser.get_state()
|
|
for p in state["players"]:
|
|
props = [state["squares"][pid]["name"] for pid in p["properties"]]
|
|
print(f" {p['name']}: ${p['money']}, loc={p['location']}, props={props}")
|
|
print(f" Log entries: {len(state['log'])}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|