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

from vidigi.utils import EventPosition, create_event_position_df
from vidigi.prep import reshape_for_animations, generate_animation_df
from vidigi.animation import generate_animation, animate_activity_log
import pandas as pd
import os
import random
from plotly.subplots import make_subplots
import plotly.io as pio
import plotly.graph_objects as go
import plotly.express as px
pio.renderers.default = "notebook"

View 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.

"""

import itertools
import random

from vidigi.resources import VidigiStore
from vidigi.animation import animate_activity_log
from vidigi.logging import EventLogger
from vidigi.utils import EventPosition, create_event_position_df

import simpy

# 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)
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")

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

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
    )

Iteration through time-unit-by-time-unit logs complete 17:55:30
Snapshot df concatenation complete at 17:55:30
Placement dataframe finished construction at 17:55:30

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 + "]"

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

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 17:55:39

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

Explore incorporating the fuel level bar plot as an additional synchronised plot

## 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 17:55:53

# Set up the desired subplot layout
ROWS = 2

sp = make_subplots(
    rows=ROWS,
    cols=1,
    row_heights=[0.75, 0.25],
    vertical_spacing=0.05,
    subplot_titles=(
        "", # Original Animation
        "Station Tank Fuel Level", # Fuel Tank Level
        )
    )

# Overwrite the domain of our original x and y axis with domain from the new axis
fig.layout['xaxis']['domain'] = sp.layout['xaxis']['domain']
fig.layout['yaxis']['domain'] = sp.layout['yaxis']['domain']

for i in range(2, ROWS+1):

    # Add in the attributes for the secondary axis from our subplot
    fig.layout[f'xaxis{i}'] = sp.layout[f'xaxis{i}']
    fig.layout[f'yaxis{i}'] = sp.layout[f'yaxis{i}']

fig._grid_ref = sp._grid_ref

# First, extract the trace containing the resource icons
# icon_trace = fig.data[1]

# Now keep our figure data as just the initial trace.
fig.data = (fig.data[0],)

# 1. RESOURCE ICONS TRACE
# Readd the resource icons trace in a consistent manner
# Confusingly, when we start messing with the naimation frames, we lose the resource icon trace
# even though it appeared fine until this point - so we have to handle it here
# fig.add_trace(icon_trace)

# 2. BAR PLOT ON SECONDARY AXIS (animated barplot in subplot)
# Initialize with a single point and assign it to subplot axes (x2/y2)

# Get unique time points
time_points = fuel_level_change_df["time"].unique()

# Initial frame (first time point)
initial_time = time_points[0]
initial_df = fuel_level_change_df[fuel_level_change_df["time"] == initial_time]

# Create the initial bar trace

fig.add_trace(go.Bar(
    x=[fuel_level_change_df["entity_id"].values[0]],
    y=[fuel_level_change_df["value"].values[0]],
    showlegend=False
    # We place it in our new subplot using the following line
), row=2, col=1)

# # Now ensure we tell it which traces we are animating
# # (as per https://chart-studio.plotly.com/~empet/15243/animating-traces-in-subplotsbr/#/)
for i, frame in enumerate(fig.frames):
    # Your original frame.data
    # This will be a tuple
    # We'll ensure we only take the first entry
    # original_data = (frame.data[0], )

    original_data = frame.data

    # The new data you want to add for this specific frame
    new_data = (
        # 0: resource icons
        # icon_trace,

        go.Bar(
            x= [fuel_level_change_df.sort_values('time')['entity_id'].values[i]],
            y= [fuel_level_change_df.sort_values('time')['value'].values[i]]
        ) ,  # This needs to be a tuple even if we're only adding a single additional trace, hence the comma
    )

    frame.data = original_data + new_data

fig