"""NEMOCITY event-type specs — validation/clamping for the 4 city event types. Semantics (godseed lineage): * out-of-range NUMBERS are CLAMPED, never rejected; * unknown kinds are repaired via SYNONYMS (anything unknown -> house); * malformed shapes (bad cells, missing geometry) are rejected with a terse observation string — but the only module that COMPOSES these events is the engine itself (placement/traffic), so rejections mean engine bugs, not model noise. The LLM never outputs coordinates. Pure stdlib. world.py validates through `validate_call`; queue_worker uses `resolve_kind` / `sanitize_name` / `default_name` on model-supplied fields. """ from __future__ import annotations import re import zlib from typing import Any, Optional from . import constants as C from .moderation import Moderator _moderator = Moderator() # wordlist layers only; no judge TOOL_NAMES: tuple[str, ...] = ("place_building", "lay_road", "apply_fix", "note") FIX_ACTIONS = ("new_road", "upgrade_avenue") NOTE_KINDS = ("milestone", "infill", "fix") _FALLBACK_KIND = "house" # ----------------------------------------------------------------- kind resolver def resolve_kind(value: Any) -> str: """Map model-supplied kind text to a BUILDINGS key. Never fails: synonyms repair, substring rescues, anything else becomes a humble house.""" if not isinstance(value, str): return _FALLBACK_KIND v = re.sub(r"[\s\-]+", "_", value.strip().lower()).strip("_") if v in C.BUILDINGS: return v if v in C.SYNONYMS: return C.SYNONYMS[v] if v.endswith("s") and v[:-1] in C.BUILDINGS: return v[:-1] for token, kind in C.SYNONYMS.items(): if token in v: return kind for kind in C.BUILDINGS: if kind in v: return kind return _FALLBACK_KIND # ------------------------------------------------------------------ name helpers _STOPWORDS = frozenset(( "a", "an", "the", "and", "or", "of", "for", "with", "near", "by", "at", "in", "on", "to", "please", "build", "make", "add", "put", "want", "like", "new", "some", "my", "our", "me", "us", "city", "town", )) def sanitize_name(value: Any, default: str = "") -> str: """Printable, wordlist-clean, <=24 chars; falls back to `default`.""" if not isinstance(value, str): return default name = "".join(ch for ch in value if ch.isprintable()) name = re.sub(r"\s+", " ", name).strip()[: C.NAME_MAX_LEN].strip() if not name or not _moderator.check_content(name).allowed: return default return name def default_name(kind: str, petition_text: str, wish_id: str) -> str: """Deterministic pleasant default derived from petition words.""" words = [ w for w in re.findall(r"[A-Za-z]+", str(petition_text or "")) if len(w) > 2 and w.lower() not in _STOPWORDS and w.lower() not in C.BUILDINGS and w.lower() not in C.SYNONYMS ] label = kind.replace("_", " ").title() if words: pick = words[zlib.crc32(f"{wish_id}:{kind}".encode()) % len(words)].title() candidate = f"{pick} {label}"[: C.NAME_MAX_LEN].strip() if _moderator.check_content(candidate).allowed: return candidate return f"The {label}"[: C.NAME_MAX_LEN] def street_name_for(key: str) -> str: """Curated street name from a deterministic key (e.g. 'w_000005:1:road').""" return C.STREET_NAMES[zlib.crc32(str(key).encode()) % len(C.STREET_NAMES)] # ------------------------------------------------------------------- primitives def _err(msg: str) -> tuple[None, str]: return None, f"rejected: {msg}" def _int_in(value: Any, lo: int, hi: int) -> Optional[int]: if isinstance(value, bool) or not isinstance(value, (int, float)): try: value = float(str(value).strip()) except (ValueError, TypeError): return None return int(round(min(max(float(value), lo), hi))) def _coerce_cell(cell: Any) -> Optional[list[int]]: if not isinstance(cell, (list, tuple)) or len(cell) < 2: return None cx = _int_in(cell[0], C.COORD_MIN, C.COORD_MAX) cz = _int_in(cell[1], C.COORD_MIN, C.COORD_MAX) if cx is None or cz is None: return None return [cx, cz] def _coerce_cells(raw: Any, max_cells: int = 256) -> Optional[list[list[int]]]: if not isinstance(raw, (list, tuple)) or not raw: return None out: list[list[int]] = [] seen: set[tuple[int, int]] = set() for c in list(raw)[:max_cells]: cell = _coerce_cell(c) if cell is None: return None if (cell[0], cell[1]) in seen: continue seen.add((cell[0], cell[1])) out.append(cell) return out def _text(value: Any, max_len: int) -> str: s = "".join(ch for ch in str(value or "") if ch.isprintable()) return re.sub(r"\s+", " ", s).strip()[:max_len] # ------------------------------------------------------------------- validators def _validate_place_building(args: dict): kind = resolve_kind(args.get("kind")) spec = C.BUILDINGS[kind] cx = _int_in(args.get("cx"), C.COORD_MIN, C.COORD_MAX - spec["w"] + 1) cz = _int_in(args.get("cz"), C.COORD_MIN, C.COORD_MAX - spec["d"] + 1) if cx is None or cz is None or args.get("cx") is None or args.get("cz") is None: return _err("place_building needs cx, cz") lo, hi = spec["floors"] floors = _int_in(args.get("floors"), lo, hi) if floors is None: floors = lo hue = _int_in(args.get("hue"), 0, 360) if hue is None: hue = 35 variant = _int_in(args.get("variant"), 0, 7) if variant is None: variant = 0 name = sanitize_name(args.get("name"), default=f"The {kind.replace('_', ' ').title()}") return { "kind": kind, "name": name, "cx": cx, "cz": cz, "w": spec["w"], "d": spec["d"], "floors": floors, "hue": hue, "variant": variant, }, None def _validate_lay_road(args: dict): cells = _coerce_cells(args.get("cells")) if cells is None: return _err("lay_road needs cells [[cx,cz],...]") klass = str(args.get("klass") or "street").strip().lower() if klass not in C.ROAD_CLASSES: klass = "street" name = sanitize_name(args.get("name"), default="") out = {"cells": cells, "klass": klass} if name: out["name"] = name return out, None def _coerce_metrics(raw: Any) -> dict: out: dict[str, Any] = {} if isinstance(raw, dict): for k, v in list(raw.items())[:6]: key = _text(k, 24) if not key: continue if isinstance(v, bool): continue if isinstance(v, (int, float)): num = round(float(v), 2) out[key] = int(num) if num.is_integer() else num elif isinstance(v, str): out[key] = _text(v, 48) return out def _validate_apply_fix(args: dict): action = str(args.get("action") or "").strip().lower() if action not in FIX_ACTIONS: return _err(f"unknown fix action '{_text(args.get('action'), 24)}'") cells = _coerce_cells(args.get("cells")) if cells is None: return _err("apply_fix needs cells [[cx,cz],...]") klass = str(args.get("klass") or "avenue").strip().lower() if klass not in C.ROAD_CLASSES: klass = "avenue" name = sanitize_name(args.get("name"), default=street_name_for(repr(cells[0]))) diagnosis = _text(args.get("diagnosis"), 200) return { "action": action, "cells": cells, "klass": klass, "name": name, "diagnosis": diagnosis, "metrics_before": _coerce_metrics(args.get("metrics_before")), "metrics_predicted": _coerce_metrics(args.get("metrics_predicted")), }, None def _validate_note(args: dict): text = _text(args.get("text"), 140) if not text: return _err("note needs text") if not _moderator.check_content(text).allowed: return _err("those words may not be posted on the city ledger") kind = str(args.get("kind") or "milestone").strip().lower() if kind not in NOTE_KINDS: kind = "milestone" return {"text": text, "kind": kind}, None _VALIDATORS = { "place_building": _validate_place_building, "lay_road": _validate_lay_road, "apply_fix": _validate_apply_fix, "note": _validate_note, } def validate_call(tool: Any, args: Any) -> tuple[Optional[dict], Optional[str]]: """Validate one event. Returns (canonical_args, None) or (None, rejection).""" if not isinstance(tool, str) or tool not in _VALIDATORS: return _err(f"unknown tool '{_text(tool, 32)}'") if args is None: args = {} if not isinstance(args, dict): return _err("args must be an object") return _VALIDATORS[tool](args) def as_dict() -> dict: """Plain-JSON view of the event surface (prompt builders may render this).""" return { "place_building": { "kind": {"type": "enum", "values": list(C.BUILDINGS)}, "name": {"type": "string", "max_len": C.NAME_MAX_LEN}, "cx": {"type": "int", "min": C.COORD_MIN, "max": C.COORD_MAX}, "cz": {"type": "int", "min": C.COORD_MIN, "max": C.COORD_MAX}, "floors": {"type": "int", "note": "clamped to the kind's range"}, "hue": {"type": "int", "min": 0, "max": 360}, }, "lay_road": { "cells": {"type": "cells", "note": "[[cx,cz],...] engine-routed only"}, "klass": {"type": "enum", "values": list(C.ROAD_CLASSES)}, "name": {"type": "string", "max_len": C.NAME_MAX_LEN}, }, "apply_fix": { "action": {"type": "enum", "values": list(FIX_ACTIONS)}, "cells": {"type": "cells"}, "klass": {"type": "enum", "values": list(C.ROAD_CLASSES)}, "diagnosis": {"type": "string", "max_len": 200}, }, "note": { "text": {"type": "string", "max_len": 140}, "kind": {"type": "enum", "values": list(NOTE_KINDS)}, }, }