Hospital ER Patient Flow

Queuing theory demo of patient flow through an emergency department.

Level:Intermediate

queue healthcare resource-management prioritization

Stocks:beds
Flows:arrivals, departures
Feedback Loops:staffing effects
Probes:wait_low, wait_med, wait_high, bed_util, lwbs, served

simulation.py

Patient flow through a busy ER

This sketch uses SimPy to model an emergency department. Patients arrive randomly, receive an acuity level, then wait on scarce Beds, Nurses, and Doctors. Some cases escalate to Imaging or the OR, creating re-entrant flow. We probe wait times by severity, utilisation of Beds, and how often patients leave before being seen (LWBS). Try bumping the nurse count to see throughput improve.


import random
from tys import probe, progress


def simulate(cfg: dict):

    import simpy

    env = simpy.Environment()
    random.seed(cfg.get("seed", 42))

    arrival_rate = cfg["arrival_rate"]  # mean arrivals per minute
    sim_time = cfg["sim_time"]

    beds = simpy.Resource(env, capacity=cfg["beds"])
    nurses = simpy.Resource(env, capacity=cfg["nurses"])
    doctors = simpy.Resource(env, capacity=cfg["doctors"])
    imaging = simpy.Resource(env, capacity=cfg.get("imaging_capacity", 1))
    operating = simpy.Resource(env, capacity=cfg.get("or_capacity", 1))

    severities = [1, 2, 3]
    severity_probs = cfg["severity_probs"]

    lwbs_threshold = cfg["lwbs_threshold"]  # leave without being seen wait limit

    nurse_time = cfg["nurse_time_mean"]
    doctor_time = cfg["doctor_time_mean"]
    imaging_time = cfg["imaging_time"]
    or_time = cfg["or_time"]

    escalate_img = cfg["escalate_to_imaging"]
    escalate_or = cfg["escalate_to_or"]
    post_img = cfg["post_imaging_bed_time"]
    post_or = cfg["post_or_bed_time"]

    total_patients = 0
    served = 0
    lwbs = 0

    done = env.event()

Record bed utilisation periodically

    def utilisation():
        while True:
            probe("bed_util", env.now, len(beds.users) / beds.capacity)
            progress(min(int(100 * env.now / sim_time), 100))
            yield env.timeout(1)

    env.process(utilisation())

Full visit lifecycle for a single patient

    def patient(name: str):
        nonlocal served, lwbs
        arrival = env.now
        severity = random.choices(severities, weights=severity_probs)[0]

First wait for a bed; impatience leads to LWBS

        with beds.request() as req:
            results = yield req | env.timeout(lwbs_threshold)
            if req not in results:
                lwbs += 1
                probe("lwbs", env.now, lwbs)
                return
            wait = env.now - arrival
            label = {1: "wait_low", 2: "wait_med", 3: "wait_high"}[severity]
            probe(label, env.now, wait)

Nurse assessment

            with nurses.request() as nreq:
                yield nreq
                yield env.timeout(random.expovariate(1.0 / nurse_time))

Doctor treatment

            with doctors.request() as dreq:
                yield dreq
                yield env.timeout(random.expovariate(1.0 / doctor_time))

Possible escalation to imaging

        if random.random() < escalate_img:
            with imaging.request() as img:
                yield img
                yield env.timeout(imaging_time)
            with beds.request() as req2:
                yield req2
                yield env.timeout(post_img)

Possible escalation to OR

        if random.random() < escalate_or:
            with operating.request() as opr:
                yield opr
                yield env.timeout(or_time)
            with beds.request() as req3:
                yield req3
                yield env.timeout(post_or)

        served += 1
        probe("served", env.now, served)

Incoming patient generator

    def arrivals():
        nonlocal total_patients
        i = 0
        while env.now < sim_time:
            env.process(patient(f"Patient{i}"))
            i += 1
            total_patients += 1
            delay = random.expovariate(arrival_rate)
            yield env.timeout(delay)
        done.succeed({"arrived": total_patients, "served": served, "lwbs": lwbs})

    env.process(arrivals())
    env.run(until=done)
    progress(100)
    return done.value


def requirements():
    return {
        "builtin": ["micropip", "pyyaml"],
        "external": ["simpy==4.1.1"],
    }

With Extra Nurse.yaml

arrival_rate: 0.1
sim_time: 480
beds: 5
nurses: 3  # one additional nurse
doctors: 1
imaging_capacity: 1
or_capacity: 1
lwbs_threshold: 60
severity_probs:
  - 0.5
  - 0.3
  - 0.2
nurse_time_mean: 10
doctor_time_mean: 15
imaging_time: 30
or_time: 90
post_imaging_bed_time: 20
post_or_bed_time: 60
escalate_to_imaging: 0.2
escalate_to_or: 0.05
seed: 42

Charts (With Extra Nurse)

bed_util

Samples	483 @ 0.00–482.00
Values	min 0.00, mean 0.89, median 1.00, max 1.00, σ 0.26

wait_low

Samples	14 @ 0.00–461.50
Values	min 0.00, mean 7.49, median 0.00, max 58.86, σ 15.64

served

Samples	20 @ 7.01–469.96
Values	min 1.00, mean 10.50, median 10.50, max 20.00, σ 5.77

wait_med

Samples	5 @ 61.35–469.96
Values	min 0.00, mean 15.86, median 0.00, max 47.68, σ 20.08

wait_high

Samples	7 @ 79.19–381.57
Values	min 0.00, mean 21.37, median 20.02, max 49.62, σ 15.73

lwbs

Samples	17 @ 250.71–460.04
Values	min 1.00, mean 9.00, median 9.00, max 17.00, σ 4.90

Baseline.yaml

arrival_rate: 0.1  # patients per minute (~6 per hour)
sim_time: 480      # minutes (8 hours)
beds: 5
nurses: 2
doctors: 1
imaging_capacity: 1
or_capacity: 1
lwbs_threshold: 60
severity_probs:
  - 0.5  # low
  - 0.3  # medium
  - 0.2  # high
nurse_time_mean: 10
doctor_time_mean: 15
imaging_time: 30
or_time: 90
post_imaging_bed_time: 20
post_or_bed_time: 60
escalate_to_imaging: 0.2
escalate_to_or: 0.05
seed: 42

Charts (Baseline)