Visualising a more complex status alongside entity icons - gas station with individual fuel tank level

Warning

This example has not yet been updated to reflect all of the new features and recommendations from vidigi 2.0.0, which have further simplified the process of adding vidigi to your model and accessing and modifying your animation.

For now, all of the code below will still work, but check out the getting started page for a full guide to the recommended way to use vidigi 2.0.0.

TipAdding synchronised charts to an animation

The bottom half of this notebook builds a second, synchronised chart panel underneath the animation. Since vidigi 2.0.0 there are helpers for this - vidigi.animation.add_subplot_panels, add_synchronised_trace and add_synchronised_trace_from_dataframe - which handle the fiddly parts (keeping the extra trace in step with every frame without losing the stage labels or resource icons). This example uses them. See Synchronised traces for a focused walkthrough.

import os
import random

import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
import plotly.io as pio

from vidigi.animation import (
    add_subplot_panels,
    add_synchronised_trace_from_dataframe,
    generate_animation,
)
from vidigi.prep import generate_animation_df, reshape_for_animations
from vidigi.utils import EventPosition, create_event_position_df

# "iframe" keeps the executed notebook small (the heavy Plotly HTML is written to a
# gitignored iframe_figures/ folder instead of embedded inline). Use "notebook" if you
# are running this interactively and want the animation inline.
pio.renderers.default = "iframe"

First we bring in our code.

This example is a modified version of the gas station example created by team SimPy. Updates have been done to incorporate vidigi resources to allow for resource tracking.

NoteView Imported Code, which has had logging steps added at the appropriate points in the ‘model’ class

"""
Gas Station Refueling example

Covers:

- Resources: Resource
- Resources: Container
- Waiting for other processes

Scenario:
  A gas station has a limited number of gas pumps that share a common
  fuel reservoir. Cars randomly arrive at the gas station, request one
  of the fuel pumps and start refueling from that reservoir.

  A gas station control process observes the gas station's fuel level
  and calls a tank truck for refueling if the station's level drops
  below a threshold.

  simpy model (simpy_gas_stations.py) written by Team SimPy as part of SimPy documentation.

"""

import itertools
import random

import simpy

from vidigi.logging import EventLogger
from vidigi.resources import VidigiStore

# fmt: off
RANDOM_SEED = 42
STATION_TANK_SIZE = 600    # MODIFIED FROM EXAMPLE: Size of the gas station tank (liters)
THRESHOLD = 25             # Station tank minimum level (% of full)
CAR_TANK_SIZE = 50         # Size of car fuel tanks (liters)
CAR_TANK_LEVEL = [5, 25]   # Min/max levels of car fuel tanks (liters)
PAYMENT_TIME = [30, 90]    # MODIFICATION: Time it takes to pay
REFUELING_SPEED = 1        # MODIFIED FROM EXAMPLE: Rate of refuelling car fuel tank (liters / second)
TANK_TRUCK_ARRIVAL_TIME = 300  # Time it takes tank truck to arrive (seconds)
TANK_TRUCK_REFUEL_TIME = 1000  # MODIFICATION: Time it takes tank truck to fill the station tank (seconds)
T_INTER = [30, 300]        # Interval between car arrivals [min, max] (seconds)
SIM_TIME = 60*60*6           # Simulation duration (seconds)
# fmt: on


