1class Direction(Enum):
2    UP = "UP"
3    DOWN = "DOWN"
4    IDLE = "IDLE"
5
6class RequestSource(Enum):
7    INTERNAL = "INTERNAL"  # From inside the cabin
8    EXTERNAL = "EXTERNAL"  # From the hall/floor

1@dataclass
2class Request:
3    target_floor: int
4    direction: Direction  # Primarily for External requests
5    source: RequestSource
6
7    def __str__(self):
8        if self.source == RequestSource.EXTERNAL:
9            return f"{self.source.value} Request to floor {self.target_floor} going {self.direction.value}"
10        else:
11            return f"{self.source.value} Request to floor {self.target_floor}"

1class ElevatorObserver(ABC):
2    @abstractmethod
3    def update(self, elevator):
4        pass
5
6class Display(ElevatorObserver):
7    def update(self, elevator):
8        print(f"[DISPLAY] Elevator {elevator.get_id()} | Current Floor: {elevator.get_current_floor()} | Direction: {elevator.get_direction().value}")

1class ElevatorSelectionStrategy(ABC):
2    @abstractmethod
3    def select_elevator(self, elevators: List, request: Request) -> Optional:
4        pass
5
6class NearestElevatorStrategy(ElevatorSelectionStrategy):
7    def select_elevator(self, elevators: List, request: Request) -> Optional:
8        best_elevator = None
9        min_distance = float('inf')
10
11        for elevator in elevators:
12            if self._is_suitable(elevator, request):
13                distance = abs(elevator.get_current_floor() - request.target_floor)
14                if distance < min_distance:
15                    min_distance = distance
16                    best_elevator = elevator
17
18        return best_elevator
19
20    def _is_suitable(self, elevator, request: Request) -> bool:
21        if elevator.get_direction() == Direction.IDLE:
22            return True
23        if elevator.get_direction() == request.direction:
24            if request.direction == Direction.UP and elevator.get_current_floor() <= request.target_floor:
25                return True
26            if request.direction == Direction.DOWN and elevator.get_current_floor() >= request.target_floor:
27                return True
28        return False

1class ElevatorState(ABC):
2    @abstractmethod
3    def move(self, elevator):
4        pass
5
6    @abstractmethod
7    def add_request(self, elevator, request: Request):
8        pass
9
10    @abstractmethod
11    def get_direction(self) -> Direction:
12        pass

1class IdleState(ElevatorState):
2    def move(self, elevator):
3        if elevator.get_up_requests():
4            elevator.set_state(MovingUpState())
5        elif elevator.get_down_requests():
6            elevator.set_state(MovingDownState())
7        # Else stay idle
8
9    def add_request(self, elevator, request: Request):
10        if request.target_floor > elevator.get_current_floor():
11            elevator.get_up_requests().add(request.target_floor)
12        elif request.target_floor < elevator.get_current_floor():
13            elevator.get_down_requests().add(request.target_floor)
14        # If request is for current floor, doors would open (handled implicitly by moving to that floor)
15
16    def get_direction(self) -> Direction:
17        return Direction.IDLE

1class MovingUpState(ElevatorState):
2    def move(self, elevator):
3        if not elevator.get_up_requests():
4            elevator.set_state(IdleState())
5            return
6
7        next_floor = min(elevator.get_up_requests())
8        elevator.set_current_floor(elevator.get_current_floor() + 1)
9
10        if elevator.get_current_floor() == next_floor:
11            print(f"Elevator {elevator.get_id()} stopped at floor {next_floor}")
12            elevator.get_up_requests().remove(next_floor)
13
14        if not elevator.get_up_requests():
15            elevator.set_state(IdleState())
16
17    def add_request(self, elevator, request: Request):
18        # Internal requests always get added to the appropriate queue
19        if request.source == RequestSource.INTERNAL:
20            if request.target_floor > elevator.get_current_floor():
21                elevator.get_up_requests().add(request.target_floor)
22            else:
23                elevator.get_down_requests().add(request.target_floor)
24            return
25
26        # External requests
27        if request.direction == Direction.UP and request.target_floor >= elevator.get_current_floor():
28            elevator.get_up_requests().add(request.target_floor)
29        elif request.direction == Direction.DOWN:
30            elevator.get_down_requests().add(request.target_floor)
31
32    def get_direction(self) -> Direction:
33        return Direction.UP

