I'm using SimPy in Python to create a Discrete Event Simulation that requires resources to be available based on a schedule input by the user in my case in a csv file. The aim is to represent different numbers of the same resource (e.g. staff) being available at different times of day. As far I as I can tell this isn't something that is available in base SimPy - like resource priorities.
I have managed to get this working and have included the code below to show how. However I wanted to ask the community if there is a better way to achieve this functionality in SimPy?
The code below works by requesting the resources at the start of each day for the times they are not supposed to be available - with a much higher priority to ensure they get the resource. The resources are then released at the appropriate times for use by other events/processes. As I say it works but seems wasteful with a lot of dummy processes working to ensure the correct true availability of resources. Any comments which would lead to improvements would be welcomed.
so the csv looks like:
Number time
0 23
50 22
100 17
50 10
20 8
5 6
where Number represents the number of staff that are the become available at the defined time. For example: There will be 5 staff from 6-8, 20 from 8-10, 50 from 10-17 and so on until the end of the day.
The code:
import csv
import simpy
# empty list ready to hold the input data in the csv
input_list = []
# a dummy process that "uses" staff until the end of the current day
def take_res():
req = staff.request(priority=-100)
yield req # Request a staff resource at set priority
yield test_env.timeout(24 - test_env.now)
# A dummy process that "uses" staff for the time those staff should not
# be available for the real processes
def request_res(delay, avail_time):
req = staff.request(priority=-100)
yield req # Request a staff resource at set priority
yield test_env.timeout(delay)
yield staff.release(req)
# pass time it is avail for
yield test_env.timeout(avail_time)
test_env.process(take_res())
# used to print current levels of resource usage
def print_usage():
print('At time %0.2f %d res are in use' % (test_env.now, staff.count))
yield test_env.timeout(0.5)
test_env.process(print_usage())
# used to open the csv and read the data into a list
with open('staff_schedule.csv', mode="r") as infile:
reader = csv.reader(infile)
next(reader, None) # ignore header
for row in reader:
input_list.append(row[:2])
# calculates the time the current number of resources will be
# available for and adds to the list
i = 0
for row in the_list:
if i == 0:
row.append(24 - int(input_list[i][1]))
else:
row.append(int(input_list[i-1][1]) - int(input_list[i][1]))
i += 1
# converts list to tuple of tuples to prevent any accidental
# edits from this point in
staff_tuple = tuple(tuple(row) for row in input_list)
print(staff_tuple)
# define environment and creates resources
test_env = simpy.Environment()
staff = simpy.PriorityResource(test_env, capacity=sum(int(l[0]) for l in staff_tuple))
# for each row in the tuple run dummy processes to hold resources
# according to schedule in the csv
for item in the_tuple:
print(item[0])
for i in range(int(item[0])):
test_env.process(request_res(int(item[1]), int(item[2])))
# run event to print usage over time
test_env.process(print_usage())
# run for 25 hours - so 1 day
test_env.run(until=25)
This is how I solved it for my application. It's not perfect but was the best I could do given my basic level of skill with Python and SimPy.
The result is the correct number of Advisers are available at the desired times.
First I define a store and set the capacity to be equal to the total number of adviser instances that will exist within the simulation.
self.adviser_store = simpy.FilterStore(self.env,
capacity=self.total_ad_instances)
The instances of the Adviser class required are created in an initialization step which for brevity I have not included. I actually use a JSON file to customize the individual adviser instances which are then placed in a list.
The run parameter in the class definition below is actually another class that contains all info related to the current run of the simulation - so for example it contains the start and end dates for the simulation. self.start_date therefore defines the date that adviser starts working. self.run.start_date is the start date for the simulation.
class Adviser(object):
def __init__(self, run, id_num, start_time, end_time, start_date, end_date):
self.env = run.env
self.run = run
self.id_num = id_num
self.start_time = start_time
self.end_time = end_time
self.start_date = datetime.datetime.strptime(start_date, '%Y, %m, %d')
self.end_date = datetime.datetime.strptime(end_date, '%Y, %m, %d')
self.ad_type = ad_type
self.avail = False
self.run.env.process(self.set_availability())
So as you can see creating the adviser class also starts the process to set the availability. In the example below I've simplified it to set the same availability each day for a given date range. You could of course set different availabilities depending on date/day etc.
def set_availability(self):
start_delay = self.start_time + (self.start_date - self.run.start_date).total_seconds()/3600 # this returns the time in hours until the resource becomes available and is applied below.
end_delay = self.end_time + (self.start_date - self.run.start_date).total_seconds()/3600
repeat = (self.end_date - self.start_date).days + 1 # defines how man days to repeat it for
for i in range(repeat):
start_delayed(self.run.env, self.add_to_store(), start_delay)
start_delayed(self.run.env, self.remove_from_store(), end_delay)
start_delay += 24
end_delay += 24
yield self.run.env.timeout(0)
def add_to_store(self):
self.run.ad_avail.remove(self) # take adviser from a list
self.run.adviser_store.put(self) # and put it in the store
yield self.run.env.timeout(0)
def remove_from_store(self):
current_ad = yield self.run.adviser_store.get(lambda item: item.id_num == self.id_num) # get itself from the store
self.run.ad_avail.append(current_ad) # and put it back in the list
yield self.run.env.timeout(0)
So essentially customers can only request advisers from the store and the advisers will only be in the store at certain times. the rest of the time they are in the list attached to the current run of the simulation.
I think there is still a pitfall here. The adviser object may be in use when it is due to become unavailable. I haven't noticed if this happens as yet or the impact if it does.