def car(name, env, gas_station, station_tank, logger):
    """A car arrives at the gas station for refueling.

    It requests one of the gas station's fuel pumps and tries to get the
    desired amount of fuel from it. If the station's fuel tank is
    depleted, the car has to wait for the tank truck to arrive.

    """
    car_tank_level = random.randint(*CAR_TANK_LEVEL)
    logger.log_arrival(entity_id=name)
    print(f"{env.now:6.1f} s: {name} arrived at gas station")
    logger.log_queue(
        entity_id=name,
        event="pump_queue_wait_begins",
        fuel_level_start=car_tank_level,
        fuel_level_end=CAR_TANK_SIZE,
    )
    with gas_station.request() as req:
        # Request one of the gas pumps
        gas_pump = yield req

        # Get the required amount of fuel
        fuel_required = CAR_TANK_SIZE - car_tank_level
        yield station_tank.get(fuel_required)

        logger.log_resource_use_start(
            entity_id=name,
            event="payment_begins",
            resource_id=gas_pump.id_attribute,
            fuel_level_start=car_tank_level,
            fuel_level_end=CAR_TANK_SIZE,
        )

        yield env.timeout(random.randint(*PAYMENT_TIME))

        logger.log_resource_use_end(
            entity_id=name,
            event="payment_ends",
            resource_id=gas_pump.id_attribute,
            fuel_level_start=car_tank_level,
            fuel_level_end=CAR_TANK_SIZE,
        )

        logger.log_resource_use_start(
            entity_id=name,
            event="pumping_begins",
            resource_id=gas_pump.id_attribute,
            fuel_level_start=car_tank_level,
            fuel_level_end=CAR_TANK_SIZE,
        )

        # The "actual" refueling process takes some time
        yield env.timeout(fuel_required / REFUELING_SPEED)

        logger.log_resource_use_end(
            entity_id=name,
            event="pumping_ends",
            resource_id=gas_pump.id_attribute,
            fuel_level_start=car_tank_level,
            fuel_level_end=CAR_TANK_SIZE,
        )

        print(f"{env.now:6.1f} s: {name} refueled with {fuel_required:.1f}L")
        logger.log_departure(entity_id=name)


def gas_station_control(env, station_tank, logger):
    """Periodically check the level of the gas station tank and call the tank
    truck if the level falls below a threshold."""
    truck_call_id = 0

    while True:
        if station_tank.level / station_tank.capacity * 100 < THRESHOLD:
            # We need to call the tank truck now!
            logger.log_arrival(entity_id=f"Call {truck_call_id}")
            logger.log_queue(entity_id=f"Call {truck_call_id}", event="calling_truck")
            print(f"{env.now:6.1f} s: Calling tank truck")
            # Wait for the tank truck to arrive and refuel the station tank
            yield env.process(tank_truck(env, station_tank, logger, truck_call_id))

            truck_call_id += 1

        yield env.timeout(120)  # Check every 120 seconds


# def tank_truck(env, station_tank, logger, truck_call_id):
#     """Arrives at the gas station after a certain delay and refuels it."""
#     yield env.timeout(TANK_TRUCK_ARRIVAL_TIME)
#     logger.log_departure(entity_id=f"Call {truck_call_id}")
#     logger.log_arrival(entity_id=f"Truck {truck_call_id}")
#     amount = station_tank.capacity - station_tank.level
#     logger.log_queue(entity_id=f"Truck {truck_call_id}", event="refueling")
#     yield env.timeout(TANK_TRUCK_REFUEL_TIME)
#     station_tank.put(amount)
#     print(
#         f'{env.now:6.1f} s: Tank truck arrived and refuelled station with {amount:.1f}L'
#     )
#     logger.log_departure(entity_id=f"Truck {truck_call_id}")


# Modification to make refuelling a smooth, loggable process
def tank_truck(env, station_tank, logger, truck_call_id):
    """Tank truck arrives and refuels the station tank for a fixed duration."""
    yield env.timeout(TANK_TRUCK_ARRIVAL_TIME)
    logger.log_departure(entity_id=f"Call {truck_call_id}")
    logger.log_arrival(entity_id=f"Truck {truck_call_id}")
    logger.log_queue(entity_id=f"Truck {truck_call_id}", event="refuelling")

    refuel_time = TANK_TRUCK_REFUEL_TIME  # total time truck stays
    refuel_rate = 10  # L/s (or adjust based on need)
    step = 1  # seconds between each refill step

    total_refueled = 0
    elapsed = 0

    while (elapsed < refuel_time) | station_tank.level < (
        STATION_TANK_SIZE - (STATION_TANK_SIZE * 0.02)
    ):
        yield env.timeout(step)
        elapsed += step

        increment = refuel_rate * step
        space_available = station_tank.capacity - station_tank.level
        actual_increment = min(increment, space_available)

        if actual_increment > 0:
            station_tank.put(actual_increment)
            total_refueled += actual_increment

    print(
        f"{env.now:6.1f} s: Truck {truck_call_id} refueled station with {total_refueled:.1f}L"
    )
    logger.log_departure(entity_id=f"Truck {truck_call_id}")


def car_generator(env, gas_station, station_tank, logger):
    """Generate new cars that arrive at the gas station."""
    for i in itertools.count():
        yield env.timeout(random.randint(*T_INTER))
        env.process(car(f"Car {i}", env, gas_station, station_tank, logger))


