Feature Example: Filtering which resource a request is granted

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.

A pool of resources is not always interchangeable. Some nurses are senior, some beds are isolation-capable, one packing station has the heavy-duty press. By default VidigiStore / VidigiPriorityStore give a request whatever unit is at the front of the pool, so there is no way to say “I need a senior nurse”.

filter_fn= closes that gap. Pass a callable that takes one pool unit and returns True to accept it, and the request is granted only a matching unit - queuing until one is free if it has to. Omitting it (the default) is exactly the old behaviour.

with nurses.request(filter_fn=lambda nurse: nurse.grade == "senior") as req:
    nurse = yield req   # guaranteed to be a senior nurse

It works on request(), get(), get_direct() and request_direct() for both store types. This notebook uses the EventLogger and the store’s built-in logger= auto-logging.

Why not just use separate pools?

The obvious alternative is to split the resource into one pool per kind - a junior_nurses store and a senior_nurses store - and route each request with an if. That works for some models, but it has costs that filter_fn avoids:

Eligibility usually overlaps, and overlap doesn’t partition. A senior nurse can take routine patients too; an isolation bed is still a bed. Put seniors in their own pool and you have to decide up front whether each senior is “for high-acuity only” or “in the general pool” - you can’t have them fall back to routine work only when no high-acuity patient is waiting. One pool with a filter keeps every unit available to every request it qualifies for.
Routing has to commit before the wait, not after. With two pools a routine patient must pick a queue to join. If they queue for juniors and a senior frees first, they sit there anyway. Getting “whichever frees first, as long as it’s allowed” means requesting from both pools at once, cancelling the loser, and handling the race - which is exactly the code the original issue reporter was hand-rolling. filter_fn is one request against one queue.
One pool is one set of numbers. Utilisation, queue length, and the animation all read a single resource. Split into N pools and you reconcile N sets of stats, and the animation shows N lanes for what may be one physical group of staff.
Attributes multiply. Grade on its own is two pools; grade x site x speciality is eight, most of them nearly empty. Predicates compose - lambda n: n.grade == "senior" and n.site == "north" - without adding pools.

Separate pools are still the right call when the resources genuinely never substitute for each other, or when each group needs its own priority rules and queue discipline. When the groups are the same resource seen at different resolutions, reach for filter_fn.

import random

import pandas as pd
import plotly.io as pio
import simpy
from sim_tools.distributions import Bernoulli, Exponential, Lognormal

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

pio.renderers.default = "notebook"

A pool whose units are not interchangeable

VidigiStore(num_resources=5, label="nurse") builds five units numbered 1-5 with identical attributes. extra_attributes= can only set the same value on every unit, so to make the pool heterogeneous we populate it normally and then tag each unit’s grade in place - the first three juniors, the last two seniors:

Show the global parameter class

class g:
    n_junior_nurses = 3
    n_senior_nurses = 2

    # 35% of arrivals are high-acuity and must be seen by a senior nurse
    high_acuity_prob = 0.35

    arrival_rate = 5.0  # mean minutes between arrivals
    treat_mean = 18.0  # mean consultation length
    treat_var = 6.0

    sim_duration = 600
    number_of_runs = 10
    base_seed = 42

Show the patient class

class Patient:
    def __init__(self, p_id, high_acuity):
        self.identifier = p_id
        self.high_acuity = high_acuity
        self.acuity = "high" if high_acuity else "routine"

In the model, the pool is built in init_resources() and tagged straight afterwards. The only filter_fn line is in attend_clinic(): a high-acuity patient passes filter_fn=lambda nurse: nurse.grade == "senior"; a routine patient passes nothing and takes whoever is free.

Show the model class

class Model:
    def __init__(self, run_number):
        self.env = simpy.Environment()
        self.run_number = run_number
        self.logger = EventLogger(env=self.env, run_number=run_number)

        seed = (run_number + 1) * g.base_seed
        self.arrival_dist = Exponential(mean=g.arrival_rate, random_seed=seed)
        self.treat_dist = Lognormal(
            mean=g.treat_mean, stdev=g.treat_var, random_seed=seed + 1
        )
        self.acuity_dist = Bernoulli(p=g.high_acuity_prob, random_seed=seed + 2)

        self.patient_counter = 0
        self.init_resources()

    def init_resources(self):
        n = g.n_junior_nurses + g.n_senior_nurses
        self.nurses = VidigiStore(
            self.env, num_resources=n, label="nurse", logger=self.logger
        )
        # Make the pool heterogeneous: tag each unit's grade in place.
        grades = ["junior"] * g.n_junior_nurses + ["senior"] * g.n_senior_nurses
        for nurse, grade in zip(self.nurses.items, grades):
            nurse.grade = grade

    def generator_patient_arrivals(self):
        while True:
            self.patient_counter += 1
            p = Patient(self.patient_counter, bool(self.acuity_dist.sample()))
            self.env.process(self.attend_clinic(p))
            yield self.env.timeout(self.arrival_dist.sample())

    def attend_clinic(self, patient):
        self.logger.log_arrival(entity_id=patient.identifier, pathway=patient.acuity)
        self.logger.log_queue(
            entity_id=patient.identifier,
            event="nurse_wait_begins",
            pathway=patient.acuity,
        )

        # High-acuity patients must be seen by a senior nurse; routine patients
        # take whoever is free.
        need_senior = (
            (lambda nurse: nurse.grade == "senior") if patient.high_acuity else None
        )

        with self.nurses.request(
            entity_id=patient.identifier, pathway=patient.acuity, filter_fn=need_senior
        ) as req:
            yield req  # resource_use / resource_use_end are auto-logged by the store
            yield self.env.timeout(self.treat_dist.sample())

        self.logger.log_departure(entity_id=patient.identifier, pathway=patient.acuity)

    def run(self):
        self.env.process(self.generator_patient_arrivals())
        self.env.run(until=g.sim_duration)
        return self.logger

