I have been trying to write a simulation script where entities go though a process of events. Each event is independent for each entity and this is where I want to implement a timeout function for the entity that is present in that event.
I don't know if I have understood the documentation correctly but in all the examples I have seen pauses the environment aka env.timeout(). I am wondering how timeout works and how I could apply a timeout for a specific entity?
Maybe I am not even using the right library or have to combine with different ones, so I am open to suggestions! Please take a look at my code below :)
FYI pretty new to programming
import simpy
class G:
# The properties for the system
cycle_time = 10
interrupt_time = 3
load_time = 4
class Delivery:
def __init__(self, d_id):
self.id = d_id
class System:
def __init__(self):
self.env = simpy.Environment()
self.load_station = simpy.Resource(self.env, capacity=1)
self.active_time = simpy.Container(self.env, init=G.cycle_time, capacity=G.cycle_time)
self.new_cycle = self.env.process(self.new_cycle())
def new_cycle(self):
while True:
if self.active_time.level <= 10:
"""
here I want the timeout to only have effect on the delivery trucks that are
in the process of traveling on the road but not loading the trucks, like a red
light.
"""
yield self.env.timeout(G.interrupt_time)
next_cycle = yield self.active_time.put(G.interrupt_time)
self.env.process(next_cycle)
def load(self):
with self.load_station.request() as req:
yield req
"""
here I want the timeout to only have effect on the delivery truck that is
loading are
"""
yield self.env.timeout(G.load_time)
"""
there are several more functions that describe the system but I hope this will suffice
"""
Here is a quick example of modeling a entity with multiple step process
The basic process is a entity travels to a warehouse and gets loaded by one of two loading processes.
The first process (the travel process) has simple logic to assign the entity to its next task (one of the two loading tasks).
This is a common way to chain tasks together. I think it makes maintenance a lot easier then trying to maintain one massive control process with a lot of nested if statements. This way, the logic is simply, "I'm know I'm in this task, therefore, what is the next step?" Of course, your entity can also have attributes that track "state", that can be used in the selection of the next task.
"""
simple entity flow of a entity
enity has a common first task, but
chooses which second task it does
scenario being model is:
entity travels to a warehouse.
When arries entity is either
unloaded quickly,
or unloaded slowly
unloading requires a loader resource
Programer: Michael R. Gibbs
"""
import simpy
import random
class Entity():
"""
Simple entity, has a id
"""
next_id:int = 1
def __init__(self):
self.id = self.__class__.next_id
self.__class__.next_id += 1
def task_1_travel(env, entity, loader_pool):
"""
models the time a entity takes to get to a warehouse
This process will also select the entitys next task
"""
print(f'{env.now:.2f} entity {entity.id} has started traveling')
# travel time
yield env.timeout(random.triangular(1, 10, 5))
print(f'{env.now:.2f} entity {entity.id} has finished traveling')
# pick next task
# note we do not yield here
# so this task can end
if random.random() < 0.3:
env.process(task_2_load_slow(env, entity, loader_pool))
else:
env.process(task_2_load_fast(env, entity, loader_pool))
def task_2_load_slow(env, entity, loader_pool):
"""
slow loading
"""
print(f'{env.now:.2f} entity {entity.id} has queued for slow loader')
with loader_pool.request() as req:
yield req
print(f'{env.now:.2f} entity {entity.id} has seized a slow loader')
yield env.timeout(random.triangular(5,15, 10))
print(f'{env.now:.2f} entity {entity.id} has finished loading and released a slow loader')
def task_2_load_fast(env, entity, loader_pool):
"""
fast loading
"""
print(f'{env.now:.2f} entity {entity.id} has queued for fast loader')
with loader_pool.request() as req:
yield req
print(f'{env.now:.2f} entity {entity.id} has seized a fast loader')
yield env.timeout(random.triangular(1, 10, 4))
print(f'{env.now:.2f} entity {entity.id} has finished loading and released a fast loader')
def gen_entites(env, loader_pool):
"""
Endless loop to generate entites
and to assign them to their first task
"""
while True:
yield env.timeout(random.triangular(1, 8, 2))
entity = Entity()
# note no yield here since we are not waiting on
# the entity to finish its task before next entity start
# Also not we are starting a new process for each entity
env.process(task_1_travel(env, entity, loader_pool))
# boot up
env = simpy.Environment()
loader_pool = simpy.Resource(env,2)
env.process(gen_entites(env, loader_pool))
env.run(100)