# Class representing patients coming in to the clinic. Here, patients have
# two attributes that they carry with them - their ID, and the amount of time
# they spent queuing for the nurse. The ID is passed in when a new patient is
# created.
class Patient:
def __init__(self, p_id):
self.id = p_id
self.q_time_nurse = 0
self.time_with_nurse = 0 ## NEW17 Tracking Resource Utilisation
When running a model, it’s useful to be able to track a few things about your resources.
- How much are resources being utilised overall
- How much are resources being utilised at specific points in time
For example, it would be useful to know that your nurse resources are being utilised 89% of the time.
However, it may also be useful to know that due to fluctuations in demand throughout the day, at 4pm the average utilisation across several runs is 60%, but at 8pm the average utilisation is at 95%.
Monitoring utilisation of resources in both of these ways can be helpful to show the performance of your model.
Tip
You often don’t want utilisation to be too close to 100% - though exactly where you want utilisation to sit can depend a lot on the kind of service you run, and how big it is.
In services where rapid access is crucial - such as a cancer service, a high dependency unit, a crisis hotline, or an emergency department - a slightly lower average utilisation may be seen, but could be necessary to safely cope with spikes in demand. At that point, you may need to look elsewhere in the system to see if it is possible to make that demand less ‘spiky’, or whether it is just an inherent feature of that kind of work that can’t be avoided.
Chasing very high utilisation figures can also be a false economy - for example, is it cheaper to have a 12 bed mental health ward that usually runs at an average 70% occupancy, or a 9 bed ward running at an average 95% occupancy, but with high levels of staff burnout, infection control issues, and a constant need to purchase additional capacity from the private sector or place patients in other wards owned by the trust, which are further from the patient’s home and support network, potentially slowing recovery and leading to a longer stay, and also incurring additional costs for patient transport. While you don’t want lots of beds sitting empty, understanding the knock-on effects can be crucial to ensuring utilisation metrics are used and interpreted well - and this often requires you having conversations with your stakeholders to understand the wider implications.
Keep in mind that they themselves may not be that confident with using anything other than ‘average demand’ for capacity planning. This video from the demand and capacity team (formerly part of NHS England) is a great resource for helping stakeholders to understand the impact of variation in demand and how it might impact their decisions on the capacity planning of the service.
Note that this is slightly different from the resource utilisation, but the ideas all tie together!
17.1 Looking at average resource utilisation across a run
To calculate the percentage of time a resource was being utilised overall, we need to start tracking how long each entity in our model spent using a resource.
What we can then do is - add this up across every entity who passes through our model - divide it by the total amount of time that has elapsed in the model, multiplied by the number of resources in the model
We can track this by adding an additional attribute to our patient class.
Let’s return to the very simplest version of the model for this.
17.1.1 A code example
17.1.1.1 The g class
The g class is unchanged.
17.1.1.2 The patient class
First, we add an attribute to the patient class.
17.1.1.3 The model class
17.1.1.3.1 The init method
Here we’re just going to add an empty list to store our patient objects.
We’re also going to add a space to store the utilisation in that run.
def __init__(self, run_number):
# Create a SimPy environment in which everything will live
self.env = simpy.Environment()
# Create a patient counter (which we'll use as a patient ID)
self.patient_counter = 0
self.patient_objects = [] ##NEW
# Create a SimPy resource to represent a nurse, that will live in the
# environment created above. The number of this resource we have is
# specified by the capacity, and we grab this value from our g class.
self.nurse = simpy.Resource(self.env, capacity=g.number_of_nurses)
# Store the passed in run number
self.run_number = run_number
# Create a new Pandas DataFrame that will store some results against
# the patient ID (which we'll use as the index).
self.results_df = pd.DataFrame()
self.results_df["Patient ID"] = [1]
self.results_df["Q Time Nurse"] = [0.0]
self.results_df["Time with Nurse"] = [0.0]
self.results_df.set_index("Patient ID", inplace=True)
# Create an attribute to store the mean queuing time for the nurse
# across this run of the model
self.mean_q_time_nurse = 0
self.nurse_utilisation = 0.0 ##NEW17.1.1.3.2 The generate_patient_arrivals method
Here we just want to add a line to add the patient objects to our patient_objects list when they are created.
Don’t worry that we haven’t yet set the attributes of the patient throughout their journeys - when we access the patient list at the very end of the model runs, we will get the most up-to-date version of the attribute for each patient.
def generator_patient_arrivals(self):
# We use an infinite loop here to keep doing this indefinitely whilst
# the simulation runs
while True:
# Increment the patient counter by 1 (this means our first patient
# will have an ID of 1)
self.patient_counter += 1
# Create a new patient - an instance of the Patient Class we
# defined above. Remember, we pass in the ID when creating a
# patient - so here we pass the patient counter to use as the ID.
p = Patient(self.patient_counter)
self.patient_objects.append(p) ##NEW
# Tell SimPy to start up the attend_clinic generator function with
# this patient (the generator function that will model the
# patient's journey through the system)
self.env.process(self.attend_clinic(p))
# Randomly sample the time to the next patient arriving. Here, we
# sample from an exponential distribution (common for inter-arrival
# times), and pass in a lambda value of 1 / mean. The mean
# inter-arrival time is stored in the g class.
sampled_inter = random.expovariate(1.0 / g.patient_inter)
# Freeze this instance of this function in place until the
# inter-arrival time we sampled above has elapsed. Note - time in
# SimPy progresses in "Time Units", which can represent anything
# you like (just make sure you're consistent within the model)
yield self.env.timeout(sampled_inter)17.1.1.3.3 The attend_clinic method
In here, we will add in a step to set the value of the time_with_nurse attribute of the patient class.
def attend_clinic(self, patient):
# Record the time the patient started queuing for a nurse
start_q_nurse = self.env.now
# This code says request a nurse resource, and do all of the following
# block of code with that nurse resource held in place (and therefore
# not usable by another patient)
with self.nurse.request() as req:
# Freeze the function until the request for a nurse can be met.
# The patient is currently queuing.
yield req
# When we get to this bit of code, control has been passed back to
# the generator function, and therefore the request for a nurse has
# been met. We now have the nurse, and have stopped queuing, so we
# can record the current time as the time we finished queuing.
end_q_nurse = self.env.now
# Calculate the time this patient was queuing for the nurse, and
# record it in the patient's attribute for this.
patient.q_time_nurse = end_q_nurse - start_q_nurse
# Now we'll randomly sample the time this patient with the nurse.
# Here, we use an Exponential distribution for simplicity, but you
# would typically use a Log Normal distribution for a real model
# (we'll come back to that). As with sampling the inter-arrival
# times, we grab the mean from the g class, and pass in 1 / mean
# as the lambda value.
sampled_nurse_act_time = random.expovariate(1.0 /
g.mean_n_consult_time)
# Here we'll store the queuing time for the nurse and the sampled
# time to spend with the nurse in the results DataFrame against the
# ID for this patient. In real world models, you may not want to
# bother storing the sampled activity times - but as this is a
# simple model, we'll do it here.
# We use a handy property of pandas called .at, which works a bit
# like .loc. .at allows us to access (and therefore change) a
# particular cell in our DataFrame by providing the row and column.
# Here, we specify the row as the patient ID (the index), and the
# column for the value we want to update for that patient.
self.results_df.at[patient.id, "Q Time Nurse"] = (
patient.q_time_nurse)
self.results_df.at[patient.id, "Time with Nurse"] = (
sampled_nurse_act_time)
patient.time_with_nurse = sampled_nurse_act_time ##NEW
# Freeze this function in place for the activity time we sampled
# above. This is the patient spending time with the nurse.
yield self.env.timeout(sampled_nurse_act_time)
# When the time above elapses, the generator function will return
# here. As there's nothing more that we've written, the function
# will simply end. This is a sink. We could choose to add
# something here if we wanted to record something - e.g. a counter
# for number of patients that left, recording something about the
# patients that left at a particular sink etc.17.1.1.3.4 A new method: audit_utilisation
In our model class, we can now create a new method.
Remember that thanks to the additional attribute we added to the model class, we have now stored the time_with_nurse for each patient and can easily access it.
As we are only tracking a single resource in this model, we could write our function like this:
def audit_utilisation(self):
activity_durations = [i.time_with_nurse for i in self.patient_objects]
return (sum(activity_durations) / (g.number_of_nurses * g.sim_duration))
Tip
A more generalisable form of this function is given below.
def audit_utilisation(self, activity_attribute, resource_attribute):
activity_durations = [i.getattr(activity_attribute) for i in self.patient_objects]
return (sum(activity_durations) / (g.getattr(resource_attribute) * g.sim_duration))This will then be called like this.
self.audit_utilisation("time_with_nurse", "number_of_nurses")If we were also recording time with a doctor, we might call that as
self.audit_utilisation("time_with_doctor", "number_of_doctors")17.1.1.3.5 The calculate_run_results method
Now we are going to use this new method to add the utilisation metric to our results dataframe.
def calculate_run_results(self):
# Take the mean of the queuing times for the nurse across patients in
# this run of the model.
self.mean_q_time_nurse = self.results_df["Q Time Nurse"].mean()
# This assumes we're not using the generalised version of the function
# which doesn't take any arguments
self.nurse_utilisation = self.audit_utilisation() ##NEWFinally, we can put this all together when we run the model.
17.1.1.4 The trial class
17.1.1.4.1 The init method
First we add a
def __init__(self):
self.df_trial_results = pd.DataFrame()
self.df_trial_results["Run Number"] = [0]
self.df_trial_results["Mean Q Time Nurse"] = [0.0]
self.df_trial_results["Average Nurse Utilisation"] = [0.0] ##NEW
self.df_trial_results.set_index("Run Number", inplace=True)17.1.1.4.2 The run_trial class
Finally, we add this to the output dataframe.
Note that we have had to tweak the step for adding results to our dataframe as there are now two possible columns for the result to go in - previously we could get away with just specifying the row in loc, but now we have to pass the column name too.
def run_trial(self):
# Run the simulation for the number of runs specified in g class.
# For each run, we create a new instance of the Model class and call its
# run method, which sets everything else in motion. Once the run has
# completed, we grab out the stored run results (just mean queuing time
# here) and store it against the run number in the trial results
# dataframe.
for run in range(g.number_of_runs):
my_model = Model(run)
my_model.run()
self.df_trial_results.loc[run, "Mean Q Time Nurse"] = [my_model.mean_q_time_nurse] ##EDITED
self.df_trial_results.loc[run, "Average Nurse Utilisation"] = [my_model.nurse_utilisation] ##NEW
# Once the trial (ie all runs) has completed, print the final results
self.print_trial_results()
Important
This method may slightly overestimate the utilisation.
Why?
Because we record the time spent with the resource prior to that time elapsing.
If we have a model run time of 600 time units, and someone reaches the nurse at unit 595 but has an activity time of 30, we will record that and use it in our calculations - even though only 5 minutes of that time actually took place during our model run.
You could adapt the code above to account for that - for example, you could check whether the time remaining in the model run (g.sim_duration - self.env.now) is greater than the activity time that is sampled, and record whichever is the smaller of the two as the patient attribute.
17.1.2 The full code
Click here to view the full code
import simpy
import random
import pandas as pd
# Class to store global parameter values. We don't create an instance of this
# class - we just refer to the class blueprint itself to access the numbers
# inside.
class g:
patient_inter = 5
mean_n_consult_time = 6
number_of_nurses = 2
sim_duration = 1200
number_of_runs = 5
# Class representing patients coming in to the clinic. Here, patients have
# two attributes that they carry with them - their ID, and the amount of time
# they spent queuing for the nurse. The ID is passed in when a new patient is
# created.
class Patient:
def __init__(self, p_id):
self.id = p_id
self.q_time_nurse = 0
self.time_with_nurse = 0 ## NEW
# Class representing our model of the clinic.
class Model:
# Constructor to set up the model for a run. We pass in a run number when
# we create a new model.
def __init__(self, run_number):
# Create a SimPy environment in which everything will live
self.env = simpy.Environment()
# Create a patient counter (which we'll use as a patient ID)
self.patient_counter = 0
self.patient_objects = [] ## NEW
# Create a SimPy resource to represent a nurse, that will live in the
# environment created above. The number of this resource we have is
# specified by the capacity, and we grab this value from our g class.
self.nurse = simpy.Resource(self.env, capacity=g.number_of_nurses)
# Store the passed in run number
self.run_number = run_number
# Create a new Pandas DataFrame that will store some results against
# the patient ID (which we'll use as the index).
self.results_df = pd.DataFrame()
self.results_df["Patient ID"] = [1]
self.results_df["Q Time Nurse"] = [0.0]
self.results_df["Time with Nurse"] = [0.0]
self.results_df.set_index("Patient ID", inplace=True)
# Create an attribute to store the mean queuing time for the nurse
# across this run of the model
self.mean_q_time_nurse = 0
self.nurse_utilisation = 0.0 ##NEW
# A generator function that represents the DES generator for patient
# arrivals
def generator_patient_arrivals(self):
# We use an infinite loop here to keep doing this indefinitely whilst
# the simulation runs
while True:
# Increment the patient counter by 1 (this means our first patient
# will have an ID of 1)
self.patient_counter += 1
# Create a new patient - an instance of the Patient Class we
# defined above. Remember, we pass in the ID when creating a
# patient - so here we pass the patient counter to use as the ID.
p = Patient(self.patient_counter)
self.patient_objects.append(p) ##NEW
# Tell SimPy to start up the attend_clinic generator function with
# this patient (the generator function that will model the
# patient's journey through the system)
self.env.process(self.attend_clinic(p))
# Randomly sample the time to the next patient arriving. Here, we
# sample from an exponential distribution (common for inter-arrival
# times), and pass in a lambda value of 1 / mean. The mean
# inter-arrival time is stored in the g class.
sampled_inter = random.expovariate(1.0 / g.patient_inter)
# Freeze this instance of this function in place until the
# inter-arrival time we sampled above has elapsed. Note - time in
# SimPy progresses in "Time Units", which can represent anything
# you like (just make sure you're consistent within the model)
yield self.env.timeout(sampled_inter)
# A generator function that represents the pathway for a patient going
# through the clinic. Here the pathway is extremely simple - a patient
# arrives, waits to see a nurse, and then leaves.
# The patient object is passed in to the generator function so we can
# extract information from / record information to it
def attend_clinic(self, patient):
# Record the time the patient started queuing for a nurse
start_q_nurse = self.env.now
# This code says request a nurse resource, and do all of the following
# block of code with that nurse resource held in place (and therefore
# not usable by another patient)
with self.nurse.request() as req:
# Freeze the function until the request for a nurse can be met.
# The patient is currently queuing.
yield req
# When we get to this bit of code, control has been passed back to
# the generator function, and therefore the request for a nurse has
# been met. We now have the nurse, and have stopped queuing, so we
# can record the current time as the time we finished queuing.
end_q_nurse = self.env.now
# Calculate the time this patient was queuing for the nurse, and
# record it in the patient's attribute for this.
patient.q_time_nurse = end_q_nurse - start_q_nurse
# Now we'll randomly sample the time this patient with the nurse.
# Here, we use an Exponential distribution for simplicity, but you
# would typically use a Log Normal distribution for a real model
# (we'll come back to that). As with sampling the inter-arrival
# times, we grab the mean from the g class, and pass in 1 / mean
# as the lambda value.
sampled_nurse_act_time = random.expovariate(1.0 /
g.mean_n_consult_time)
# Here we'll store the queuing time for the nurse and the sampled
# time to spend with the nurse in the results DataFrame against the
# ID for this patient. In real world models, you may not want to
# bother storing the sampled activity times - but as this is a
# simple model, we'll do it here.
# We use a handy property of pandas called .at, which works a bit
# like .loc. .at allows us to access (and therefore change) a
# particular cell in our DataFrame by providing the row and column.
# Here, we specify the row as the patient ID (the index), and the
# column for the value we want to update for that patient.
self.results_df.at[patient.id, "Q Time Nurse"] = (
patient.q_time_nurse)
self.results_df.at[patient.id, "Time with Nurse"] = (
sampled_nurse_act_time)
patient.time_with_nurse = sampled_nurse_act_time ##NEW
# Freeze this function in place for the activity time we sampled
# above. This is the patient spending time with the nurse.
yield self.env.timeout(sampled_nurse_act_time)
# When the time above elapses, the generator function will return
# here. As there's nothing more that we've written, the function
# will simply end. This is a sink. We could choose to add
# something here if we wanted to record something - e.g. a counter
# for number of patients that left, recording something about the
# patients that left at a particular sink etc.
def audit_utilisation(self):
activity_durations = [i.time_with_nurse for i in self.patient_objects]
return (sum(activity_durations) / (g.number_of_nurses * g.sim_duration))
# This method calculates results over a single run. Here we just calculate
# a mean, but in real world models you'd probably want to calculate more.
def calculate_run_results(self):
# Take the mean of the queuing times for the nurse across patients in
# this run of the model.
self.mean_q_time_nurse = self.results_df["Q Time Nurse"].mean()
self.nurse_utilisation = self.audit_utilisation() ##NEW
# The run method starts up the DES entity generators, runs the simulation,
# and in turns calls anything we need to generate results for the run
def run(self):
# Start up our DES entity generators that create new patients. We've
# only got one in this model, but we'd need to do this for each one if
# we had multiple generators.
self.env.process(self.generator_patient_arrivals())
# Run the model for the duration specified in g class
self.env.run(until=g.sim_duration)
# Now the simulation run has finished, call the method that calculates
# run results
self.calculate_run_results()
# We are not printing the patient-level results in this case
# so this code has been removed
class Trial:
# The constructor sets up a pandas dataframe that will store the key
# results from each run (just the mean queuing time for the nurse here)
# against run number, with run number as the index.
def __init__(self):
self.df_trial_results = pd.DataFrame()
self.df_trial_results["Run Number"] = [0]
self.df_trial_results["Mean Q Time Nurse"] = [0.0]
self.df_trial_results["Average Nurse Utilisation"] = [0.0] ##NEW
self.df_trial_results.set_index("Run Number", inplace=True)
# Method to print out the results from the trial. In real world models,
# you'd likely save them as well as (or instead of) printing them
def print_trial_results(self):
print ("Trial Results")
print (self.df_trial_results)
# Method to run a trial
def run_trial(self):
# Run the simulation for the number of runs specified in g class.
# For each run, we create a new instance of the Model class and call its
# run method, which sets everything else in motion. Once the run has
# completed, we grab out the stored run results (just mean queuing time
# here) and store it against the run number in the trial results
# dataframe.
for run in range(g.number_of_runs):
my_model = Model(run)
my_model.run()
self.df_trial_results.loc[run, "Mean Q Time Nurse"] = [my_model.mean_q_time_nurse] ##EDITED
self.df_trial_results.loc[run, "Average Nurse Utilisation"] = [my_model.nurse_utilisation] ##NEW
# Once the trial (ie all runs) has completed, print the final results
self.print_trial_results()17.1.3 Evaluating the output
# Create an instance of the Trial class
my_trial = Trial()
# Call the run_trial method of our Trial object
my_trial.run_trial()Trial Results
Mean Q Time Nurse Average Nurse Utilisation
Run Number
0 2.760811 0.637456
1 4.501722 0.648550
2 2.332940 0.565586
3 1.610495 0.574371
4 5.404386 0.657577
Tip
A nice enhancement to this output would be to format the average nurse utilisation column as a percentage rather than keeping it in decimal format.
17.2 Interval Audits
Another option for monitoring resource usage is to set up a new interval audit process.
We create a new interval_audit_utilisation() function that we pass - a list of resources we want to monitor - a time interval at which to snapshot the utilisation in the process
17.2.1 A code example
17.2.1.1 The g class
We add a parameter to the g class which will tell us how often to record the current utilisation.
# Class to store global parameter values. We don't create an instance of this
# class - we just refer to the class blueprint itself to access the numbers
# inside.
class g:
patient_inter = 5
mean_n_consult_time = 35
number_of_nurses = 9
sim_duration = 600
number_of_runs = 5
audit_interval = 5 ##NEW17.2.1.2 The Model Class
17.2.1.2.1 The init method
First, we create an empty list to hold our utilisation audit in our Model class.
def __init__(self, run_number):
# Create a SimPy environment in which everything will live
self.env = simpy.Environment()
# Create a patient counter (which we'll use as a patient ID)
self.patient_counter = 0
# Create a SimPy resource to represent a nurse, that will live in the
# environment created above. The number of this resource we have is
# specified by the capacity, and we grab this value from our g class.
self.nurse = simpy.Resource(self.env, capacity=g.number_of_nurses)
# Store the passed in run number
self.run_number = run_number
# Create a new Pandas DataFrame that will store some results against
# the patient ID (which we'll use as the index).
self.results_df = pd.DataFrame()
self.results_df["Patient ID"] = [1]
self.results_df["Q Time Nurse"] = [0.0]
self.results_df["Time with Nurse"] = [0.0]
self.results_df.set_index("Patient ID", inplace=True)
# Create an attribute to store the mean queuing time for the nurse
# across this run of the model
self.mean_q_time_nurse = 0
self.utilisation_audit = [] ##NEW17.2.1.2.2 A new method - interval_audit_utilisation
Then we add the interval_audit_utilisation function to our Model class.
We set it up so either a list of resources can be passed as a dictionary, or just a single simpy resource object.
def interval_audit_utilisation(self, resources, interval=1):
'''
Record utilisation at defined intervals.
Needs to be passed to env.process when running model.
Parameters:
------
resource: SimPy resource object
The resource to monitor
OR
a list of dictionaries containing simpy resource objects in the format
[{'resource_name':'my_resource', 'resource_object': resource},
{'resource_name':'my_second_resource', 'resource_object': resource_2}
]
where resource and resource_2 in the examples above are simpy resources.
interval: int:
Time between audits.
In simpy time units
Default: 1
'''
# Keep doing the below as long as the simulation is running
while True:
# Code for what to do if we pass in a list of resources
if isinstance(resources, list):
for i in range(len(resources)):
self.utilisation_audit.append({
'resource_name': resources[i]['resource_name'], # The provided name for the resource
'simulation_time': self.env.now, # The current simulation time
'number_utilised': len(resources[i]['resource_object'].count), # The number of users
'number_available': resources[i]['resource_object'].capacity, # The total resource available,
'queue_length': len(resources[i]['resource_object'].queue)
})
else:
# Code for what to do if we just pass in a single resource to monitor
self.utilisation_audit.append({
'simulation_time': self.env.now, # The current simulation time
'number_utilised': resources.count, # The number of users
'number_available': resources.capacity, # The total resource available
'queue_length': len(resources.queue)
})
# Trigger next audit after desired interval has passed.
yield self.env.timeout(interval)
Tip
If you are using a simpy PriorityResource instead of a standard resource, the code should work the same.
If you are using a simpy store instead of a simpy resource, it’s possibly to adapt by using len(resources.items) to get the number of resources currently being utilised, and resources.capacity will still return the total number of resources that are in the system.
17.2.1.2.3 the run method
We now add the interval audit process to our run.
Remember - this will update the self.utilisation_audit in our method class.
Here we have passed in the nurse resource that gets created during the init method of the model class.
# The run method starts up the DES entity generators, runs the simulation,
# and in turns calls anything we need to generate results for the run
def run(self):
# Start up our DES entity generators that create new patients. We've
# only got one in this model, but we'd need to do this for each one if
# we had multiple generators.
self.env.process(
self.generator_patient_arrivals()
)
self.env.process(
self.interval_audit_utilisation(resources=self.nurse,interval=g.audit_interval) ##NEW
)
# Run the model for the duration specified in g class
self.env.run(until=g.sim_duration)
# Now the simulation run has finished, call the method that calculates
# run results
self.calculate_run_results()
# Print the run number with the patient-level results from this run of
# the model
# EDIT: We've skipped printing the individual-level results here
# print (f"Run Number {self.run_number}")
# print (self.results_df)
Tip
If we wanted to conduct an interval audit of multiple resources, we could do it like this:
self.env.process(
self.interval_audit_utilisation(
resources= [
{'resource_name':'receptionists', 'resource_object': self.receptionist},
{'resource_name':'nurses', 'resource_object': self.nurse},
{'resource_name':'doctors', 'resource_object': self.doctor}
],
interval=g.audit_interval) ##NEW
)This assumes our additional resources were set up in the init method of the model class like so:
self.receptionist = simpy.Resource(
self.env, capacity=g.number_of_receptionists
)
self.nurse = simpy.Resource(
self.env, capacity=g.number_of_nurses
)
self.doctor = simpy.Resource(
self.env, capacity=g.number_of_doctors
)17.2.1.3 The trial class
17.2.1.3.1 The init method
We can add an empty list where we will store all of the interval audits
class Trial:
# The constructor sets up a pandas dataframe that will store the key
# results from each run (just the mean queuing time for the nurse here)
# against run number, with run number as the index.
def __init__(self):
self.df_trial_results = pd.DataFrame()
self.df_trial_results["Run Number"] = [0]
self.df_trial_results["Mean Q Time Nurse"] = [0.0]
self.df_trial_results.set_index("Run Number", inplace=True)
self.interval_audit_list = [] ##NEW17.2.1.3.2 The run_trial method
# Method to run a trial
def run_trial(self):
# Run the simulation for the number of runs specified in g class.
# For each run, we create a new instance of the Model class and call its
# run method, which sets everything else in motion. Once the run has
# completed, we grab out the stored run results (just mean queuing time
# here) and store it against the run number in the trial results
# dataframe.
for run in range(g.number_of_runs):
my_model = Model(run)
my_model.run() ##EDITED
self.df_trial_results.loc[run] = [my_model.mean_q_time_nurse]
interval_audit_df = pd.DataFrame(my_model.utilisation_audit)
interval_audit_df["run"] = run
interval_audit_df["perc_utilisation"] = (
interval_audit_df["number_utilised"]/interval_audit_df["number_available"]
)
self.interval_audit_list.append(interval_audit_df)
# For now, we're going to skip printing the trial results, so the line
# below has been commented out
# self.print_trial_results()17.2.1.3.3 New method - get_interval_audits
Finally we are going to create a new method that just joins all of the interval audit dataframes together and returns them as a single dataframe.
def get_interval_audits(self):
return pd.concat(self.interval_audit_list)17.2.2 The full code
Click here to view the full code
# Class to store global parameter values. We don't create an instance of this
# class - we just refer to the class blueprint itself to access the numbers
# inside.
class g:
patient_inter = 5
mean_n_consult_time = 35
number_of_nurses = 9
sim_duration = 600
number_of_runs = 5
audit_interval = 5 ##NEW
# Class representing patients coming in to the clinic. Here, patients have
# two attributes that they carry with them - their ID, and the amount of time
# they spent queuing for the nurse. The ID is passed in when a new patient is
# created.
class Patient:
def __init__(self, p_id):
self.id = p_id
self.q_time_nurse = 0
# Class representing our model of the clinic.
class Model:
# Constructor to set up the model for a run. We pass in a run number when
# we create a new model.
def __init__(self, run_number):
# Create a SimPy environment in which everything will live
self.env = simpy.Environment()
# Create a patient counter (which we'll use as a patient ID)
self.patient_counter = 0
# Create a SimPy resource to represent a nurse, that will live in the
# environment created above. The number of this resource we have is
# specified by the capacity, and we grab this value from our g class.
self.nurse = simpy.Resource(self.env, capacity=g.number_of_nurses)
# Store the passed in run number
self.run_number = run_number
# Create a new Pandas DataFrame that will store some results against
# the patient ID (which we'll use as the index).
self.results_df = pd.DataFrame()
self.results_df["Patient ID"] = [1]
self.results_df["Q Time Nurse"] = [0.0]
self.results_df["Time with Nurse"] = [0.0]
self.results_df.set_index("Patient ID", inplace=True)
# Create an attribute to store the mean queuing time for the nurse
# across this run of the model
self.mean_q_time_nurse = 0
self.utilisation_audit = [] ##NEW
# A generator function that represents the DES generator for patient
# arrivals
def generator_patient_arrivals(self):
# We use an infinite loop here to keep doing this indefinitely whilst
# the simulation runs
while True:
# Increment the patient counter by 1 (this means our first patient
# will have an ID of 1)
self.patient_counter += 1
# Create a new patient - an instance of the Patient Class we
# defined above. Remember, we pass in the ID when creating a
# patient - so here we pass the patient counter to use as the ID.
p = Patient(self.patient_counter)
# Tell SimPy to start up the attend_clinic generator function with
# this patient (the generator function that will model the
# patient's journey through the system)
self.env.process(self.attend_clinic(p))
# Randomly sample the time to the next patient arriving. Here, we
# sample from an exponential distribution (common for inter-arrival
# times), and pass in a lambda value of 1 / mean. The mean
# inter-arrival time is stored in the g class.
sampled_inter = random.expovariate(1.0 / g.patient_inter)
# Freeze this instance of this function in place until the
# inter-arrival time we sampled above has elapsed. Note - time in
# SimPy progresses in "Time Units", which can represent anything
# you like (just make sure you're consistent within the model)
yield self.env.timeout(sampled_inter)
# A generator function that represents the pathway for a patient going
# through the clinic. Here the pathway is extremely simple - a patient
# arrives, waits to see a nurse, and then leaves.
# The patient object is passed in to the generator function so we can
# extract information from / record information to it
def attend_clinic(self, patient):
# Record the time the patient started queuing for a nurse
start_q_nurse = self.env.now
# This code says request a nurse resource, and do all of the following
# block of code with that nurse resource held in place (and therefore
# not usable by another patient)
with self.nurse.request() as req:
# Freeze the function until the request for a nurse can be met.
# The patient is currently queuing.
yield req
# When we get to this bit of code, control has been passed back to
# the generator function, and therefore the request for a nurse has
# been met. We now have the nurse, and have stopped queuing, so we
# can record the current time as the time we finished queuing.
end_q_nurse = self.env.now
# Calculate the time this patient was queuing for the nurse, and
# record it in the patient's attribute for this.
patient.q_time_nurse = end_q_nurse - start_q_nurse
# Now we'll randomly sample the time this patient with the nurse.
# Here, we use an Exponential distribution for simplicity, but you
# would typically use a Log Normal distribution for a real model
# (we'll come back to that). As with sampling the inter-arrival
# times, we grab the mean from the g class, and pass in 1 / mean
# as the lambda value.
sampled_nurse_act_time = random.expovariate(1.0 /
g.mean_n_consult_time)
# Here we'll store the queuing time for the nurse and the sampled
# time to spend with the nurse in the results DataFrame against the
# ID for this patient. In real world models, you may not want to
# bother storing the sampled activity times - but as this is a
# simple model, we'll do it here.
# We use a handy property of pandas called .at, which works a bit
# like .loc. .at allows us to access (and therefore change) a
# particular cell in our DataFrame by providing the row and column.
# Here, we specify the row as the patient ID (the index), and the
# column for the value we want to update for that patient.
self.results_df.at[patient.id, "Q Time Nurse"] = (
patient.q_time_nurse)
self.results_df.at[patient.id, "Time with Nurse"] = (
sampled_nurse_act_time)
# Freeze this function in place for the activity time we sampled
# above. This is the patient spending time with the nurse.
yield self.env.timeout(sampled_nurse_act_time)
# When the time above elapses, the generator function will return
# here. As there's nothing more that we've written, the function
# will simply end. This is a sink. We could choose to add
# something here if we wanted to record something - e.g. a counter
# for number of patients that left, recording something about the
# patients that left at a particular sink etc.
# This method calculates results over a single run. Here we just calculate
# a mean, but in real world models you'd probably want to calculate more.
def calculate_run_results(self):
# Take the mean of the queuing times for the nurse across patients in
# this run of the model.
self.mean_q_time_nurse = self.results_df["Q Time Nurse"].mean()
def interval_audit_utilisation(self, resources, interval=1):
'''
Record utilisation at defined intervals.
Needs to be passed to env.process when running model.
Parameters:
------
resource: SimPy resource object
The resource to monitor
OR
a list of dictionaries containing simpy resource objects in the format
[{'resource_name':'my_resource', 'resource_object': resource},
{'resource_name':'my_second_resource', 'resource_object': resource_2}
]
where resource and resource_2 in the examples above are simpy resources.
interval: int:
Time between audits.
In simpy time units
Default: 1
'''
# Keep doing the below as long as the simulation is running
while True:
# Code for what to do if we pass in a list of resources
if isinstance(resources, list):
for i in range(len(resources)):
self.utilisation_audit.append({
'resource_name': resources[i]['resource_name'], # The provided name for the resource
'simulation_time': self.env.now, # The current simulation time
'number_utilised': len(resources[i]['resource_object'].count), # The number of users
'number_available': resources[i]['resource_object'].capacity, # The total resource available,
'queue_length': len(resources[i]['resource_object'].queue)
})
else:
# Code for what to do if we just pass in a single resource to monitor
self.utilisation_audit.append({
'simulation_time': self.env.now, # The current simulation time
'number_utilised': resources.count, # The number of users
'number_available': resources.capacity, # The total resource available
'queue_length': len(resources.queue)
})
# Trigger next audit after desired interval has passed.
yield self.env.timeout(interval)
# The run method starts up the DES entity generators, runs the simulation,
# and in turns calls anything we need to generate results for the run
def run(self):
# Start up our DES entity generators that create new patients. We've
# only got one in this model, but we'd need to do this for each one if
# we had multiple generators.
self.env.process(
self.generator_patient_arrivals()
)
self.env.process(
self.interval_audit_utilisation(resources=self.nurse,interval=g.audit_interval) ##NEW
)
# Run the model for the duration specified in g class
self.env.run(until=g.sim_duration)
# Now the simulation run has finished, call the method that calculates
# run results
self.calculate_run_results()
# Print the run number with the patient-level results from this run of
# the model
# EDIT: We've skipped printing the individual-level results here
# print (f"Run Number {self.run_number}")
# print (self.results_df)
# Class representing a Trial for our simulation - a batch of simulation runs.
class Trial:
# The constructor sets up a pandas dataframe that will store the key
# results from each run (just the mean queuing time for the nurse here)
# against run number, with run number as the index.
def __init__(self):
self.df_trial_results = pd.DataFrame()
self.df_trial_results["Run Number"] = [0]
self.df_trial_results["Mean Q Time Nurse"] = [0.0]
self.df_trial_results.set_index("Run Number", inplace=True)
self.interval_audit_list = [] ##NEW
# Method to print out the results from the trial. In real world models,
# you'd likely save them as well as (or instead of) printing them
def print_trial_results(self):
print ("Trial Results")
print (self.df_trial_results)
# Method to run a trial
def run_trial(self):
# Run the simulation for the number of runs specified in g class.
# For each run, we create a new instance of the Model class and call its
# run method, which sets everything else in motion. Once the run has
# completed, we grab out the stored run results (just mean queuing time
# here) and store it against the run number in the trial results
# dataframe.
for run in range(g.number_of_runs):
my_model = Model(run)
my_model.run() ##EDITED
self.df_trial_results.loc[run] = [my_model.mean_q_time_nurse]
interval_audit_df = pd.DataFrame(my_model.utilisation_audit)
interval_audit_df["run"] = run
interval_audit_df["perc_utilisation"] = (
interval_audit_df["number_utilised"]/interval_audit_df["number_available"]
)
self.interval_audit_list.append(interval_audit_df)
# For now, we're going to skip printing the trial results, so the line
# below has been commented out
# self.print_trial_results()
def get_interval_audits(self):
return pd.concat(self.interval_audit_list)17.2.3 Evaluating the outputs
# Create an instance of the Trial class
my_trial = Trial()
# Call the run_trial method of our Trial object
my_trial.run_trial()
interval_audits = my_trial.get_interval_audits()
interval_audits| simulation_time | number_utilised | number_available | queue_length | run | perc_utilisation | |
|---|---|---|---|---|---|---|
| 0 | 0 | 0 | 9 | 0 | 0 | 0.000000 |
| 1 | 5 | 1 | 9 | 0 | 0 | 0.111111 |
| 2 | 10 | 1 | 9 | 0 | 0 | 0.111111 |
| 3 | 15 | 1 | 9 | 0 | 0 | 0.111111 |
| 4 | 20 | 1 | 9 | 0 | 0 | 0.111111 |
| ... | ... | ... | ... | ... | ... | ... |
| 115 | 575 | 7 | 9 | 0 | 4 | 0.777778 |
| 116 | 580 | 6 | 9 | 0 | 4 | 0.666667 |
| 117 | 585 | 5 | 9 | 0 | 4 | 0.555556 |
| 118 | 590 | 5 | 9 | 0 | 4 | 0.555556 |
| 119 | 595 | 6 | 9 | 0 | 4 | 0.666667 |
600 rows × 6 columns
We can explore this output as a graph, with the vertical axis showing the % utilisation over time and the horizontal axis showing the progression of simulation time. Each coloured line represents a different run. We can see that the % utilisation is different at a given time point across different runs, which we would expect due to the randomness.
import plotly.express as px
fig = px.line(interval_audits, x="simulation_time", y="perc_utilisation", color="run")
fig.show()We could look at the median utilisation across different runs.
If we get to a high enough number of runs, we may expect the average utilisation to become quite consistent over time.
Tip
This may be of more use when running a model with variable arrival rates at different times of day - as we would then expect to see some variation.
Tip
This can also form part of your strategy for working out how long the warm-up period of a model should be.
g.number_of_runs = 1000
# Create an instance of the Trial class
my_trial = Trial()
# Call the run_trial method of our Trial object
my_trial.run_trial()
interval_audits = my_trial.get_interval_audits()
interval_audits_median = interval_audits.groupby("simulation_time").median().reset_index()
interval_audits_median| simulation_time | number_utilised | number_available | queue_length | run | perc_utilisation | |
|---|---|---|---|---|---|---|
| 0 | 0 | 0.0 | 9.0 | 0.0 | 499.5 | 0.000000 |
| 1 | 5 | 2.0 | 9.0 | 0.0 | 499.5 | 0.222222 |
| 2 | 10 | 2.0 | 9.0 | 0.0 | 499.5 | 0.222222 |
| 3 | 15 | 3.0 | 9.0 | 0.0 | 499.5 | 0.333333 |
| 4 | 20 | 4.0 | 9.0 | 0.0 | 499.5 | 0.444444 |
| ... | ... | ... | ... | ... | ... | ... |
| 115 | 575 | 7.0 | 9.0 | 0.0 | 499.5 | 0.777778 |
| 116 | 580 | 7.0 | 9.0 | 0.0 | 499.5 | 0.777778 |
| 117 | 585 | 7.0 | 9.0 | 0.0 | 499.5 | 0.777778 |
| 118 | 590 | 7.0 | 9.0 | 0.0 | 499.5 | 0.777778 |
| 119 | 595 | 7.0 | 9.0 | 0.0 | 499.5 | 0.777778 |
120 rows × 6 columns
import plotly.express as px
fig = px.line(interval_audits_median, x="simulation_time", y="perc_utilisation")
fig.show()