Show the trial class

class Trial:
    def __init__(self):
        self.all_event_logs = []

    def run_trial(self):
        for run in range(g.number_of_runs):
            random.seed(run)
            self.all_event_logs.append(Model(run).run())
        return self.all_event_logs

trial = Trial()
logs = trial.run_trial()

logs[0].to_dataframe().head(12)

	entity_id	event_type	event	time	pathway	resource_id	unique_resource_id
0	1	arrival_departure	arrival	0.000000	routine	NaN	NaN
1	1	queue	nurse_wait_begins	0.000000	routine	NaN	NaN
2	1	resource_use	nurse_start	0.000000	routine	1.0	nurse_1
3	2	arrival_departure	arrival	12.021043	routine	NaN	NaN
4	2	queue	nurse_wait_begins	12.021043	routine	NaN	NaN
5	2	resource_use	nurse_start	12.021043	routine	2.0	nurse_2
6	1	resource_use_end	nurse_end	18.485130	routine	1.0	nurse_1
7	1	arrival_departure	depart	18.485130	routine	NaN	NaN
8	3	arrival_departure	arrival	23.701991	routine	NaN	NaN
9	3	queue	nurse_wait_begins	23.701991	routine	NaN	NaN
10	3	resource_use	nurse_start	23.701991	routine	3.0	nurse_3
11	2	resource_use_end	nurse_end	33.302228	routine	2.0	nurse_2

nurse_start / nurse_end (and their resource_id) are written for us by the store, because it was constructed with logger= and we passed entity_id= to request().

What the filter changed

High-acuity patients can only use two of the five nurses, so they queue more. Pooling every run, compare the wait from joining the queue to being seen:

waits = []
for lg in logs:
    df = lg.to_dataframe()
    joined = df[df["event"] == "nurse_wait_begins"][["entity_id", "pathway", "time"]]
    seen = df[df["event"] == "nurse_start"][["entity_id", "time"]]
    merged = joined.merge(seen, on="entity_id", suffixes=("_joined", "_seen"))
    merged["wait"] = merged["time_seen"] - merged["time_joined"]
    waits.append(merged)

wait_by_acuity = (
    pd.concat(waits)
    .groupby("pathway")["wait"]
    .agg(["mean", "median", "count"])
    .round(2)
)
wait_by_acuity

	mean	median	count
pathway
high	9.42	8.64	416
routine	1.69	0.00	760

ratio = wait_by_acuity.loc["high", "mean"] / wait_by_acuity.loc["routine", "mean"]
print(
    f"High-acuity patients wait {ratio:.1f}x as long as routine patients on average - "
    f"purely because the senior-nurse filter restricts them to {g.n_senior_nurses} "
    f"of the {g.n_junior_nurses + g.n_senior_nurses} nurses."
)

High-acuity patients wait 5.6x as long as routine patients on average - purely because the senior-nurse filter restricts them to 2 of the 5 nurses.

And the senior nurses carry more of the load - they take every high-acuity patient plus their share of routine ones:

grades = ["junior"] * g.n_junior_nurses + ["senior"] * g.n_senior_nurses
grade_of = {i + 1: grade for i, grade in enumerate(grades)}

use = pd.concat(lg.to_dataframe() for lg in logs)
use = use[use["event"] == "nurse_start"].copy()
use["grade"] = use["resource_id"].map(grade_of)

consultations = use.groupby(["grade", "resource_id"]).size().rename("consultations")
consultations.reset_index()

	grade	resource_id	consultations
0	junior	1.0	210
1	junior	2.0	212
2	junior	3.0	200
3	senior	4.0	279
4	senior	5.0	275

per_nurse = consultations.groupby("grade").mean().round(0)
per_nurse.rename("mean consultations per nurse (10 runs)")

grade
junior    207.0
senior    277.0
Name: mean consultations per nurse (10 runs), dtype: float64

