I am simulating people movements and their elevator usage. An elevator can take up multiple persons before moving to another floor. The default process has a capacity parameter, however, these indicate the number of processes and not the number of people using the elevator at the same time.
I have tried to use multiple of the resources available, such as Container, Store, and Base. The elevator should be requested and these objects do not have the functionality to be requested. Hence, the only suitable solution is to inherent from the base.Resource class. I have tried to create a subclass Elevator, implementing from base.Resource and adjusting the function _do_get to take multiple elements from the queue. I am pretty confident that this is not the proper way to implement it and it gives an error as well: RuntimeError: <Request() object at 0x1ffb4474be0> has already been triggered. I have no clue which files to adjust to make Simpy happy. Could someone point me in the right direction?
@dataclass
class Elevator(simpy.Resource):
current_floor: int = 0
available_floors: List[int] = field(default_factory=lambda: [0, 1])
capacity: int = 3
# load_carriers: List[LoadCarrier] = field(default_factory=list)
move_time: int = 5
def __init__(self, env: Environment, capacity: int = 1, elevator_capacity: int = 1):
self.elevator_capacity = elevator_capacity
if capacity <= 0:
raise ValueError('"capacity" must be > 0.')
super().__init__(env, capacity)
self.users: List[Request] = []
"""List of :class:`Request` events for the processes that are currently
using the resource."""
self.queue = self.put_queue
"""Queue of pending :class:`Request` events. Alias of
:attr:`~simpy.resources.base.BaseResource.put_queue`.
"""
@property
def count(self) -> int:
"""Number of users currently using the resource."""
return len(self.users)
if TYPE_CHECKING:
def request(self) -> Request:
"""Request a usage slot."""
return Request(self)
def release(self, request: Request) -> Release:
"""Release a usage slot."""
return Release(self, request)
else:
request = BoundClass(Request)
release = BoundClass(Release)
def _do_put(self, event: Request) -> None:
if len(self.users) < self.capacity:
self.users.append(event)
event.usage_since = self._env.now
event.succeed()
def _do_get(self, event: Release) -> None:
for i in range(min(self.elevator_capacity, len(self.users))):
try:
event = self.users.pop(0)
event.succeed()
# self.users.remove(event.request) # type: ignore
except ValueError:
pass
# event.succeed()
So here is the solution I came up with. The tricky bit is I chained two events together. When you queue up for the elevator you get a event that fires when the elevator arrives. This event also returns a second event that fires when you get to your destination floor. This second event is a common event shared by all the passengers that are on the elevator and going to the same floor. Firing this one event notifies a bunch of passengers. This subscribe broadcast pattern can greatly reduce the number of events the model needs to process which in turn improves performance. I use the chained events because if you are in a queue, and the guy in front of you gets on and you do not, then that guy is also going to get off before you, requiring a different destination arrive event. Put another way, I do not know when you will get off until you get on, so I need to defer that part till you actually get onto the elevator.
"""
Simple elevator demo using events to implements a subscribe, broadcast pattern to let passengers know when
they have reached there floor. All the passengers getting off on the same floor are waiting on the
same one event.
Programmer: Michael R. Gibbs
"""
import simpy
import random
class Passenger():
"""
simple class with unique id per passenger
"""
next_id = 1
@classmethod
def get_next_id(cls):
id = cls.next_id
cls.next_id += 1
return id
def __init__(self):
self.id = self.get_next_id()
class Elevator():
""""
Elevator that move people from floor to floor
Has a max compatity
Uses a event to notifiy passengers when they can get on the elevator
and when they arrive at their destination floor
"""
class Move_Goal():
"""
wrapps passengers so we can track where they are going to
"""
def __init__(self, passenger, start_floor, dest_floor, onboard_event):
self.passenger = passenger
self.start_floor = start_floor
self.dest_floor = dest_floor
self.onboard_event = onboard_event
self.arrive_event = None
def __init__(self,env, passenger_cap, floors):
self.env = env
self.passenger_cap = passenger_cap
self.floors = floors
self.on_floor = 0
self.move_inc = 1
# list of passengers on elevator, one per floor
self.on_board = {f:[] for f in range(1,floors + 1)}
# queue for passengers waitting to get on elevator, one queue per floor
self.boarding_queues = {f:[] for f in range(1,floors + 1)}
# events to notify passengers when they have arrived at their floor, one per floor
self.arrive_events = {f: simpy.Event(env) for f in range(1, floors + 1)}
# start elevator
env.process(self._move_next_floor())
def _move_next_floor(self):
"""
Moves the elevator up and down
Elevator stops at every floor
"""
while True:
# move time to next floor
yield self.env.timeout(5)
# update floor elevator is at
self.on_floor = self.on_floor + self.move_inc
# check if elevator needs to change direction
if self.on_floor == self.floors:
self.move_inc = -1
elif self.on_floor == 1:
self.move_inc = 1
# unload and notify passengers that want to get of at this floor
arrive_event = self.arrive_events[self.on_floor]
self.arrive_events[self.on_floor] = simpy.Event(self.env)
arrive_event.succeed()
self.on_board[self.on_floor] = []
# load new passengers
# get open capacity
used_cap = 0
for p in self.on_board.values():
used_cap += len(p)
open_cap = self.passenger_cap - used_cap
# get boarding passengers
boarding = self.boarding_queues[self.on_floor][:open_cap]
self.boarding_queues[self.on_floor] = self.boarding_queues[self.on_floor][open_cap:]
# sort bording into dest floors
for p in boarding:
# give passenger common event for arriving at destination floor
p.arrive_event = self.arrive_events[p.dest_floor]
# notify passeger that they are onboard the elevator
p.onboard_event.succeed()
self.on_board[p.dest_floor].append(p)
def move_to(self, passenger, from_floor, to_floor):
"""
Return a event that fires when the passenger gets on the elevator
The event returns another event that fires when the passager
arrives at their destination floor
(uses the env.process() to convert a process to a event)
"""
return self.env.process(self._move_to(passenger, from_floor, to_floor))
def _move_to(self, passenger, from_floor, to_floor):
"""
Puts the passenger into a queue for the elevator
"""
# creat event to notify passenger when they can get onto the elemator
onboard_event = simpy.Event(self.env)
# save move data in a wrapper and put passenger into queue
move_goal = self.Move_Goal(passenger, from_floor, to_floor, onboard_event)
self.boarding_queues[from_floor].append(move_goal)
# wait for elevator to arrive, and have space for passenger
yield onboard_event
# get destination arrival event
dest_event = self.arrive_events[to_floor]
move_goal.arrive_event = dest_event
return dest_event
def use_elevator(env, elevator, passenger, start_floor, end_floor):
"""
process for using a elevator to move from one floor to another
"""
print(f'{env.now:.2f} passenger {passenger.id} has queued on floor {start_floor}')
arrive_event = yield elevator.move_to(passenger, start_floor, end_floor)
print(f'{env.now:.2f} passenger {passenger.id} has boarded on floor {start_floor}')
yield arrive_event
print(f'{env.now:.2f} passenger {passenger.id} has arrived on floor {end_floor}')
def gen_passengers(env, elevator):
"""
creates passengers to use a elevatore
"""
floor_set = {f for f in range(1, elevator.floors + 1)}
while True:
# time between arrivals
yield env.timeout(random.uniform(0,5))
# get passenger and where they want to go
passenger = Passenger()
start_floor, end_floor = random.sample(floor_set, 2)
# use the elevator to get there
env.process(use_elevator(env, elevator, passenger, start_floor, end_floor))
# boot up
env = simpy.Environment()
elevator = Elevator(env, 20, 3)
env.process(gen_passengers(env, elevator))
env.run(100)