A slightly more complex example with multiple servers

When dealing with multiple servers, we need to have a way to monitor which resource is in use. This is where vidigi’s custom resource classes come in, such as VidigiStore.

VidigiStore is designed to mimic the Resource class in Simpy - so you can continue to use the same patterns of resource requesting you are familiar with. It also exposes store.count (units currently in use), store.capacity / store.num_resources (pool size) and store.n_waiting (requests queued for a unit), mirroring simpy.Resource.count / .capacity / .queue for conditional model logic or manual logging. Returning more units than the pool holds raises ValueError (pass strict_capacity=False to opt out).

Note

When setting up a VidigiStore, we use the parameters ‘num_resources’ and ‘label’.

‘num_resources’ is analogous to ‘capacity’ in a SimPy resource. It’s worth noting that capacity means something different in a VidigiStore, so be careful to not mix them up!

A label is optional but highly recommended.

servers = VidigiStore(env=env, num_resources=params.num_servers, label="server")

Step 1. Import required libraries

• vidigi
• simpy - for simulation model (or see the Ciw functions and examples elsewhere in this documentation)
• random - for generating random arrivals
• pandas - for managing dataframes

import simpy
import pandas as pd
import random
from vidigi.logging import EventLogger
from vidigi.utils import EventPosition, create_event_position_df
from vidigi.resources import VidigiStore

Step 2. Set up simulation parameters

# Simple simulation parameters
SIM_DURATION = 50
NUM_SERVERS = 3
ARRIVAL_RATE = 4.0
SERVICE_TIME = 3.0

Step 3. Write model code with event logs

Here we create a simple simulation model using simpy.

On the left is a basic simpy model. If you’re not familiar with this notation, check out the simpy documentation for an introduction to SimPy, or the DES book from the Health Service Modelling Associate programme.

On the right is how we incorporate vidigi.

We set up a logger using the EventLogger class. This then gives us methods for recording arrivals, departures and queueing steps.

Warning

Every entity must have a single arrival and a single departure event. Reusing an entity ID for a second arrival or departure - rather than every entity having its own unique ID - averages the duplicate times together instead of keeping either one. vidigi now raises a ValueError naming the offending entity ID when this happens, rather than silently producing a wrong animation.

They may have as many queueing events as desired (and they don’t have to just be what you’re traditionally think of as a ‘queue’ - they can be any point where an individual is waiting for something or doing something).

We can also record the points at which an entity starts using/blocking a resource/server, and when this ends. For this we also need to pass in an identifier that tracks which resource they are using. When using SimPy resources, it is not possible to track this information. However, if we instead use a VidigiStore, we get access to a resource ID, which can then be recorded and helps us visualise resource use clearly.

We also need to pass in the resource/server counts as a dictionary or as a class. If passing as a dictionary, we will pass the resource name - which we choose ourselves - as the key, and the count of that resource as the value. If passing as a class, we create the resource name - which we again choose ourselves - as an attribute of this class. If you already have a class in which you store your scenario, it is recommended to use this class, as demonstrated in the HSMA structured example.

Option 1: Manually Logging Resource Use Start and End

Simple SimPy Model

def patient_generator(env, server, event_log):
    """Generate patients arriving"""
    patient_id = 0

    while True:
        patient_id += 1

        # Start the patient process
        env.process(patient_process(env, patient_id, server, event_log))

        # Wait for next arrival
        yield env.timeout(random.expovariate(ARRIVAL_RATE))

def patient_process(env, patient_id, server, event_log):
    """Process a single patient through the system"""

    # Request server
    with server.request() as request:
        yield request

        # Service time
        service_duration = random.expovariate(1.0/SERVICE_TIME)
        yield env.timeout(service_duration)

# Run the simulation
def run_simulation():
    env = simpy.Environment()
    server = simpy.Resource(env, capacity=NUM_SERVERS)
    event_log = []

    # Start patient generator
    env.process(patient_generator(env, server, event_log))

    # Run simulation
    env.run(until=SIM_DURATION)