def fuel_monitor(env, station_tank, logger, interval=1):
    """Logs the fuel level at regular intervals."""
    while True:
        logger.log_queue(
            entity_id="StationTank",
            event_type="fuel_level_change",
            event="fuel_level_change",
            value=station_tank.level,
        )
        yield env.timeout(interval)


# Setup and start the simulation
print("Gas Station refuelling")
random.seed(RANDOM_SEED)

# Create environment and start processes
env = simpy.Environment()
gas_station = VidigiStore(env, num_resources=2, label="pump")
station_tank = simpy.Container(env, capacity=STATION_TANK_SIZE, init=STATION_TANK_SIZE)
logger = EventLogger(env=env)
logger.log_queue(
    entity_id="parameter",
    event_type="parameter",
    event="tank_size",
    value=STATION_TANK_SIZE,
)
env.process(gas_station_control(env, station_tank, logger))
env.process(car_generator(env, gas_station, station_tank, logger))
env.process(fuel_monitor(env, station_tank, logger))


# Execute!
env.run(until=SIM_TIME)

logger.to_csv("gas_station_log.csv")

First, we build our animation in the normal way.

# Define positions for animation
event_positions = create_event_position_df(
    [
        EventPosition(event="arrival", x=0, y=350, label="Entrance"),
        EventPosition(event="pump_queue_wait_begins", x=400, y=350, label="Queue"),
        EventPosition(
            event="payment_begins",
            x=340,
            y=175,
            resource="num_pumps",
            label="Pumping Gas",
        ),
        EventPosition(
            event="pumping_begins",
            x=340,
            y=175,
            resource="num_pumps",
            label="Pumping Gas",
        ),
        EventPosition(event="calling_truck", x=140, y=50, label="Calling Truck"),
        EventPosition(event="refuelling", x=340, y=50, label="Truck Filling Tank"),
        EventPosition(event="depart", x=250, y=50, label="Exit"),
    ]
)


class Params:
    def __init__(self):
        self.num_pumps = 2


icon_list = [
    "🚗",
    "🚙",
    "🚓",
    "🚗",
    "🚙",
    "🏍️",
    "🏍️",
    "🚗",
    "🚙",
    "🚑",
    "🚗",
    "🚙",
    "🛻",
    "🚗",
    "🚙",
    "🚛",
    "🚗",
    "🚙",
    "🚕",
    "🚗",
    "🚙",
    "🚒",
    "🚗",
    "🚙",
    "🚑",
]

random.shuffle(icon_list)

/home/runner/work/vidigi/vidigi/src/vidigi/utils.py:246: UserWarning: `create_event_position_df` places different events at identical coordinates: (340, 175): 'payment_begins', 'pumping_begins'.

