jupedsim_scenarios

jupedsim_scenarios#

Scenario layer — high-level Scenario / run_scenario API.

Wraps the simulation primitives in utils.simulation_init and shared.direct_steering_runtime into a load-and-run API used by the trajectory regression test and the scripts/run_scenario.py CLI. The web runtime itself goes through services.simulation_service.

This package replaced the former backend/core/ mirror. Once jupedsim.internal.scenarios lands upstream (jupedsim PR #1565), this module’s contents will migrate to thin wrappers around the upstream API.

Submodules#

Classes#

`Scenario`	A loaded scenario ready for inspection and execution.
`ScenarioResult`	Results from running a scenario.
`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 ZIP or directory exported from the JuPedSim web UI.
`run_scenario`(→ ScenarioResult)	Run a scenario with the same shared setup/runtime semantics as the web app.
`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

Package Contents#

class Scenario[source]#

A loaded scenario ready for inspection and execution.

copy(**overrides) → Scenario[source]#: Return an independent deep copy of this scenario, with optional field overrides.

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)[source]#

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

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

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

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

Set agent parameters for a distribution.

Supported keys: radius, desired_speed (or v0), radius_distribution, radius_std, desired_speed_distribution (or v0_distribution), desired_speed_std (or v0_std), use_flow_spawning, flow_start_time, flow_end_time, distribution_mode, number.

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_max_time(seconds: float)[source]#

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

set_model_type(model_type: str)[source]#

set_seed(seed: int)[source]#

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

summary() → str[source]#

property distributions: dict[str, Any]#

property exits: dict[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: dict[str, Any]#

walkable_area_wkt: str#

property walkable_polygon#

property zones: dict[str, Any]#

class ScenarioResult[source]#

Results from running a scenario.

cleanup()[source]#: Delete the temporary SQLite trajectory file.

trajectory_dataframe()[source]#

Load trajectory data into a pandas DataFrame.

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

property agents_evacuated: int#

property agents_remaining: int#

property dt: float#: Simulation timestep in seconds.

property evacuation_time: float#

property frame_rate: float#

Trajectory frame rate in frames per second.

Sourced from metrics (computed from the actual dt and writer stride used at simulation time). Falls back to 10.0 only when metrics are absent — that fallback is a safety value, not the true rate.

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 SweepResult[source]#

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() → None[source]#: Remove every trial’s sqlite trajectory file.

save(path: str | pathlib.Path) → None[source]#

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. load() reattaches metadata to the sqlite files; if the sqlites are gone, the loaded result is metadata-only.

to_dataframe()[source]#

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[source]#

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#

property success: bool#

load_scenario(path: str) → Scenario[source]#: Load a scenario ZIP or directory exported from the JuPedSim web UI.

run_scenario(scenario: Scenario, *, seed: int | None = None) → ScenarioResult[source]#: Run a scenario with the same shared setup/runtime semantics as the web app.

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[source]#

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[source]#

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