89. An emergency department model in SimPy, with patient prioritisation and capacity limited by doctor availability (object based)

ed_object

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)

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out /  Change )

Google+ photo

You are commenting using your Google+ account. Log Out /  Change )

Twitter picture

You are commenting using your Twitter account. Log Out /  Change )

Facebook photo

You are commenting using your Facebook account. Log Out /  Change )

Connecting to %s