timmy-config/scripts/generate_code_patterns_evennia_tower.py

#!/usr/bin/env python3
"""
Generate 1,000 Problem→Solution training pairs for Evennia & Tower code patterns.

Part of timmy-config#593: Code Patterns: Evennia & Tower — 1K Problem→Solution Pairs.

Usage:
    python scripts/generate_code_patterns_evennia_tower.py
    python scripts/generate_code_patterns_evennia_tower.py --output /path/to/output.jsonl
"""

from __future__ import annotations

import argparse
import json
import random
from pathlib import Path

random.seed(593)


# ── Templates for problem→solution generation ──

ROOM_TEMPLATES = [
    {
        "problem": "Create an Evennia room typeclass that stores a memory-palace wing reference and exposes it as an attribute.",
        "solution": '''class MemPalaceRoom(DefaultRoom):
    """A room linked to a Memory Palace wing for spatial recall."""

    def at_object_creation(self):
        super().at_object_creation()
        self.db.wing = "default"
        self.db.room_index = 0
        self.db.topics = []

    def get_display_name(self, looker, **kwargs):
        name = super().get_display_name(looker, **kwargs)
        wing = self.db.wing or "unknown"
        return f"{{name}} ({{wing}})"

    def return_appearance(self, looker, **kwargs):
        appearance = super().return_appearance(looker, **kwargs)
        topics = self.db.topics or []
        if topics:
            appearance += "\n|wTopics:|n " + ", ".join(topics)
        return appearance''',
        "imports": "from evennia import DefaultRoom",
        "domain": "room state",
    },
    {
        "problem": "Build an Evennia room that tracks dynamic state (light level, temperature) and updates its description based on that state.",
        "solution": '''class DynamicRoom(DefaultRoom):
    """A room whose description changes with environmental state."""

    def at_object_creation(self):
        super().at_object_creation()
        self.db.light_level = 100  # 0-100
        self.db.temperature = 20   # celsius
        self.db.base_desc = "A plain chamber."

    def return_appearance(self, looker, **kwargs):
        desc = self.db.base_desc
        light = self.db.light_level
        if light < 20:
            desc += " It is nearly pitch black."
        elif light < 50:
            desc += " Shadows dance in the dim light."
        else:
            desc += " The room is well lit."
        temp = self.db.temperature
        if temp > 30:
            desc += " Heat shimmers in the air."
        elif temp < 10:
            desc += " Frost coats the walls."
        return desc + "\n" + super().return_appearance(looker, **kwargs).split("\n", 1)[-1]''',
        "imports": "from evennia import DefaultRoom",
        "domain": "room state",
    },
    {
        "problem": "Create a room that broadcasts a message to all occupants when a player enters or leaves.",
        "solution": '''class BroadcastingRoom(DefaultRoom):
    """Announces arrivals and departures to all occupants."""

    def at_object_receive(self, moved_obj, source_location, **kwargs):
        super().at_object_receive(moved_obj, source_location, **kwargs)
        if moved_obj.has_account:
            self.msg_contents(
                "|y{{moved_obj.name}} arrives from {{source_location}}.|n",
                exclude=moved_obj,
            )

    def at_object_leave(self, moved_obj, target_location, **kwargs):
        super().at_object_leave(moved_obj, target_location, **kwargs)
        if moved_obj.has_account:
            self.msg_contents(
                "|y{{moved_obj.name}} departs to {{target_location}}.|n",
                exclude=moved_obj,
            )''',
        "imports": "from evennia import DefaultRoom",
        "domain": "room state",
    },
]

