The elevator system has the following functions:
- There is one building, and the number of elevators can be set arbitrarily.
- Requests can be sent from outside the elevators, in this case the request will have its original floor, and also the direction (going up or down). Requests of this kind are first sent to a "central control system", then will be assigned to different elevators according to the distance between the elevator and the original floor, and also the moving direction of elevators.
- Requests can also be sent from inside each elevator. In this case, the request will be sent only to the current elevator. As in real life, you can only send requests that have the same direction as the elevator. i.e., you can only tell the elevator to go to the floors below you when the elevator is going down.
- The elevator can go up, down or be idle.
You can make evaluations from an interviewer's perspective because I am preparing for an interview. Please give some advice.
What do you recommend in terms of the completeness of classes, attributes and functions? Do you think it is a complete design for an interview?
import heapq
from time import sleep
from enum import Enum
from collections import deque
class Direction(Enum):
UP = 1
DOWN = 2
class Request:
def __init__(self, level, time):
self.level = level
self.time = time
class RequestType(Enum):
EXTERNAL = 1
INTERNAL = 2
class InternalRequest(Request):
def __init__(self, level, elevator, time) -> None:
super().__init__(level, time)
self.type = RequestType.INTERNAL
self.elevator = elevator
def __str__(self) -> str:
return f"Internal request: floor {self.level}, elevator {self.elevator}, time {self.time}"
class ExternalRequest(Request):
def __init__(self, level, direction, time):
super().__init__(level, time)
self.direction = direction
self.type = RequestType.EXTERNAL
def __str__(self) -> str:
return f"External request: floor {self.level}, {self.direction}, time {self.time}"
class State(Enum):
UP = 1
DOWN = 2
IDLE = 3
class Elevator:
def __init__(self, id):
self.id = id
self.current_floor = 1
self.running_state = State.IDLE
self.stops = []
def __str__(self) -> str:
return f"id {self.id}, current floor {self.current_floor}, running state {self.running_state}, stops list {self.stops}"
def send_internal_request(self, internal_request: InternalRequest):# send internal requests directly to elevators
print(f"Elevator {self.id} received {internal_request}")
if self.running_state == State.UP and internal_request.level >= self.current_floor:
heapq.heappush(self.stops, internal_request.level)
elif self.running_state == State.DOWN and internal_request.level <= self.current_floor:
heapq.heappush(self.stops, -internal_request.level)
elif self.running_state == State.IDLE and internal_request.level != self.current_floor:
if internal_request.level < self.current_floor:
self.stops.append(-internal_request.level)
self.running_state = State.DOWN
elif internal_request.level > self.current_floor:
self.stops.append(internal_request.level)
self.running_state = State.UP
else:
print("Sorry, your request is illeagal.")
def dispatch(self, level):# called by Building to dispatch elevators based on external requests
if self.running_state == State.UP:
heapq.heappush(self.stops, level)
elif self.running_state == State.DOWN:
heapq.heappush(self.stops, -level)
elif self.running_state == State.IDLE:
if level >= self.current_floor:
self.running_state = State.UP
self.stops.append(level)
else:
self.running_state = State.DOWN
self.stops.append(-level)
def run(self):# called every loop to update elevator's state
if self.running_state == State.UP:
self.current_floor += 1
if self.stops and self.stops[0] == self.current_floor:
print(f"Elevator {self.id} reaches floor {self.current_floor}")
while self.stops and self.stops[0] == self.current_floor:
heapq.heappop(self.stops)
elif self.running_state == State.DOWN:
self.current_floor -= 1
if self.stops and self.stops[0] == -self.current_floor:
print(f"Elevator {self.id} reaches floor {self.current_floor}")
while self.stops and -self.stops[0] == self.current_floor:
heapq.heappop(self.stops)
if not self.stops:
self.running_state = State.IDLE
class Building:
def __init__(self, height, elevator_count, requests):
self.time = 0
self.request_queue = deque(requests)
self.height = height
self.elevators = [Elevator(i) for i in range(elevator_count)]
self.not_assigned_requests = []
def send_external_request(self, external_request: ExternalRequest):# receive external requests and either assign to specific elevators or store them
print(f"Received {external_request}")
idle_elevators = list(filter(lambda e: e.running_state == State.IDLE,
self.elevators))
if external_request.direction == Direction.UP:
up_elevators = list(filter(lambda e: e.running_state == State.UP and e.current_floor < external_request.level,
self.elevators))
if up_elevators:
min(up_elevators, key=lambda e: external_request.level - e.current_floor).dispatch(external_request.level)
elif idle_elevators:
min(idle_elevators, key=lambda e: abs(external_request.level - e.current_floor)).dispatch(external_request.level)
else:
self.not_assigned_requests.append(external_request)
elif external_request.direction == Direction.DOWN:
down_elevators = list(filter(lambda e: e.running_state == State.DOWN and e.current_floor > external_request.level,
self.elevators))
if down_elevators:
min(down_elevators, key=lambda e: e.current_floor - external_request.level).dispatch(external_request.level)
elif idle_elevators:
min(idle_elevators, key=lambda e: abs(external_request.level - e.current_floor)).dispatch(external_request.level)
else:
self.not_assigned_requests.append(external_request)
def handle_not_assigned_requests(self):# in the last step of each loop, we check the stored requests to see if some elevators are ready to handle these requests
if self.not_assigned_requests:
tmp = self.not_assigned_requests.copy()
self.not_assigned_requests.clear()
for request in tmp:
self.send_external_request(request)
def run(self):# this method starts the main loop, in each loop, it first sends the requests, then call run() on each elevator to update their states
while True:
print(f"Now is {self.time}")
while self.request_queue and self.time == self.request_queue[0].time:
request = self.request_queue.popleft()
if request.type == RequestType.EXTERNAL:
self.send_external_request(request)
elif request.type == RequestType.INTERNAL:
self.elevators[request.elevator].send_internal_request(request)
for elevator in self.elevators:
print(elevator)
self.time += 1
print("-------------------------------------------------------------")
for elevator in self.elevators:
elevator.run()
sleep(1)
self.handle_not_assigned_requests()
if __name__ == "__main__":
requests = [
ExternalRequest(3, Direction.UP, 0),
ExternalRequest(4, Direction.UP, 0),
InternalRequest(level=4, elevator=1, time=4),
InternalRequest(level=2, elevator=0, time=4),
ExternalRequest(level=1, direction=Direction.DOWN, time=6),
ExternalRequest(level=5, direction=Direction.DOWN, time=6),
InternalRequest(level=1, elevator=0, time=6),
InternalRequest(level=1, elevator=1, time=6),
]
building = Building(height=5, elevator_count=2, requests=requests)
building.run()
