Elevator/docs/events.rst

Event-Driven Simulation and Tick-Based Execution
================================================

Elevator Saga uses an **event-driven, tick-based** discrete simulation model. The simulation progresses in discrete time steps (ticks), and events are generated to notify the controller about state changes.

Simulation Overview
-------------------

.. code-block:: text

   ┌──────────────────────────────────────────────────────────┐
   │                  Simulation Loop                         │
   │                                                          │
   │  Tick N                                                  │
   │    1. Update elevator status (START_UP → CONSTANT_SPEED) │
   │    2. Process arrivals (new passengers)                  │
   │    3. Move elevators (physics simulation)                │
   │    4. Process stops (boarding/alighting)                 │
   │    5. Generate events                                    │
   │                                                          │
   │  Events sent to client → Client processes → Commands     │
   │                                                          │
   │  Tick N+1                                                │
   │    (repeat...)                                           │
   └──────────────────────────────────────────────────────────┘

Tick-Based Execution
--------------------

What is a Tick?
~~~~~~~~~~~~~~~

A **tick** is the fundamental unit of simulation time. Each tick represents one discrete time step where:

1. Physics is updated (elevators move)
2. State changes occur (passengers board/alight)
3. Events are generated
4. Controller receives events and makes decisions

Think of it like frames in a video game - the simulation updates at discrete intervals.

Tick Processing Flow
~~~~~~~~~~~~~~~~~~~~

In ``simulator.py``, the ``step()`` method processes ticks:

.. code-block:: python

   def step(self, num_ticks: int = 1) -> List[SimulationEvent]:
       """Process one or more simulation ticks"""
       with self.lock:
           new_events = []
           for _ in range(num_ticks):
               self.state.tick += 1
               tick_events = self._process_tick()
               new_events.extend(tick_events)

               # Force complete passengers if max duration reached
               if self.tick >= self.max_duration_ticks:
                   completed_count = self.force_complete_remaining_passengers()

           return new_events

Each ``_process_tick()`` executes the four-phase cycle:

.. code-block:: python

   def _process_tick(self) -> List[SimulationEvent]:
       """Process one simulation tick"""
       events_start = len(self.state.events)

       # Phase 1: Update elevator status
       self._update_elevator_status()

       # Phase 2: Add new passengers from traffic queue
       self._process_arrivals()

       # Phase 3: Move elevators
       self._move_elevators()

       # Phase 4: Process elevator stops and passenger boarding/alighting
       self._process_elevator_stops()

       # Return events generated this tick
       return self.state.events[events_start:]

Elevator State Machine
-----------------------

Elevators transition through states each tick:

.. code-block:: text

   STOPPED ──(target set)──► START_UP ──(1 tick)──► CONSTANT_SPEED
                                                           │
                                                    (near target)
                                                           ▼
                                                      START_DOWN
                                                           │
                                                      (1 tick)
                                                           ▼
                               (arrived)               STOPPED

State Transitions
~~~~~~~~~~~~~~~~~

**Phase 1: Update Elevator Status** (``_update_elevator_status()``):

.. code-block:: python

   def _update_elevator_status(self) -> None:
       """Update elevator operational state"""
       for elevator in self.elevators:
           # If no direction, check for next target
           if elevator.target_floor_direction == Direction.STOPPED:
               if elevator.next_target_floor is not None:
                   self._set_elevator_target_floor(elevator, elevator.next_target_floor)
                   self._process_passenger_in()
                   elevator.next_target_floor = None
               else:
                   continue

           # Transition state machine
           if elevator.run_status == ElevatorStatus.STOPPED:
               # Start acceleration
               elevator.run_status = ElevatorStatus.START_UP
           elif elevator.run_status == ElevatorStatus.START_UP:
               # Switch to constant speed after 1 tick
               elevator.run_status = ElevatorStatus.CONSTANT_SPEED

**Important Notes**:

- ``START_UP`` = acceleration (not direction!)
- ``START_DOWN`` = deceleration (not direction!)
- Actual movement direction is ``target_floor_direction`` (UP/DOWN)
- State transitions happen **before** movement

Movement Physics
----------------

Speed by State
~~~~~~~~~~~~~~

Elevators move at different speeds depending on their state:

.. code-block:: python

   def _move_elevators(self) -> None:
       """Move all elevators towards their destinations"""
       for elevator in self.elevators:
           # Determine speed based on state
           movement_speed = 0
           if elevator.run_status == ElevatorStatus.START_UP:
               movement_speed = 1      # Accelerating: 0.1 floors/tick
           elif elevator.run_status == ElevatorStatus.START_DOWN:
               movement_speed = 1      # Decelerating: 0.1 floors/tick
           elif elevator.run_status == ElevatorStatus.CONSTANT_SPEED:
               movement_speed = 2      # Full speed: 0.2 floors/tick

           if movement_speed == 0:
               continue

           # Apply movement in appropriate direction
           if elevator.target_floor_direction == Direction.UP:
               new_floor = elevator.position.floor_up_position_add(movement_speed)
           elif elevator.target_floor_direction == Direction.DOWN:
               new_floor = elevator.position.floor_up_position_add(-movement_speed)