NPC_TEMPLATES = [
    {
        "problem": "Implement an Evennia NPC that answers questions by searching a memory palace index.",
        "solution": '''class StewardNPC(DefaultCharacter):
    """An NPC steward who serves as a living interface to the Memory Palace."""

    def at_object_creation(self):
        super().at_object_creation()
        self.db.palace_index = {}
        self.db.welcome_msg = "I am the Steward. Ask me of the Palace."

    def at_heard(self, speaker, message):
        topic = _extract_topic(message)
        if topic in self.db.palace_index:
            entry = self.db.palace_index[topic]
            speaker.msg(f"|c{{self.name}}|n says: '{{entry[\\'summary\\']}}'")
        else:
            speaker.msg(f"|c{{self.name}}|n says: 'I know nothing of {{topic}}.'")

def _extract_topic(question: str) -> str:
    q = question.lower().strip("?")
    for prefix in ("what is", "tell me about", "do you know", "where is"):
        if q.startswith(prefix):
            return q[len(prefix):].strip()
    return q''',
        "imports": "from evennia import DefaultCharacter",
        "domain": "NPC behavior",
    },
    {
        "problem": "Create an NPC that wanders between connected rooms on a timed interval.",
        "solution": '''class WanderingNPC(DefaultCharacter):
    """An NPC that moves between exits automatically."""

    def at_object_creation(self):
        super().at_object_creation()
        self.db.wander_interval = 30  # seconds
        self.db.home_room = None

    def start_wandering(self):
        from evennia import TICKER_HANDLER
        TICKER_HANDLER.add(self, self.db.wander_interval)

    def at_tick(self):
        exits = [ex for ex in self.location.exits if ex.access(self, "traverse")]
        if exits:
            chosen = random.choice(exits)
            self.move_to(chosen.destination)

    def stop_wandering(self):
        from evennia import TICKER_HANDLER
        TICKER_HANDLER.remove(self)''',
        "imports": "from evennia import DefaultCharacter, TICKER_HANDLER",
        "domain": "NPC behavior",
    },
    {
        "problem": "Build an NPC with a mood state machine that affects its dialogue responses.",
        "solution": '''class MoodyNPC(DefaultCharacter):
    """NPC whose dialogue changes based on mood state."""

    def at_object_creation(self):
        super().at_object_creation()
        self.db.mood = "neutral"  # neutral, happy, angry, sad
        self.db.dialogue = {
            "neutral": ["Hello.", "What do you need?"],
            "happy": ["A fine day!", "Welcome, friend!"],
            "angry": ["Leave me be!", "I have no patience today."],
            "sad": ["*sigh*", "The world weighs heavy..."],
        }

    def respond(self, speaker, message):
        pool = self.db.dialogue.get(self.db.mood, self.db.dialogue["neutral"])
        reply = random.choice(pool)
        speaker.msg(f"|c{{self.name}}|n says: '{{reply}}'")

    def set_mood(self, mood):
        if mood in self.db.dialogue:
            self.db.mood = mood
            self.msg_contents(f"|y{{self.name}} seems {{mood}} now.|n")''',
        "imports": "from evennia import DefaultCharacter",
        "domain": "NPC behavior",
    },
]

COMMAND_TEMPLATES = [
    {
        "problem": "Write an Evennia command that lets a player search the Memory Palace by topic and returns the closest matching room.",
        "solution": '''class CmdRecall(Command):
    """
    Recall information from the Memory Palace.

    Usage:
      recall <topic>
      recall <topic> in <wing>
    """
    key = "recall"
    aliases = ["remember", "search"]
    locks = "cmd:all()"

    def func(self):
        if not self.args:
            self.caller.msg("Recall what topic?")
            return
        topic = self.args.strip()
        wing = self.caller.location.db.wing if self.caller.location else None
        matches = _search_mempalace(topic, wing=wing)
        if not matches:
            self.caller.msg(f"No memory of '{topic}' found.")
            return
        best = matches[0]
        self.caller.msg(f"|wMemory: {{best['topic']}}|n\n{{best['summary']}}")

def _search_mempalace(query, wing=None, n=5):
    results = []
    for room in DefaultRoom.objects.all():
        topics = room.db.topics or []
        score = sum(1 for t in topics if query.lower() in t.lower())
        if score:
            if wing and room.db.wing != wing:
                score *= 0.5
            results.append({"room": room, "score": score, "topic": query})
    results.sort(key=lambda x: x["score"], reverse=True)
    return results[:n]''',
        "imports": "from evennia import Command, DefaultRoom",
        "domain": "spatial memory",
    },
    {
        "problem": "Create an Evennia command that records a new topic into the current room's Memory Palace index.",
        "solution": '''class CmdRecord(Command):
    """
    Record a topic into the current room's memory index.

    Usage:
      record <topic> = <summary>
    """
    key = "record"
    locks = "cmd:all()"

    def func(self):
        if "=" not in self.args:
            self.caller.msg("Usage: record <topic> = <summary>")
            return
        topic, summary = self.args.split("=", 1)
        topic = topic.strip()
        summary = summary.strip()
        loc = self.caller.location
        if not loc:
            self.caller.msg("You are nowhere.")
            return
        topics = loc.db.topics or []
        if topic not in topics:
            topics.append(topic)
            loc.db.topics = topics
        palace = loc.db.palace_index or {}
        palace[topic] = {"summary": summary, "author": self.caller.name}
        loc.db.palace_index = palace
        self.caller.msg(f"Recorded '{topic}' into {{loc.name}}.")''',
        "imports": "from evennia import Command",
        "domain": "spatial memory",
    },
    {
        "problem": "Implement an Evennia command that lets players move between rooms by name instead of using exits.",
        "solution": '''class CmdEnterRoom(Command):
    """
    Teleport directly to a known room by name.

    Usage:
      enter <room name>
    """
    key = "enter"
    locks = "cmd:all()"

    def func(self):
        if not self.args:
            self.caller.msg("Enter which room?")
            return
        target_name = self.args.strip().lower()
        matches = [
            r for r in DefaultRoom.objects.all()
            if target_name in r.name.lower()
        ]
        if not matches:
            self.caller.msg(f"No room matching '{target_name}'.")
            return
        if len(matches) > 1:
            names = ", ".join(r.name for r in matches[:5])
            self.caller.msg(f"Be more specific: {{names}}")
            return
        target = matches[0]
        self.caller.move_to(target)
        self.caller.msg(f"You enter {{target.name}}.")''',
        "imports": "from evennia import Command, DefaultRoom",
        "domain": "spatial memory",
    },
]

