scripts/record_fallback.py

#!/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()