"""NEMOCITY derived city state — a pure function of the event list. CityState is cheap to rebuild (from_events) and supports incremental apply() so the queue worker can keep one instance current while a petition lands. Nothing here is persisted: occupancy, the road graph, street names, tallies, growth radius and road_version are all derived. Determinism notes (the JS renderer mirrors parts of this): * street names: lay_road/apply_fix args carry an explicit name when the engine composed one; events without a name fall back to the curated pool indexed by crc32(event_id); * a re-laid road cell takes the LATER event's klass/name (upgrade_avenue repaints in place); * building "door" = the road cell 4-adjacent to the footprint with the smallest (cz, cx) — traffic.js must use the same rule. Pure stdlib. """ from __future__ import annotations import zlib from dataclasses import dataclass, field from typing import Any, Iterable, Optional from . import constants as C Cell = tuple[int, int] # (cx, cz) _NEIGHBORS_4: tuple[Cell, ...] = ((0, -1), (1, 0), (0, 1), (-1, 0)) WATER_CELLS: frozenset[Cell] = frozenset( (cx, cz) for cz in range(C.COORD_MIN, C.COORD_MAX + 1) for cx in C.river_cols(cz) ) def in_bounds(cx: int, cz: int) -> bool: return C.COORD_MIN <= cx <= C.COORD_MAX and C.COORD_MIN <= cz <= C.COORD_MAX def neighbors4(cell: Cell) -> list[Cell]: cx, cz = cell return [ (cx + dx, cz + dz) for dx, dz in _NEIGHBORS_4 if in_bounds(cx + dx, cz + dz) ] @dataclass class RoadCell: klass: str name: str bridge: bool @dataclass class Building: id: str kind: str name: str cx: int cz: int w: int d: int floors: int seed: int t: float cells: tuple[Cell, ...] = field(default_factory=tuple) @property def centroid(self) -> tuple[float, float]: return (self.cx + (self.w - 1) / 2.0, self.cz + (self.d - 1) / 2.0) def progress(self, now_s: float) -> float: dur = C.BUILDINGS[self.kind]["duration_s"] return min(max((now_s - self.t) / dur, 0.0), 1.0) def _event_fields(ev: Any) -> tuple[str, str, dict, int, float]: if isinstance(ev, dict): return (ev["id"], ev["tool"], ev.get("args") or {}, int(ev.get("seed", 0)), float(ev.get("t", 0))) return (ev.id, ev.tool, ev.args or {}, int(ev.seed), float(ev.t)) class CityState: """Occupancy + road graph + registries derived from an event list.""" def __init__(self) -> None: self.roads: dict[Cell, RoadCell] = {} self.buildings: list[Building] = [] self.building_cells: dict[Cell, Building] = {} self.road_version: int = 0 self._adjacency: Optional[dict[Cell, list[Cell]]] = None self._street_cells: Optional[dict[str, list[Cell]]] = None @classmethod def from_events(cls, events: Iterable[Any]) -> "CityState": city = cls() for ev in events or (): city.apply(ev) return city # ------------------------------------------------------------------ apply def apply(self, ev: Any) -> None: ev_id, tool, args, seed, t = _event_fields(ev) if tool in ("lay_road", "apply_fix"): name = args.get("name") or C.STREET_NAMES[ zlib.crc32(ev_id.encode()) % len(C.STREET_NAMES) ] klass = args.get("klass", "street") for cx, cz in args.get("cells") or (): cell = (int(cx), int(cz)) if not in_bounds(*cell) or cell in self.building_cells: continue self.roads[cell] = RoadCell( klass=klass, name=name, bridge=cell in WATER_CELLS ) self.road_version += 1 self._adjacency = None self._street_cells = None elif tool == "place_building": b = Building( id=ev_id, kind=args["kind"], name=args.get("name", ""), cx=int(args["cx"]), cz=int(args["cz"]), w=int(args.get("w") or C.BUILDINGS[args["kind"]]["w"]), d=int(args.get("d") or C.BUILDINGS[args["kind"]]["d"]), floors=int(args.get("floors", 1)), seed=seed, t=t, ) b.cells = tuple( (b.cx + dx, b.cz + dz) for dx in range(b.w) for dz in range(b.d) ) self.buildings.append(b) for cell in b.cells: self.building_cells[cell] = b # "note" mutates nothing # ------------------------------------------------------------------ views def is_empty(self, cell: Cell) -> bool: return ( in_bounds(*cell) and cell not in WATER_CELLS and cell not in self.roads and cell not in self.building_cells ) def occupancy(self, cell: Cell) -> Optional[str]: """'road' | 'water' | building kind | None (empty/off-grid).""" if cell in self.roads: return "road" if cell in WATER_CELLS: return "water" b = self.building_cells.get(cell) return b.kind if b else None @property def adjacency(self) -> dict[Cell, list[Cell]]: """Road graph: node = road cell, edges between 4-adjacent road cells.""" if self._adjacency is None: self._adjacency = { cell: [n for n in neighbors4(cell) if n in self.roads] for cell in self.roads } return self._adjacency @property def street_cells(self) -> dict[str, list[Cell]]: """Street-name registry; cells in insertion order per name.""" if self._street_cells is None: reg: dict[str, list[Cell]] = {} for cell, rc in self.roads.items(): reg.setdefault(rc.name, []).append(cell) self._street_cells = reg return self._street_cells def street_midpoint(self, name: str) -> Optional[Cell]: cells = self.street_cells.get(name) if not cells: return None return cells[len(cells) // 2] def door(self, b: Building) -> Optional[Cell]: """Frontage road cell: smallest (cz, cx) among cells adjacent to the footprint. Mirrored by traffic.js — keep the ordering rule.""" adjacent = { n for cell in b.cells for n in neighbors4(cell) if n in self.roads } if not adjacent: return None return min(adjacent, key=lambda c: (c[1], c[0])) def population(self, now_s: float) -> int: return int(sum( C.BUILDINGS[b.kind]["residents"] * b.progress(now_s) for b in self.buildings )) @property def housing_capacity(self) -> int: return sum(C.BUILDINGS[b.kind]["residents"] for b in self.buildings) @property def jobs(self) -> int: return sum(C.BUILDINGS[b.kind]["jobs"] for b in self.buildings) @property def growth_radius(self) -> int: r = C.GROWTH_RADIUS_MIN for cell in self.roads: r = max(r, abs(cell[0]), abs(cell[1])) for cell in self.building_cells: r = max(r, abs(cell[0]), abs(cell[1])) return r def counts_by_kind(self) -> dict[str, int]: counts: dict[str, int] = {} for b in self.buildings: counts[b.kind] = counts.get(b.kind, 0) + 1 return counts def find_building(self, text: str) -> Optional[Building]: """Most recent building whose name matches `text` (folded substring, either direction). Returns None when nothing matches.""" needle = str(text or "").strip().lower() if not needle: return None for b in reversed(self.buildings): name = b.name.lower() if name and (name in needle or needle in name): return b return None def find_street(self, text: str) -> Optional[str]: needle = str(text or "").strip().lower() if not needle: return None for name in self.street_cells: low = name.lower() if low and (low in needle or needle in low): return name return None