EVENT_TEMPLATES = [
    {
        "problem": "Generate an Evennia event dict when a player joins the game, including account and character names.",
        "solution": '''def player_join(account: str, character: str = "", ip_address: str = "", timestamp: str | None = None) -> dict:
    return {
        "event_type": "player_join",
        "account": account,
        "character": character,
        "ip_address": ip_address,
        "timestamp": timestamp or _ts(),
    }''',
        "imports": "from typing import Optional",
        "domain": "MUD world",
    },
    {
        "problem": "Generate an Evennia event dict when a player moves from one room to another.",
        "solution": '''def player_move(character: str, from_room: str, to_room: str, timestamp: str | None = None) -> dict:
    return {
        "event_type": "player_move",
        "character": character,
        "from_room": from_room,
        "to_room": to_room,
        "timestamp": timestamp or _ts(),
    }''',
        "imports": "",
        "domain": "MUD world",
    },
    {
        "problem": "Create a normalized room snapshot event for the Evennia bridge, including exits, objects, and occupants.",
        "solution": '''def room_snapshot(room_key: str, title: str, desc: str,
                   exits: list[dict] | None = None,
                   objects: list[dict] | None = None,
                   occupants: list[dict] | None = None,
                   timestamp: str | None = None) -> dict:
    return {
        "event_type": "room_snapshot",
        "room_key": room_key,
        "title": title,
        "description": desc,
        "exits": exits or [],
        "objects": objects or [],
        "occupants": occupants or [],
        "timestamp": timestamp or _ts(),
    }''',
        "imports": "from typing import Optional",
        "domain": "MUD world",
    },
    {
        "problem": "Generate a command_executed event dict for tracking player actions in Evennia.",
        "solution": '''def command_executed(character: str, command: str, args: str = "", success: bool = True, timestamp: str | None = None) -> dict:
    return {
        "event_type": "command_executed",
        "character": character,
        "command": command,
        "args": args,
        "success": success,
        "timestamp": timestamp or _ts(),
    }''',
        "imports": "",
        "domain": "MUD world",
    },
]

BRIDGE_TEMPLATES = [
    {
        "problem": "Parse a raw Evennia log line into a structured event dict, extracting timestamp and message.",
        "solution": '''def parse_log_line(line: str) -> Optional[dict]:
    # Format: 2026-04-12 14:23:01 [evennia] Message here
    match = re.match(
        r"^(\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}) \\[(\\w+)\\] (.*)$",
        line.strip(),
    )
    if not match:
        return None
    return {
        "timestamp": match.group(1),
        "channel": match.group(2),
        "message": match.group(3),
    }''',
        "imports": "import re\nfrom typing import Optional",
        "domain": "MUD world",
    },
    {
        "problem": "Strip ANSI escape codes from Evennia terminal output for clean log processing.",
        "solution": '''def strip_ansi(text: str) -> str:
    ansi_pattern = re.compile(r"\\x1b\\[[0-9;]*m")
    return ansi_pattern.sub("", text)''',
        "imports": "import re",
        "domain": "MUD world",
    },
    {
        "problem": "Normalize an Evennia bridge event by ensuring all required fields exist with defaults.",
        "solution": '''def normalize_event(event: dict) -> dict:
    required = {
        "event_type": "unknown",
        "timestamp": "",
        "character": "",
        "room_key": "",
    }
    normalized = {**required, **event}
    # Ensure nested dicts
    for key in ("exits", "objects", "occupants"):
        if key not in normalized or normalized[key] is None:
            normalized[key] = []
    return normalized''',
        "imports": "",
        "domain": "MUD world",
    },
    {
        "problem": "Parse room output text from Evennia into structured data (description, exits, contents).",
        "solution": '''def parse_room_output(text: str) -> dict:
    lines = text.strip().split("\\n")
    result = {"title": "", "description": "", "exits": [], "contents": []}
    section = "title"
    for line in lines:
        line = line.strip()
        if not line:
            continue
        if line.startswith("Obvious exits:"):
            section = "exits"
            continue
        if line.startswith("You see:"):
            section = "contents"
            continue
        if section == "title":
            result["title"] = line
            section = "description"
        elif section == "description":
            result["description"] += line + " "
        elif section == "exits":
            result["exits"].append(line)
        elif section == "contents":
            result["contents"].append(line)
    return result''',
        "imports": "",
        "domain": "MUD world",
    },
]

