""" 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 from .monop_parser import 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