jupedsim_scenarios.runner

jupedsim_scenarios.runner#

High-level helpers for loading and running JuPedSim web-UI scenario JSON files.

A thin scenario layer on top of the simulation primitives in utils.simulation_init and shared.direct_steering_runtime. Used by the trajectory regression test and the scripts/run_scenario.py CLI; the web runtime itself goes through services.simulation_service.

This module replaced the previous backend/core/scenario.py mirror — see the chore/drop-core-mirror PR. The longer-term plan is to migrate to jupedsim.internal.scenarios (jupedsim PR #1565) once it lands upstream.

Usage:

from scenarios import load_scenario, run_scenario

scenario = load_scenario("scenario.zip")
print(scenario.summary())

result = run_scenario(scenario)
print(result.metrics)

df = result.trajectory_dataframe()

Classes#

`Scenario`	A loaded scenario ready for inspection and execution.
`ScenarioResult`	Results from running a scenario.
`ScenarioRunner`	Drive a scenario tick-by-tick with inspection between steps.

Functions#

`load_scenario`(→ Scenario)	Load a scenario from a ZIP archive, a directory, or a self-contained JSON file.
`run_scenario`(→ ScenarioResult)	Run a scenario to completion. Thin wrapper around `ScenarioRunner`.
`save_scenario`(→ None)	Write `scenario` to `path` as self-contained JSON.

Module Contents#

class Scenario[source]#

A loaded scenario ready for inspection and execution.

add_distribution(coordinates, *, key: str | None = None, number: int = 10, **agent_params) → str[source]#

Add a spawn distribution. Returns its id.

coordinates accepts a shapely Polygon or any iterable of (x, y) pairs. The polygon is automatically closed. key picks a specific id; None auto-generates one as jps-distributions_{n}. Extra keyword arguments are validated through the same allow-list as set_agent_params so typos surface immediately.

add_exit(coordinates, *, key: str | None = None, max_throughput: float = 0.0) → str[source]#: Add an exit polygon. Returns its id.

add_stage(coordinates, *, key: str | None = None, waiting_time: float = 0.0) → str[source]#

Add a waypoint / checkpoint stage. Returns its id.

The web-UI JSON calls these “checkpoints”; the Python API uses “stage” to match jupedsim’s runtime vocabulary. raw keeps the JSON name, so existing web exports load unchanged.

add_zone(coordinates, *, key: str | None = None, speed_factor: float = 1.0) → str[source]#: Add a speed-modifier zone. Returns its id.

copy() → Scenario[source]#

Return an independent deep copy of this scenario.

The clone shares no mutable state with the original — mutating the clone (adding distributions, changing the seed, etc.) does not affect the source. Used internally by run_sweep to isolate per-trial scenarios.

For a copy with one or two fields changed, copy first then assign:

clone = base.copy()
clone.seed = 99
clone.max_simulation_time = 60

list_distributions() → list[dict][source]#: Return a list of {"index", "id", "agents", "flow"} dicts.

list_stages() → list[dict][source]#: Return a list of {"index", "id", "waiting_time"} dicts.

list_zones() → list[dict][source]#: Return a list of {"index", "id", "speed_factor"} dicts.

plot(ax=None, *, show_journeys: bool = True, trajectories=None, show_trajectories: bool | None = None)[source]#

Plot the scenario geometry with labeled distributions, exits, zones, and checkpoints.

When show_journeys is true (default) and the scenario defines journeys, curved arrows connect the elements of each journey in route order.

Pass trajectories — a ScenarioResult from a completed run, or a pedpy.TrajectoryData — to overlay the agent paths on the geometry via pedpy.plot_trajectories(). A single call then shows the plan (exits, journeys) and what the agents actually did:

result = run_scenario(scenario, seed=42)
scenario.plot(trajectories=result)

show_trajectories forces the overlay on/off; left at None it is drawn whenever trajectories is given.

Returns the matplotlib Axes so callers can further customise the figure.

remove_distribution(key: int | str) → None[source]#

remove_exit(key: int | str) → None[source]#

remove_stage(key: int | str) → None[source]#

remove_zone(key: int | str) → None[source]#

set_agent_count(distribution_id: int | str, count: int)[source]#

Switch a distribution to by-number spawning with count agents.

Sets number=count AND unconditionally forces distribution_mode="by_number" — so a distribution previously configured as by_density will be flipped. If you want to update the count without changing the spawning mode, call set_agent_params(id, number=count) directly.

set_agent_params(distribution_id: int | str, **kwargs)[source]#

Set agent parameters for a distribution.

Supported keys: radius, desired_speed, radius_distribution, radius_std, desired_speed_distribution, desired_speed_std, use_flow_spawning, flow_start_time, flow_end_time, distribution_mode, number.

v0, v0_std, and v0_distribution are accepted as deprecated aliases for the desired_speed* keys and emit a DeprecationWarning. They will be removed in a future release.

set_checkpoint_waiting_time(checkpoint_id: int | str, waiting_time: float)[source]#: Set the waiting time for a checkpoint/stage.

set_flow_schedule(distribution_id: int | str, schedule: list[dict], *, keep_initial_agents: bool = False)[source]#: Attach a time-windowed inflow schedule to one source distribution.

set_model_params(**kwargs)[source]#: Set model-specific parameters (e.g. strength_neighbor_repulsion, range_neighbor_repulsion).

set_zone_speed_factor(zone_id: int | str, factor: float)[source]#: Set the speed factor for a zone.

summary() → str[source]#

to_json() → str[source]#

Serialize the scenario as self-contained JSON.

Returns the JSON string. To write to disk, use save_scenario() (mirrors load_scenario()).

The output embeds walkable_area_wkt at the top level so load_scenario() can read it back via the self-contained-JSON branch — completing the build → run → persist loop introduced by R2.1’s add_* methods.

Round-trip is value-preserving for everything the public API touches (geometry, seed, model_type, sim_params, exits, distributions, stages, zones, journeys) provided every value is JSON-serializable. The default json.dumps() encoder is used with no fallback, so an unsupported type stuffed into scenario.raw (or sim_params) raises TypeError at save time rather than silently stringifying — surfacing the mistake before the round-trip lies about what was preserved.

property distributions: types.MappingProxyType[str, Any][source]#

property exits: types.MappingProxyType[str, Any][source]#

property journeys: list[dict[str, Any]][source]#

property max_simulation_time: float[source]#

model_type: str[source]#

raw: dict[str, Any][source]#

seed: int[source]#

sim_params: dict[str, Any][source]#

source_path: str | None = None[source]#

property stages: types.MappingProxyType[str, Any][source]#

walkable_area_wkt: str[source]#

property walkable_polygon[source]#

property zones: types.MappingProxyType[str, Any][source]#

class ScenarioResult[source]#

Results from running a scenario.

as_pedpy_trajectory()[source]#

Return the trajectory as a pedpy.TrajectoryData.

Thin adapter so scientists doing pedpy analysis don’t have to rebuild the dataframe and look up the frame rate themselves:

result = run_scenario(scenario, seed=42)
traj = result.as_pedpy_trajectory()
pedpy.compute_classic_density(traj=traj, ...)

pedpy is already a hard dependency of this package (used internally for WalkableArea etc.) so the import is direct.

cleanup() → int[source]#

Delete the temporary SQLite trajectory file.

Returns the number of files removed (0 or 1) so callers and SweepResult.cleanup can report totals without re-checking.

trajectory_dataframe()[source]#

Load trajectory data into a pandas DataFrame.

Columns: frame, id, x, y, ori_x, ori_y

visualise(*, every_nth_frame: int | None = None, width: int = 800, height: int = 800, radius: float = 0.2, title_note: str = '', save_path: str | pathlib.Path | None = None)[source]#

Interactive plotly playback of the trajectory.

Thin wrapper around jupedsim’s own notebook animation (jupedsim.internal.notebook_utils.animate()): agents are drawn as circles coloured by speed, with orientation arrows, a play button and a time slider. Returns a plotly.graph_objects.Figure that renders inline in Jupyter with no extra steps — just result.visualise() as the last line of a cell.

Trajectory, frame rate and geometry are read straight from the run’s SQLite file (which carries the orientation columns this animation needs). radius is the drawn agent radius in metres.

every_nth_frame subsamples frames so long runs stay light. Left at None (the default) it is chosen automatically so the animation lands near _TARGET_ANIMATION_FRAMES frames: short runs play every frame, huge runs are thinned to stay responsive. Pass an explicit positive integer to override.

If save_path is given the figure is written to disk: .html produces a self-contained, interactive page; any other suffix is passed to plotly’s static image export.

property agents_evacuated: int[source]#

property agents_remaining: int[source]#

property dt: float[source]#: Simulation timestep in seconds, as reported by jupedsim.

property evacuation_time: float[source]#

property frame_rate: float[source]#: Trajectory frame rate (Hz), computed from the writer stride and dt at simulation time. KeyError if the metrics dict doesn’t have it — that’s a runner bug, not something to paper over with a default.

metrics: dict[str, Any][source]#

property seed: int[source]#: Random seed used for this run.

sqlite_file: str | None = None[source]#

property success: bool[source]#

property total_agents: int[source]#

property walkable_polygon[source]#: Walkable area as a Shapely Polygon (for pedpy analysis).

class ScenarioRunner(scenario: Scenario, *, seed: int | None = None, dt: float | None = None, every_nth_frame: int = 10, output_path: str | pathlib.Path | None = None)[source]#

Drive a scenario tick-by-tick with inspection between steps.

Matches the imperative shape of jupedsim.Simulation so users familiar with the lower-level API feel at home:

with ScenarioRunner(scenario, seed=42) as runner:
    runner.run_until(10.0)
    print(runner.elapsed_time, runner.agent_count)
    # mutate the simulation or scenario here, then continue:
    runner.run_until(20.0)
    result = runner.result()

To run all the way to scenario.max_simulation_time, call run_until() with no argument (the loop also stops early once every agent has evacuated):

with ScenarioRunner(scenario, seed=42) as runner:
    runner.run_until()
    result = runner.result()

Outside a context manager you can call close() directly; the sqlite trajectory file survives close() so the returned ScenarioResult keeps working.

agents()[source]#: Iterate over live agents (delegates to jupedsim.Simulation.agents).

close() → None[source]#

Close the trajectory writer and clean up the config tempfile.

Idempotent. Does NOT remove the sqlite trajectory file — call ScenarioResult.cleanup() for that, or just keep the file.

result() → ScenarioResult[source]#

Build a ScenarioResult reflecting the runner’s current state.

Safe to call multiple times — each invocation snapshots the live metrics. The runner does NOT close the writer when result() is called, so subsequent step() / run_until calls keep appending to the same sqlite file.

run_until(target_time: float | None = None) → None[source]#

Run forward until elapsed_time >= target_time.

target_time=None (default) means scenario.max_simulation_time — i.e. “run to completion”. Stops early once every agent has evacuated (and any flow spawning has emptied its budget).

target_time is clamped to scenario.max_simulation_time so callers can’t drive the simulation past the scenario’s configured horizon by accident. target_time=0 is accepted and is a no-op at t=0.

step() → None[source]#: Advance one iteration tick (all per-tick helpers + simulation.iterate).

property agent_count: int[source]#

property elapsed_time: float[source]#

property seed: int[source]#

property simulation[source]#: The underlying jupedsim.Simulation. Use this when the higher-level API doesn’t cover what you need; mutate at your own risk.

load_scenario(path: str) → Scenario[source]#

Load a scenario from a ZIP archive, a directory, or a self-contained JSON file.

Three input shapes are supported:

Directory — contains one *.json and one *.wkt file.
ZIP archive — same two files packed together.
Self-contained JSON — a single .json file whose top-level object embeds the walkable geometry as "walkable_area_wkt". This is what the CLI consumes.

run_scenario(scenario: Scenario, *, seed: int | None = None, dt: float | None = None, every_nth_frame: int = 10, output_path: str | pathlib.Path | None = None) → ScenarioResult[source]#: Run a scenario to completion. Thin wrapper around ScenarioRunner.

save_scenario(scenario: Scenario, path: str | pathlib.Path) → None[source]#

Write scenario to path as self-contained JSON.

Side-effecting counterpart to Scenario.to_json() (which returns the JSON as a string), mirroring load_scenario(). Parent directories are created if missing.