Design an elevator system for a building that can:
Handle multiple elevators
Optimize elevator assignment for efficiency
Support different scheduling algorithms
Handle concurrent requests from multiple floors
Why This Problem? Elevator design tests your ability to handle state machines, scheduling algorithms, and concurrency. It’s a favorite at companies like Uber and Amazon.
Key Insight: The elevator controller uses Strategy Pattern for scheduling algorithms - you can swap between FCFS, SCAN, and LOOK algorithms without changing the core logic.
@dataclassclass FloorRequest: """Request from a floor button (external)""" floor: int direction: Direction timestamp: float def __hash__(self): return hash((self.floor, self.direction))@dataclassclass ElevatorRequest: """Request from inside elevator (internal)""" destination_floor: int elevator_id: int timestamp: float
class Elevator: def __init__(self, elevator_id: int, total_floors: int): self.id = elevator_id self.total_floors = total_floors self._current_floor = 1 self._direction = Direction.IDLE self._state = ElevatorState.STOPPED self.door = Door() # Two sorted lists for efficient destination management self._up_stops: SortedList = SortedList() self._down_stops: SortedList = SortedList() self._lock = threading.Lock() @property def current_floor(self) -> int: return self._current_floor @property def direction(self) -> Direction: return self._direction @property def state(self) -> ElevatorState: return self._state def add_destination(self, floor: int): """Add a floor to visit""" with self._lock: if floor == self._current_floor: return if floor > self._current_floor: if floor not in self._up_stops: self._up_stops.add(floor) else: if floor not in self._down_stops: self._down_stops.add(floor) # Update direction if idle if self._direction == Direction.IDLE: self._update_direction() def _update_direction(self): """Determine next direction based on pending stops""" if self._up_stops: self._direction = Direction.UP elif self._down_stops: self._direction = Direction.DOWN else: self._direction = Direction.IDLE def get_next_stop(self) -> Optional[int]: """Get the next floor to stop at""" if self._direction == Direction.UP and self._up_stops: return self._up_stops[0] # Nearest up elif self._direction == Direction.DOWN and self._down_stops: return self._down_stops[-1] # Nearest down return None def move(self): """Move elevator one step towards destination""" if self._state == ElevatorState.MAINTENANCE: return if self.door.is_open(): return # Can't move with open doors next_stop = self.get_next_stop() if next_stop is None: self._update_direction() return self._state = ElevatorState.MOVING # Move one floor if self._direction == Direction.UP: self._current_floor += 1 elif self._direction == Direction.DOWN: self._current_floor -= 1 print(f"Elevator {self.id} at floor {self._current_floor}") # Check if we should stop if self._should_stop(): self._stop() def _should_stop(self) -> bool: """Check if elevator should stop at current floor""" if self._direction == Direction.UP: return self._current_floor in self._up_stops elif self._direction == Direction.DOWN: return self._current_floor in self._down_stops return False def _stop(self): """Stop at current floor""" self._state = ElevatorState.STOPPED # Remove from destination lists if self._current_floor in self._up_stops: self._up_stops.remove(self._current_floor) if self._current_floor in self._down_stops: self._down_stops.remove(self._current_floor) # Open door self.door.open() print(f"Elevator {self.id} stopped at floor {self._current_floor}, doors open") time.sleep(DOOR_OPERATION_TIME) self.door.close() print(f"Elevator {self.id} doors closed") # Update direction self._update_direction() def get_status(self) -> dict: return { "id": self.id, "current_floor": self._current_floor, "direction": self._direction.name, "state": self._state.name, "pending_stops": len(self._up_stops) + len(self._down_stops) } def distance_to(self, floor: int) -> int: """Calculate distance to a floor considering direction""" return abs(floor - self._current_floor) def can_serve(self, floor: int, direction: Direction) -> bool: """Check if elevator can efficiently serve this request""" if self._state == ElevatorState.MAINTENANCE: return False if self._direction == Direction.IDLE: return True # Going up and floor is above, or going down and floor is below if self._direction == Direction.UP: return floor >= self._current_floor and direction == Direction.UP else: return floor <= self._current_floor and direction == Direction.DOWN
import threading# Create controller with 3 elevators and 10 floorscontroller = ElevatorController( num_elevators=3, num_floors=10, scheduler=SCANScheduler())# Create displaydisplay = DisplayPanel(controller)# Start elevator system in backgroundcontroller_thread = threading.Thread(target=controller.run)controller_thread.start()# Simulate requestscontroller.request_elevator(floor=5, direction=Direction.UP)controller.request_elevator(floor=3, direction=Direction.DOWN)controller.select_floor(elevator_id=0, floor=8)time.sleep(2)display.show()# Person gets in on floor 5, wants to go to floor 9controller.select_floor(elevator_id=0, floor=9)time.sleep(5)display.show()# Stop the systemcontroller.stop()controller_thread.join()
Different buildings have different needs. A hospital may want FCFS for fairness, while an office may want SCAN for efficiency. Strategy pattern allows swapping algorithms without changing core logic.
Why Two Sorted Lists for Stops?
Separating up and down stops allows efficient retrieval of the next stop in each direction. A single list would require filtering by direction each time.
Why Thread Locks?
Multiple threads can request elevators simultaneously (buttons on different floors). Locks prevent race conditions when modifying elevator state.
Interview Extension: Be ready to discuss VIP/priority floors, weight limits, fire emergency mode, or energy-efficient algorithms that minimize empty trips.
