217 lines
7.3 KiB
Python
217 lines
7.3 KiB
Python
|
"""
|
||
|
|
Cardinal plugin that watches a monop-irc game and tracks state.
|
||
|
|
|
||
|
|
Listens to all messages in the configured channel, feeds them to
|
||
|
|
MonopParser, and writes game-state.json on every state change.
|
||
|
|
Provides commands to query the current game state.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import logging
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
|
||
|
|
from cardinal.decorators import command, event, help
|
||
|
|
|
||
|
|
# Import parser from same directory
|
||
|
|
import os
|
||
|
|
import importlib.util
|
||
|
|
_parser_path = os.path.join(os.path.dirname(__file__), "monop_parser.py")
|
||
|
|
_spec = importlib.util.spec_from_file_location("monop_parser", _parser_path)
|
||
|
|
_mod = importlib.util.module_from_spec(_spec)
|
||
|
|
_spec.loader.exec_module(_mod)
|
||
|
|
MonopParser = _mod.MonopParser
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
DEFAULT_STATE_PATH = "game-state.json"
|
||
|
|
DEFAULT_BOT_NICK = "monop"
|
||
|
|
|
||
|
|
|
||
|
|
class MonopPlugin:
|
||
|
|
def __init__(self, cardinal, config):
|
||
|
|
self.cardinal = cardinal
|
||
|
|
self.config = config or {}
|
||
|
|
|
||
|
|
self.bot_nick = self.config.get("bot_nick", DEFAULT_BOT_NICK)
|
||
|
|
self.state_path = self.config.get(
|
||
|
|
"state_path",
|
||
|
|
os.path.join(cardinal.storage_path, DEFAULT_STATE_PATH)
|
||
|
|
)
|
||
|
|
self.watched_channels = self.config.get("channels", [])
|
||
|
|
|
||
|
|
self.parser = MonopParser()
|
||
|
|
self._last_state_hash = None
|
||
|
|
|
||
|
|
logger.info(
|
||
|
|
"MonopPlugin loaded: watching %s, bot_nick=%s, state_path=%s",
|
||
|
|
self.watched_channels, self.bot_nick, self.state_path,
|
||
|
|
)
|
||
|
|
|
||
|
|
def _should_watch(self, channel):
|
||
|
|
"""Check if we should watch this channel."""
|
||
|
|
if not self.watched_channels:
|
||
|
|
return True # watch all channels if none configured
|
||
|
|
return channel.lower() in [c.lower() for c in self.watched_channels]
|
||
|
|
|
||
|
|
def _feed_line(self, sender, channel, message):
|
||
|
|
"""Feed a line to the parser and save state if changed."""
|
||
|
|
ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
|
log_line = f"{ts}\t{sender}\t{message}"
|
||
|
|
self.parser.parse_line(log_line)
|
||
|
|
|
||
|
|
state = self.parser.get_state()
|
||
|
|
if state is None:
|
||
|
|
return
|
||
|
|
|
||
|
|
# Only write if state changed
|
||
|
|
state_hash = json.dumps(state, sort_keys=True)
|
||
|
|
if state_hash != self._last_state_hash:
|
||
|
|
self._last_state_hash = state_hash
|
||
|
|
state["lastUpdated"] = datetime.now(timezone.utc).isoformat()
|
||
|
|
self._save_state(state)
|
||
|
|
|
||
|
|
def _save_state(self, state):
|
||
|
|
"""Write game state to JSON file."""
|
||
|
|
try:
|
||
|
|
with open(self.state_path, "w") as f:
|
||
|
|
json.dump(state, f, indent=2)
|
||
|
|
logger.debug("State saved to %s", self.state_path)
|
||
|
|
except Exception:
|
||
|
|
logger.exception("Failed to save state to %s", self.state_path)
|
||
|
|
|
||
|
|
@event("irc.privmsg")
|
||
|
|
def on_msg(self, cardinal, user, channel, message):
|
||
|
|
"""Watch all channel messages and feed to parser."""
|
||
|
|
if not self._should_watch(channel):
|
||
|
|
return
|
||
|
|
|
||
|
|
self._feed_line(user.nick, channel, message)
|
||
|
|
|
||
|
|
@command("monop")
|
||
|
|
@help("Show current monop game state summary.")
|
||
|
|
@help("Syntax: .monop [player|board|log]")
|
||
|
|
def monop_cmd(self, cardinal, user, channel, msg):
|
||
|
|
parts = msg.split()
|
||
|
|
subcmd = parts[1].lower() if len(parts) > 1 else "status"
|
||
|
|
|
||
|
|
state = self.parser.get_state()
|
||
|
|
if state is None:
|
||
|
|
cardinal.sendMsg(channel, "No active monop game being tracked.")
|
||
|
|
return
|
||
|
|
|
||
|
|
if subcmd == "status":
|
||
|
|
self._show_status(cardinal, channel, state)
|
||
|
|
elif subcmd == "player" or subcmd == "players":
|
||
|
|
self._show_players(cardinal, channel, state)
|
||
|
|
elif subcmd == "board":
|
||
|
|
self._show_board(cardinal, channel, state)
|
||
|
|
elif subcmd == "owned":
|
||
|
|
self._show_owned(cardinal, channel, state)
|
||
|
|
else:
|
||
|
|
cardinal.sendMsg(
|
||
|
|
channel,
|
||
|
|
"Usage: .monop [status|players|board|owned]"
|
||
|
|
)
|
||
|
|
|
||
|
|
def _show_status(self, cardinal, channel, state):
|
||
|
|
players = state.get("players", [])
|
||
|
|
cp = state.get("currentPlayer")
|
||
|
|
|
||
|
|
if not players:
|
||
|
|
cardinal.sendMsg(channel, "No players registered yet.")
|
||
|
|
return
|
||
|
|
|
||
|
|
current_name = "?"
|
||
|
|
for p in players:
|
||
|
|
if p["number"] == cp:
|
||
|
|
current_name = p["name"]
|
||
|
|
|
||
|
|
lines = [
|
||
|
|
f"Monop game: {len(players)} players, "
|
||
|
|
f"current turn: {current_name}"
|
||
|
|
]
|
||
|
|
for p in players:
|
||
|
|
loc_name = self._location_name(state, p["location"])
|
||
|
|
jail = " [IN JAIL]" if p.get("inJail") else ""
|
||
|
|
lines.append(
|
||
|
|
f" {p['name']} (${p['money']}) on {loc_name}{jail}"
|
||
|
|
)
|
||
|
|
|
||
|
|
for line in lines:
|
||
|
|
cardinal.sendMsg(channel, line)
|
||
|
|
|
||
|
|
def _show_players(self, cardinal, channel, state):
|
||
|
|
for p in state.get("players", []):
|
||
|
|
props = []
|
||
|
|
for sq in state.get("squares", []):
|
||
|
|
if sq.get("owner") == p["number"]:
|
||
|
|
name = sq["name"]
|
||
|
|
if sq.get("mortgaged"):
|
||
|
|
name += " [M]"
|
||
|
|
if sq.get("houses", 0) > 0:
|
||
|
|
h = sq["houses"]
|
||
|
|
name += f" [{h}H]" if h < 5 else " [Hotel]"
|
||
|
|
props.append(name)
|
||
|
|
|
||
|
|
gojf = p.get("getOutOfJailFreeCards", 0)
|
||
|
|
gojf_str = f", {gojf} GOJF" if gojf else ""
|
||
|
|
prop_str = ", ".join(props) if props else "none"
|
||
|
|
|
||
|
|
cardinal.sendMsg(
|
||
|
|
channel,
|
||
|
|
f"{p['name']} (${p['money']}{gojf_str}): {prop_str}"
|
||
|
|
)
|
||
|
|
|
||
|
|
def _show_board(self, cardinal, channel, state):
|
||
|
|
"""Show owned properties summary."""
|
||
|
|
self._show_owned(cardinal, channel, state)
|
||
|
|
|
||
|
|
def _show_owned(self, cardinal, channel, state):
|
||
|
|
"""Show all owned properties grouped by player."""
|
||
|
|
players = {p["number"]: p["name"] for p in state.get("players", [])}
|
||
|
|
owned = {}
|
||
|
|
for sq in state.get("squares", []):
|
||
|
|
owner = sq.get("owner")
|
||
|
|
if owner is not None:
|
||
|
|
owned.setdefault(owner, []).append(sq)
|
||
|
|
|
||
|
|
if not owned:
|
||
|
|
cardinal.sendMsg(channel, "No properties owned yet.")
|
||
|
|
return
|
||
|
|
|
||
|
|
for pnum, squares in sorted(owned.items()):
|
||
|
|
name = players.get(pnum, f"Player {pnum}")
|
||
|
|
parts = []
|
||
|
|
for sq in squares:
|
||
|
|
s = sq["name"]
|
||
|
|
if sq.get("mortgaged"):
|
||
|
|
s += "*"
|
||
|
|
h = sq.get("houses", 0)
|
||
|
|
if h == 5:
|
||
|
|
s += "(H)"
|
||
|
|
elif h > 0:
|
||
|
|
s += f"({h})"
|
||
|
|
parts.append(s)
|
||
|
|
cardinal.sendMsg(channel, f"{name}: {', '.join(parts)}")
|
||
|
|
|
||
|
|
def _location_name(self, state, loc_id):
|
||
|
|
"""Get square name from location ID."""
|
||
|
|
squares = state.get("squares", [])
|
||
|
|
if 0 <= loc_id < len(squares):
|
||
|
|
return squares[loc_id]["name"]
|
||
|
|
elif loc_id == 40:
|
||
|
|
return "JAIL"
|
||
|
|
return f"square {loc_id}"
|
||
|
|
|
||
|
|
def close(self):
|
||
|
|
"""Save final state on plugin unload."""
|
||
|
|
state = self.parser.get_state()
|
||
|
|
if state:
|
||
|
|
state["lastUpdated"] = datetime.now(timezone.utc).isoformat()
|
||
|
|
self._save_state(state)
|
||
|
|
logger.info("MonopPlugin unloaded")
|
||
|
|
|
||
|
|
|
||
|
|
entrypoint = MonopPlugin
|