Spaces:
Runtime error
Runtime error
File size: 5,009 Bytes
e7a9f02 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | #!/usr/bin/env python3
"""Record a scenario to a replayable file.
Demo insurance. The primary demonstration is always the live simulation; this
exists so that if a live run cannot be created on the day — a broken
dependency, a machine that cannot carry the agent count — the dashboard can
still show the complete result rather than an error.
A recording is a sequence of the same frames the WebSocket would have sent,
plus the strategy comparison captured at the decision point, so a replay is
visually and numerically identical to the live run it was made from. It is a
recording of a real run, never a hand-written script.
Run: python scripts/record_fallback.py [--scenario ...] [--interval 10]
"""
from __future__ import annotations
import argparse
import asyncio
import datetime as dt
import json
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "backend"))
from flowtwin.config import FALLBACK_DIR, SETTINGS # noqa: E402
from flowtwin.runtime.session import SessionConfig, SimulationSession # noqa: E402
from flowtwin.venue import load_scenario # noqa: E402
async def record(scenario_id: str, interval_s: float, decision_t_s: float | None,
seed: int | None) -> Path:
scenario = load_scenario(scenario_id)
cfg = SessionConfig(
venue_id=scenario.venue_id,
scenario_id=scenario_id,
seed=seed if seed is not None else scenario.default_seed,
speed=10,
)
session = SimulationSession(cfg, SETTINGS)
steps_per_frame = max(1, int(round(interval_s / session.sim.dt)))
frames: list[dict] = []
strategy_run: dict | None = None
applied_branches: dict[str, int] = {}
agents_affected = 0
# Pick the decision point automatically: the first moment the primary
# bottleneck is projected to go critical.
auto_decision = decision_t_s is None
target = decision_t_s
while session.sim.time < scenario.duration_s and not session.sim.is_complete:
frame = session.frame()
frames.append(frame)
if strategy_run is None:
fire = False
if auto_decision:
alerts = frame.get("alerts") or []
fire = any(a["severity"] == "critical" and a["time_to_critical_s"] is not None
for a in alerts)
elif target is not None and session.sim.time >= target:
fire = True
if fire:
print(f" decision point at T+{session.sim.time:.0f}s — evaluating strategies")
strategy_run = await session.evaluate_strategies(horizon_s=300)
if strategy_run.get("available"):
rec = strategy_run["recommendation"]["strategy_id"]
applied_branches[rec] = len(frames)
result = await session.apply_strategy(rec)
agents_affected = result.get("agents_affected", 0)
print(f" applied {rec} · {agents_affected:,} people rerouted")
session._advance(steps_per_frame)
if session.finished:
break
frames.append(session.frame())
FALLBACK_DIR.mkdir(parents=True, exist_ok=True)
path = FALLBACK_DIR / f"{scenario_id}.json"
payload = {
"meta": {
"venue_id": scenario.venue_id,
"scenario_id": scenario_id,
"seed": session.sim.seed,
"crowd_size": session.sim.n_agents,
"speed": 10,
"interval_s": interval_s,
"frames": len(frames),
"applied_branches": applied_branches,
"agents_affected": agents_affected,
"recorded_utc": dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"),
"note": ("Recording of a real seeded run of this scenario. Used only "
"if a live simulation cannot be created."),
},
"frames": frames,
"strategy_run": strategy_run,
}
path.write_text(json.dumps(payload), encoding="utf-8")
await session.close()
return path
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--scenario", action="append", default=None)
ap.add_argument("--interval", type=float, default=10.0,
help="simulated seconds between recorded frames")
ap.add_argument("--decision", type=float, default=None,
help="force the strategy decision at this sim time")
ap.add_argument("--seed", type=int, default=None)
args = ap.parse_args()
scenarios = args.scenario or ["circuit_alpha_post_race", "barcelona_2022_egress"]
for scenario_id in scenarios:
print(f"recording {scenario_id} …")
path = asyncio.run(record(scenario_id, args.interval, args.decision, args.seed))
size_mb = path.stat().st_size / 1e6
print(f" saved {path} ({size_mb:.1f} MB)")
if __name__ == "__main__":
main()
|