| """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] |
|
|
| _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 |
|
|
| |
|
|
| 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 |
| |
|
|
| |
|
|
| 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 |
|
|