With Vidigi Modifications

class Params: 
    def __init__(self): 
        self.num_servers = 3 

def patient_generator(env, servers, logger):
    """Generate patients arriving"""
    patient_id = 0

    while True:
        patient_id += 1

        # Log arrival
        logger.log_arrival(entity_id=patient_id) 

        # Start the patient process
        env.process(patient_process(env, patient_id, servers, logger))

        # Wait for next arrival
        yield env.timeout(random.expovariate(ARRIVAL_RATE))

def patient_process(env, patient_id, servers, logger):
    """Process a single patient through the system"""

    # Log start of queue wait 
    logger.log_queue(entity_id=patient_id, event='queue_wait_begins') 

    # Request server
    with servers.request() as request:
        server = yield request  

        # Log service start
        logger.log_resource_use_start(entity_id=patient_id, event="service_begins", resource_id=server.id) 

        # Service time
        service_duration = random.expovariate(1.0/SERVICE_TIME)
        yield env.timeout(service_duration)

        # Log service start
        logger.log_resource_use_end(entity_id=patient_id, event="service_complete", resource_id=server.id) 

    # Log departure 
    logger.log_departure(entity_id=patient_id)  

# Run the simulation
def run_simulation():
    params = Params()
    env = simpy.Environment()
    servers = VidigiStore(env=env, num_resources=params.num_servers, label="server") 
    logger = EventLogger(env=env) 

    # Start patient generator
    env.process(patient_generator(env, servers, logger))

    # Run simulation
    env.run(until=SIM_DURATION)

    return logger 

Option 2: An alternative - automatic resource-use logging

Calling log_resource_use_start and log_resource_use_end by hand around every resource request works, but it’s easy to get wrong - forgetting the closing call, or passing the wrong resource_id.

If you pass a logger to VidigiStore (or VidigiPriorityStore) when you create it, and pass entity_id when you request a resource, both events are logged for you automatically - no manual log_resource_use_start/log_resource_use_end calls needed. This is entirely opt-in: a store built without logger= behaves exactly as before, and you can still log some steps manually while auto-logging others.

Tip

If you have passed both a label and logger, your resource_use_start and resource_use_end events will be generated automatically with the label you provided plus the suffix _start or _end.

So for the example above, you’d have nurse_examination_start and nurse_examination_end.

def patient_generator_auto(env, servers, logger):
    """Generate patients arriving - identical to patient_generator above,
    just calling patient_process_auto instead."""
    patient_id = 0

    while True:
        patient_id += 1

        logger.log_arrival(entity_id=patient_id)

        env.process(patient_process_auto(env, patient_id, servers, logger))

        yield env.timeout(random.expovariate(ARRIVAL_RATE))

def patient_process_auto(env, patient_id, servers, logger):
    """Process a single patient through the system, letting VidigiStore
    log resource use automatically instead of calling
    log_resource_use_start/log_resource_use_end ourselves."""

    logger.log_queue(entity_id=patient_id, event='queue_wait_begins')

    # entity_id (and, optionally, start_event/end_event) is all VidigiStore
    # needs to log resource_use/resource_use_end itself
    with servers.request(
        entity_id=patient_id, start_event="service_begins", end_event="service_complete"
    ) as request:
        yield request

        service_duration = random.expovariate(1.0/SERVICE_TIME)
        yield env.timeout(service_duration)

    logger.log_departure(entity_id=patient_id)

def run_simulation_auto():
    params = Params()
    env = simpy.Environment()
    logger = EventLogger(env=env)
    # Pass the logger straight to VidigiStore instead of calling
    # log_resource_use_start/log_resource_use_end by hand in patient_process
    servers = VidigiStore(
        env=env, num_resources=params.num_servers, label="server", logger=logger
    )

    env.process(patient_generator_auto(env, servers, logger))
    env.run(until=SIM_DURATION)

    return logger.to_dataframe()