Animating one run

Because the store logged resource_id for us, the animation shows individual nurses - you can watch the two senior nurses (the higher-numbered dots) stay busy while juniors sometimes idle.

class scenario:
    n_nurses = g.n_junior_nurses + g.n_senior_nurses


event_position_df = create_event_position_df(
    [
        EventPosition(event="arrival", x=60, y=200, label="Arrival"),
        EventPosition(
            event="nurse_wait_begins", x=280, y=150, label="Waiting for a nurse"
        ),
        EventPosition(
            event="nurse_start", x=280, y=90, label="With a nurse", resource="n_nurses"
        ),
        EventPosition(event="depart", x=480, y=50, label="Departed"),
    ]
)
event_position_df

	event	x	y	label	resource	direction	flip_icons	resource_icon
0	arrival	60	200	Arrival	NaN	None	None	None
1	nurse_wait_begins	280	150	Waiting for a nurse	NaN	None	None	None
2	nurse_start	280	90	With a nurse	n_nurses	None	None	None
3	depart	480	50	Departed	NaN	None	None	None

animate_activity_log(
    event_log=logs[0],
    event_position_df=event_position_df,
    scenario=scenario(),
    every_x_time_units=5,
    limit_duration=300,
    include_play_button=True,
    entity_icon_size=18,
    resource_icon_size=18,
    gap_between_entities=8,
    gap_between_queue_rows=25,
    plotly_height=500,
    plotly_width=900,
    override_x_max=560,
    override_y_max=260,
)

Priority and filter together

On VidigiPriorityStore, filter_fn and priority combine. A returned unit goes to the highest-priority queued request that accepts it - so a lower-priority request whose filter matches can be served ahead of a higher-priority one whose filter the unit fails. That is the point of the parameter, not a bug.

Here three beds are available; one is isolation-capable. Two routine patients and one infectious patient arrive together and fill them. Then an infectious patient (priority=0, needs the isolation bed) and a routine patient (priority=2, any bed) join the queue:

env = simpy.Environment()
beds = VidigiPriorityStore(env, num_resources=3, label="bed")
for bed, kind in zip(beds.items, ["ward", "ward", "isolation"]):
    bed.kind = kind

timeline = []


def patient(name, priority, needs_isolation, arrive, stay):
    yield env.timeout(arrive)
    only_isolation = (lambda b: b.kind == "isolation") if needs_isolation else None
    timeline.append((env.now, name, "joins queue"))
    with beds.request(priority=priority, filter_fn=only_isolation) as req:
        bed = yield req
        timeline.append(
            (env.now, name, f"admitted to {bed.kind} bed {bed.id_attribute}")
        )
        yield env.timeout(stay)
    timeline.append((env.now, name, "discharged"))


# three patients fill the beds at t=0
env.process(
    patient("A (routine)", priority=1, needs_isolation=False, arrive=0, stay=30)
)
env.process(
    patient("B (routine)", priority=1, needs_isolation=False, arrive=0, stay=20)
)
env.process(
    patient("C (infectious)", priority=0, needs_isolation=True, arrive=0, stay=40)
)
# then the queue builds
env.process(
    patient("D (infectious)", priority=0, needs_isolation=True, arrive=5, stay=10)
)
env.process(
    patient("E (routine)", priority=2, needs_isolation=False, arrive=6, stay=10)
)

env.run()
pd.DataFrame(timeline, columns=["time", "patient", "event"])

	time	patient	event
0	0	A (routine)	joins queue
1	0	B (routine)	joins queue
2	0	C (infectious)	joins queue
3	0	A (routine)	admitted to ward bed 1
4	0	B (routine)	admitted to ward bed 2
5	0	C (infectious)	admitted to isolation bed 3
6	5	D (infectious)	joins queue
7	6	E (routine)	joins queue
8	20	B (routine)	discharged
9	20	E (routine)	admitted to ward bed 2
10	30	A (routine)	discharged
11	30	E (routine)	discharged
12	40	C (infectious)	discharged
13	40	D (infectious)	admitted to isolation bed 3
14	50	D (infectious)	discharged

At t=20 a ward bed frees. D is higher priority but a ward bed fails its filter, so the bed goes to E (lower priority, no filter). D waits until t=40, when the isolation bed frees. Drop the filter_fn and D - being higher priority - would take that ward bed instead.

Recap

filter_fn= on request() / get() / get_direct() / request_direct() grants only a pool unit for which the callable returns True, queuing until one is free.
Omitting it, or passing None, is the pre-existing behaviour exactly.
Build a heterogeneous pool by tagging store.items after num_resources= populates it.
On VidigiPriorityStore, a freed unit goes to the highest-priority waiter that accepts it - priority and filter_fn compose.
A filter_fn that never matches waits forever, like any unsatisfiable request. Mixing filter_fn with a finite capacity smaller than the number of units is not fully supported; the default capacity is infinite.