API Reference

Core Modules

Elevator Saga Data Models 统一的数据模型定义，用于客户端和服务器的类型一致性和序列化

class elevator_saga.core.models.APIRequest(request_id: str = <factory>, timestamp: str = <factory>)

Bases: SerializableModel

API请求基类

request_id: str

timestamp: str

class elevator_saga.core.models.APIResponse(success: bool, request_id: str | None = None, error_message: str | None = None, timestamp: str = <factory>)

Bases: SerializableModel

API响应基类

error_message: str | None = None

request_id: str | None = None

success: bool

timestamp: str

class elevator_saga.core.models.Direction(value)

Bases: Enum

电梯方向枚举

DOWN = 'down'

STOPPED = 'stopped'

UP = 'up'

class elevator_saga.core.models.ElevatorCommand(elevator_id: int, command_type: str, parameters: ~typing.Dict[str, ~typing.Any] = <factory>, request_id: str = <factory>, timestamp: str = <factory>)

Bases: SerializableModel

电梯命令

command_type: str

elevator_id: int

parameters: Dict[str, Any]

request_id: str

timestamp: str

class elevator_saga.core.models.ElevatorCommandResponse(success: bool, elevator_id: int)

Bases: SerializableModel

电梯命令响应

elevator_id: int

success: bool

class elevator_saga.core.models.ElevatorIndicators(up: bool = False, down: bool = False)

Bases: SerializableModel

电梯指示灯状态

down: bool = False

set_direction(direction: Direction) → None

根据方向设置指示灯

up: bool = False

class elevator_saga.core.models.ElevatorState(id: int, position: ~elevator_saga.core.models.Position, next_target_floor: int | None = None, passengers: ~typing.List[int] = <factory>, max_capacity: int = 10, speed_pre_tick: float = 0.5, run_status: ~elevator_saga.core.models.ElevatorStatus = ElevatorStatus.STOPPED, last_tick_direction: ~elevator_saga.core.models.Direction = Direction.STOPPED, indicators: ~elevator_saga.core.models.ElevatorIndicators = <factory>, passenger_destinations: ~typing.Dict[int, int] = <factory>, energy_consumed: float = 0.0, energy_rate: float = 1.0, last_update_tick: int = 0)

Bases: SerializableModel

电梯状态

clear_destinations() → None

清空目标队列

property current_floor: int

当前楼层

property current_floor_float: float

当前楼层

energy_consumed: float = 0.0

energy_rate: float = 1.0

id: int

indicators: ElevatorIndicators

property is_full: bool

是否满载

property is_idle: bool

是否空闲

property is_running: bool

是否正在运行

last_tick_direction: Direction = 'stopped'

last_update_tick: int = 0

property load_factor: float

载重系数

max_capacity: int = 10

next_target_floor: int | None = None

passenger_destinations: Dict[int, int]

passengers: List[int]

position: Position

property pressed_floors: List[int]

按下的楼层（基于当前乘客的目的地动态计算）

run_status: ElevatorStatus = 'stopped'

speed_pre_tick: float = 0.5

property target_floor: int

当前楼层

property target_floor_direction: Direction

目标方向

class elevator_saga.core.models.ElevatorStatus(value)

Bases: Enum

电梯运行状态机：stopped -1tick> start_up -1tick> constant_speed -?tick> start_down -1tick> stopped 注意：START_UP/START_DOWN表示加速/减速状态，不表示移动方向 实际移动方向由target_floor_direction属性决定

CONSTANT_SPEED = 'constant_speed'

START_DOWN = 'start_down'

START_UP = 'start_up'

STOPPED = 'stopped'

class elevator_saga.core.models.EventType(value)

Bases: Enum

事件类型枚举

DOWN_BUTTON_PRESSED = 'down_button_pressed'

ELEVATOR_APPROACHING = 'elevator_approaching'

ELEVATOR_MOVE = 'elevator_move'

IDLE = 'idle'

PASSENGER_ALIGHT = 'passenger_alight'

PASSENGER_BOARD = 'passenger_board'