Position System
~~~~~~~~~~~~~~~

Positions use a **10-unit sub-floor** system:

- ``current_floor = 2, floor_up_position = 0`` → exactly at floor 2
- ``current_floor = 2, floor_up_position = 5`` → halfway between floors 2 and 3
- ``current_floor = 2, floor_up_position = 10`` → advances to ``current_floor = 3, floor_up_position = 0``

This granularity allows smooth movement and precise deceleration timing.

Deceleration Logic
~~~~~~~~~~~~~~~~~~

Elevators must decelerate before stopping:

.. code-block:: python

   def _should_start_deceleration(self, elevator: ElevatorState) -> bool:
       """Check if should start decelerating"""
       distance = self._calculate_distance_to_target(elevator)
       return distance == 1  # Start deceleration 1 position unit before target

   # In _move_elevators():
   if elevator.run_status == ElevatorStatus.CONSTANT_SPEED:
       if self._should_start_deceleration(elevator):
           elevator.run_status = ElevatorStatus.START_DOWN

This ensures elevators don't overshoot their target floor.

Event System
------------

Event Types
~~~~~~~~~~~

The simulation generates 9 types of events defined in ``EventType`` enum:

.. code-block:: python

   class EventType(Enum):
       UP_BUTTON_PRESSED = "up_button_pressed"
       DOWN_BUTTON_PRESSED = "down_button_pressed"
       PASSING_FLOOR = "passing_floor"
       STOPPED_AT_FLOOR = "stopped_at_floor"
       ELEVATOR_APPROACHING = "elevator_approaching"
       IDLE = "idle"
       PASSENGER_BOARD = "passenger_board"
       PASSENGER_ALIGHT = "passenger_alight"
       ELEVATOR_MOVE = "elevator_move"

Event Generation
~~~~~~~~~~~~~~~~

Events are generated during tick processing:

**Passenger Arrival**:

.. code-block:: python

   def _process_arrivals(self) -> None:
       """Process new passenger arrivals"""
       while self.traffic_queue and self.traffic_queue[0].tick <= self.tick:
           traffic_entry = self.traffic_queue.pop(0)
           passenger = PassengerInfo(
               id=traffic_entry.id,
               origin=traffic_entry.origin,
               destination=traffic_entry.destination,
               arrive_tick=self.tick,
           )
           self.passengers[passenger.id] = passenger

           if passenger.destination > passenger.origin:
               self.floors[passenger.origin].up_queue.append(passenger.id)
               # Generate UP_BUTTON_PRESSED event
               self._emit_event(
                   EventType.UP_BUTTON_PRESSED,
                   {"floor": passenger.origin, "passenger": passenger.id}
               )
           else:
               self.floors[passenger.origin].down_queue.append(passenger.id)
               # Generate DOWN_BUTTON_PRESSED event
               self._emit_event(
                   EventType.DOWN_BUTTON_PRESSED,
                   {"floor": passenger.origin, "passenger": passenger.id}
               )

**Elevator Movement**:

.. code-block:: python

   def _move_elevators(self) -> None:
       for elevator in self.elevators:
           # ... movement logic ...

           # Elevator moves
           if elevator.target_floor_direction != Direction.STOPPED:
               self._emit_event(
                   EventType.ELEVATOR_MOVE,
                   {
                       "elevator": elevator.id,
                       "from_position": old_position,
                       "to_position": elevator.position.current_floor_float,
                       "direction": elevator.target_floor_direction.value,
                       "status": elevator.run_status.value,
                   }
               )

           # Passing a floor
           if old_floor != new_floor and new_floor != target_floor:
               self._emit_event(
                   EventType.PASSING_FLOOR,
                   {
                       "elevator": elevator.id,
                       "floor": new_floor,
                       "direction": elevator.target_floor_direction.value
                   }
               )

           # About to arrive (during deceleration)
           if self._near_next_stop(elevator):
               self._emit_event(
                   EventType.ELEVATOR_APPROACHING,
                   {
                       "elevator": elevator.id,
                       "floor": int(round(elevator.position.current_floor_float)),
                       "direction": elevator.target_floor_direction.value
                   }
               )

           # Arrived at target
           if target_floor == new_floor and elevator.position.floor_up_position == 0:
               elevator.run_status = ElevatorStatus.STOPPED
               self._emit_event(
                   EventType.STOPPED_AT_FLOOR,
                   {
                       "elevator": elevator.id,
                       "floor": new_floor,
                       "reason": "move_reached"
                   }
               )