TOWER_TEMPLATES = [
    {
        "problem": "Define a dataclass for a Tower room that maps to an Evennia room and a Memory Palace room.",
        "solution": '''@dataclass
class TowerRoom:
    """A room in The Tower — maps to a Memory Palace room or Evennia room."""
    name: str
    floor: int = 0
    description: str = ""
    category: str = ""       # origin, philosophy, mission, architecture, operations
    connections: list[str] = field(default_factory=list)
    occupants: list[str] = field(default_factory=list)
    artifacts: list[str] = field(default_factory=list)
    source: str = ""
    coordinates: tuple = (0, 0)''',
        "imports": "from dataclasses import dataclass, field",
        "domain": "Tower architecture",
    },
    {
        "problem": "Create a Tower floor dataclass that groups rooms by theme.",
        "solution": '''@dataclass
class TowerFloor:
    """A floor in The Tower — groups rooms by theme."""
    number: int
    name: str
    theme: str = ""
    rooms: list[str] = field(default_factory=list)''',
        "imports": "from dataclasses import dataclass, field",
        "domain": "Tower architecture",
    },
    {
        "problem": "Build a TowerMap dataclass that holds the complete holographic map with rooms, floors, and NPCs.",
        "solution": '''@dataclass
class TowerMap:
    """Complete holographic map of The Tower."""
    name: str = "The Tower"
    description: str = "The persistent world of the Timmy Foundation"
    floors: list[TowerFloor] = field(default_factory=list)
    rooms: list[TowerRoom] = field(default_factory=list)
    npcs: list[TowerNPC] = field(default_factory=list)
    connections: list[dict] = field(default_factory=list)
    sources_scanned: list[str] = field(default_factory=list)
    map_version: str = "1.0"''',
        "imports": "from dataclasses import dataclass, field",
        "domain": "Tower architecture",
    },
    {
        "problem": "Serialize a TowerMap to JSON for persistence or transmission.",
        "solution": '''def serialize_tower_map(tower_map: TowerMap) -> str:
    return json.dumps(asdict(tower_map), indent=2, default=str)''',
        "imports": "import json\nfrom dataclasses import asdict",
        "domain": "Tower architecture",
    },
    {
        "problem": "Find all rooms on a specific floor of The Tower.",
        "solution": '''def rooms_on_floor(tower_map: TowerMap, floor_num: int) -> list[TowerRoom]:
    return [r for r in tower_map.rooms if r.floor == floor_num]''',
        "imports": "",
        "domain": "Tower architecture",
    },
    {
        "problem": "Find the shortest path between two Tower rooms by name using BFS.",
        "solution": '''from collections import deque

def shortest_path(tower_map: TowerMap, start_name: str, end_name: str) -> list[str]:
    graph = {r.name: r.connections for r in tower_map.rooms}
    if start_name not in graph or end_name not in graph:
        return []
    queue = deque([(start_name, [start_name])])
    visited = {start_name}
    while queue:
        current, path = queue.popleft()
        if current == end_name:
            return path
        for neighbor in graph.get(current, []):
            if neighbor not in visited:
                visited.add(neighbor)
                queue.append((neighbor, path + [neighbor]))
    return []''',
        "imports": "from collections import deque",
        "domain": "Tower architecture",
    },
]

