RiMEA 13: Fundamental Diagram on Stairs¶

This notebook approximates RiMEA Test 13 with a zone-based stair model. The stair segment is represented by a slowdown zone inside the corridor. Agents keep a shared flat-ground demographic around $1.2\,\mathrm{m/s}$, while the stair behavior is introduced through zone speed factors 0.5 for upwards and 0.7 for downwards.

The outputs focus on three questions:

  • do the upstairs and downstairs speed-density points stay within the coded band?
  • is the downstairs curve faster than the upstairs curve at the same density?
  • what do the trajectories look like in the stair geometry?

Deviation from RiMEA¶

RiMEA asks for stair speeds within the spread of Abb. 16. The test only requires at least 25% of the points inside the coded band, so the inside-band fractions below are published rather than gated. The band (the suite's own, source pending; compare RiMEA Abb. 16) and the RiMEA standard population (section 3.2.4) are not yet checked against the standard.

In [1]:
from datetime import datetime
print(f"Executed on {datetime.now().astimezone().strftime('%d %B %Y, %H:%M %Z')}")

Executed on 16 October 2026, 12:31 UTC
In [2]:
import json
import tempfile
from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import pedpy
from shapely.geometry import Point, Polygon

from scenario_builders.rimea13_stairs import (
    EXIT_COORDINATES,
    FRUIN_STAIR_SPEEDS,
    RIEMA_STANDARD_DEMOGRAPHIC,
    STAIR_WALKABLE_AREA_WKT,
    STAIR_ZONE_COORDINATES,
    STAIR_ZONE_SPEED_FACTORS,
    build_raw_scenario,
    corbetta_envelope_bounds,
)
from jupedsim_scenarios import load_scenario, run_sweep_from_factory
In [3]:
plt.rcParams.update(
    {
        "figure.facecolor": "white",
        "axes.facecolor": "#f8fafc",
        "axes.edgecolor": "#334155",
        "axes.labelcolor": "#0f172a",
        "axes.titleweight": "bold",
        "xtick.color": "#0f172a",
        "ytick.color": "#0f172a",
        "font.size": 11,
    }
)

STAIR_ZONE = Polygon(STAIR_ZONE_COORDINATES)
COLORS = {"up": "#2563EB", "down": "#DC2626"}

Build and Run Both Scenarios¶

In [4]:
def _build_direction(params):
    """Factory that constructs a Scenario for the given direction.
    The scenario reference is paired into extras so downstream
    post-processing can read it back off the Trial."""
    direction = params["direction"]
    raw = build_raw_scenario(direction=direction)
    tmpdir = tempfile.TemporaryDirectory()
    scenario_dir = Path(tmpdir.name)
    (scenario_dir / "config.json").write_text(json.dumps(raw, indent=2), encoding="utf-8")
    (scenario_dir / "geometry.wkt").write_text(STAIR_WALKABLE_AREA_WKT, encoding="utf-8")

    scenario = load_scenario(str(scenario_dir))
    return scenario, scenario


def _process_direction(trial):
    """Compute the per-direction pedpy artefacts from a Trial.
    Replaces the post-processing tail that used to live inside
    run_direction."""
    result = trial.result
    traj_df = result.trajectory_dataframe()[["id", "frame", "x", "y"]].copy()
    traj = pedpy.TrajectoryData(traj_df, frame_rate=result.frame_rate)
    speed_df = pedpy.compute_individual_speed(
        traj_data=traj,
        frame_step=10,
        speed_calculation=pedpy.SpeedCalculation.BORDER_EXCLUDE,
    )
    merged = traj.data.merge(speed_df[["id", "frame", "speed"]], on=["id", "frame"], how="inner")
    stair_points = merged[
        merged.apply(lambda row: STAIR_ZONE.covers(Point(row["x"], row["y"])), axis=1)
    ].copy()
    frame_counts = stair_points.groupby("frame")["id"].nunique().rename("count")
    stair_points = stair_points.merge(frame_counts, on="frame", how="left")
    stair_points["density"] = stair_points["count"] / STAIR_ZONE.area
    stair_points = stair_points[(stair_points["density"] >= 0.6) & (stair_points["density"] <= 1.5)]
    direction = trial.axis_values["direction"]
    scenario = trial.extras
    return {
        "direction": direction,
        "scenario": scenario,
        "result": result,
        "traj": traj,
        "walkable_area": pedpy.WalkableArea(scenario.walkable_polygon),
        "stair_points": stair_points,
    }
In [5]:
%%capture
# Run both directions in one factory-driven sweep so they execute
# in parallel.
sweep = run_sweep_from_factory(
    _build_direction,
    trials=[{"direction": "up"}, {"direction": "down"}],
    seeds=[42],
    workers=2,
)
# Look up trials by direction so the assignment is independent of
# the order run_sweep_from_factory returns them in.
by_direction = {t.axis_values["direction"]: t for t in sweep.trials}
up = _process_direction(by_direction["up"])
down = _process_direction(by_direction["down"])
sweep.cleanup()
In [6]:
summary = pd.DataFrame(
    [
        {
            "direction": run["direction"],
            "zone_speed_factor": STAIR_ZONE_SPEED_FACTORS[run["direction"]],
            "evacuation_time [s]": run["result"].evacuation_time,
            "stair samples": len(run["stair_points"]),
            "mean stair speed [m/s]": run["stair_points"]["speed"].mean(),
        }
        for run in [up, down]
    ]
)
summary
Out[6]:
direction zone_speed_factor evacuation_time [s] stair samples mean stair speed [m/s]
0 up 0.5 52.62 918 0.553609
1 down 0.7 45.19 853 0.756988

Geometry, Stair Zone, and Trajectories¶

In [7]:
fig, axes = plt.subplots(1, 2, figsize=(15.5, 5.4), constrained_layout=True)

for ax, run in zip(axes, [up, down]):
    pedpy.plot_trajectories(
        traj=run["traj"],
        walkable_area=run["walkable_area"],
        axes=ax,
        traj_alpha=0.18,
        traj_width=0.35,
        traj_color=COLORS[run["direction"]],
    ).set_aspect("equal")
    ax.fill(
        [xy[0] for xy in STAIR_ZONE_COORDINATES],
        [xy[1] for xy in STAIR_ZONE_COORDINATES],
        color="#22C55E",
        alpha=0.22,
        zorder=2,
    )
    ax.fill(
        [xy[0] for xy in EXIT_COORDINATES],
        [xy[1] for xy in EXIT_COORDINATES],
        color="#15803D",
        alpha=0.28,
        zorder=2,
    )
    ax.text(14.5, 6.25, "stair zone", color="#166534", fontsize=10, ha="center")
    ax.text(18.9, 6.25, "goal", color="#166534", fontsize=10, ha="center")
    ax.set_title(f"{run['direction'].title()} scenario", pad=12)
    ax.set_xlim(-0.5, 19.5)
    ax.set_ylim(-0.5, 10.5)
    ax.set_xlabel("x [m]")
    ax.set_ylabel("y [m]")
    ax.grid(True, linestyle="--", linewidth=0.7, alpha=0.35)
    ax.spines["top"].set_visible(False)
    ax.spines["right"].set_visible(False)

plt.show()
No description has been provided for this image

Fundamental Diagram Against the Coded Band¶

In [8]:
fig, axes = plt.subplots(1, 2, figsize=(15.5, 5.4), constrained_layout=True)

for ax, run in zip(axes, [up, down]):
    rho = np.linspace(0.6, 1.5, 400)
    low, high = corbetta_envelope_bounds(run["direction"], rho)
    ax.fill_between(rho, low, high, color="#BFDBFE", alpha=0.7, label="Suite band (source pending)")
    ax.scatter(
        run["stair_points"]["density"],
        run["stair_points"]["speed"],
        s=16,
        alpha=0.22,
        color=COLORS[run["direction"]],
        label="simulation points",
    )
    binned = run["stair_points"].groupby("count", observed=True)["speed"].mean()
    ax.plot(binned.index.to_numpy(dtype=float) / STAIR_ZONE.area, binned.to_numpy(), color="#0F172A", linewidth=2.2, marker="o", label="binned mean")
    ax.set_title(f"{run['direction'].title()} stairs", pad=12)
    ax.set_xlabel("Density [1/m²]")
    ax.set_ylabel("Speed [m/s]")
    ax.set_xlim(0.55, 1.55)
    ax.set_ylim(0.15, 1.0)
    ax.grid(True, linestyle="--", linewidth=0.7, alpha=0.35)
    ax.spines["top"].set_visible(False)
    ax.spines["right"].set_visible(False)
    ax.legend(frameon=False)

plt.show()
No description has been provided for this image
In [9]:
def inside_band_fraction(run):
    points = run["stair_points"]
    low, high = corbetta_envelope_bounds(run["direction"], points["density"].to_numpy())
    speed = points["speed"].to_numpy()
    return float(((speed >= low) & (speed <= high)).mean())


pd.DataFrame(
    [
        {
            "direction": run["direction"],
            "stair samples": len(run["stair_points"]),
            "fraction inside coded band": round(inside_band_fraction(run), 3),
        }
        for run in [up, down]
    ]
)
Out[9]:
direction stair samples fraction inside coded band
0 up 918 0.998
1 down 853 0.549

Density-Matched Comparison¶

In [10]:
up_binned = up["stair_points"].groupby("count", observed=True)["speed"].mean()
down_binned = down["stair_points"].groupby("count", observed=True)["speed"].mean()
common_counts = up_binned.index.intersection(down_binned.index)
area = STAIR_ZONE.area
comparison = pd.DataFrame(
    {
        "count": common_counts.astype(int),
        "density": common_counts.astype(float) / area,
        "upstairs_speed": up_binned.loc[common_counts].to_numpy(),
        "downstairs_speed": down_binned.loc[common_counts].to_numpy(),
    }
).sort_values("count")
comparison["difference"] = comparison["downstairs_speed"] - comparison["upstairs_speed"]

fig, ax = plt.subplots(figsize=(9, 4.8))
ax.plot(comparison["density"], comparison["upstairs_speed"], color=COLORS["up"], marker="o", linewidth=2.0, label="upstairs")
ax.plot(comparison["density"], comparison["downstairs_speed"], color=COLORS["down"], marker="o", linewidth=2.0, label="downstairs")
ax.fill_between(
    comparison["density"],
    comparison["upstairs_speed"],
    comparison["downstairs_speed"],
    color="#FDE68A",
    alpha=0.35,
)
ax.set_title("Stair fundamental diagram: downstairs is faster across the FD range", pad=12)
ax.set_xlabel("Density [1/m²]")
ax.set_ylabel("Mean speed [m/s]")
ax.grid(True, linestyle="--", linewidth=0.7, alpha=0.35)
ax.spines["top"].set_visible(False)
ax.spines["right"].set_visible(False)
ax.legend(frameon=False)
plt.show()

comparison
No description has been provided for this image
Out[10]:
count density upstairs_speed downstairs_speed difference
0 6 0.6 0.500578 0.689080 0.188503
1 7 0.7 0.597739 0.768939 0.171201
2 8 0.8 0.481706 0.625440 0.143735
3 9 0.9 0.611225 0.695758 0.084533
4 10 1.0 0.508669 0.612071 0.103402
5 11 1.1 0.575517 0.802071 0.226554
6 12 1.2 0.517502 0.725480 0.207978
7 13 1.3 0.489394 0.677406 0.188012
8 14 1.4 0.514621 0.711225 0.196604
9 15 1.5 0.594185 0.834138 0.239954

Flat-Ground Demographic v0 Used In Both Scenarios¶

In [11]:
pd.DataFrame(RIEMA_STANDARD_DEMOGRAPHIC)
Out[11]:
label count v0
0 under_30_internal 15 1.34
1 under_30_external 15 1.30
2 age_30_50_internal 15 1.24
3 age_30_50_external 15 1.20
4 over_50_internal 15 1.14
5 over_50_external 15 1.10
6 impaired_mobility 10 0.96

Fruin Stair-Speed Reference Used To Motivate The Zone Factors¶

In [12]:
pd.DataFrame(
    [
        {
            "direction": direction,
            "group": row["label"],
            "stair_speed": row["speed"],
        }
        for direction, rows in FRUIN_STAIR_SPEEDS.items()
        for row in rows
    ]
)
Out[12]:
direction group stair_speed
0 down under_30_internal 0.76
1 down under_30_external 0.81
2 down age_30_50_internal 0.65
3 down age_30_50_external 0.78
4 down over_50_internal 0.55
5 down over_50_external 0.59
6 down impaired_mobility 0.42
7 up under_30_internal 0.55
8 up under_30_external 0.58
9 up age_30_50_internal 0.50
10 up age_30_50_external 0.58
11 up over_50_internal 0.42
12 up over_50_external 0.42
13 up impaired_mobility 0.32
In [13]:
# sweep.cleanup() above already released the per-trial result and
# temp scenario directories. Nothing extra needed here.