PASSING_FLOOR = 'passing_floor'

STOPPED_AT_FLOOR = 'stopped_at_floor'

UP_BUTTON_PRESSED = 'up_button_pressed'

class elevator_saga.core.models.FloorState(floor: int, up_queue: ~typing.List[int] = <factory>, down_queue: ~typing.List[int] = <factory>)

Bases: SerializableModel

楼层状态

add_waiting_passenger(passenger_id: int, direction: Direction) → None

添加等待乘客

down_queue: List[int]

floor: int

property has_waiting_passengers: bool

是否有等待的乘客

remove_waiting_passenger(passenger_id: int) → bool

移除等待乘客

property total_waiting: int

总等待人数

up_queue: List[int]

class elevator_saga.core.models.GoToFloorCommand(elevator_id: int, floor: int, immediate: bool = False, request_id: str = <factory>, timestamp: str = <factory>)

Bases: SerializableModel

前往楼层命令

command_type: str = 'go_to_floor'

elevator_id: int

floor: int

immediate: bool = False

property parameters: Dict[str, Any]

request_id: str

timestamp: str

class elevator_saga.core.models.PassengerInfo(id: int, origin: int, destination: int, arrive_tick: int, pickup_tick: int = 0, dropoff_tick: int = 0, arrived: bool = False, elevator_id: int | None = None)

Bases: SerializableModel

乘客信息

property arrival_wait_time: int

总等待时间（从到达到下电梯）

arrive_tick: int

arrived: bool = False

destination: int

dropoff_tick: int = 0

elevator_id: int | None = None

property floor_wait_time: int

在楼层等待的时间（从到达到上电梯）

id: int

origin: int

pickup_tick: int = 0

property status: PassengerStatus

乘客状态

property travel_direction: Direction

移动方向

class elevator_saga.core.models.PassengerStatus(value)

Bases: Enum

乘客状态枚举

CANCELLED = 'cancelled'

COMPLETED = 'completed'

IN_ELEVATOR = 'in_elevator'

WAITING = 'waiting'

class elevator_saga.core.models.PerformanceMetrics(completed_passengers: int = 0, total_passengers: int = 0, average_floor_wait_time: float = 0.0, p95_floor_wait_time: float = 0.0, average_arrival_wait_time: float = 0.0, p95_arrival_wait_time: float = 0.0, total_energy_consumption: float = 0.0)

Bases: SerializableModel

性能指标

average_arrival_wait_time: float = 0.0

average_floor_wait_time: float = 0.0

completed_passengers: int = 0

property completion_rate: float

完成率

p95_arrival_wait_time: float = 0.0

p95_floor_wait_time: float = 0.0

total_energy_consumption: float = 0.0

total_passengers: int = 0

class elevator_saga.core.models.Position(current_floor: int = 0, target_floor: int = 0, floor_up_position: int = 0)

Bases: SerializableModel

位置信息

current_floor: int = 0

property current_floor_float: float

floor_up_position: int = 0

floor_up_position_add(num: int) → int

target_floor: int = 0

class elevator_saga.core.models.SerializableModel

Bases: object

可序列化模型基类

classmethod from_dict(data: Dict[str, Any]) → T

从字典创建实例

classmethod from_json(json_str: str) → T

从JSON字符串创建实例

to_dict() → Dict[str, Any]

转换为字典

to_json() → str

转换为JSON字符串

class elevator_saga.core.models.SimulationEvent(tick: int, type: EventType, data: Dict[str, Any], timestamp: str | None = None)

Bases: SerializableModel

模拟事件

data: Dict[str, Any]

tick: int

timestamp: str | None = None

type: EventType

class elevator_saga.core.models.SimulationState(tick: int, elevators: ~typing.List[~elevator_saga.core.models.ElevatorState], floors: ~typing.List[~elevator_saga.core.models.FloorState], passengers: ~typing.Dict[int, ~elevator_saga.core.models.PassengerInfo] = <factory>, metrics: ~elevator_saga.core.models.PerformanceMetrics = <factory>, events: ~typing.List[~elevator_saga.core.models.SimulationEvent] = <factory>)

