This model mimics the random arrival of patients at an emergency department (ED). The patients are assessed as being low, medium or high priority. Higher priority patients are always selected next (but do not displace lower priority patients already being seen by a doctor).
This model requires some understanding of object-oriented programming in Python.
There are four classes of object in the model:
1) A global variables class. These are stored directly, and may edited, in the class definition.
2) A model class. There is one instance of this class created. This holds the SimPy model.
3) A patient class. A new instance of this class is triggered with each patient arrival. The patient object holds all relevant information about individual patients (such as their priority level).
4) A resources class. There is one instance of this class created which holds the doctors (if other resources were required they could also be held here).
There is a warm-up period before the auditing of results starts.
"""This simpy model mimics the arrival and treatment of patients in an
emergency department with a limited number of doctors. Patients are generated,
wait for a doc (if none available), take up the doc resources for the time of
consulation, and exit the ED straight after seeing doc. Patients are of three
priorities (1=high to 3=low) and docs will always see higher priority patients
first (though they do not interrupt lower priority patients already being seen).
The model has four classes:
Global_vars: holds global variables
Model: holds model environemnt and th emain model run methods
Patient: each patient is an object instance of this class. This class also holds
a static variable (dictionary) containing all patient objects currently in
the simulation model.
Resources: defines the doc resources required to see/treat patient.
"""
import simpy
import random
import pandas as pd
import matplotlib.pyplot as plt
class Global_vars:
"""Storage of global variables. No object instance created. All times are
in minutes"""
# Simulation run time and warm-up (warm-up is time before audit results are
# collected)
sim_duration = 5000
warm_up = 1000
# Average time between patients arriving
inter_arrival_time = 10
# Number of doctors in ED
number_of_docs = 2
# Time between audits
audit_interval = 100
# Average and standard deviation of time patients spends being treated in ED
# (does not include any queuing time, this si the time a doc is occupied
# with a patient)
appointment_time_mean = 18
appointment_time_sd = 7
# Lists used to store audit results
audit_time = []
audit_patients_in_ED = []
audit_patients_waiting = []
audit_patients_waiting_p1 = []
audit_patients_waiting_p2 = []
audit_patients_waiting_p3 = []
audit_reources_used = []
# Set up dataframes to store results (will be transferred from lists)
patient_queuing_results = pd.DataFrame(
columns=['priority', 'q_time', 'consult_time'])
results = pd.DataFrame()
# Set up counter for number fo aptients entering simulation
patient_count = 0
# Set up running counts of patients waiting (total and by priority)
patients_waiting = 0
patients_waiting_by_priority = [0, 0, 0]
class Model:
"""
Model class contains the following methods:
__init__: constructor for initiating simpy simulation environment.
build_audit_results: At end of model run, transfers results held in lists
into a pandas DataFrame.
chart: At end of model run, plots model results using MatPlotLib.
perform_audit: Called at each audit interval. Records simulation time, total
patients waiting, patients waiting by priority, and number of docs
occupied. Will then schedule next audit.
run: Called immediately after initialising simulation object. This method:
1) Calls method to set up doc resources.
2) Initialises the two starting processes: patient admissions and audit.
3) Starts model envrionment.
4) Save individual patient level results to csv
5) Calls the build_audit_results metha and saves to csv
6) Calls the chart method to plot results
see_doc: After a patient arrives (generated in the trigger_admissions
method of this class), this see_doc process method is called (with
patient object passed to process method). This process requires a free
doc resource (resource objects held in this model class). The request is
prioritised by patient priority (lower priority numbers grab resources
first). The number of patients waiting is incremented, and doc resources
are requested. Once doc resources become available queuing times are
recorded (these are saved to global results if warm up period has been
completed). The patient is held for teh required time with doc (held in
patient object) and then time with doc recorded. The patient is then
removed from the Patient calss dictionary (which triggers Python to
remove the patient object).
trigger_admissions: Generates new patient admissions. Each patient is an
instance of the Patient obect class. This method allocates each
patient an ID, adds the patient to the dictionary of patients held by
the Patient class (static class variable), initiates a simpy process
(in this model class) to see a doc, and schedules the next admission.
"""
def __init__(self):
"""constructor for initiating simpy simulation environment"""
self.env = simpy.Environment()
def build_audit_results(self):
"""At end of model run, transfers results held in lists into a pandas
DataFrame."""
Global_vars.results['time'] = Global_vars.audit_time
Global_vars.results['patients in ED'] = Global_vars.audit_patients_in_ED
Global_vars.results['all patients waiting'] = \
Global_vars.audit_patients_waiting
Global_vars.results['priority 1 patients waiting'] = \
Global_vars.audit_patients_waiting_p1
Global_vars.results['priority 2 patients waiting'] = \
Global_vars.audit_patients_waiting_p2
Global_vars.results['priority 3 patients waiting'] = \
Global_vars.audit_patients_waiting_p3
Global_vars.results['resources occupied'] = \
Global_vars.audit_reources_used
def chart(self):
"""At end of model run, plots model results using MatPlotLib."""
# Define figure size and defintion
fig = plt.figure(figsize=(12, 4.5), dpi=75)
# Create two charts side by side
# Figure 1: patient perspective results
ax1 = fig.add_subplot(131) # 1 row, 3 cols, chart position 1
x = Global_vars.patient_queuing_results.index
# Chart loops through 3 priorites
markers = ['o', 'x', '^']
for priority in range(1, 4):
x = (Global_vars.patient_queuing_results
[Global_vars.patient_queuing_results['priority'] ==
priority].index)
y = (Global_vars.patient_queuing_results
[Global_vars.patient_queuing_results['priority'] ==
priority]['q_time'])
ax1.scatter(x, y,
marker=markers[priority - 1],
label='Priority ' + str(priority))
ax1.set_xlabel('Patient')
ax1.set_ylabel('Queuing time')
ax1.legend()
ax1.grid(True, which='both', lw=1, ls='--', c='.75')
# Figure 2: ED level queuing results
ax2 = fig.add_subplot(132) # 1 row, 3 cols, chart position 2
x = Global_vars.results['time']
y1 = Global_vars.results['priority 1 patients waiting']
y2 = Global_vars.results['priority 2 patients waiting']
y3 = Global_vars.results['priority 3 patients waiting']
y4 = Global_vars.results['all patients waiting']
ax2.plot(x, y1, marker='o', label='Priority 1')
ax2.plot(x, y2, marker='x', label='Priority 2')
ax2.plot(x, y3, marker='^', label='Priority 3')
ax2.plot(x, y4, marker='s', label='All')
ax2.set_xlabel('Time')
ax2.set_ylabel('Patients waiting')
ax2.legend()
ax2.grid(True, which='both', lw=1, ls='--', c='.75')
# Figure 3: ED staff usage
ax3 = fig.add_subplot(133) # 1 row, 3 cols, chart position 3
x = Global_vars.results['time']
y = Global_vars.results['resources occupied']
ax3.plot(x, y, label='Docs occupied')
ax3.set_xlabel('Time')
ax3.set_ylabel('Doctors occupied')
ax3.legend()
ax3.grid(True, which='both', lw=1, ls='--', c='.75')
# Create plot
plt.tight_layout(pad=3)
plt.show()
def perform_audit(self):
"""Called at each audit interval. Records simulation time, total
patients waiting, patients waiting by priority, and number of docs
occupied. Will then schedule next audit."""
# Delay before first aurdit if length of warm-up
yield self.env.timeout(Global_vars.warm_up)
# The trigger repeated audits
while True:
# Record time
Global_vars.audit_time.append(self.env.now)
# Record patients waiting by referencing global variables
Global_vars.audit_patients_waiting.append(
Global_vars.patients_waiting)
Global_vars.audit_patients_waiting_p1.append(
Global_vars.patients_waiting_by_priority[0])
Global_vars.audit_patients_waiting_p2.append(
Global_vars.patients_waiting_by_priority[1])
Global_vars.audit_patients_waiting_p3.append(
Global_vars.patients_waiting_by_priority[2])
# Record patients waiting by asking length of dictionary of all
# patients (another way of doing things)
Global_vars.audit_patients_in_ED.append(len(Patient.all_patients))
# Record resources occupied
Global_vars.audit_reources_used.append(
self.doc_resources.docs.count)
# Trigger next audit after interval
yield self.env.timeout(Global_vars.audit_interval)
def run(self):
"""Called immediately after initialising simulation object. This method:
1) Calls method to set up doc resources.
2) Initialises the two starting processes: patient admissions and audit.
3) Starts model envrionment.
4) Save individual patient level results to csv
5) Calls the build_audit_results metha and saves to csv
6) Calls the chart method to plot results
"""
# Set up resources using Resouces class
self.doc_resources = Resources(self.env, Global_vars.number_of_docs)
# Initialise processes that will run on model run
self.env.process(self.trigger_admissions())
self.env.process(self.perform_audit())
# Run
self.env.run(until=Global_vars.sim_duration)
# End of simulation run. Build and save results
Global_vars.patient_queuing_results.to_csv('patient results.csv')
self.build_audit_results()
Global_vars.results.to_csv('operational results.csv')
# plot results
self.chart()
def see_doc(self, p):
"""After a patient arrives (generated in the trigger_admissions
method of this class), this see_doc process method is called (with
patient object passed to process method). This process requires a free
doc resource (resource objects held in this model class). The request is
prioritised by patient priority (lower priority numbers grab resources
first). The number of patients waiting is incremented, and doc resources
are requested. Once doc resources become available queuing times are
recorded (these are saved to global results if warm up period has been
completed). The patient is held for teh required time with doc (held in
patient object) and then time with doc recorded. The patient is then
removed from the Patient calss dictionary (which triggers Python to
remove the patient object).
"""
# See doctor requires doc_resources
with self.doc_resources.docs.request(priority=p.priority) as req:
# Increment count of number of patients waiting. 1 is subtracted
# from priority to align priority (1-3) with zero indexed list.
Global_vars.patients_waiting += 1
Global_vars.patients_waiting_by_priority[p.priority - 1] += 1
# Wait for resources to become available
yield req
# Resources now available. Record time patient starts to see doc
p.time_see_doc = self.env.now
# Record patient queuing time in patient object
p.queuing_time = self.env.now - p.time_in
# Reduce count of number of patients (waiting)
Global_vars.patients_waiting_by_priority[p.priority - 1] -= 1
Global_vars.patients_waiting -= 1
# Create a temporary results list with patient priority and queuing
# time
_results = [p.priority, p.queuing_time]
# Hold patient (with doc) for consulation time required
yield self.env.timeout(p.consulation_time)
# At end of consultation add time spent with doc to temp results
_results.append(self.env.now - p.time_see_doc)
# Record results in global results data if warm-up complete
if self.env.now >= Global_vars.warm_up:
Global_vars.patient_queuing_results.loc[p.id] = _results
# Delete patient (removal from patient dictionary removes only
# reference to patient and Python then automatically cleans up)
del Patient.all_patients[p.id]
def trigger_admissions(self):
"""Generates new patient admissions. Each patient is an instance of the
Patient obect class. This method allocates each patient an ID, adds the
patient to the dictionary of patients held by the Patient class (static
class variable), initiates a simpy process (in this model class) to see
a doc, and then schedules the next admission"""
# While loop continues generating new patients throughout model run
while True:
# Initialise new patient (pass environment to be used to record
# current simulation time)
p = Patient(self.env)
# Add patient to dictionary of patients
Patient.all_patients[p.id] = p
# Pass patient to see_doc method
self.env.process(self.see_doc(p))
# Sample time for next asmissions
next_admission = random.expovariate(
1 / Global_vars.inter_arrival_time)
# Schedule next admission
yield self.env.timeout(next_admission)
class Patient:
"""The Patient class is for patient objects. Each patient is an instance of
this class. This class also holds a static dictionary which holds all
patient objects (a patient is removed after exiting ED).
Methods are:
__init__: constructor for new patient
"""
# The following static class dictionary stores all patient objects
# This is not actually used further but shows how patients may be tracked
all_patients = {}
def __init__(self, env):
"""Constructor for new patient object.
"""
# Increment global counts of patients
Global_vars.patient_count += 1
# Set patient id and priority (random between 1 and 3)
self.id = Global_vars.patient_count
self.priority = random.randint(1, 3)
# Set consultation time (time spent with doc) by random normal
# distribution. If value <0 then set to 0
self.consulation_time = random.normalvariate(
Global_vars.appointment_time_mean, Global_vars.appointment_time_sd)
self.consulation_time = 0 if self.consulation_time < 0 \
else self.consulation_time
# Set initial queuing time as zero (this will be adjusted in model if
# patient has to waiti for doc)
self.queuing_time = 0
# record simulation time patient enters simulation
self.time_in = env.now
# Set up variables to record simulation time that patient see doc and
# exit simulation
self.time_see_doc = 0
self.time_out = 0
class Resources:
"""Resources class for simpy. Only resource used is docs"""
def __init__(self, env, number_of_docs):
self.docs = simpy.PriorityResource(env, capacity=number_of_docs)
# Run model
if __name__ == '__main__':
# Initialise model environment
model = Model()
# Run model
model.run()
One thought on “89. An emergency department model in SimPy, with patient prioritisation and capacity limited by doctor availability (object based)”