"""The FlowTwin crowd simulator. A mesoscopic, capacity-constrained pedestrian network model. Agents are individuals with their own walking speed, destination, route and compliance, but they move along graph edges rather than in free 2-D space. That choice is deliberate: it keeps 40,000 agents inside a few milliseconds per step, which is what makes counterfactual simulation — running five alternative futures from the same frozen state while an operator waits — actually possible. What the model reproduces, and why each part is needed: * speed collapse under density -> queues form instead of dots piling up * per-minute throughput at gates -> a degraded exit really is a bottleneck * physical storage limits per corridor -> congestion spills back upstream * first-come-first-served admission -> queues behave like queues * per-agent compliance -> a reroute instruction is not obeyed by all Every run is fully determined by (venue, scenario, seed, overrides). The RNG state travels with the snapshot, so a counterfactual branch is reproducible and two strategies are always compared against an identical starting state. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any import numpy as np from ..config import Settings from ..crowd.state import CrowdStateEngine from ..routing.costs import CostModel from ..routing.graph import RoutingTables, static_assignment from ..venue.models import CompiledVenue, NodeType from ..venue.scenario import Scenario, TimelineEvent from .agents import ( POLICY_ADAPTIVE, POLICY_SHORTEST, POLICY_STATIC, STATUS_ARRIVED, STATUS_ON_EDGE, STATUS_WAITING, AgentPopulation, build_population, ) from .movement import CapacityBudget, admit, weidmann_speed HEAD_EPSILON_M = 0.35 @dataclass class RunOverrides: """Per-run parameters the operator can change from the What-If panel.""" crowd_size: int | None = None release_ramp_s: float | None = None compliance_scale: float = 1.0 routing_policy: int = POLICY_SHORTEST capacity_overrides: dict[str, float] = field(default_factory=dict) #: Replacement factors for scripted timeline events, keyed by event target. #: This is how the What-If panel retunes the scripted failure: the event #: still fires when the scenario says it does, but with the operator's #: severity instead of the authored one. event_factor_overrides: dict[str, float] = field(default_factory=dict) disable_timeline: bool = False def as_dict(self) -> dict[str, Any]: return { "crowd_size": self.crowd_size, "release_ramp_s": self.release_ramp_s, "compliance_scale": self.compliance_scale, "routing_policy": int(self.routing_policy), "capacity_overrides": dict(self.capacity_overrides), "event_factor_overrides": dict(self.event_factor_overrides), "disable_timeline": self.disable_timeline, } @dataclass class AppliedIntervention: """Record of an intervention actually applied to this simulation.""" strategy_id: str label: str t_s: float detail: dict[str, Any] = field(default_factory=dict) agents_affected: int = 0 class Simulator: """Discrete-time crowd simulation over a venue graph.""" def __init__( self, venue: CompiledVenue, scenario: Scenario, settings: Settings, seed: int | None = None, overrides: RunOverrides | None = None, ) -> None: self.venue = venue self.scenario = scenario self.settings = settings self.overrides = overrides or RunOverrides() self.seed = int(seed if seed is not None else scenario.default_seed) self.dt = settings.simulation.dt_s crowd = self.overrides.crowd_size or scenario.crowd_size if crowd > settings.simulation.max_agents: raise ValueError( f"crowd_size {crowd} exceeds the configured maximum " f"{settings.simulation.max_agents}" ) self.rng = np.random.default_rng(self.seed) # A separate stream for interventions so that applying a strategy never # perturbs the population's own random draws. self.action_rng = np.random.default_rng(self.seed ^ 0x5F3759DF) self.pop, self.dest_indices, self.dest_ids = build_population( venue, scenario, self.rng, settings.movement, crowd_size=crowd, release_ramp_s=self.overrides.release_ramp_s, compliance_scale=self.overrides.compliance_scale, initial_policy=self.overrides.routing_policy, ) self.n_agents = self.pop.size self.costs = CostModel(venue, settings.routing, settings.movement.free_speed_mps) self.tables = RoutingTables(venue, self.costs, self.dest_indices, settings.routing) self._prepare_static_routing() self.node_budget = CapacityBudget(venue.node_service_ppm) self.edge_budget = CapacityBudget(venue.edge_capacity_ppm) for target, factor in self.overrides.capacity_overrides.items(): self._scale_capacity(target, factor) self.state = CrowdStateEngine( venue, settings.risk, settings.movement, settings.prediction.history_window, settings.prediction.growth_window_s, self.dt, ) self.time = 0.0 self.step_count = 0 self.total_arrived = 0 self.travel_time_sum = 0.0 self.total_rerouted = 0 self.total_reroute_decisions = 0 self.fired_events: set[int] = set() self.event_log: list[dict[str, Any]] = [] self.applied_interventions: list[AppliedIntervention] = [] self.critical_edge_seconds = 0.0 self.risk_integral = 0.0 self.blocked_agents = 0 # Warm the state engine so the first frame is not all zeros. self._cell_density = np.zeros(venue.n_cells) self._queue_len_m = np.zeros(venue.n_edges) self._queued_count = np.zeros(venue.n_edges) self._measure(np.zeros(venue.n_edges), np.zeros(venue.n_edges), np.zeros(venue.n_nodes)) # ------------------------------------------------------------------ # setup # ------------------------------------------------------------------ def _prepare_static_routing(self) -> None: """Build the frozen baseline routing tables. The static baseline runs a small method-of-successive-averages traffic assignment using the scenario's expected demand. It is a real pre-event plan: capacity-aware, but blind to what actually happens. """ self.tables.costs.compute_static_costs( np.zeros(self.venue.n_edges), np.zeros(self.venue.n_nodes) ) self.tables.build_static_tables() demand: list[tuple[int, int, float]] = [] ramp = self.overrides.release_ramp_s or self.scenario.release.ramp_s window_min = max(ramp / 60.0, 1.0) total = self.pop.size for group, share in self.scenario.normalised_demand(): origin = self.venue.node_index[group.origin] weight_sum = sum(group.destinations.values()) for dest_id, w in group.destinations.items(): dest_node = self.venue.node_index[dest_id] slot = self.dest_indices.index(dest_node) people = total * share * (w / weight_sum) demand.append((origin, slot, people / window_min)) edge_vol, node_vol = static_assignment(self.venue, self.tables, demand) self.costs.compute_static_costs(edge_vol, node_vol) self.tables.build_static_tables() self.expected_edge_volume = edge_vol self.expected_node_volume = node_vol def _scale_capacity(self, target: str, factor: float) -> None: if target in self.venue.node_index: self.node_budget.multiplier[self.venue.node_index[target]] *= factor return touched = False for i, base in enumerate(self.venue.edge_base_id): if base == target: self.edge_budget.multiplier[i] *= factor touched = True if not touched: raise KeyError(f"unknown capacity target {target!r}") # ------------------------------------------------------------------ # main loop # ------------------------------------------------------------------ def step(self) -> None: dt = self.dt t = self.time pop = self.pop v = self.venue if not self.overrides.disable_timeline: self._fire_timeline_events(t) # -- 1. local density and walking speed, per cell ------------------ # # Density is evaluated over ~12-metre cells rather than over a whole # corridor. A queue backing up from a degraded gate therefore slows # only the people who have actually reached it, and the congested # region grows upstream cell by cell — which is what a real queue does, # and what makes "peak local density" a meaningful operational number. on_edge = pop.status == STATUS_ON_EDGE edge_idx = pop.edge idx_on = np.flatnonzero(on_edge) occ = np.bincount(edge_idx[idx_on], minlength=v.n_edges).astype(np.float64) pair = v.pair_of has_pair = pair >= 0 combined = occ.copy() combined[has_pair] += occ[pair[has_pair]] # The standing queue at the head of an edge is everyone who has stopped # or is barely shuffling — not only those formally at the stop line. # # This distinction is load-bearing. Discharge is governed by the gate's # throughput, so the queue must be a first-come-first-served pool that # the gate drains. If only the handful of agents literally at the stop # line counted, the queue would occupy almost no length, and everyone # behind would have to *walk* through a near-jammed corridor at a few # centimetres per second to reach it — throttling a 500/min gate to # under 200/min. Measuring the queue by who has actually stopped makes # its physical extent, and therefore where walkers join the back of it, # match what the crowd is really doing. n_queued = self._queued_count if self._queued_count is not None else np.zeros(v.n_edges) pack = self.settings.movement.queue_pack_density queue_len = np.minimum(n_queued / np.maximum(pack * v.edge_width, 1e-6), v.edge_length * 0.99) self._queue_len_m = queue_len queue_start = v.edge_length - queue_len cell_of = np.zeros(0, dtype=np.int64) if idx_on.size: e = edge_idx[idx_on] eff_pos = pop.pos_m[idx_on].astype(np.float64) q = pop.blocked[idx_on] if np.any(q): spread = ((idx_on[q] * 40503) % 997) / 997.0 eff_pos[q] = queue_start[e[q]] + spread * queue_len[e[q]] within = np.clip((eff_pos / v.edge_cell_size[e]).astype(np.int64), 0, v.edge_n_cells[e] - 1) cell_of = v.edge_cell_offset[e] + within cell_occ = np.bincount(cell_of, minlength=v.n_cells).astype(np.float64) cell_comb = cell_occ.copy() cp = v.cell_pair valid_pair = cp >= 0 cell_comb[valid_pair] += cell_occ[cp[valid_pair]] cell_density = cell_comb / np.maximum(v.cell_area, 1e-6) cell_speed = weidmann_speed(cell_density, self.settings.movement) self._cell_density = cell_density # -- 2. advance the walking agents -------------------------------- if idx_on.size: e = edge_idx[idx_on] free_mask = ~pop.blocked[idx_on] speed = cell_speed[cell_of] * pop.speed_factor[idx_on] new_pos = pop.pos_m[idx_on] + speed.astype(np.float32) * np.float32(dt) # A walker cannot step into a cell that is already packed solid. # Without this the model lets people accumulate past the physical # jam density at the head of a corridor; with it, the congestion # front propagates backwards one cell at a time, as it does in a # real crowd. within_now = (cell_of - v.edge_cell_offset[e]).astype(np.int64) has_next = within_now < (v.edge_n_cells[e] - 1) next_full = np.zeros(idx_on.size, dtype=bool) if np.any(has_next): nxt = cell_of[has_next] + 1 next_full[has_next] = cell_density[nxt] >= (self.settings.movement.jam_density * 0.90) cell_ceiling = ((within_now + 1) * v.edge_cell_size[e] - 0.05).astype(np.float32) new_pos = np.where(next_full, np.minimum(new_pos, cell_ceiling), new_pos) # A walker stops when it reaches the back of the standing queue. stop_at = queue_start[e].astype(np.float32) reached = free_mask & (new_pos >= stop_at) pop.pos_m[idx_on] = np.where(free_mask, np.minimum(new_pos, stop_at), pop.pos_m[idx_on]) pop.speed_now[idx_on] = np.where(free_mask & ~reached, speed, 0.0).astype(np.float32) newly = idx_on[reached] if newly.size: pop.blocked[newly] = True pop.pos_m[newly] = v.edge_length[edge_idx[newly]].astype(np.float32) # -- 3. build the transition candidate set ---------------------- released = (pop.status == STATUS_WAITING) & (pop.release_t <= t) at_head = (pop.status == STATUS_ON_EDGE) & pop.blocked cand = np.flatnonzero(released | at_head) edge_inflow = np.zeros(v.n_edges, dtype=np.float64) edge_outflow = np.zeros(v.n_edges, dtype=np.float64) node_throughput = np.zeros(v.n_nodes, dtype=np.float64) if cand.size: fresh = np.isinf(pop.queue_since[cand]) pop.queue_since[cand[fresh]] = np.float32(t) from_node = np.where( pop.status[cand] == STATUS_WAITING, pop.origin[cand], v.edge_dst[np.maximum(pop.edge[cand], 0)], ).astype(np.int32) arriving = from_node == pop.dest_node[cand] target = np.full(cand.size, -1, dtype=np.int32) moving = ~arriving if np.any(moving): target[moving] = self.tables.next_hop[ pop.policy[cand][moving], pop.dest_slot[cand][moving], from_node[moving] ] # No U-turns. A routing table that has just been re-weighted can # briefly make the corridor an agent is standing in look like the # cheapest way onward, which sends people back the way they came # and, with repeated interventions, leaves a residue bouncing # between two nodes. Crowds do not do this; fall back to the # baseline hop unless reversing is genuinely the only option. came_from = np.where(pop.status[cand] == STATUS_ON_EDGE, v.pair_of[np.maximum(pop.edge[cand], 0)], np.int32(-1)) u_turn = moving & (target >= 0) & (target == came_from) if np.any(u_turn): fallback = self.tables.next_hop[ POLICY_SHORTEST, pop.dest_slot[cand][u_turn], from_node[u_turn]] keep = (fallback >= 0) & (fallback != came_from[u_turn]) patched = target[u_turn] patched[keep] = fallback[keep] target[u_turn] = patched # Agents with no onward route are treated as arrived at a dead end # rather than being silently stuck forever. stranded = moving & (target < 0) arriving = arriving | stranded prio = pop.queue_since[cand] # Node throughput budget (gates, exits, transport interfaces). node_allow = self.node_budget.accrue(dt) self.node_budget.clamp_carry(3.0, dt) pass_node = admit(from_node, prio, node_allow) # Edge entry budget, then the receiving limit. # # A link does not accept people at its nominal capacity right up # until it is physically full. As it fills, the rate at which it # can take anyone new falls to zero — the congestion propagates # backwards at `backward_wave_mps`. This is what turns a degraded # exit into a queue that grows up the corridor and then out into # the concourse behind it, instead of a corridor that quietly # absorbs an impossible number of people. edge_allow = self.edge_budget.accrue(dt) self.edge_budget.clamp_carry(3.0, dt) space = np.maximum(v.edge_jam_occupancy - combined, 0.0) receiving_ppm = (self.settings.movement.backward_wave_mps * 60.0 * space / np.maximum(v.edge_length, 1e-6)) receiving = np.floor(receiving_ppm * dt / 60.0).astype(np.int64) edge_allow = np.minimum(edge_allow, np.maximum(receiving, 0)) headroom = np.floor(space).astype(np.int64) edge_allow = np.minimum(edge_allow, headroom) movers_mask = pass_node & ~arriving pass_edge = np.zeros(cand.size, dtype=bool) if np.any(movers_mask): sub = np.flatnonzero(movers_mask) ok = admit(target[sub], prio[sub], edge_allow) pass_edge[sub] = ok absorbers = pass_node & arriving movers = pass_edge used_nodes = np.bincount(from_node[absorbers | movers], minlength=v.n_nodes) self.node_budget.consume(used_nodes.astype(np.float64)) if np.any(movers): used_edges = np.bincount(target[movers], minlength=v.n_edges) self.edge_budget.consume(used_edges.astype(np.float64)) edge_inflow += used_edges node_throughput += used_nodes # -- apply absorptions ------------------------------------- if np.any(absorbers): a = cand[absorbers] prev_edge = pop.edge[a] left = prev_edge >= 0 if np.any(left): edge_outflow += np.bincount(prev_edge[left], minlength=v.n_edges) pop.status[a] = STATUS_ARRIVED pop.arrive_t[a] = np.float32(t) pop.edge[a] = -1 pop.node[a] = from_node[absorbers] pop.pos_m[a] = 0.0 pop.speed_now[a] = 0.0 pop.blocked[a] = False pop.queue_since[a] = np.inf entered = pop.enter_t[a] valid = ~np.isnan(entered) self.travel_time_sum += float(np.sum(t - entered[valid])) self.total_arrived += int(valid.sum()) # -- apply moves -------------------------------------------- if np.any(movers): m = cand[movers] prev_edge = pop.edge[m] left = prev_edge >= 0 if np.any(left): edge_outflow += np.bincount(prev_edge[left], minlength=v.n_edges) tgt = target[movers] # A route change is a decision that differs from the # shortest-path plan the agent would otherwise have followed. baseline_hop = self.tables.next_hop[ POLICY_SHORTEST, pop.dest_slot[m], from_node[movers] ] diverted = (pop.policy[m] != POLICY_SHORTEST) & (tgt != baseline_hop) & (baseline_hop >= 0) if np.any(diverted): n_div = int(diverted.sum()) self.total_reroute_decisions += n_div first_time = pop.reroute_count[m][diverted] == 0 self.total_rerouted += int(first_time.sum()) counts = pop.reroute_count[m] counts[diverted] += 1 pop.reroute_count[m] = counts pop.status[m] = STATUS_ON_EDGE pop.edge[m] = tgt pop.pos_m[m] = 0.0 pop.node[m] = from_node[movers] pop.blocked[m] = False pop.queue_since[m] = np.inf nan_enter = np.isnan(pop.enter_t[m]) if np.any(nan_enter): ent = pop.enter_t[m] ent[nan_enter] = np.float32(t) pop.enter_t[m] = ent # -- 4. measure ------------------------------------------------- self._measure(edge_inflow, edge_outflow, node_throughput, None) # -- 5. refresh adaptive routing -------------------------------- if (self.time - self.tables.last_refresh_t) >= self.settings.routing.refresh_interval_s: self.refresh_routing() self.time += dt self.step_count += 1 def _measure( self, edge_inflow: np.ndarray, edge_outflow: np.ndarray, node_throughput: np.ndarray, _unused: Any = None, ) -> None: v = self.venue pop = self.pop on_edge = pop.status == STATUS_ON_EDGE idx_on = np.flatnonzero(on_edge) occ = np.bincount(pop.edge[idx_on], minlength=v.n_edges).astype(np.float64) speed_sum = np.bincount(pop.edge[idx_on], weights=pop.speed_now[idx_on].astype(np.float64), minlength=v.n_edges) # "Queueing" means moving materially slower than a walk, not merely # standing on the stop line. A corridor where 3,000 people are shuffling # forward at 0.2 m/s is a queue of 3,000, and that is the number an # operator needs. queue_count = np.zeros(v.n_edges, dtype=np.float64) node_queue = np.zeros(v.n_nodes, dtype=np.float64) peak_local = np.zeros(v.n_edges, dtype=np.float64) if idx_on.size: e = pop.edge[idx_on] slow_cut = 0.35 * self.settings.movement.free_speed_mps stuck = pop.blocked[idx_on] | (pop.speed_now[idx_on] < slow_cut) if np.any(stuck): queue_count = np.bincount(e[stuck], minlength=v.n_edges).astype(np.float64) node_queue = np.bincount(v.edge_dst[e[stuck]], minlength=v.n_nodes).astype(np.float64) self._queued_count = queue_count cell_d = getattr(self, "_cell_density", None) if cell_d is not None and cell_d.size: peak_local = np.maximum.reduceat(cell_d, v.edge_cell_offset[:-1]) waiting = pop.status == STATUS_WAITING node_occ = np.bincount(pop.origin[waiting], minlength=v.n_nodes).astype(np.float64) # People held at an origin whose departure time has passed are queueing # to leave, not sitting in a seat. ready = waiting & (pop.release_t <= self.time) if np.any(ready): node_queue += np.bincount(pop.origin[ready], minlength=v.n_nodes).astype(np.float64) self.state.update( edge_occupancy=occ, edge_speed_sum=speed_sum, edge_inflow_count=edge_inflow, edge_outflow_count=edge_outflow, edge_queue_count=queue_count, node_occupancy=node_occ, node_queue=node_queue, node_throughput_count=node_throughput, edge_peak_local=peak_local, warning_density=self.venue.venue.warning_density, critical_density=self.venue.venue.critical_density, ) crit = self.state.critical_edge_count(self.venue.venue.critical_density) self.critical_edge_seconds += crit * self.dt self.risk_integral += float(np.sum(self.state.edge_risk)) * self.dt self.blocked_agents = int(queue_count.sum()) def refresh_routing(self) -> None: """Recompute the adaptive next-hop table from the live crowd state. Intervention penalties relax back towards neutral each refresh. An operator who intervenes repeatedly would otherwise leave a permanently distorted cost surface, and the routing would keep chasing assets that recovered long ago. """ self.costs.relax_penalties(self.settings.routing.penalty_decay) edge_cost = self.costs.dynamic_edge_cost( self.state.edge_velocity, self.state.phys_occupancy, self.state.edge_risk ) node_cost = self.costs.dynamic_node_cost(self.state.node_queue) self.tables.refresh_adaptive(edge_cost, node_cost, apply_hysteresis=True) self.tables.last_refresh_t = self.time def run_for(self, seconds: float) -> None: steps = int(round(seconds / self.dt)) for _ in range(steps): self.step() def run_until_complete(self, max_seconds: float | None = None) -> None: limit = max_seconds if max_seconds is not None else self.scenario.duration_s while self.time < limit and not self.is_complete: self.step() @property def is_complete(self) -> bool: return bool(np.all(self.pop.status == STATUS_ARRIVED)) @property def remaining(self) -> int: return int(np.sum(self.pop.status != STATUS_ARRIVED)) # ------------------------------------------------------------------ # timeline # ------------------------------------------------------------------ def _fire_timeline_events(self, t: float) -> None: for i, ev in enumerate(self.scenario.timeline): if i in self.fired_events or not ev.automatic or ev.t_s > t: continue self.trigger_event(i) def trigger_event(self, index: int) -> dict[str, Any]: """Apply a scenario timeline event (scripted or operator-triggered).""" if index in self.fired_events: return {"applied": False, "reason": "already fired"} ev: TimelineEvent = self.scenario.timeline[index] self.fired_events.add(index) factor = self.overrides.event_factor_overrides.get(ev.target, ev.factor) if ev.type == "capacity" and ev.target: self._scale_capacity(ev.target, factor) record = { "t_s": round(self.time, 1), "scheduled_t_s": ev.t_s, "type": ev.type, "target": ev.target, "factor": factor, "authored_factor": ev.factor, "label": (ev.label if factor == ev.factor else f"{ev.target.replace('_', ' ')} throughput set to " f"{factor * 100:.0f}% of nominal"), "detail": ev.detail, "severity": ev.severity, "index": index, } self.event_log.append(record) return {"applied": True, "event": record} # ------------------------------------------------------------------ # interventions (used by the strategy engine) # ------------------------------------------------------------------ def divert_flow( self, fraction: float, target_edges: set[int], target_nodes: set[int], penalty: float = 6.0, ) -> int: """Move a fraction of the affected crowd onto the adaptive routing plan. "Affected" means an agent whose current shortest-path route actually traverses the congested asset. Sending an instruction to people who were never going that way would inflate the intervention's apparent reach without changing anything. Compliance is per agent: an instruction reaches everyone selected, but only agents whose personal compliance clears a random draw act on it. """ if fraction <= 0: return 0 for e in target_edges: self.costs.penalise_edge(int(e), penalty) pair = int(self.venue.pair_of[int(e)]) if pair >= 0: self.costs.penalise_edge(pair, penalty) for n in target_nodes: self.costs.penalise_node(int(n), penalty) matrix = self.tables.traversal_matrix(POLICY_SHORTEST, target_edges, target_nodes) pop = self.pop active = pop.status != STATUS_ARRIVED at_node = np.where(pop.status == STATUS_WAITING, pop.origin, self.venue.edge_dst[np.maximum(pop.edge, 0)]) affected = active & matrix[pop.dest_slot, at_node] & (pop.policy != POLICY_ADAPTIVE) candidates = np.flatnonzero(affected) if candidates.size == 0: self.refresh_routing() return 0 self.action_rng.shuffle(candidates) take = int(round(fraction * candidates.size)) chosen = candidates[:take] if chosen.size == 0: self.refresh_routing() return 0 complies = self.action_rng.random(chosen.size) < pop.compliance[chosen] accepted = chosen[complies] pop.policy[accepted] = np.int8(POLICY_ADAPTIVE) self.refresh_routing() return int(accepted.size) def stagger_release(self, origin_ids: list[str], fraction: float, delay_s: float) -> int: """Hold back a fraction of not-yet-departed spectators. This is the demand-side lever: it flattens the departure peak instead of moving people sideways through the network. """ if fraction <= 0 or delay_s <= 0: return 0 pop = self.pop if origin_ids: origins = {self.venue.node_index[o] for o in origin_ids if o in self.venue.node_index} in_scope = np.isin(pop.origin, list(origins)) else: in_scope = np.ones(self.n_agents, dtype=bool) eligible = np.flatnonzero((pop.status == STATUS_WAITING) & in_scope & (pop.release_t >= self.time - 1.0)) if eligible.size == 0: return 0 self.action_rng.shuffle(eligible) take = int(round(fraction * eligible.size)) chosen = eligible[:take] if chosen.size == 0: return 0 # Spread the held-back group across the delay window rather than # releasing them all at once when the hold ends. jitter = self.action_rng.random(chosen.size) * delay_s pop.release_t[chosen] = (pop.release_t[chosen] + np.float32(delay_s * 0.5) + jitter.astype(np.float32)) return int(chosen.size) def open_alternate(self, node_id: str, factor: float) -> bool: """Bring contingency capacity online at an exit or transport interface.""" if node_id not in self.venue.node_index: return False idx = self.venue.node_index[node_id] self.node_budget.multiplier[idx] *= factor # Make the newly opened asset attractive to the router. self.costs.penalise_node(idx, 1.0 / max(factor, 1e-6)) self.refresh_routing() return True def redistribute_destinations( self, from_dest: str, to_dest: str, fraction: float ) -> int: """Send a fraction of one destination's demand to another. Operationally this is "your coach has been moved to the south apron": a change of where people are going, not merely how they get there. """ if fraction <= 0: return 0 vi = self.venue.node_index if from_dest not in vi or to_dest not in vi: return 0 from_node, to_node = vi[from_dest], vi[to_dest] if to_node not in self.dest_indices: return 0 to_slot = self.dest_indices.index(to_node) pop = self.pop eligible = np.flatnonzero((pop.status != STATUS_ARRIVED) & (pop.dest_node == from_node)) if eligible.size == 0: return 0 self.action_rng.shuffle(eligible) take = int(round(fraction * eligible.size)) chosen = eligible[:take] if chosen.size == 0: return 0 complies = self.action_rng.random(chosen.size) < pop.compliance[chosen] accepted = chosen[complies] pop.dest_node[accepted] = np.int32(to_node) pop.dest_slot[accepted] = np.int32(to_slot) pop.policy[accepted] = np.int8(POLICY_ADAPTIVE) self.refresh_routing() return int(accepted.size) def record_intervention(self, applied: AppliedIntervention) -> None: self.applied_interventions.append(applied) # ------------------------------------------------------------------ # snapshot / restore # ------------------------------------------------------------------ def snapshot(self) -> dict[str, Any]: """Exact, restorable copy of the entire simulation state.""" return { "pop": self.pop.copy(), "time": self.time, "step_count": self.step_count, "total_arrived": self.total_arrived, "travel_time_sum": self.travel_time_sum, "total_rerouted": self.total_rerouted, "total_reroute_decisions": self.total_reroute_decisions, "critical_edge_seconds": self.critical_edge_seconds, "risk_integral": self.risk_integral, "blocked_agents": self.blocked_agents, "queued_count": self._queued_count.copy(), "fired_events": set(self.fired_events), "event_log": [dict(e) for e in self.event_log], "applied_interventions": list(self.applied_interventions), "node_budget": self.node_budget.state(), "edge_budget": self.edge_budget.state(), "costs": self.costs.state(), "tables": self.tables.state(), "crowd_state": self.state.state(), "rng": self.rng.bit_generator.state, "action_rng": self.action_rng.bit_generator.state, } def restore(self, snap: dict[str, Any]) -> None: self.pop = snap["pop"].copy() self.n_agents = self.pop.size self.time = snap["time"] self.step_count = snap["step_count"] self.total_arrived = snap["total_arrived"] self.travel_time_sum = snap["travel_time_sum"] self.total_rerouted = snap["total_rerouted"] self.total_reroute_decisions = snap["total_reroute_decisions"] self.critical_edge_seconds = snap["critical_edge_seconds"] self.risk_integral = snap["risk_integral"] self.blocked_agents = snap["blocked_agents"] self._queued_count = snap["queued_count"].copy() self.fired_events = set(snap["fired_events"]) self.event_log = [dict(e) for e in snap["event_log"]] self.applied_interventions = list(snap["applied_interventions"]) self.node_budget.restore(snap["node_budget"]) self.edge_budget.restore(snap["edge_budget"]) self.costs.restore(snap["costs"]) self.tables.restore(snap["tables"]) self.state.restore(snap["crowd_state"]) self.rng.bit_generator.state = snap["rng"] self.action_rng.bit_generator.state = snap["action_rng"] def branch(self) -> "Simulator": """A detached copy of this simulation, for counterfactual roll-out.""" clone = object.__new__(Simulator) clone.venue = self.venue clone.scenario = self.scenario clone.settings = self.settings clone.overrides = self.overrides clone.seed = self.seed clone.dt = self.dt clone.dest_indices = list(self.dest_indices) clone.dest_ids = list(self.dest_ids) clone.expected_edge_volume = self.expected_edge_volume clone.expected_node_volume = self.expected_node_volume clone.rng = np.random.default_rng(self.seed) clone.action_rng = np.random.default_rng(self.seed) clone.costs = CostModel(self.venue, self.settings.routing, self.settings.movement.free_speed_mps) clone.tables = RoutingTables(self.venue, clone.costs, self.dest_indices, self.settings.routing) clone.node_budget = CapacityBudget(self.venue.node_service_ppm) clone.edge_budget = CapacityBudget(self.venue.edge_capacity_ppm) clone.state = CrowdStateEngine( self.venue, self.settings.risk, self.settings.movement, self.settings.prediction.history_window, self.settings.prediction.growth_window_s, self.dt, ) clone.pop = self.pop.copy() clone.n_agents = clone.pop.size clone.restore(self.snapshot()) return clone # ------------------------------------------------------------------ # metrics # ------------------------------------------------------------------ def metrics(self) -> dict[str, float]: """Cumulative run metrics. All measured, none assumed.""" pop = self.pop arrived = pop.status == STATUS_ARRIVED travel = np.where(arrived & ~np.isnan(pop.enter_t) & ~np.isnan(pop.arrive_t), pop.arrive_t - pop.enter_t, np.nan) finite = travel[~np.isnan(travel)] return { "sim_time_s": round(self.time, 2), "agents_total": int(self.n_agents), "agents_waiting": int(np.sum(pop.status == STATUS_WAITING)), "agents_moving": int(np.sum(pop.status == STATUS_ON_EDGE)), "agents_arrived": int(arrived.sum()), "throughput": int(arrived.sum()), "avg_travel_time_s": round(float(np.mean(finite)), 2) if finite.size else 0.0, "p95_travel_time_s": round(float(np.percentile(finite, 95)), 2) if finite.size else 0.0, "peak_density": round(float(np.max(self.state.peak_edge_density)), 3), "current_peak_density": round(float(np.max(self.state.edge_density)), 3), "critical_edge_seconds": round(self.critical_edge_seconds, 1), "max_queue": int(np.max(self.state.peak_node_queue)) if self.venue.n_nodes else 0, "current_max_queue": int(np.max(self.state.node_queue)) if self.venue.n_nodes else 0, "aggregate_risk": round(self.risk_integral, 1), "rerouted_agents": int(self.total_rerouted), "reroute_decisions": int(self.total_reroute_decisions), "blocked_agents": int(self.blocked_agents), "completion_pct": round(100.0 * float(arrived.sum()) / max(self.n_agents, 1), 1), } def dispersal_time(self, quantile: float = 0.95) -> float | None: """Sim time by which `quantile` of the crowd had reached a destination.""" arrive = self.pop.arrive_t[~np.isnan(self.pop.arrive_t)] if arrive.size < max(1, int(quantile * self.n_agents)): return None return float(np.percentile(arrive, quantile * 100.0)) # ------------------------------------------------------------------ # rendering support # ------------------------------------------------------------------ def agent_sample(self, budget: int) -> dict[str, list]: """A deterministic thinned sample of moving agents, for the map. Rendering every one of 40,000 agents is a browser problem, not a simulation problem. The simulation always runs the full population; the map draws an evenly spaced subset and reports the sampling ratio so the UI can be honest about what is on screen. """ pop = self.pop idx = np.flatnonzero(pop.status == STATUS_ON_EDGE) total = idx.size if total == 0: return {"x": [], "y": [], "v": [], "sampled": 0, "total": 0, "ratio": 1.0} if total > budget: stride = int(np.ceil(total / budget)) idx = idx[::stride] e = pop.edge[idx] frac = np.clip(pop.pos_m[idx] / np.maximum(self.venue.edge_length[e], 1e-6), 0.0, 1.0) # Queued agents are all held at pos == length internally. On the map # they are spread across the physical extent the queue actually # occupies, so a growing queue is visible as it backs up the corridor. qlen = getattr(self, "_queue_len_m", None) if qlen is not None: q = pop.blocked[idx] if np.any(q): spread = ((idx[q] * 40503) % 997) / 997.0 length = np.maximum(self.venue.edge_length[e[q]], 1e-6) frac[q] = np.clip(1.0 - spread * (qlen[e[q]] / length), 0.0, 1.0) xs = np.empty(idx.size, dtype=np.float64) ys = np.empty(idx.size, dtype=np.float64) for edge_id in np.unique(e): m = e == edge_id x, y = self.venue.positions_on_edge(int(edge_id), frac[m]) # Lateral spread across the corridor width, deterministic per agent. half = self.venue.edge_width[int(edge_id)] * 0.42 dx, dy = self.venue.edge_direction(int(edge_id)) offs = (((idx[m] * 2654435761) % 1000) / 1000.0 - 0.5) * 2.0 * half xs[m] = x - dy * offs ys[m] = y + dx * offs speed = pop.speed_now[idx] / max(self.settings.movement.free_speed_mps, 1e-6) return { "x": [round(float(a), 1) for a in xs], "y": [round(float(a), 1) for a in ys], "v": [round(float(a), 2) for a in np.clip(speed, 0.0, 1.0)], "sampled": int(idx.size), "total": int(total), "ratio": round(float(total) / max(idx.size, 1), 2), }