This is usually a copy-paste slip - the events will be drawn directly on top of each other, so entities at those steps look like they share a position. Give each event its own x/y if that was not intended.
  _warn_on_duplicate_event_positions(

event_log_df = pd.read_csv("gas_station_log.csv")

STEP_SNAPSHOT_MAX = 6
LIMIT_DURATION = 60 * 60 * 3
WRAP_QUEUES_AT = 3

full_entity_df = reshape_for_animations(
    event_log=event_log_df,
    every_x_time_units=5,
    step_snapshot_max=STEP_SNAPSHOT_MAX,
    limit_duration=LIMIT_DURATION,
    debug_mode=True,
)

full_entity_df_plus_pos = generate_animation_df(
    full_entity_df=full_entity_df,
    event_position_df=event_positions,
    wrap_queues_at=WRAP_QUEUES_AT,
    step_snapshot_max=STEP_SNAPSHOT_MAX,
    gap_between_entities=150,
    gap_between_resources=180,
    gap_between_queue_rows=150,
    # gap_between_resource_rows=60,
    debug_mode=True,
    custom_entity_icon_list=icon_list,
)

/tmp/ipykernel_4283/1532434859.py:1: UserWarning: 2 entities ('parameter', 'StationTank') have events in the event log but no 'arrival' event, so they will be missing from every frame of the animation.

vidigi works out who is present at each snapshot from the arrival and departure rows, so an entity without an arrival is never drawn.

The usual cause is discarding a warm-up period by filtering the log, e.g. `event_log[event_log['time'] >= warm_up]`, which removes the arrival rows of everyone already in the system - including entities that are still queuing.

To skip a warm-up period, pass the whole event log and set `warm_up` to the end of the warm-up instead. That trims the animation window without discarding the history it needs.
  full_entity_df = reshape_for_animations(

Iteration through time-unit-by-time-unit logs complete 13:45:53
Snapshot df concatenation complete at 13:45:54
Placement dataframe started construction at 13:45:54
Placement dataframe finished construction at 13:45:54

/home/runner/work/vidigi/vidigi/src/vidigi/utils.py:1228: UserWarning: `event_position_df` places different events at identical coordinates: (340, 175): 'payment_begins', 'pumping_begins'.

This is usually a copy-paste slip - the events will be drawn directly on top of each other, so entities at those steps look like they share a position. Give each event its own x/y if that was not intended.
  return func(*bound.args, **bound.kwargs)

Let’s now define a custom function that uses the amount of fuel to generate a bar for each individual car that can update as they fill up.

def build_fuel_bar(value, max_value=50, length=10):
    """Create an ASCII bar to show fuel level."""
    try:
        if value is None or (
            isinstance(value, float) and (value != value)
        ):  # check for None or NaN
            proportion = 0
        else:
            proportion = min(max(value / max_value, 0), 1)
    except Exception:
        proportion = 0  # fallback

    filled = int(proportion * length)
    empty = length - filled
    filled_icon = "█"
    empty_icon = "░"
    return "[" + filled_icon * filled + empty_icon * empty + "]"

Now we can combine this with a function that will apply a custom icon to different kinds of entities - the refill trucks, the action of calling the truck (both of which we’ll display as a string of text plus an icon), and the custom icons that reflect the stage of transaction individual entities are at, along with their fuel level at that point.

def custom_icon_rules(row):
    icon = row.get("icon", "")
    entity_id = row.get("entity_id", "")
    event = row.get("event", "")
    fuel_level_start = row.get("fuel_level_start", None)  # Only for cars

    if "more" not in str(icon):
        if isinstance(entity_id, str):
            if "Truck" in entity_id:
                return "🚚 Truck is refilling the tank..."
            elif "Call" in entity_id:
                return "☎️ Calling Truck!"
            elif "Car" in entity_id:
                bar = ""
                if (
                    event == "arrival" or event == "pump_queue_wait_begins"
                ) and fuel_level_start is not None:
                    bar = " " + build_fuel_bar(fuel_level_start)
                    return icon + "<br>" + bar + "<br><br>"
                elif event == "payment_begins" and fuel_level_start is not None:
                    bar = " " + build_fuel_bar(fuel_level_start)
                    return icon + "<br>" + bar + "<br> Paying"
                elif event == "pumping_begins" and fuel_level_start is not None:
                    arrival_time = row["time"]
                    elapsed = max(float(row["snapshot_time"]) - float(arrival_time), 0)
                    current_fuel = min(fuel_level_start + elapsed * 1, 50)
                    bar = " " + build_fuel_bar(current_fuel)
                    return icon + "<br>" + bar + "<br> Pumping"
                elif event == "departure" or event == "pumping_ends":
                    bar = " " + build_fuel_bar(50)  # Car is full when it leaves
                    return icon + "<br>" + bar + "<br> <br>"

            else:
                return icon
    return icon


full_entity_df_plus_pos = full_entity_df_plus_pos.assign(
    icon=full_entity_df_plus_pos.apply(custom_icon_rules, axis=1)
)

Finally we create our animation. Note that instead of including a resource icon in the normal way, we’ve added the resources as part of our background. This can be a more visually pleasing option when the number of resources is something that isn’t going to change.

fig = generate_animation(
    full_entity_df_plus_pos=full_entity_df_plus_pos.sort_values(
        ["entity_id", "snapshot_time"]
    ),
    event_position_df=event_positions,
    scenario=Params(),
    simulation_time_unit="seconds",
    plotly_height=900,
    plotly_width=1200,
    override_x_max=500,
    override_y_max=750,
    entity_icon_size=30,
    gap_between_resources=180,
    display_stage_labels=False,
    # resource_opacity=1,
    resource_opacity=0,
    setup_mode=False,
    # custom_resource_icon="⛽",
    resource_icon_size=40,
    add_background_image="https://raw.githubusercontent.com/hsma-tools/vidigi/refs/heads/main/examples/example_15_gas_station_refuelling/gas_station.png",
    background_image_opacity=1,  # New parameter in 1.1.0
    overflow_text_color="white",  # New parameter in 1.1.0
    start_time="09:00:00",
    time_display_units="%H:%M:%S",
    debug_mode=True,
    frame_duration=100,
    frame_transition_duration=100,
)

fig

Output animation generation complete at 13:46:05

Now let’s explore visualising the total amount of fuel available in the station’s tank.

fuel_level_change_df = event_log_df[
    (event_log_df["event_type"] == "fuel_level_change")
    & (event_log_df["time"] % 5 == 0)
    & (event_log_df["time"] < LIMIT_DURATION)
]

px.bar(
    fuel_level_change_df,
    x="entity_id",
    y="value",
    animation_frame="time",
    range_y=[0, 400],
)

Next, we can incorporate the tank fuel level as an additional synchronised chart beneath the animation, using add_subplot_panels and add_synchronised_trace_from_dataframe.

We’ll first regenerate the animation, with a little more height to leave room for the second panel.

## Same as before, but increase the height to give space for some of it to be taken up by the bar plot later

fig = generate_animation(
    full_entity_df_plus_pos=full_entity_df_plus_pos.sort_values(
        ["entity_id", "snapshot_time"]
    ),
    event_position_df=event_positions,
    scenario=Params(),
    simulation_time_unit="seconds",
    plotly_height=1000,
    plotly_width=1200,
    override_x_max=500,
    override_y_max=750,
    entity_icon_size=30,
    gap_between_resources=180,
    display_stage_labels=False,
    # resource_opacity=1,
    resource_opacity=0,
    setup_mode=False,
    # custom_resource_icon="⛽",
    resource_icon_size=40,
    add_background_image="https://raw.githubusercontent.com/hsma-tools/vidigi/refs/heads/main/examples/example_15_gas_station_refuelling/gas_station.png",
    background_image_opacity=1,  # New parameter in 1.1.0
    overflow_text_color="white",  # New parameter in 1.1.0
    start_time="09:00:00",
    time_display_units="%H:%M:%S",
    debug_mode=True,
    frame_duration=100,
    frame_transition_duration=100,
)

Output animation generation complete at 13:46:19

Step 1: make room for the second panel

add_subplot_panels turns the single-axis animation into a two-row subplot grid, with the animation in the top row. row_heights sets how the vertical space is split. We keep the new panel’s axes visible (hide_new_panel_axes=False) so the fuel scale can be read.

fig = add_subplot_panels(
    fig,
    row_heights=[0.78, 0.22],
    subplot_titles=("", "Station Tank Fuel Level"),
    hide_new_panel_axes=False,
)

# One tank reading per animation snapshot (forward-filled between fuel_level_change events)
tank_level = (
    event_log_df.loc[
        event_log_df["event_type"] == "fuel_level_change", ["time", "value"]
    ]
    .drop_duplicates("time")
    .set_index("time")["value"]
)

station_fuel_df = pd.DataFrame(
    {"snapshot_time": sorted(full_entity_df_plus_pos["snapshot_time"].unique())}
)
station_fuel_df["fuel_level"] = (
    station_fuel_df["snapshot_time"].map(tank_level).ffill().bfill()
)


def fuel_bar(rows):
    return go.Bar(
        x=["Station Tank"],
        y=list(rows["fuel_level"]),
        marker_color="#4c78a8",
        showlegend=False,
        xaxis="x2",
        yaxis="y2",
    )


fig = add_synchronised_trace_from_dataframe(
    fig,
    station_fuel_df,
    fuel_bar,
    frame_time_col="snapshot_time",
    match="index",
    accumulate=False,
)

fig.update_yaxes(range=[0, station_fuel_df["fuel_level"].max() * 1.1], row=2, col=1)

Step 2: add the synchronised bar

The fuel_level_change events record the tank level once per simulated second. We take one reading per animation snapshot - the same snapshot_time values generate_animation_df produced - so there is exactly one row per frame, then let add_synchronised_trace_from_dataframe build a bar for each.

match="index" pairs the i-th row with the i-th frame, so it doesn’t matter that the frames are labelled as clock times while our data is in seconds. accumulate=False means each frame sees only its own row (a snapshot, not a running total).

We now have a complex animation showing a lot of different information for entities and the situation - and the tank fuel level panel stays in step across every frame, not just the first.

Drag the slider and you can watch the tank draining as cars fill up, then jumping back up each time the delivery truck refills it.

fig