mirror of
https://github.com/ZGCA-Forge/Elevator.git
synced 2026-02-04 21:35:17 +00:00
added other examples
This commit is contained in:
@@ -14,7 +14,6 @@ from elevator_saga.core.models import (
|
||||
GoToFloorCommand,
|
||||
PassengerInfo,
|
||||
PerformanceMetrics,
|
||||
SetIndicatorsCommand,
|
||||
SimulationEvent,
|
||||
SimulationState,
|
||||
StepResponse,
|
||||
@@ -132,7 +131,7 @@ class ElevatorAPIClient:
|
||||
else:
|
||||
raise RuntimeError(f"Step failed: {response_data.get('error')}")
|
||||
|
||||
def send_elevator_command(self, command: Union[GoToFloorCommand, SetIndicatorsCommand]) -> bool:
|
||||
def send_elevator_command(self, command: Union[GoToFloorCommand]) -> bool:
|
||||
"""发送电梯命令"""
|
||||
endpoint = self._get_elevator_endpoint(command)
|
||||
debug_log(f"Sending elevator command: {command.command_type} to elevator {command.elevator_id} To:F{command.floor}")
|
||||
@@ -155,25 +154,12 @@ class ElevatorAPIClient:
|
||||
debug_log(f"Go to floor failed: {e}")
|
||||
return False
|
||||
|
||||
def set_indicators(self, elevator_id: int, up: Optional[bool] = None, down: Optional[bool] = None) -> bool:
|
||||
"""设置电梯指示灯"""
|
||||
command = SetIndicatorsCommand(elevator_id=elevator_id, up=up, down=down)
|
||||
|
||||
try:
|
||||
response = self.send_elevator_command(command)
|
||||
return response
|
||||
except Exception as e:
|
||||
debug_log(f"Set indicators failed: {e}")
|
||||
return False
|
||||
|
||||
def _get_elevator_endpoint(self, command: Union[GoToFloorCommand, SetIndicatorsCommand]) -> str:
|
||||
def _get_elevator_endpoint(self, command: Union[GoToFloorCommand]) -> str:
|
||||
"""获取电梯命令端点"""
|
||||
base = f"/api/elevators/{command.elevator_id}"
|
||||
|
||||
if isinstance(command, GoToFloorCommand):
|
||||
return f"{base}/go_to_floor"
|
||||
else: # SetIndicatorsCommand
|
||||
return f"{base}/set_indicators"
|
||||
|
||||
def _send_get_request(self, endpoint: str) -> Dict[str, Any]:
|
||||
"""发送GET请求"""
|
||||
|
||||
@@ -10,48 +10,36 @@ class ProxyFloor(FloorState):
|
||||
直接使用 FloorState 数据模型实例,提供完整的类型安全访问
|
||||
"""
|
||||
|
||||
init_ok = False
|
||||
_init_ok = False
|
||||
|
||||
def __init__(self, floor_id: int, api_client: ElevatorAPIClient):
|
||||
self._floor_id = floor_id
|
||||
self._api_client = api_client
|
||||
self._cached_instance = None
|
||||
self.init_ok = True
|
||||
self._init_ok = True
|
||||
|
||||
def _get_floor_state(self) -> FloorState:
|
||||
"""获取 FloorState 实例"""
|
||||
# 获取当前状态
|
||||
state = self._api_client.get_state()
|
||||
floor_data = next((f for f in state.floors if f.floor == self._floor_id), None)
|
||||
return floor_data
|
||||
|
||||
if floor_data is None:
|
||||
raise AttributeError(f"Floor {self._floor_id} not found")
|
||||
|
||||
# 如果是字典,转换为 FloorState 实例
|
||||
if isinstance(floor_data, dict):
|
||||
return FloorState.from_dict(floor_data)
|
||||
def __getattribute__(self, name: str) -> Any:
|
||||
if not name.startswith("_") and self._init_ok and name not in self.__class__.__dict__:
|
||||
try:
|
||||
self_attr = object.__getattribute__(self, name)
|
||||
if callable(self_attr):
|
||||
return object.__getattribute__(self, name)
|
||||
except AttributeError:
|
||||
pass
|
||||
floor_state = self._get_floor_state()
|
||||
return floor_state.__getattribute__(name)
|
||||
else:
|
||||
# 如果已经是 FloorState 实例,直接返回
|
||||
return floor_data
|
||||
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
"""动态获取楼层属性"""
|
||||
floor_state = self._get_floor_state()
|
||||
try:
|
||||
if hasattr(floor_state, name):
|
||||
attr = getattr(floor_state, name)
|
||||
# 如果是 property 或方法,调用并返回结果
|
||||
if callable(attr):
|
||||
return attr()
|
||||
else:
|
||||
return attr
|
||||
except AttributeError:
|
||||
|
||||
raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")
|
||||
return object.__getattribute__(self, name)
|
||||
|
||||
def __setattr__(self, name: str, value: Any) -> None:
|
||||
"""禁止修改属性,保持只读特性"""
|
||||
if not self.init_ok:
|
||||
if not self._init_ok:
|
||||
object.__setattr__(self, name, value)
|
||||
else:
|
||||
raise AttributeError(f"Cannot modify read-only attribute '{name}'")
|
||||
@@ -66,54 +54,40 @@ class ProxyElevator(ElevatorState):
|
||||
直接使用 ElevatorState 数据模型实例,提供完整的类型安全访问和操作方法
|
||||
"""
|
||||
|
||||
init_ok = False
|
||||
_init_ok = False
|
||||
|
||||
def __init__(self, elevator_id: int, api_client: ElevatorAPIClient):
|
||||
self._elevator_id = elevator_id
|
||||
self._api_client = api_client
|
||||
self.init_ok = True
|
||||
self._init_ok = True
|
||||
|
||||
def _get_elevator_state(self) -> ElevatorState:
|
||||
"""获取 ElevatorState 实例"""
|
||||
# 获取当前状态
|
||||
state = self._api_client.get_state()
|
||||
elevator_data = next((e for e in state.elevators if e.id == self._elevator_id), None)
|
||||
return elevator_data
|
||||
|
||||
if elevator_data is None:
|
||||
raise AttributeError(f"Elevator {self._elevator_id} not found")
|
||||
|
||||
# 如果是字典,转换为 ElevatorState 实例
|
||||
if isinstance(elevator_data, dict):
|
||||
return ElevatorState.from_dict(elevator_data)
|
||||
else:
|
||||
# 如果已经是 ElevatorState 实例,直接返回
|
||||
return elevator_data
|
||||
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
"""动态获取电梯属性"""
|
||||
try:
|
||||
def __getattribute__(self, name: str) -> Any:
|
||||
if not name.startswith("_") and self._init_ok and name not in self.__class__.__dict__:
|
||||
try:
|
||||
self_attr = object.__getattribute__(self, name)
|
||||
if callable(self_attr):
|
||||
return object.__getattribute__(self, name)
|
||||
except AttributeError:
|
||||
pass
|
||||
elevator_state = self._get_elevator_state()
|
||||
# 直接从 ElevatorState 实例获取属性
|
||||
return getattr(elevator_state, name)
|
||||
|
||||
except AttributeError:
|
||||
raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")
|
||||
return elevator_state.__getattribute__(name)
|
||||
else:
|
||||
return object.__getattribute__(self, name)
|
||||
|
||||
def go_to_floor(self, floor: int, immediate: bool = False) -> bool:
|
||||
"""前往指定楼层"""
|
||||
return self._api_client.go_to_floor(self._elevator_id, floor, immediate)
|
||||
|
||||
def set_up_indicator(self, value: bool) -> None:
|
||||
"""设置上行指示器"""
|
||||
self._api_client.set_indicators(self._elevator_id, up=value)
|
||||
|
||||
def set_down_indicator(self, value: bool) -> None:
|
||||
"""设置下行指示器"""
|
||||
self._api_client.set_indicators(self._elevator_id, down=value)
|
||||
|
||||
def __setattr__(self, name: str, value: Any) -> None:
|
||||
"""禁止修改属性,保持只读特性"""
|
||||
if not self.init_ok:
|
||||
if not self._init_ok:
|
||||
object.__setattr__(self, name, value)
|
||||
else:
|
||||
raise AttributeError(f"Cannot modify read-only attribute '{name}'")
|
||||
@@ -128,43 +102,35 @@ class ProxyPassenger(PassengerInfo):
|
||||
直接使用 PassengerInfo 数据模型实例,提供完整的类型安全访问
|
||||
"""
|
||||
|
||||
init_ok = False
|
||||
_init_ok = False
|
||||
|
||||
def __init__(self, passenger_id: int, api_client: ElevatorAPIClient):
|
||||
self._passenger_id = passenger_id
|
||||
self._api_client = api_client
|
||||
self.init_ok = True
|
||||
self._init_ok = True
|
||||
|
||||
def _get_passenger_info(self) -> PassengerInfo:
|
||||
"""获取 PassengerInfo 实例"""
|
||||
# 获取当前状态
|
||||
state = self._api_client.get_state()
|
||||
|
||||
# 在乘客字典中查找
|
||||
passenger_data = state.passengers.get(self._passenger_id)
|
||||
if passenger_data is None:
|
||||
raise AttributeError(f"Passenger {self._passenger_id} not found")
|
||||
return passenger_data
|
||||
|
||||
# 如果是字典,转换为 PassengerInfo 实例
|
||||
if isinstance(passenger_data, dict):
|
||||
return PassengerInfo.from_dict(passenger_data)
|
||||
def __getattribute__(self, name: str) -> Any:
|
||||
if not name.startswith("_") and self._init_ok and name not in self.__class__.__dict__:
|
||||
try:
|
||||
self_attr = object.__getattribute__(self, name)
|
||||
if callable(self_attr):
|
||||
return object.__getattribute__(self, name)
|
||||
except AttributeError:
|
||||
pass
|
||||
psg_info = self._get_passenger_info()
|
||||
return psg_info.__getattribute__(name)
|
||||
else:
|
||||
# 如果已经是 PassengerInfo 实例,直接返回
|
||||
return passenger_data
|
||||
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
"""动态获取乘客属性"""
|
||||
try:
|
||||
passenger_info = self._get_passenger_info()
|
||||
# 直接从 PassengerInfo 实例获取属性
|
||||
return getattr(passenger_info, name)
|
||||
|
||||
except AttributeError:
|
||||
raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")
|
||||
return object.__getattribute__(self, name)
|
||||
|
||||
def __setattr__(self, name: str, value: Any) -> None:
|
||||
"""禁止修改属性,保持只读特性"""
|
||||
if not self.init_ok:
|
||||
if not self._init_ok:
|
||||
object.__setattr__(self, name, value)
|
||||
else:
|
||||
raise AttributeError(f"Cannot modify read-only attribute '{name}'")
|
||||
|
||||
Reference in New Issue
Block a user