**Boarding and Alighting**:

.. code-block:: python

   def _process_elevator_stops(self) -> None:
       for elevator in self.elevators:
           if elevator.run_status != ElevatorStatus.STOPPED:
               continue

           current_floor = elevator.current_floor

           # Passengers alight
           passengers_to_remove = []
           for passenger_id in elevator.passengers:
               passenger = self.passengers[passenger_id]
               if passenger.destination == current_floor:
                   passenger.dropoff_tick = self.tick
                   passengers_to_remove.append(passenger_id)

           for passenger_id in passengers_to_remove:
               elevator.passengers.remove(passenger_id)
               self._emit_event(
                   EventType.PASSENGER_ALIGHT,
                   {"elevator": elevator.id, "floor": current_floor, "passenger": passenger_id}
               )

**Idle Detection**:

.. code-block:: python

   # If elevator stopped with no direction, it's idle
   if elevator.last_tick_direction == Direction.STOPPED:
       self._emit_event(
           EventType.IDLE,
           {"elevator": elevator.id, "floor": current_floor}
       )

Event Processing in Controller
-------------------------------

The ``ElevatorController`` base class automatically routes events to handler methods:

.. code-block:: python

   class ElevatorController(ABC):
       def _execute_events(self, events: List[SimulationEvent]) -> None:
           """Process events and route to handlers"""
           for event in events:
               if event.type == EventType.UP_BUTTON_PRESSED:
                   passenger_id = event.data["passenger"]
                   floor = self.floors[event.data["floor"]]
                   passenger = ProxyPassenger(passenger_id, self.api_client)
                   self.on_passenger_call(passenger, floor, "up")

               elif event.type == EventType.DOWN_BUTTON_PRESSED:
                   passenger_id = event.data["passenger"]
                   floor = self.floors[event.data["floor"]]
                   passenger = ProxyPassenger(passenger_id, self.api_client)
                   self.on_passenger_call(passenger, floor, "down")

               elif event.type == EventType.STOPPED_AT_FLOOR:
                   elevator = self.elevators[event.data["elevator"]]
                   floor = self.floors[event.data["floor"]]
                   self.on_elevator_stopped(elevator, floor)

               elif event.type == EventType.IDLE:
                   elevator = self.elevators[event.data["elevator"]]
                   self.on_elevator_idle(elevator)

               elif event.type == EventType.ELEVATOR_MOVE:
                   elevator = self.elevators[event.data["elevator"]]
                   from_position = event.data["from_position"]
                   to_position = event.data["to_position"]
                   direction = event.data["direction"]
                   status = event.data["status"]
                   self.on_elevator_move(elevator, from_position, to_position, direction, status)

               # ... other event types ...

Control Flow: Bus Example
--------------------------

The ``bus_example.py`` demonstrates a simple "bus route" algorithm:

.. code-block:: python

   class ElevatorBusExampleController(ElevatorController):
       def __init__(self):
           super().__init__("http://127.0.0.1:8000", True)
           self.all_passengers = []
           self.max_floor = 0

       def on_init(self, elevators: List[ProxyElevator], floors: List[ProxyFloor]) -> None:
           """Initialize elevators to starting positions"""
           self.max_floor = floors[-1].floor
           self.floors = floors

           for i, elevator in enumerate(elevators):
               # Distribute elevators evenly across floors
               target_floor = (i * (len(floors) - 1)) // len(elevators)
               elevator.go_to_floor(target_floor, immediate=True)

       def on_event_execute_start(self, tick: int, events: List[SimulationEvent],
                                   elevators: List[ProxyElevator], floors: List[ProxyFloor]) -> None:
           """Print state before processing events"""
           print(f"Tick {tick}: Processing {len(events)} events {[e.type.value for e in events]}")
           for elevator in elevators:
               print(
                   f"\t{elevator.id}[{elevator.target_floor_direction.value},"
                   f"{elevator.current_floor_float}/{elevator.target_floor}]"
                   + "👦" * len(elevator.passengers),
                   end=""
               )
           print()

       def on_elevator_stopped(self, elevator: ProxyElevator, floor: ProxyFloor) -> None:
           """Implement bus route logic"""
           print(f"🛑 Elevator E{elevator.id} stopped at F{floor.floor}")

           # Bus algorithm: go up to top, then down to bottom, repeat
           if elevator.last_tick_direction == Direction.UP and elevator.current_floor == self.max_floor:
               # At top, start going down
               elevator.go_to_floor(elevator.current_floor - 1)
           elif elevator.last_tick_direction == Direction.DOWN and elevator.current_floor == 0:
               # At bottom, start going up
               elevator.go_to_floor(elevator.current_floor + 1)
           elif elevator.last_tick_direction == Direction.UP:
               # Continue upward
               elevator.go_to_floor(elevator.current_floor + 1)
           elif elevator.last_tick_direction == Direction.DOWN:
               # Continue downward
               elevator.go_to_floor(elevator.current_floor - 1)

       def on_elevator_idle(self, elevator: ProxyElevator) -> None:
           """Send idle elevator to floor 1"""
           elevator.go_to_floor(1)

