Tram Route Model (Continuous State Example)#
Summary#
This model demonstrates Modelly’s continuous state capability.
Overview of continuous states: Most agent-based models advance in discrete ticks, and anything that changes smoothly – a position, a temperature, a stock level – has to be re-integrated by hand on every tick, with the model polling on each tick to check whether some threshold has been crossed yet. That is both wasteful (nothing interesting happens on most ticks) and imprecise (the crossing is only ever detected at the end of the tick in which it happened).
A ContinuousState instead stores a value as a trajectory: a baseline value, a timestamp, and a rate of change. Reading the attribute extrapolates from that baseline to the current model time, so the value is always exact and is never wrong between ticks. Rates can be chained – here position' = speed and speed' = acceleration – in which case the extrapolation picks up the second-order term and becomes piecewise-quadratic.
A Threshold builds on this. Because the trajectory is known analytically, the exact time at which a state will reach a limit can be solved for rather than watched for, and a single event scheduled at that moment. When the trajectory changes – a new acceleration, a new limit – the threshold re-solves and re-schedules itself automatically.
To provide a simple demonstration of this capability is a tram route model.
A single tram runs an ordered route of stations. Departing a station it accelerates at a fixed rate; on reaching cruise speed it coasts; at a brake point computed from v^2 / 2a it applies the brakes; on reaching zero speed it has arrived, dwells for a fixed time, and departs for the next station. Every one of those transitions is a Threshold crossing:
_cruise_threshold = Threshold(
state=speed, limit=cruise_speed, callback="start_coasting", direction="rising"
)
_braking_threshold = Threshold(
state=position, limit=brake_point, callback="brake", direction="rising"
)
_stop_threshold = Threshold(
state=speed, limit=0.0, callback="arrive_at_station", direction="falling"
)
Note that the first two limits are not constants but the cruise_speed and brake_point Observables. brake_point is rewritten at the start of every segment and cruise_speed comes from the scenario, and in both cases the threshold observes the attribute and re-solves for the new crossing time on its own; nothing has to re-arm it.
The brake point is computed from the segment’s peak speed rather than the cruise speed. On a short segment, or with weak acceleration, the tram never reaches cruise speed, so it accelerates to sqrt(2d / (1/a + 1/b)) and brakes straight from there. Assuming cruise speed instead puts the brake point outside the segment, and a rising threshold that starts out already past its limit never fires at all.
The result is a model with an empty agent step. TransitSystem.step only samples the tram’s states for the data collector – if you deleted it, the tram would still run its route with identical timings, driven entirely by the event queue.
How to Run#
Install Modelly with recommended dependencies:
pip install "modelly-abm[rec]"
Then run the example:
solara run app.py
Open the displayed local URL in your browser.
To run the model without visualization and print the collected trace:
python model.py
Files#
model.py: Contains theTransitSystemmodel class and itsTramScenario.agents.py: Contains theTramagent class, its continuous states and its thresholds.app.py: Code for the interactive visualization.utils.py: Palette and matplotlib drawing helpers used byapp.py.
Further Reading#
The continuous state and threshold API used here lives in modelly.behaviour.states, and builds on the reactive observables in modelly.time.signals. Both follow semantic versioning like the rest of Modelly.
Agents#
"""Agents for the Tram Route Model."""
import math
from modelly import Agent, Model
from modelly.behaviour.states import ContinuousState, Threshold
from modelly.time.signals import HasEmitters, Observable
class Tram(Agent, HasEmitters):
"""A tram travelling an ordered route of station positions.
Attributes:
acceleration (Observable): Current acceleration, m/s^2.
speed (ContinuousState): Current speed, m/s.
position (ContinuousState): Current position along the route, m.
Chained off speed (position' = speed).
brake_point (Observable): Position at which braking begins for the
current segment, m. Updated at the start of each departure;
the _brake_point threshold reacts to changes automatically.
cruise_speed (Observable): Speed at which the tram stops accelerating,
m/s. An Observable rather than a plain attribute so the _cruise
threshold tracks it instead of a fixed limit.
"""
acceleration = Observable(fallback_value=0.0)
speed = ContinuousState(fallback_value=0.0, rate=lambda a: a.acceleration)
position = ContinuousState(fallback_value=0.0, rate=lambda a: a.speed)
brake_point = Observable(fallback_value=float("inf"))
cruise_speed = Observable(fallback_value=15.0)
# Threshold names must not collide with the private backing attribute of any
# Observable they read: a Threshold called _brake_point would be found by the
# lookup of brake_point's own "_brake_point" store and silently shadow it.
_cruise_threshold = Threshold(
state=speed, limit=cruise_speed, callback="start_coasting", direction="rising"
)
_braking_threshold = Threshold(
state=position, limit=brake_point, callback="brake", direction="rising"
)
_stop_threshold = Threshold(
state=speed, limit=0.0, callback="arrive_at_station", direction="falling"
)
def __init__(
self,
model: Model,
route: list[float],
cruise_speed: float = 15.0,
acceleration_rate: float = 2.0,
deceleration_rate: float = 3.0,
dwell_time: float = 5.0,
) -> None:
"""Create a new tram.
Args:
model (Model): The model instance that contains the tram.
route (list[float]): Ordered station positions, in meters, to visit.
Must contain at least a starting position and one destination.
cruise_speed (float, optional): Target speed once accelerating
finishes, m/s. Defaults to 15.0.
acceleration_rate (float, optional): Acceleration while departing
a station, m/s^2. Defaults to 2.0.
deceleration_rate (float, optional): Deceleration while braking
into a station, m/s^2. Defaults to 3.0.
dwell_time (float, optional): Time spent stopped at each station
before departing for the next one, seconds. Defaults to 5.0.
Raises:
ValueError: If route has fewer than two stops.
"""
super().__init__(model)
if len(route) < 2:
raise ValueError("route must contain at least a start and one destination")
self.route = route
self.acceleration_rate = acceleration_rate
self.deceleration_rate = deceleration_rate
self.dwell_time = dwell_time
self._segment_index = 1
# The limits the thresholds read must exist before the first
# ContinuousState assignment, which is what binds the thresholds.
self.brake_point = float("inf")
self.cruise_speed = cruise_speed
self.acceleration = 0.0
self.speed = 0.0
self.position = route[0]
@property
def next_station(self) -> float:
"""Position of the station this tram is currently travelling towards.
Returns:
float: The target station's position, in meters.
"""
return self.route[self._segment_index]
@property
def route_complete(self) -> bool:
"""Whether the tram has visited every station on its route.
Returns:
bool: True once the final station has been reached.
"""
return self._segment_index >= len(self.route)
def peak_speed(self) -> float:
"""Highest speed the tram can reach before it must brake for the next stop.
On a long segment this is the cruise speed. On a short one the tram runs
out of room first and the profile is triangular: it accelerates to
v = sqrt(2d / (1/a + 1/b)) and brakes from there, never coasting.
Returns:
float: The peak speed for this segment, in meters per second.
"""
distance = self.next_station - self.position
triangular = math.sqrt(
2.0
* distance
/ (1.0 / self.acceleration_rate + 1.0 / self.deceleration_rate)
)
return min(self.cruise_speed, triangular)
def braking_distance(self) -> float:
"""Distance needed to decelerate to a stop from this segment's peak speed.
Returns:
float: The braking distance, in meters.
"""
return self.peak_speed() ** 2 / (2.0 * self.deceleration_rate)
def depart(self) -> None:
"""Accelerate towards the next station."""
self.brake_point = self.next_station - self.braking_distance()
self.acceleration = self.acceleration_rate
def start_coasting(self) -> None:
"""Stop accelerating once cruise speed is reached."""
self.acceleration = 0.0
def brake(self) -> None:
"""Decelerate once the brake point for this segment is reached."""
self.brake_point = float("inf")
self.acceleration = -self.deceleration_rate
def arrive_at_station(self) -> None:
"""Come to a stop and schedule departure for the next station, if any."""
self.acceleration = 0.0
self._segment_index += 1
if self.route_complete:
return
self.model.schedule_event(self.depart, after=self.dwell_time)
Model#
"""Tram Route Model.
A model of a tram running a multi-station route using continuous-time
kinematics. The tram accelerates to a cruise speed, coasts, brakes at an
analytically-computed point before each station, and dwells before departing
for the next one -- all without the model needing to step on every tick to
check whether a threshold has been crossed.
"""
from modelly import Model
from modelly.data.collector import DataCollector
from modelly.data.recorders import DataRecorder
from modelly.examples.behaviour.tram_model.agents import Tram
from modelly.scenarios import Scenario
class TramScenario(Scenario):
"""Scenario for the Tram Route model.
Args:
n_stations: Number of stations on the route, including the origin.
station_spacing: Distance between consecutive stations, in meters.
cruise_speed: Target speed once accelerating finishes, m/s.
acceleration_rate: Acceleration while departing a station, m/s^2.
deceleration_rate: Deceleration while braking into a station, m/s^2.
dwell_time: Time spent stopped at each station, in seconds.
rng: Seed for reproducibility.
"""
n_stations: int = 20
station_spacing: float = 200.0
cruise_speed: float = 15.0
acceleration_rate: float = 2.0
deceleration_rate: float = 3.0
dwell_time: float = 5.0
class TransitSystem(Model):
"""A minimal model containing a single tram running a multi-station route.
Attributes:
route (list[float]): Ordered station positions, in meters.
tram (Tram): The tram running the route.
"""
def __init__(self, scenario: TramScenario = TramScenario):
"""Create the model and its tram.
Args:
scenario: TramScenario containing model parameters.
"""
super().__init__(scenario=scenario)
self.route = [
scenario.station_spacing * i for i in range(int(scenario.n_stations))
]
self.tram = Tram(
self,
route=self.route,
cruise_speed=scenario.cruise_speed,
acceleration_rate=scenario.acceleration_rate,
deceleration_rate=scenario.deceleration_rate,
dwell_time=scenario.dwell_time,
)
self.recorder = DataRecorder(self)
self.data_registry.track_agents(
self.agents, "tram_data", ["position", "speed", "acceleration"]
).record(self.recorder)
# The DataCollector is kept alongside only to feed the plots in
# app.py, since make_plot_component reads model.datacollector.
self.datacollector = DataCollector(
model_reporters={
"Position": lambda m: m.tram.position,
"Speed": lambda m: m.tram.speed,
"Acceleration": lambda m: m.tram.acceleration,
}
)
self.datacollector.collect(self)
# The tram is idle until it is told to leave the first station. Scheduling
# the departure rather than calling it directly keeps every state change
# on the event queue, so model.time is meaningful from the very first entry.
self.schedule_event(self.tram.depart, at=0.0)
def step(self) -> None:
"""Sample the tram's continuous states once per time unit.
The tram itself needs no per-step logic: its speed and position are
extrapolated analytically and its transitions fire from the event
queue. This step exists purely to feed the DataCollector behind the
plots; the recorder collects without it.
"""
self.datacollector.collect(self)
if __name__ == "__main__":
model = TransitSystem(scenario=TramScenario())
print(f"Route: {model.route}")
print(
f"cruise_speed={model.tram.cruise_speed} m/s, "
f"acceleration={model.tram.acceleration_rate} m/s^2, "
f"deceleration={model.tram.deceleration_rate} m/s^2, "
f"braking_distance={model.tram.braking_distance():.2f}m\n"
)
model.run_until(2000.0)
print(model.recorder.get_table_dataframe("tram_data").to_string())
print(f"\nFinal time: {model.time:.2f}")
print(f"Final position: {model.tram.position:.2f} m")
print(f"Final speed: {model.tram.speed:.2f} m/s")
App#
import solara
from matplotlib.figure import Figure
from modelly.examples.behaviour.tram_model.model import TramScenario, TransitSystem
from modelly.examples.behaviour.tram_model.utils import (
POSITION_COLOR,
SCALE,
SPEED_COLOR,
TITLE_SIZE,
draw_track_view,
tram_portrayal,
)
from modelly.viz import Slider, SolaraViz, make_plot_component
from modelly.viz.utils import update_counter
model_params = {
"rng": {
"type": "InputText",
"value": 42,
"label": "Random Seed",
},
"n_stations": Slider("Number of stations", 20, 2, 40, 1),
"station_spacing": Slider("Station spacing (m)", 200, 100, 1000, 50),
"cruise_speed": Slider("Cruise speed (m/s)", 15, 5, 30, 1),
"acceleration_rate": Slider("Acceleration (m/s^2)", 2.0, 0.5, 5.0, 0.5),
"deceleration_rate": Slider("Deceleration (m/s^2)", 3.0, 0.5, 5.0, 0.5),
"dwell_time": Slider("Dwell time (s)", 5, 0, 30, 1),
}
@solara.component
def plot_track(model):
"""Draw the route as a line of stations with the tram at its current position."""
update_counter.get()
tram = model.tram
style = tram_portrayal(tram)
fig = Figure(figsize=(7 * SCALE, 2.1 * SCALE))
ax = fig.subplots()
draw_track_view(ax, model, style)
ax.set_title(
f"t={model.time:.1f}s | {tram.position:.1f} m | "
f"{tram.speed:.1f} m/s | {style['phase']}",
color=style["color"],
fontsize=TITLE_SIZE,
)
fig.tight_layout()
solara.FigureMatplotlib(fig)
SpeedPlot = make_plot_component({"Speed": SPEED_COLOR})
PositionPlot = make_plot_component({"Position": POSITION_COLOR})
model = TransitSystem(scenario=TramScenario())
page = SolaraViz(
model,
components=[SpeedPlot, PositionPlot, plot_track],
model_params=model_params,
name="Tram Route Model",
)
page # noqa