268 lines
8.4 KiB
Python
268 lines
8.4 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Generate a realistic game-state.json from a scripted monop session using pexpect."""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import json
|
||
|
|
import re
|
||
|
|
import time
|
||
|
|
import pexpect
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
|
||
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'bot'))
|
||
|
|
from board_data import SQUARES, NAME_TO_ID
|
||
|
|
|
||
|
|
STATE_FILE = os.path.join(os.path.dirname(__file__), '..', 'site', 'game-state.json')
|
||
|
|
MONOP = "/usr/games/monop"
|
||
|
|
PLAYER_NAMES = ["Alice", "Bob", "Charlie"]
|
||
|
|
|
||
|
|
|
||
|
|
def find_square(name):
|
||
|
|
name = name.strip().lower().rstrip(".")
|
||
|
|
# Try various lookups
|
||
|
|
for key in [name, name.split("(")[0].strip()]:
|
||
|
|
key = key.strip()
|
||
|
|
if key in NAME_TO_ID:
|
||
|
|
return NAME_TO_ID[key]
|
||
|
|
for k, v in NAME_TO_ID.items():
|
||
|
|
if key in k or k in key:
|
||
|
|
return v
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
class GameState:
|
||
|
|
def __init__(self, names):
|
||
|
|
self.players = []
|
||
|
|
for i, name in enumerate(names):
|
||
|
|
self.players.append({
|
||
|
|
"name": name, "number": i + 1, "money": 1500,
|
||
|
|
"location": 0, "inJail": False, "jailTurns": 0,
|
||
|
|
"goJailFreeCards": 0, "properties": [],
|
||
|
|
"numRailroads": 0, "numUtilities": 0,
|
||
|
|
})
|
||
|
|
self.squares = []
|
||
|
|
for sq in SQUARES:
|
||
|
|
self.squares.append({
|
||
|
|
"id": sq["id"], "name": sq["name"], "type": sq.get("type", "safe"),
|
||
|
|
"owner": -1, "cost": sq.get("cost", 0), "mortgaged": False,
|
||
|
|
"houses": 0, "monopoly": False, "group": sq.get("group"),
|
||
|
|
"rent": sq.get("rent", [0]),
|
||
|
|
})
|
||
|
|
self.current_player = 0
|
||
|
|
self.log = []
|
||
|
|
|
||
|
|
def add_log(self, text, player=None):
|
||
|
|
self.log.append({
|
||
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||
|
|
"text": text,
|
||
|
|
"player": player,
|
||
|
|
})
|
||
|
|
|
||
|
|
def find_player(self, name):
|
||
|
|
for i, p in enumerate(self.players):
|
||
|
|
if p["name"].lower() == name.lower():
|
||
|
|
return i
|
||
|
|
return None
|
||
|
|
|
||
|
|
def cur(self):
|
||
|
|
return self.players[self.current_player]
|
||
|
|
|
||
|
|
def to_dict(self):
|
||
|
|
return {
|
||
|
|
"lastUpdated": datetime.now(timezone.utc).isoformat(),
|
||
|
|
"currentPlayer": self.current_player,
|
||
|
|
"players": self.players,
|
||
|
|
"squares": self.squares,
|
||
|
|
"log": self.log[-100:],
|
||
|
|
}
|
||
|
|
|
||
|
|
def save(self):
|
||
|
|
d = self.to_dict()
|
||
|
|
tmp = STATE_FILE + ".tmp"
|
||
|
|
with open(tmp, "w") as f:
|
||
|
|
json.dump(d, f, indent=2)
|
||
|
|
os.replace(tmp, STATE_FILE)
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
gs = GameState(PLAYER_NAMES)
|
||
|
|
|
||
|
|
child = pexpect.spawn(MONOP, encoding='utf-8', timeout=5)
|
||
|
|
all_output = []
|
||
|
|
|
||
|
|
def interact(expect_pat, response, timeout=3):
|
||
|
|
try:
|
||
|
|
child.expect(expect_pat, timeout=timeout)
|
||
|
|
text = (child.before or "") + (child.after or "")
|
||
|
|
all_output.append(text)
|
||
|
|
process_text(text)
|
||
|
|
child.sendline(response)
|
||
|
|
time.sleep(0.15)
|
||
|
|
return True
|
||
|
|
except (pexpect.TIMEOUT, pexpect.EOF):
|
||
|
|
try:
|
||
|
|
text = child.before or ""
|
||
|
|
if text:
|
||
|
|
all_output.append(text)
|
||
|
|
process_text(text)
|
||
|
|
except:
|
||
|
|
pass
|
||
|
|
return False
|
||
|
|
|
||
|
|
def process_text(text):
|
||
|
|
for line in text.split("\n"):
|
||
|
|
line = line.strip().replace("\r", "")
|
||
|
|
if not line:
|
||
|
|
continue
|
||
|
|
|
||
|
|
cp = gs.cur()
|
||
|
|
|
||
|
|
# Roll
|
||
|
|
m = re.match(r"roll is (\d+), (\d+)", line)
|
||
|
|
if m:
|
||
|
|
gs.add_log(f"roll is {m.group(1)}, {m.group(2)}", cp["name"])
|
||
|
|
|
||
|
|
# Movement
|
||
|
|
m = re.match(r"That puts you on (.+)", line)
|
||
|
|
if m:
|
||
|
|
sq_name = m.group(1).strip()
|
||
|
|
sq_id = find_square(sq_name)
|
||
|
|
if sq_id is not None:
|
||
|
|
cp["location"] = sq_id
|
||
|
|
gs.add_log(f"Landed on {sq_name}", cp["name"])
|
||
|
|
|
||
|
|
# Pass Go
|
||
|
|
if "pass" in line.lower() and "$200" in line:
|
||
|
|
cp["money"] += 200
|
||
|
|
gs.add_log("Passed Go, collected $200", cp["name"])
|
||
|
|
|
||
|
|
# Doubles
|
||
|
|
m = re.match(r"(.+) rolled doubles", line)
|
||
|
|
if m:
|
||
|
|
gs.add_log(f"{m.group(1)} rolled doubles", m.group(1))
|
||
|
|
|
||
|
|
# Jail
|
||
|
|
if "GO DIRECTLY TO JAIL" in line or "go to jail" in line.lower():
|
||
|
|
cp["location"] = 10
|
||
|
|
cp["inJail"] = True
|
||
|
|
gs.add_log("Go to Jail!", cp["name"])
|
||
|
|
|
||
|
|
# Rent
|
||
|
|
m = re.match(r"rent is (\d+)", line)
|
||
|
|
if m:
|
||
|
|
rent = int(m.group(1))
|
||
|
|
cp["money"] -= rent
|
||
|
|
sq_id = cp["location"]
|
||
|
|
owner = gs.squares[sq_id]["owner"]
|
||
|
|
if 0 <= owner < len(gs.players):
|
||
|
|
gs.players[owner]["money"] += rent
|
||
|
|
gs.add_log(f"Paid ${rent} rent", cp["name"])
|
||
|
|
|
||
|
|
# Tax
|
||
|
|
if "Luxury tax" in line:
|
||
|
|
cp["money"] -= 75
|
||
|
|
gs.add_log("Paid $75 luxury tax", cp["name"])
|
||
|
|
if "Income tax" in line and "%" in line:
|
||
|
|
cp["money"] -= 200
|
||
|
|
gs.add_log("Paid $200 income tax", cp["name"])
|
||
|
|
|
||
|
|
# Services $25
|
||
|
|
if "Receive for Services $25" in line:
|
||
|
|
cp["money"] += 25
|
||
|
|
gs.add_log("Received $25 for services", cp["name"])
|
||
|
|
|
||
|
|
# Player turn indicator
|
||
|
|
m = re.match(r"(\w+) \(\d+\) \(cash \$(\d+)\) on (.+)", line)
|
||
|
|
if m:
|
||
|
|
name = m.group(1)
|
||
|
|
money = int(m.group(2))
|
||
|
|
idx = gs.find_player(name)
|
||
|
|
if idx is not None:
|
||
|
|
gs.current_player = idx
|
||
|
|
gs.players[idx]["money"] = money
|
||
|
|
|
||
|
|
# Bought detection (after "yes" to buy)
|
||
|
|
if "-- Loss of $" in line:
|
||
|
|
m2 = re.search(r"Loss of \$(\d+)", line)
|
||
|
|
if m2:
|
||
|
|
cp["money"] -= int(m2.group(1))
|
||
|
|
|
||
|
|
gs.save()
|
||
|
|
|
||
|
|
# Setup
|
||
|
|
print("=== Setup ===", flush=True)
|
||
|
|
interact("How many players\\?", str(len(PLAYER_NAMES)))
|
||
|
|
for i, name in enumerate(PLAYER_NAMES):
|
||
|
|
interact(f"Player {i+1}'s name:", name)
|
||
|
|
|
||
|
|
# Handle initial rolls for order
|
||
|
|
for _ in range(10):
|
||
|
|
if not interact([r"\(\d+\) rolls \d+", "goes first"], "", timeout=2):
|
||
|
|
break
|
||
|
|
|
||
|
|
# Play the game
|
||
|
|
print("\n=== Playing ===", flush=True)
|
||
|
|
turns = 0
|
||
|
|
max_turns = 80
|
||
|
|
|
||
|
|
while turns < max_turns:
|
||
|
|
# Try to match common prompts
|
||
|
|
matched = False
|
||
|
|
|
||
|
|
# Buy prompt
|
||
|
|
if interact("Do you want to buy\\?", "yes", timeout=2):
|
||
|
|
# Record the purchase
|
||
|
|
cp = gs.cur()
|
||
|
|
sq_id = cp["location"]
|
||
|
|
if sq_id is not None and gs.squares[sq_id]["owner"] == -1:
|
||
|
|
sq = gs.squares[sq_id]
|
||
|
|
if sq["cost"] > 0:
|
||
|
|
sq["owner"] = gs.current_player
|
||
|
|
cp["properties"].append(sq_id)
|
||
|
|
cp["money"] -= sq["cost"]
|
||
|
|
if sq["type"] == "railroad":
|
||
|
|
cp["numRailroads"] += 1
|
||
|
|
elif sq["type"] == "utility":
|
||
|
|
cp["numUtilities"] += 1
|
||
|
|
gs.add_log(f"Bought {sq['name']} for ${sq['cost']}", cp["name"])
|
||
|
|
gs.save()
|
||
|
|
matched = True
|
||
|
|
turns += 1
|
||
|
|
continue
|
||
|
|
|
||
|
|
# Command prompt - roll
|
||
|
|
if interact("-- Command:", "", timeout=2):
|
||
|
|
matched = True
|
||
|
|
turns += 1
|
||
|
|
continue
|
||
|
|
|
||
|
|
# Any other prompt
|
||
|
|
if interact("\\? ", "", timeout=2):
|
||
|
|
matched = True
|
||
|
|
turns += 1
|
||
|
|
continue
|
||
|
|
|
||
|
|
if not matched:
|
||
|
|
# Try just reading output
|
||
|
|
try:
|
||
|
|
child.expect(r".+", timeout=1)
|
||
|
|
text = (child.before or "") + (child.after or "")
|
||
|
|
if text:
|
||
|
|
process_text(text)
|
||
|
|
except (pexpect.TIMEOUT, pexpect.EOF):
|
||
|
|
child.sendline("")
|
||
|
|
turns += 1
|
||
|
|
|
||
|
|
print(f"\n=== Done: {turns} turns ===", flush=True)
|
||
|
|
gs.save()
|
||
|
|
child.close()
|
||
|
|
|
||
|
|
for p in gs.players:
|
||
|
|
props = [gs.squares[pid]["name"] for pid in p["properties"]]
|
||
|
|
print(f" {p['name']}: ${p['money']}, loc={p['location']}, props={props}")
|
||
|
|
print(f" Log: {len(gs.log)} entries")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|