Bases: SerializableModel

模拟状态

add_event(event_type: EventType, data: Dict[str, Any]) → None

添加事件

elevators: List[ElevatorState]

events: List[SimulationEvent]

floors: List[FloorState]

get_elevator_by_id(elevator_id: int) → ElevatorState | None

根据ID获取电梯

get_floor_by_number(floor_number: int) → FloorState | None

根据楼层号获取楼层

get_passengers_by_status(status: PassengerStatus) → List[PassengerInfo]

根据状态获取乘客

metrics: PerformanceMetrics

passengers: Dict[int, PassengerInfo]

tick: int

class elevator_saga.core.models.StateRequest(request_id: str = <factory>, timestamp: str = <factory>, include_passengers: bool = True, include_events: bool = False, since_tick: int | None = None)

Bases: APIRequest

状态请求

include_events: bool = False

include_passengers: bool = True

since_tick: int | None = None

class elevator_saga.core.models.StepRequest(request_id: str = <factory>, timestamp: str = <factory>, ticks: int = 1)

Bases: APIRequest

步进请求

ticks: int = 1

class elevator_saga.core.models.StepResponse(success: bool, tick: int, events: ~typing.List[~elevator_saga.core.models.SimulationEvent] = <factory>, request_id: str | None = None, error_message: str | None = None, timestamp: str = <factory>)

Bases: SerializableModel

步进响应

error_message: str | None = None

events: List[SimulationEvent]

request_id: str | None = None

success: bool

tick: int

timestamp: str

class elevator_saga.core.models.TrafficEntry(id: int, origin: int, destination: int, tick: int)

Bases: SerializableModel

流量条目

destination: int

id: int

origin: int

tick: int