auto_event_log_df = run_simulation_auto()
print(f"Generated {len(auto_event_log_df)} events")
auto_event_log_df.head()

Generated 517 events

	entity_id	event_type	event	time	resource_id	unique_resource_id
0	1	arrival_departure	arrival	0.000000	NaN	NaN
1	1	queue	queue_wait_begins	0.000000	NaN	NaN
2	1	resource_use	service_begins	0.000000	1.0	server_1
3	2	arrival_departure	arrival	0.195949	NaN	NaN
4	2	queue	queue_wait_begins	0.195949	NaN	NaN

The two versions above are run with independent random draws, so their event timings won’t match - but the logged event shape (resource_use/resource_use_end rows carrying entity_id, resource_id, event and time, one matched pair per patient) is identical either way. vidigi’s own test suite pins this more rigorously: the same scenario run once with manual log_resource_use_start/log_resource_use_end calls and once with logger=/entity_id= auto-logging (and no manual calls at all) is asserted to produce byte-identical event logs.

Recording extra entity-level fields with automatic logging

When you log resource use by hand, you can pass any number of extra keyword arguments to log_resource_use_start/log_resource_use_end (for example acuity=3, arrival_mode="ambulance") and they become extra columns in the event log, ready for downstream analysis. Auto-logging keeps this: any keyword argument you pass to request()/get() that vidigi doesn’t recognise is forwarded straight through to both the resource_use and the resource_use_end event.

def patient_process_auto_extra_fields(env, patient_id, servers, logger):
    """As patient_process_auto, but also recording two entity-level attributes
    on the resource-use events via request()'s **extra_fields passthrough."""

    logger.log_queue(entity_id=patient_id, event='queue_wait_begins')

    # Attributes of this specific patient, known at the point of request
    priority_score = random.randint(1, 5)
    arrival_mode = random.choice(["walk-in", "ambulance"])

    with servers.request(
        entity_id=patient_id,
        start_event="service_begins",
        end_event="service_complete",
        priority_score=priority_score,   # any extra keyword args are forwarded
        arrival_mode=arrival_mode,       # to BOTH resource_use rows
    ) as request:
        yield request

        service_duration = random.expovariate(1.0/SERVICE_TIME)
        yield env.timeout(service_duration)

    logger.log_departure(entity_id=patient_id)

def run_simulation_auto_extra_fields():
    params = Params()
    env = simpy.Environment()
    logger = EventLogger(env=env)
    servers = VidigiStore(
        env=env, num_resources=params.num_servers, label="server", logger=logger
    )

    def generator(env, servers, logger):
        patient_id = 0
        while True:
            patient_id += 1
            logger.log_arrival(entity_id=patient_id)
            env.process(
                patient_process_auto_extra_fields(env, patient_id, servers, logger)
            )
            yield env.timeout(random.expovariate(ARRIVAL_RATE))

    env.process(generator(env, servers, logger))
    env.run(until=SIM_DURATION)

    return logger

extra_fields_log_df = run_simulation_auto_extra_fields().to_dataframe()
# priority_score and arrival_mode now appear on the resource_use / resource_use_end rows
extra_fields_log_df[
    extra_fields_log_df["event_type"].isin(["resource_use", "resource_use_end"])
].head()

	entity_id	event_type	event	time	resource_id	priority_score	arrival_mode	unique_resource_id
2	1	resource_use	service_begins	0.000000	1.0	1.0	walk-in	server_1
5	2	resource_use	service_begins	0.190484	2.0	3.0	ambulance	server_2
8	3	resource_use	service_begins	0.329604	3.0	3.0	walk-in	server_3
15	1	resource_use_end	service_complete	1.256361	1.0	1.0	walk-in	server_1
17	4	resource_use	service_begins	1.256361	1.0	1.0	walk-in	server_1

