jupedsim_scenarios

High-level API for loading, building, running, and persisting JuPedSim scenarios.

The package wraps the lower-level jupedsim.Simulation primitives into a load → mutate → run → analyse flow that matches what scientists building and sweeping scenarios actually want.

The jps-scenarios CLI (entry point cli.main()) exposes the same surface for scripted pipelines.

Submodules

• jupedsim_scenarios.cli
• jupedsim_scenarios.direct_steering_runtime
• jupedsim_scenarios.runner
• jupedsim_scenarios.simulation_init
• jupedsim_scenarios.sweep

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.
SweepResult	Collection of trials produced by run_sweep.
Trial	One realised cell of the sweep: axis values + seed + result.

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.
run_sweep(, output_dir, workers, progress, int, dict], ...)	Run the scenario once per (axis combination, seed) pair.
run_sweep_from_factory(, output_dir, workers, ...)	Run one simulation per (trial-params, seed) pair, building each
save_scenario(→ None)	Write scenario to path as self-contained JSON.

Package Contents

class Scenario

A loaded scenario ready for inspection and execution.

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

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

Add an exit polygon. Returns its id.

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

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

Add a speed-modifier zone. Returns its id.

copy() → Scenario

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]

Return a list of {"index", "id", "agents", "flow"} dicts.

list_stages() → list[dict]

Return a list of {"index", "id", "waiting_time"} dicts.

list_zones() → list[dict]

Return a list of {"index", "id", "speed_factor"} dicts.

plot(ax=None, *, show_journeys: bool = True, trajectories=None, show_trajectories: bool | None = None)

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

remove_exit(key: int | str) → None

remove_stage(key: int | str) → None

remove_zone(key: int | str) → None

set_agent_count(distribution_id: int | str, count: int)

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)

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)

Set the waiting time for a checkpoint/stage.

set_flow_schedule(distribution_id: int | str, schedule: list[dict], *, keep_initial_agents: bool = False)

Attach a time-windowed inflow schedule to one source distribution.

set_model_params(**kwargs)

Set model-specific parameters (e.g. strength_neighbor_repulsion, range_neighbor_repulsion).

set_zone_speed_factor(zone_id: int | str, factor: float)

Set the speed factor for a zone.

summary() → str

to_json() → str

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]

property exits: types.MappingProxyType[str, Any]

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

property max_simulation_time: float

model_type: str

raw: dict[str, Any]

seed: int

sim_params: dict[str, Any]

source_path: str | None = None

property stages: types.MappingProxyType[str, Any]

walkable_area_wkt: str

property walkable_polygon

property zones: types.MappingProxyType[str, Any]

class ScenarioResult

Results from running a scenario.

as_pedpy_trajectory()

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

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()

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)

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

property agents_remaining: int

property dt: float

Simulation timestep in seconds, as reported by jupedsim.

property evacuation_time: float

property frame_rate: float

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]

property seed: int

Random seed used for this run.

sqlite_file: str | None = None

property success: bool

property total_agents: int

property walkable_polygon

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)

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()

Iterate over live agents (delegates to jupedsim.Simulation.agents).

close() → None

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

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

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

Advance one iteration tick (all per-tick helpers + simulation.iterate).

property agent_count: int

property elapsed_time: float

property seed: int

property simulation

The underlying jupedsim.Simulation. Use this when the higher-level API doesn’t cover what you need; mutate at your own risk.

class SweepResult

Collection of trials produced by run_sweep.

Holds the per-trial ScenarioResult objects (each pointing at its own on-disk sqlite). Call .cleanup() when done to delete the sqlites, or .save(path) first if you want to keep the metadata.

cleanup() → int

Remove every trial’s sqlite trajectory file.

Returns the number of files actually removed (trials whose sqlite was already deleted or moved don’t count).

classmethod load(path: str | pathlib.Path) → SweepResult

Rebuild a SweepResult from a JSON written by save().

Trial sqlite paths are kept verbatim — if the files moved or were cleaned up, trial.result.sqlite_file still points at the old location and trajectory_dataframe will raise FileNotFoundError. The metrics-derived properties on ScenarioResult (success, evacuation_time, …) work either way because they read from the metrics dict.

save(path: str | pathlib.Path) → None

Persist sweep metadata (axes, seeds, per-trial paths + metrics) as JSON.