class elevator_saga.core.models.TrafficPattern(name: str, description: str, entries: ~typing.List[~elevator_saga.core.models.TrafficEntry] = <factory>, metadata: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: SerializableModel

流量模式

add_entry(entry: TrafficEntry) → None

添加流量条目

description: str

property duration: int

流量持续时间

entries: List[TrafficEntry]

get_entries_for_tick(tick: int) → List[TrafficEntry]

获取指定tick的流量条目

metadata: Dict[str, Any]

name: str

property total_passengers: int

总乘客数

elevator_saga.core.models.create_empty_simulation_state(elevators: int, floors: int, max_capacity: int) → SimulationState

创建空的模拟状态

elevator_saga.core.models.create_simple_traffic_pattern(name: str, passengers: List[Tuple[int, int, int]]) → TrafficPattern

创建简单流量模式

Parameters:

• name – 模式名称

• passengers – [(origin, destination, tick), …]

Client Modules

Unified API Client for Elevator Saga 使用统一数据模型的客户端API封装

class elevator_saga.client.api_client.ElevatorAPIClient(base_url: str)

Bases: object

统一的电梯API客户端

get_state(force_reload: bool = False) → SimulationState

获取模拟状态

Parameters:

force_reload – 是否强制重新加载，忽略缓存

get_traffic_info() → Dict[str, Any] | None

获取当前流量文件信息

go_to_floor(elevator_id: int, floor: int, immediate: bool = False) → bool

电梯前往指定楼层

mark_tick_processed() → None

标记当前tick处理完成，使缓存在下次get_state时失效

next_traffic_round(full_reset: bool = False) → bool

切换到下一个流量文件

reset() → bool

重置模拟

send_elevator_command(command: GoToFloorCommand) → bool

发送电梯命令

step(ticks: int = 1) → StepResponse

执行步进

class elevator_saga.client.proxy_models.ProxyElevator(elevator_id: int, api_client: ElevatorAPIClient)

Bases: ElevatorState

电梯动态代理类 直接使用 ElevatorState 数据模型实例，提供完整的类型安全访问和操作方法

__setattr__(name: str, value: Any) → None

禁止修改属性，保持只读特性

go_to_floor(floor: int, immediate: bool = False) → bool

前往指定楼层

class elevator_saga.client.proxy_models.ProxyFloor(floor_id: int, api_client: ElevatorAPIClient)

Bases: FloorState

楼层动态代理类 直接使用 FloorState 数据模型实例，提供完整的类型安全访问

__setattr__(name: str, value: Any) → None

禁止修改属性，保持只读特性

class elevator_saga.client.proxy_models.ProxyPassenger(passenger_id: int, api_client: ElevatorAPIClient)

Bases: PassengerInfo

乘客动态代理类 直接使用 PassengerInfo 数据模型实例，提供完整的类型安全访问

__setattr__(name: str, value: Any) → None

禁止修改属性，保持只读特性

Elevator Controller Base Class 电梯调度基础控制器类 - 提供面向对象的算法开发接口

class elevator_saga.client.base_controller.ElevatorController(server_url: str = 'http://127.0.0.1:8000', debug: bool = False)

Bases: ABC

电梯调度控制器基类

用户通过继承此类并实现 abstract 方法来创建自己的调度算法

__init__(server_url: str = 'http://127.0.0.1:8000', debug: bool = False)

初始化控制器

Parameters:

• server_url – 服务器URL

• debug – 是否启用debug模式

abstract on_elevator_approaching(elevator: ProxyElevator, floor: ProxyFloor, direction: str) → None

电梯即将到达时的回调 - 可选实现

Parameters:

• elevator – 电梯代理对象

• floor – 即将到达的楼层代理对象

• direction – 移动方向

abstract on_elevator_idle(elevator: ProxyElevator) → None

电梯空闲时的回调 - 可选实现

Parameters:

elevator – 空闲的电梯代理对象

on_elevator_move(elevator: ProxyElevator, from_position: float, to_position: float, direction: str, status: str) → None

电梯移动时的回调 - 可选实现

Parameters:

• elevator – 电梯代理对象

• from_position – 起始位置（浮点数表示楼层）

• to_position – 目标位置（浮点数表示楼层）

• direction – 移动方向

• status – 电梯运行状态

abstract on_elevator_passing_floor(elevator: ProxyElevator, floor: ProxyFloor, direction: str) → None

电梯经过楼层时的回调 - 可选实现

Parameters:

• elevator – 电梯代理对象

• floor – 经过的楼层代理对象

• direction – 移动方向

abstract on_elevator_stopped(elevator: ProxyElevator, floor: ProxyFloor) → None

电梯停靠时的回调 - 可选实现

Parameters:

• elevator – 停靠的电梯代理对象

• floor – 停靠楼层代理对象

abstract on_event_execute_end(tick: int, events: List[Any], elevators: List[Any], floors: List[Any]) → None

事件执行后的回调 - 必须由子类实现

Parameters:

• tick – 当前时间tick

• events – 已执行的事件列表

• elevators – 电梯列表

• floors – 楼层列表

abstract on_event_execute_start(tick: int, events: List[Any], elevators: List[Any], floors: List[Any]) → None

事件执行前的回调 - 必须由子类实现

Parameters:

• tick – 当前时间tick

• events – 即将执行的事件列表

• elevators – 电梯列表

• floors – 楼层列表

abstract on_init(elevators: List[Any], floors: List[Any]) → None

算法初始化方法 - 必须由子类实现

Parameters:

• elevators – 电梯列表

• floors – 楼层列表

abstract on_passenger_alight(elevator: ProxyElevator, passenger: ProxyPassenger, floor: ProxyFloor) → None

乘客下车时的回调 - 可选实现

Parameters:

• elevator – 电梯代理对象

• passenger – 乘客代理对象

• floor – 下车楼层代理对象

abstract on_passenger_board(elevator: ProxyElevator, passenger: ProxyPassenger) → None

乘客上梯时的回调 - 可选实现

Parameters:

• elevator – 电梯代理对象

• passenger – 乘客代理对象

abstract on_passenger_call(passenger: ProxyPassenger, floor: ProxyFloor, direction: str) → None

乘客呼叫时的回调 - 可选实现

Parameters:

• floor – 呼叫楼层代理对象

• direction – 方向 (“up” 或 “down”)

on_simulation_complete(final_state: Dict[str, Any]) → None

模拟完成时的回调 - 可选实现

Parameters:

final_state – 最终状态数据

on_start() → None

算法启动前的回调 - 可选实现

on_stop() → None

算法停止后的回调 - 可选实现

start() → None

启动控制器

stop() → None

停止控制器

Server Modules

Elevator simulation server - tick-based discrete event simulation Provides HTTP API for controlling elevators and advancing simulation time

class elevator_saga.server.simulator.CustomJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: JSONEncoder

自定义JSON编码器，处理Enum和其他特殊类型的序列化

default(o: Any) → Any

重写默认序列化方法，处理特殊类型

Parameters:

o – 要序列化的对象

Returns:

序列化后的值

class elevator_saga.server.simulator.ElevatorSimulation(traffic_dir: str, _init_only: bool = False)

Bases: object

elevator_go_to_floor(elevator_id: int, floor: int, immediate: bool = False) → None

设置电梯去向，是生命周期开始，分配目的地

property elevators: List[ElevatorState]

电梯列表

property floors: List[FloorState]

楼层列表

force_complete_remaining_passengers() → int

强制完成所有未完成的乘客，返回完成的乘客数量

get_events(since_tick: int = 0) → List[SimulationEvent]

Get events since specified tick

get_state() → SimulationStateResponse

Get complete simulation state

get_traffic_info() → Dict[str, Any]

load_current_traffic() → None

加载当前索引对应的流量文件

load_traffic(traffic_file: str) → None

Load passenger traffic from JSON file using unified data models

max_duration_ticks: int

next_passenger_id: int

next_traffic_round(full_reset: bool = False) → bool

切换到下一个流量文件，返回是否成功切换

property passengers: Dict[int, PassengerInfo]

乘客字典

reset() → None

Reset simulation to initial state

step(num_ticks: int = 1) → List[SimulationEvent]

property tick: int

当前tick

traffic_queue: List[TrafficEntry]

class elevator_saga.server.simulator.PassengerSummary(completed: int, waiting: int, in_transit: int, total: int)

Bases: SerializableModel

乘客摘要

completed: int

in_transit: int

total: int

waiting: int

class elevator_saga.server.simulator.SimulationStateResponse(tick: int, elevators: List[ElevatorState], floors: List[FloorState], passengers: Dict[int, PassengerInfo], metrics: PerformanceMetrics)

Bases: SerializableModel

模拟状态响应

elevators: List[ElevatorState]

floors: List[FloorState]

metrics: PerformanceMetrics

passengers: Dict[int, PassengerInfo]

tick: int

elevator_saga.server.simulator.after_request(response: Response) → Response

elevator_saga.server.simulator.elevator_go_to_floor(elevator_id: int) → Response | tuple[Response, int]

elevator_saga.server.simulator.get_state() → Response | tuple[Response, int]

elevator_saga.server.simulator.get_traffic_info() → Response | tuple[Response, int]

获取当前流量文件信息

elevator_saga.server.simulator.json_response(data: Any, status: int = 200) → Response | tuple[Response, int]

创建JSON响应，使用自定义编码器处理Enum等特殊类型

Parameters:

• data – 要序列化的数据

• status – HTTP状态码

Returns:

Flask Response对象，或者Response和状态码的元组（当状态码不是200时）

elevator_saga.server.simulator.main() → None

elevator_saga.server.simulator.next_traffic_round() → Response | tuple[Response, int]

切换到下一个流量文件

elevator_saga.server.simulator.reset_simulation() → Response | tuple[Response, int]

elevator_saga.server.simulator.server_debug_log(message: str) → None

Print server debug message if debug mode is enabled

elevator_saga.server.simulator.set_server_debug_mode(enabled: bool) → None

Enable or disable server debug logging

elevator_saga.server.simulator.step_simulation() → Response | tuple[Response, int]