1class MovingDownState(ElevatorState):
2    def move(self, elevator):
3        if not elevator.get_down_requests():
4            elevator.set_state(IdleState())
5            return
6
7        next_floor = max(elevator.get_down_requests())
8        elevator.set_current_floor(elevator.get_current_floor() - 1)
9
10        if elevator.get_current_floor() == next_floor:
11            print(f"Elevator {elevator.get_id()} stopped at floor {next_floor}")
12            elevator.get_down_requests().remove(next_floor)
13
14        if not elevator.get_down_requests():
15            elevator.set_state(IdleState())
16
17    def add_request(self, elevator, request: Request):
18        # Internal requests always get added to the appropriate queue
19        if request.source == RequestSource.INTERNAL:
20            if request.target_floor > elevator.get_current_floor():
21                elevator.get_up_requests().add(request.target_floor)
22            else:
23                elevator.get_down_requests().add(request.target_floor)
24            return
25
26        # External requests
27        if request.direction == Direction.DOWN and request.target_floor <= elevator.get_current_floor():
28            elevator.get_down_requests().add(request.target_floor)
29        elif request.direction == Direction.UP:
30            elevator.get_up_requests().add(request.target_floor)
31
32    def get_direction(self) -> Direction:
33        return Direction.DOWN

1class Elevator:
2    def __init__(self, elevator_id: int):
3        self.id = elevator_id
4        self.current_floor = 1
5        self.current_floor_lock = threading.Lock()
6        self.state = IdleState()
7        self.is_running = True
8        
9        self.up_requests = set()
10        self.down_requests = set()
11        
12        # Observer Pattern: List of observers
13        self.observers = []
14
15    # --- Observer Pattern Methods ---
16    def add_observer(self, observer: ElevatorObserver):
17        self.observers.append(observer)
18        observer.update(self)  # Send initial state
19
20    def notify_observers(self):
21        for observer in self.observers:
22            observer.update(self)
23
24    # --- State Pattern Methods ---
25    def set_state(self, state: ElevatorState):
26        self.state = state
27        self.notify_observers()  # Notify observers on direction change
28
29    def move(self):
30        self.state.move(self)
31
32    # --- Request Handling ---
33    def add_request(self, request: Request):
34        print(f"Elevator {self.id} processing: {request}")
35        self.state.add_request(self, request)
36
37    # --- Getters and Setters ---
38    def get_id(self) -> int:
39        return self.id
40
41    def get_current_floor(self) -> int:
42        with self.current_floor_lock:
43            return self.current_floor
44
45    def set_current_floor(self, floor: int):
46        with self.current_floor_lock:
47            self.current_floor = floor
48        self.notify_observers()  # Notify observers on floor change
49
50    def get_direction(self) -> Direction:
51        return self.state.get_direction()
52
53    def get_up_requests(self) -> Set[int]:
54        return self.up_requests
55
56    def get_down_requests(self) -> Set[int]:
57        return self.down_requests
58
59    def is_elevator_running(self) -> bool:
60        return self.is_running
61
62    def stop_elevator(self):
63        self.is_running = False
64
65    def run(self):
66        while self.is_running:
67            self.move()
68            try:
69                time.sleep(1)  # Simulate movement time
70            except KeyboardInterrupt:
71                self.is_running = False
72                break