The same values land on both events. If you instead need different fields on the start and the end - or a value that isn’t known until the resource is released, such as a service outcome - you have two options.

Option 1: the manual get_direct()/put() pattern. Each call carries its own fields, evaluated at its own time. You give up the with block, so you’re responsible for calling put():

item = yield servers.get_direct(entity_id=patient_id, event="service_begins",
                                triage_priority=priority_score)
yield env.timeout(service_duration)
outcome = "admitted" if ... else "discharged"   # only known now
servers.put(item, entity_id=patient_id, event="service_complete",
            outcome=outcome, tests_ordered=n_tests)

Option 2: keep the with block, but pass auto_log=False and log by hand. This keeps the context manager’s automatic item return while letting you write the log_resource_use_start/log_resource_use_end calls yourself with whatever fields you like. auto_log=False also tells vidigi the missing entity_id is deliberate, so it does not emit the usual “entity_id was not passed” warning:

with servers.request(auto_log=False) as request:   # keep auto-return, log manually
    server = yield request
    logger.log_resource_use_start(
        entity_id=patient_id, resource_id=server.id,
        unique_resource_id=server.unique_id,
        event="service_begins", triage_priority=priority_score)

    yield env.timeout(service_duration)

    outcome = "admitted" if ... else "discharged"   # only known now
    logger.log_resource_use_end(
        entity_id=patient_id, resource_id=server.id,
        unique_resource_id=server.unique_id,
        event="service_complete", outcome=outcome, tests_ordered=n_tests)

auto_log=False is per-call, so other request()/get_direct() calls on the same store keep auto-logging normally. (If a whole resource pool is always logged by hand, it’s simpler not to pass logger= to that store at all - entity_id-less calls on a store with no logger never warn.)

Step 4. Run simulation

# Run simulation and get event log
event_log = run_simulation()
print(f"Generated {len(event_log.to_dataframe())} events")

Generated 549 events

Step 5. Create event positions dataframe

Here, for the service_begins step, we pass in the ‘resource’ parameter, which needs to match one of the attributes in our parameter class. This will then pull back the correct number of resources/servers for that step and display them as icons.

# Define positions for animation
event_positions = create_event_position_df([
    EventPosition(event='arrival', x=0, y=350, label="Entrance"),
    EventPosition(event='queue_wait_begins', x=250, y=250, label="Queue"),
    EventPosition(event='service_begins', x=250, y=150, resource='num_servers', label="Being Served"),
    EventPosition(event='depart', x=250, y=50, label="Exit")
])

Step 6. Create animation

• plotly_height and/or plotly_width set the actual physical size of the resulting plotly canvas in pixels
• override_x_max and/or override_y_max controls the limits of the grid within the plot. If not specified, it will set it to a value that’s just larger than the maximum coordinate that is generated within the animation. However, manually setting the extents of the plot makes it more consistent and can help if we want to later add a background image.
• setup_mode turns on and off the gridlines and coordinate grid values, which are useful when setting up the animation but may not be wanted once you are ready to share the animation
• every_x_time_units defines how often the animation will poll the position of each entity - so e.g. a value of 10 would mean that the animation would progress in jumps of 10 minutes (or whatever the relevant time unit of your simulation is)

We pass our Params() class in so that vidigi knows the number of resource icons it needs to generate.

Tip

There are three main ways to call animate_activity_log

• (not covered here) Via TrialLogger.animate_activity_log()
• Via EventLogger.animate_activity_log() - demonstrated here
• (not covered here) Called directly via from vidigi.animation import animate_activity_log and then call the function as animate_activity_log(event_log=...) where event_log is either
  • a TrialLogger (passing in run_number as well)
  • an EventLogger
  • a dataframe representing the event log

# Create animation
event_log.animate_activity_log(
    event_position_df=event_positions,
    scenario=Params(),
    every_x_time_units=1,
    plotly_height=600,
    override_x_max=360,
    limit_duration=SIM_DURATION
)