5
\$\begingroup\$

The elevator system has the following functions:

  1. There is one building, and the number of elevators can be set arbitrarily.
  2. 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.
  3. 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.
  4. 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()
\$\endgroup\$
1
  • 3
    \$\begingroup\$ You don't need to review very carefully. You can make evaluations from an interviewer's perspective - rest assured that when I perform interviews I am careful \$\endgroup\$
    – Reinderien
    Commented Oct 18, 2022 at 12:36

1 Answer 1

5
\$\begingroup\$

Let's start from the top.

import heapq

can be

from heapq import heappush, heappop

This way you avoid repetition later on in the code.

RequestType is useless since you use inheritance.
It is only used here:

if request.type == RequestType.EXTERNAL:

This can be replaced by if isinstance(request, ExternalRequest) or even better, by using polymorphism. This whole construction

if request.type == RequestType.EXTERNAL:
    self.send_external_request(request)
elif request.type == RequestType.INTERNAL:
    self.elevators[request.elevator].send_internal_request(request)

can be replaced by request.process(self). To do so create an abstract method of Request:

@abstract_method
def process(self, building: Building):
    pass

Add implementations:

External:

def process(self, building: Building):
    building.send_external_request(self)

Internal:

def process(self, building: Building):
    building.elevators[self.elevator].send_internal_request(self)

Request class should be abstract: class Request(ABC):


This logic is very confusing:

if self.running_state == State.UP:
    heapq.heappush(self.stops, level)
elif self.running_state == State.DOWN:
    heapq.heappush(self.stops, -level)

Why are we pushing negative level? Why are we using stops as a heap here and as a list in some other cases: self.stops.append(level)?

There is some repetition:

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)

Although I don't understand what this code does, you can shrink it into a function call that takes direction as a parameter (that can be represented as 1 and -1 instead of 1 and 2). This way you avoid repetition, decrease nesting and increase readability. Same applies to the logic inside send_external_request.

Leave blank lines between function declarations so they don't merge into one block visually (PEP 8: E301).

Leave 2 spaces before starting the comment ( #) for better readability (PEP 8: E261).


This code shows that you know basic OOP but there are some issues to be addressed. Good luck!

\$\endgroup\$
3
  • \$\begingroup\$ Hi, for the heap part: Since python's heapq can only maintain a min heap, that's why when the elevator is going downward, we need to store the negative value. Because when we store negative value, this heap can be seen as a "max heap". So we can go from the highest level to the lowest. Thank you for the advice, I really appreciate that. \$\endgroup\$
    – Hangji He
    Commented Oct 18, 2022 at 21:18
  • \$\begingroup\$ What do you recommend in terms of functions and classes? Do you think it is a complete design for an interview? \$\endgroup\$
    – Hangji He
    Commented Oct 18, 2022 at 21:22
  • \$\begingroup\$ @HangjiHe I don't know what complete for an interview means, it depends on what the interviewer wants from you. I think I expressed my recommendations regarding functions and classes in the answer. \$\endgroup\$ Commented Oct 19, 2022 at 6:28

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.