Execution Sequence
~~~~~~~~~~~~~~~~~~

Here's what happens in a typical tick:

.. code-block:: text

   Server: Tick 42
     Phase 1: Update status
       - Elevator 0: STOPPED → START_UP (has target)
     Phase 2: Process arrivals
       - Passenger 101 arrives at floor 0, going to floor 5
       - Event: UP_BUTTON_PRESSED
     Phase 3: Move elevators
       - Elevator 0: floor 2.0 → 2.1 (accelerating)
     Phase 4: Process stops
       - (no stops this tick)

     Events: [UP_BUTTON_PRESSED, PASSING_FLOOR]

   Client: Receive events
     on_event_execute_start(tick=42, events=[...])
       - Print "Tick 42: Processing 2 events"

     _execute_events():
       - UP_BUTTON_PRESSED → on_passenger_call()
         → Controller decides which elevator to send
       - PASSING_FLOOR → on_elevator_passing_floor()

     on_event_execute_end(tick=42, events=[...])

   Client: Send commands
     - elevator.go_to_floor(0)  → POST /api/elevators/0/go_to_floor

   Client: Step simulation
     - POST /api/step → Server processes tick 43

Key Timing Concepts
-------------------

Immediate vs. Queued
~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

   # Queued (default): Wait until current target reached
   elevator.go_to_floor(5, immediate=False)
   # ← Sets elevator.next_target_floor = 5
   # ← Processed when current_floor == target_floor

   # Immediate: Change target right away
   elevator.go_to_floor(5, immediate=True)
   # ← Sets elevator.position.target_floor = 5 immediately
   # ← May interrupt current journey

Use ``immediate=True`` for emergency redirects, ``immediate=False`` (default) for normal operation.

Performance Metrics
-------------------

Metrics are calculated from passenger data:

.. code-block:: python

   def _calculate_metrics(self) -> PerformanceMetrics:
       """Calculate performance metrics"""
       completed = [p for p in self.state.passengers.values()
                    if p.status == PassengerStatus.COMPLETED]

       floor_wait_times = [float(p.floor_wait_time) for p in completed]
       arrival_wait_times = [float(p.arrival_wait_time) for p in completed]

       def average_excluding_top_percent(data: List[float], exclude_percent: int) -> float:
           """计算排除掉最长的指定百分比后的平均值"""
           if not data:
               return 0.0
           sorted_data = sorted(data)
           keep_count = int(len(sorted_data) * (100 - exclude_percent) / 100)
           if keep_count == 0:
               return 0.0
           kept_data = sorted_data[:keep_count]
           return sum(kept_data) / len(kept_data)

       return PerformanceMetrics(
           completed_passengers=len(completed),
           total_passengers=len(self.state.passengers),
           average_floor_wait_time=sum(floor_wait_times) / len(floor_wait_times) if floor_wait_times else 0,
           p95_floor_wait_time=average_excluding_top_percent(floor_wait_times, 5),
           average_arrival_wait_time=sum(arrival_wait_times) / len(arrival_wait_times) if arrival_wait_times else 0,
           p95_arrival_wait_time=average_excluding_top_percent(arrival_wait_times, 5),
       )

Key metrics:

- **Floor wait time**: ``pickup_tick - arrive_tick`` (在楼层等待的时间，从到达到上电梯)
- **Arrival wait time**: ``dropoff_tick - arrive_tick`` (总等待时间，从到达到下电梯)
- **P95 metrics**: 排除掉最长的5%时间后，计算剩余95%的平均值

Summary
-------

The event-driven, tick-based architecture provides:

- **Deterministic**: Same inputs always produce same results
- **Testable**: Easy to create test scenarios with traffic files
- **Debuggable**: Clear event trail shows what happened when
- **Flexible**: Easy to implement different dispatch algorithms
- **Scalable**: Can simulate large buildings and many passengers

Next Steps
----------

- Study ``bus_example.py`` for a complete working example
- Implement your own controller by extending ``ElevatorController``
- Experiment with different dispatch algorithms
- Analyze performance metrics to optimize your approach