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

┌──────────────────────────────────────────────────────────┐
│                  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:

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:

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:

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()):

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:

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:

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:

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:

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:

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:

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:

# 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:

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:

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:

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

# 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:

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