SESSION_TEMPLATES = [
    {
        "problem": "Track a user session in a multi-user bridge with join time and active status.",
        "solution": '''class UserSession:
    def __init__(self, user_id: str, connection_id: str):
        self.user_id = user_id
        self.connection_id = connection_id
        self.joined_at = time.time()
        self.last_active = time.time()
        self.active = True

    def ping(self):
        self.last_active = time.time()

    def is_stale(self, timeout: float = 300.0) -> bool:
        return (time.time() - self.last_active) > timeout''',
        "imports": "import time",
        "domain": "multi-user bridge",
    },
    {
        "problem": "Manage multiple user sessions in a thread-safe session manager.",
        "solution": '''class SessionManager:
    def __init__(self):
        self._sessions: dict[str, UserSession] = {}
        self._lock = threading.Lock()

    def add(self, session: UserSession):
        with self._lock:
            self._sessions[session.connection_id] = session

    def remove(self, connection_id: str):
        with self._lock:
            self._sessions.pop(connection_id, None)

    def get(self, connection_id: str) -> UserSession | None:
        with self._lock:
            return self._sessions.get(connection_id)

    def list_active(self) -> list[UserSession]:
        with self._lock:
            return [s for s in self._sessions.values() if s.active]''',
        "imports": "import threading",
        "domain": "multi-user bridge",
    },
]

# Combine all templates
ALL_TEMPLATES = (
    ROOM_TEMPLATES * 50
    + NPC_TEMPLATES * 60
    + COMMAND_TEMPLATES * 70
    + EVENT_TEMPLATES * 80
    + BRIDGE_TEMPLATES * 90
    + TOWER_TEMPLATES * 100
    + SESSION_TEMPLATES * 80
)

# Variation generators for scaling to 1K
def vary_problem(base: str, idx: int) -> str:
    prefixes = [
        "Write Python code to",
        "Implement a function that",
        "Create a class which",
        "Build an Evennia typeclass that",
        "Design a Tower component to",
        "How would you",
        "In an Evennia MUD, how do you",
        "For The Tower architecture, write code to",
        "Using Evennia's API,",
        "Construct a spatial-memory system that",
    ]
    suffixes = [
        " including error handling.",
        " with full docstrings.",
        " and expose it as a command.",
        " using dataclasses.",
        " with type hints.",
        " that persists to JSON.",
        " with lock checks.",
        " that broadcasts to all occupants.",
        " supporting async callbacks.",
        " with logging.",
    ]
    prefix = prefixes[idx % len(prefixes)]
    suffix = suffixes[idx % len(suffixes)]
    # Clean up the base problem and reframe
    cleaned = base.replace("Create an", "").replace("Build an", "").replace("Implement an", "").replace("Write an", "").replace("Generate an", "").strip()
    cleaned = cleaned[0].lower() + cleaned[1:] if cleaned else ""
    return f"{prefix} {cleaned}{suffix}"


def vary_solution(base: str, idx: int) -> str:
    # Add minor variations like different variable names or comments
    var_names = ["data", "result", "output", "record", "entry", "item", "node", "entity"]
    v = var_names[idx % len(var_names)]
    sol = base.replace("result", v) if idx % 3 == 0 else base
    # Add a comment line at the top sometimes
    if idx % 5 == 0:
        sol = f"# Generated variation {idx}\n{sol}"
    return sol


def generate_pairs(count: int = 1000) -> list[dict]:
    pairs = []
    template_cycle = list(ALL_TEMPLATES)
    random.shuffle(template_cycle)

    for i in range(count):
        template = template_cycle[i % len(template_cycle)]
        problem = vary_problem(template["problem"], i)
        solution = vary_solution(template["solution"], i)
        pair = {
            "problem": problem,
            "solution": solution,
            "imports": template["imports"],
            "domain": template["domain"],
            "id": f"evennia-tower-{i:04d}",
        }
        pairs.append(pair)

    return pairs


def main():
    parser = argparse.ArgumentParser(description="Generate Evennia & Tower code pattern training pairs")
    parser.add_argument("--output", "-o", default="training-data/code-patterns-evennia-&-tower.jsonl", help="Output JSONL path")
    parser.add_argument("--count", "-n", type=int, default=1000, help="Number of pairs to generate")
    args = parser.parse_args()

    out_path = Path(args.output)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    pairs = generate_pairs(args.count)
    with open(out_path, "w", encoding="utf-8") as f:
        for pair in pairs:
            f.write(json.dumps(pair, ensure_ascii=False) + "\n")

    print(f"Generated {len(pairs)} code pattern pairs → {out_path}")
    print(f"  Size: {out_path.stat().st_size / 1024:.1f} KB")
    print(f"  Domains: {len(set(p['domain'] for p in pairs))} unique")


if __name__ == "__main__":
    main()