agents/master/base.py

from __future__ import annotations

from contextlib import contextmanager
import re
import warnings
from pathlib import Path


MAX_NODES = 40
MAX_ITEMS = 32
MAX_QUEST_STEPS = 64
MIN_NODES = 5
MIN_QUEST_STEPS = 2
MIN_CLUES = 3
MAX_CLUES = 5
TARGET_RATIO = 1.5
TARGET_RATIO_SIGMA = 0.4
MAX_STEP_MULTIPLIER = 5
INVENTORY_ID = "__inventory__"
STORED_ID = "__stored__"
ROOT_DIR = Path(__file__).resolve().parents[2]
ARTIFACTS_ROOT = ROOT_DIR / ".artifacts" / "dm_env"
CUSTOM_LOGIC_DIR = ROOT_DIR / "textworld_data" / "dnd" / "logic"
CUSTOM_GRAMMAR_DIR = ROOT_DIR / "textworld_data" / "dnd" / "text_grammars"
SUPPORTED_DIRECTIONS = ("north", "south", "east", "west", "up", "down", "in", "out")
OPPOSITE_DIRECTION = {
    "north": "south",
    "south": "north",
    "east": "west",
    "west": "east",
    "up": "down",
    "down": "up",
    "in": "out",
    "out": "in",
}

GO_RE = re.compile(r"^go\((?P<target>[a-z0-9_]+)\)$")
OPEN_RE = re.compile(r"^open\((?P<target>[a-z0-9_]+)\)$")
UNLOCK_RE = re.compile(r"^unlock\((?P<door>[a-z0-9_]+),(?P<key>[a-z0-9_]+)\)$")
TAKE_RE = re.compile(r"^take\((?P<item>[a-z0-9_]+),(?P<source>[a-z0-9_]+)\)$")
READ_RE = re.compile(r"^read\((?P<target>[a-z0-9_]+)\)$")
USE_RE = re.compile(r"^use\((?P<item>[a-z0-9_]+),(?P<target>[a-z0-9_]+)\)$")
COMBINE_RE = re.compile(r"^combine\((?P<item_a>[a-z0-9_]+),(?P<item_b>[a-z0-9_]+)\)$")
GIVE_RE = re.compile(r"^give\((?P<item>[a-z0-9_]+),(?P<npc>[a-z0-9_]+)\)$")
TALK_RE = re.compile(r"^talk\((?P<target>[a-z0-9_]+)\)$")
SUBMIT_RE = re.compile(r"^submit\((?P<quote>[\"'])(?P<answer>.+)(?P=quote)\)$")


class DMCompileError(RuntimeError):
    pass


class DMInterfaceError(RuntimeError):
    pass


@contextmanager
def suppress_unsupported_game_warning():
    with warnings.catch_warnings():
        warnings.filterwarnings(
            "ignore",
            message=r"Game '.*' is not fully supported\..*",
            category=Warning,
        )
        yield


def normalize_snake_id(value: str, kind: str) -> str:
    if not re.fullmatch(r"[a-z][a-z0-9_]*", value):
        raise DMCompileError(f"{kind} '{value}' must be snake_case.")
    return value


def parser_safe_text(value: str) -> str:
    collapsed = re.sub(r"[^A-Za-z0-9 ]+", " ", value).strip().lower()
    collapsed = re.sub(r"\s+", " ", collapsed)
    if not collapsed:
        raise DMCompileError(f"Unable to derive a parser-safe name from '{value}'.")
    return collapsed


def normalize_answer_text(value: str) -> str:
    collapsed = re.sub(r"[^A-Za-z0-9 ]+", " ", value).strip().lower()
    return re.sub(r"\s+", " ", collapsed)