1class ElevatorSystem:
2    _instance = None
3    _lock = threading.Lock()
4
5    def __new__(cls, num_elevators: int):
6        if cls._instance is None:
7            with cls._lock:
8                if cls._instance is None:
9                    cls._instance = super().__new__(cls)
10                    cls._instance._initialized = False
11        return cls._instance
12
13    def __init__(self, num_elevators: int):
14        if self._initialized:
15            return
16        
17        self.selection_strategy = NearestElevatorStrategy()
18        self.executor_service = ThreadPoolExecutor(max_workers=num_elevators)
19        
20        elevator_list = []
21        display = Display()  # Create the observer
22
23        for i in range(1, num_elevators + 1):
24            elevator = Elevator(i)
25            elevator.add_observer(display)  # Attach the observer
26            elevator_list.append(elevator)
27
28        self.elevators = {elevator.get_id(): elevator for elevator in elevator_list}
29        self._initialized = True
30
31    @classmethod
32    def get_instance(cls, num_elevators: int):
33        return cls(num_elevators)
34
35    def start(self):
36        for elevator in self.elevators.values():
37            self.executor_service.submit(elevator.run)
38
39    # --- Facade Methods ---
40
41    # EXTERNAL Request (Hall Call)
42    def request_elevator(self, floor: int, direction: Direction):
43        print(f"\n>> EXTERNAL Request: User at floor {floor} wants to go {direction.value}")
44        request = Request(floor, direction, RequestSource.EXTERNAL)
45
46        # Use strategy to find the best elevator
47        selected_elevator = self.selection_strategy.select_elevator(list(self.elevators.values()), request)
48
49        if selected_elevator:
50            selected_elevator.add_request(request)
51        else:
52            print("System busy, please wait.")
53
54    # INTERNAL Request (Cabin Call)
55    def select_floor(self, elevator_id: int, destination_floor: int):
56        print(f"\n>> INTERNAL Request: User in Elevator {elevator_id} selected floor {destination_floor}")
57        request = Request(destination_floor, Direction.IDLE, RequestSource.INTERNAL)
58
59        elevator = self.elevators.get(elevator_id)
60        if elevator:
61            elevator.add_request(request)
62        else:
63            print("Invalid elevator ID.", file=sys.stderr)
64
65    def shutdown(self):
66        print("Shutting down elevator system...")
67        for elevator in self.elevators.values():
68            elevator.stop_elevator()
69        self.executor_service.shutdown()

1class ElevatorSystemDemo:
2    @staticmethod
3    def main():
4        import sys
5        
6        # Setup: A building with 2 elevators
7        num_elevators = 2
8        # The get_instance method now initializes the elevators and attaches the Display (Observer).
9        elevator_system = ElevatorSystem.get_instance(num_elevators)
10
11        # Start the elevator system
12        elevator_system.start()
13        print("Elevator system started. ConsoleDisplay is observing.\n")
14
15        # --- SIMULATION START ---
16
17        # 1. External Request: User at floor 5 wants to go UP.
18        # The system will dispatch this to the nearest elevator (likely E1 or E2, both at floor 1).
19        elevator_system.request_elevator(5, Direction.UP)
20        time.sleep(0.1)  # Wait for the elevator to start moving
21
22        # 2. Internal Request: Assume E1 took the previous request.
23        # The user gets in at floor 5 and presses 10.
24        # We send this request directly to E1.
25
26        # Note: In a real simulation, we'd wait until E1 reaches floor 5, but for this demo,
27        # we simulate the internal button press shortly after the external one.
28        elevator_system.select_floor(1, 10)
29        time.sleep(0.2)
30
31        # 3. External Request: User at floor 3 wants to go DOWN.
32        # E2 (likely still idle at floor 1) might take this, or E1 if it's convenient.
33        elevator_system.request_elevator(3, Direction.DOWN)
34        time.sleep(0.3)
35
36        # 4. Internal Request: User in E2 presses 1.
37        elevator_system.select_floor(2, 1)
38
39        # Let the simulation run for a while to observe the display updates
40        print("\n--- Letting simulation run for 1 second ---")
41        time.sleep(1)
42
43        # Shutdown the system
44        elevator_system.shutdown()
45        print("\n--- SIMULATION END ---")
46
47if __name__ == "__main__":
48    ElevatorSystemDemo.main()