The trajectory sqlites themselves are NOT moved — they stay where run_sweep’s output_dir put them. Pair with SweepResult.load() to reattach the metadata to the on-disk sqlites later; if the sqlites are gone, the loaded result is still useful for the metrics dataframe.

to_dataframe()

Return a pandas DataFrame with one row per trial.

Columns: every axis name, seed, success, evacuation_time, total_agents, agents_evacuated, agents_remaining, sqlite_path.

axes: dict[str, list[Any]]

seeds: list[int | None] = []

trials: list[Trial]

class Trial

One realised cell of the sweep: axis values + seed + result.

extras is an opaque per-trial payload. run_sweep always leaves it None; run_sweep_from_factory lets the factory attach anything it likes (geometry, label, computed metadata) so downstream code can pick it up via for t in sweep.trials: t.extras.

axis_values: dict[str, Any]

extras: Any = None

index: int

result: jupedsim_scenarios.runner.ScenarioResult

seed: int

load_scenario(path: str) → Scenario

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

Run a scenario to completion. Thin wrapper around ScenarioRunner.

run_sweep(scenario: jupedsim_scenarios.runner.Scenario, *, axes: collections.abc.Mapping[str, collections.abc.Sequence[Any]] | None = None, apply: collections.abc.Mapping[str, AxisApplyFn] | None = None, seeds: collections.abc.Iterable[int | None] = (None,), output_dir: str | pathlib.Path | None = None, workers: int = 1, progress: collections.abc.Callable[[int, int, dict], None] | None = None) → SweepResult

Run the scenario once per (axis combination, seed) pair.

Parameters:

• scenario – The base scenario. .copy() is taken per trial; the caller’s scenario is never mutated.

• axes – Mapping of axis name → list of values. Trials cover the full cartesian product. Empty / None ⇒ no parameter sweep (seed-only).

• apply – Mapping of axis name → callable (Scenario, value) -> None. Mutates the trial’s scenario copy in place. Required for each axis in axes.

• seeds – Seeds to replicate every axis combination over. Default (None,) ⇒ one trial per combination with whatever seed the scenario carries.

• output_dir – If given, every trial’s sqlite trajectory is placed inside it with a deterministic name (trial_<index>.sqlite). If omitted, each trial gets its own tempfile (cleaned by SweepResult.cleanup).

• workers – Number of parallel worker processes. 1 runs sequentially in the calling process; >1 dispatches trials via joblib.Parallel (loky backend); 0 selects os.cpu_count(). Trial-level mutations are applied in the parent process, so user apply callables don’t need any special pickling treatment.

• progress – Optional callback invoked after each trial with (trial_index, total_trials, axis_values_with_seed).

Return type:

SweepResult

run_sweep_from_factory(factory: ScenarioFactoryFn, *, trials: collections.abc.Iterable[collections.abc.Mapping[str, Any]], seeds: collections.abc.Iterable[int | None] = (None,), output_dir: str | pathlib.Path | None = None, workers: int = 1, progress: collections.abc.Callable[[int, int, dict], None] | None = None) → SweepResult

Run one simulation per (trial-params, seed) pair, building each scenario fresh via a user-supplied factory.

Use this when the scenario can’t be expressed as a single base mutated by axis values — typically because the geometry itself depends on trial parameters (e.g. a loop track whose radius scales with agent count). Each call to factory(trial_params) is expected to construct a fresh Scenario.

Parameters:

• factory – Callable (trial_params) -> Scenario or (trial_params) -> (Scenario, extras). Called once per trial-parameters dict in the parent process; the resulting Scenario is then pickled to a worker for the actual simulation. extras (if returned) is attached to Trial.extras for the caller to read after the sweep completes.

• trials – Iterable of trial-parameters mappings. The mapping’s keys become the DataFrame columns when you call SweepResult.to_dataframe(), so name them meaningfully.

• seeds – Seeds to replicate every trial-params combination over. Default (None,) ⇒ one run per trial-params dict using the seed embedded in the factory’s Scenario.

• output_dir – Same semantics as run_sweep.

• workers – Same semantics as run_sweep.

• progress – Same semantics as run_sweep.

Return type:

SweepResult

save_scenario(scenario: Scenario, path: str | pathlib.Path) → None

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.