open_pub/Elevator
1
0
Fork 0
mirror of https://github.com/ZGCA-Forge/Elevator.git synced 2025-12-19 14:01:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix downstairs

Browse Source
This commit is contained in:
Xuwznln
2025-09-28 20:23:11 +08:00
parent 16d2fec0ac
commit 6a3b312e41
6 changed files with 203 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
elevator_saga/scripts/client_examples/simple_example.py
View File

@@ -7,7 +7,7 @@ from typing import Dict, List
from elevator_saga.client.base_controller import ElevatorController
from elevator_saga.client.proxy_models import ProxyElevator, ProxyFloor, ProxyPassenger
from elevator_saga.core.models import SimulationEvent
from elevator_saga.core.models import SimulationEvent, Direction
class ElevatorBusController(ElevatorController):
@@ -43,9 +43,9 @@ class ElevatorBusController(ElevatorController):
self, tick: int, events: List[SimulationEvent], elevators: List[ProxyElevator], floors: List[ProxyFloor]
) -> None:
"""事件执行前的回调"""
print(f"Tick {tick}: 即将处理 {len(events)} 个事件 {[e.type for e in events]}", end="")
print(f"Tick {tick}: 即将处理 {len(events)} 个事件 {[e.type.value for e in events]}")
for i in elevators:
print(f"电梯{i.id}[{i.target_floor_direction.value}] 位置{i.current_floor_float}/{i.target_floor}, ", end="")
print(f"\t{i.id}[{i.target_floor_direction.value},{i.current_floor_float}/{i.target_floor}]" + "👦" * len(i.passengers), end="")
print()
def on_event_execute_end(
@@ -55,29 +55,31 @@ class ElevatorBusController(ElevatorController):
# print(f"✅ Tick {tick}: 已处理 {len(events)} 个事件")
pass
def on_passenger_call(self, floor: ProxyFloor, direction: str) -> None:
def on_passenger_call(self, passenger:ProxyPassenger, floor: ProxyFloor, direction: str) -> None:
"""
乘客呼叫时的回调
公交车模式下，电梯已经在循环运行，无需特别响应呼叫
"""
print(f"📞 楼层 {floor.floor} 有乘客呼叫 ({direction}) - 公交车将按既定路线服务")
print(f"乘客 {passenger.id} F{floor.floor} 请求 {passenger.origin} -> {passenger.destination} ({direction})")
def on_elevator_idle(self, elevator: ProxyElevator) -> None:
"""
电梯空闲时的回调
让空闲的电梯继续执行公交车循环路线，每次移动一层楼
"""
print(f"🛑 电梯 {elevator.id} 空闲 {elevator.current_floor} = {elevator.target_floor}")
print(f"🛑 电梯 E{elevator.id} 在 F{elevator.current_floor} 层空闲")
# 设置指示器让乘客知道电梯的行进方向
if self.elevator_directions[elevator.id] == "down" and elevator.current_floor != 0:
elevator.go_to_floor(elevator.current_floor - 1, immediate=True)
current_direction = self.elevator_directions[elevator.id]
if current_direction == "up":
elevator.set_up_indicator(True)
elevator.set_down_indicator(False)
else:
elevator.set_up_indicator(False)
elevator.set_down_indicator(True)
# elevator.set_up_indicator(True)
elevator.go_to_floor(1)
# current_direction = self.elevator_directions[elevator.id]
# if current_direction == "up":
# elevator.set_up_indicator(True)
# elevator.set_down_indicator(False)
# else:
# elevator.set_up_indicator(False)
# elevator.set_down_indicator(True)
def on_elevator_stopped(self, elevator: ProxyElevator, floor: ProxyFloor) -> None:
"""
@@ -85,7 +87,7 @@ class ElevatorBusController(ElevatorController):
公交车模式下，在每一层都停下，然后继续下一站
需要注意的是，stopped会比idle先触发
"""
print(f"🛑 电梯 {elevator.id} 停靠在 {floor.floor} 楼")
print(f"🛑 电梯 E{elevator.id} 停靠在 F{floor.floor}")
if self.elevator_directions[elevator.id] == "up" and elevator.current_floor == self.max_floor:
elevator.go_to_floor(elevator.current_floor - 1, immediate=True)
self.elevator_directions[elevator.id] = "down"
@@ -93,41 +95,47 @@ class ElevatorBusController(ElevatorController):
elevator.go_to_floor(elevator.current_floor + 1, immediate=True)
self.elevator_directions[elevator.id] = "up"
elif self.elevator_directions[elevator.id] == "up":
if elevator.id == 0:
raise ValueError("这里故意要求0号电梯不可能触发非两端停止，通过on_elevator_approaching实现")
elevator.go_to_floor(elevator.current_floor + 1, immediate=True)
# 这里故意少写下降的情况，用于了解stopped会先于idle触发
# elif self.elevator_directions[elevator.id] == "down":
# elevator.go_to_floor(elevator.current_floor - 1, immediate=True)
# self.elevator_directions[elevator.id] = "down"
def on_passenger_board(self, elevator: ProxyElevator, passenger: ProxyPassenger) -> None:
"""
乘客上车时的回调
打印乘客上车信息
乘客上梯时的回调
打印乘客上梯信息
"""
print(f"⬆️ 乘客 {passenger.id} 上车 - 电梯 {elevator.id} - 楼层 {elevator.current_floor} - 目标楼层: {passenger.destination}")
print(
f" 乘客{passenger.id} E{elevator.id}⬆️ F{elevator.current_floor} -> F{passenger.destination}"
)
def on_passenger_alight(self, elevator: ProxyElevator, passenger: ProxyPassenger, floor: ProxyFloor) -> None:
"""
乘客下车时的回调
打印乘客下车信息
"""
print(f"⬇️ 乘客 {passenger.id} 在 {floor.floor} 楼下车 - 电梯 {elevator.id}")
print(f" 乘客{passenger.id} E{elevator.id}⬇️ F{floor.floor}")
def on_elevator_passing_floor(self, elevator: ProxyElevator, floor: ProxyFloor, direction: str) -> None:
"""
电梯经过楼层时的回调
打印经过楼层的信息
"""
print(f"🔄 电梯 {elevator.id} 经过 {floor.floor} 楼 (方向: {direction})")
print(f"🔄 电梯 E{elevator.id} 经过 F{floor.floor} (方向: {direction})")
def on_elevator_approaching(self, elevator: ProxyElevator, floor: ProxyFloor, direction: str) -> None:
"""
电梯即将到达时的回调 (START_DOWN事件)
电梯开始减速，即将到达目标楼层
"""
print(f"🎯 电梯 {elevator.id} 即将到达 {floor.floor} 楼 (方向: {direction})")
print(f"🎯 电梯 E{elevator.id} 即将到达 F{floor.floor} (方向: {direction})")
if elevator.target_floor == floor.floor and elevator.target_floor_direction == Direction.UP: # 电梯的目标楼层就是即将停靠的楼层
if elevator.id == 0: # 这里为了测试，让0号电梯往上一层就新加一层，上行永远不会开门
elevator.go_to_floor(elevator.target_floor + 1, immediate=True)
print(f" 不让0号电梯上行停站，设定新目标楼层 {elevator.target_floor + 1}")
if __name__ == "__main__":
algorithm = ElevatorBusController(debug=True)