3. 电梯控制系统

任务目标

设计一个单电梯控制系统的有限状态机(FSM),覆盖空闲、运行、开门、关门四个核心状态,并处理楼层请求与方向选择逻辑。系统接收外部楼层召唤与轿厢内目标楼层请求,通过状态迁移控制电梯完成响应。

状态定义与迁移逻辑

当前状态触发事件条件下一状态动作
IDLErequest_added存在待响应请求MOVING确定方向,开始移动
MOVINGarrived_at_floor当前楼层有请求或为顺路目标DOOR_OPEN停靠,清除该楼层请求
MOVINGno_more_requests所有请求已处理IDLE停止
DOOR_OPENdoor_closed开门计时结束IDLE关门
DOOR_OPENdoor_closed仍有同向请求MOVING关门后继续运行

完整实现代码

from enum import Enum, auto
from collections import deque

class State(Enum):
    IDLE = auto()
    MOVING = auto()
    DOOR_OPEN = auto()

class Direction(Enum):
    UP = 1
    DOWN = -1
    NONE = 0

class ElevatorFSM:
    def __init__(self, total_floors):
        self.total_floors = total_floors
        self.current_floor = 1
        self.direction = Direction.NONE
        self.state = State.IDLE
        self.up_requests = set()    # 上行请求楼层
        self.down_requests = set()  # 下行请求楼层
        self.pending = deque()      # 待处理请求队列

    def request(self, floor, direction=None):
        """外部召唤或内部目标楼层请求"""
        if floor < 1 or floor > self.total_floors:
            return
        if direction == Direction.UP:
            self.up_requests.add(floor)
        elif direction == Direction.DOWN:
            self.down_requests.add(floor)
        else:  # 内部目标楼层,按当前方向归类
            if self.direction == Direction.UP and floor > self.current_floor:
                self.up_requests.add(floor)
            elif self.direction == Direction.DOWN and floor < self.current_floor:
                self.down_requests.add(floor)
            else:
                self.pending.append(floor)
        self._process()

    def _has_requests(self):
        return bool(self.up_requests or self.down_requests or self.pending)

    def _next_target(self):
        """根据当前方向选择下一个目标楼层"""
        if self.direction == Direction.UP:
            candidates = [f for f in self.up_requests if f > self.current_floor]
            if candidates:
                return min(candidates)
            # 上方无请求,切换方向
            self.direction = Direction.DOWN
        elif self.direction == Direction.DOWN:
            candidates = [f for f in self.down_requests if f < self.current_floor]
            if candidates:
                return max(candidates)
            self.direction = Direction.UP
        # 处理待定请求
        if self.pending:
            return self.pending.popleft()
        # 从剩余请求中选最近楼层
        all_reqs = self.up_requests | self.down_requests
        if all_reqs:
            return min(all_reqs, key=lambda f: abs(f - self.current_floor))
        return None

    def _process(self):
        if self.state == State.IDLE and self._has_requests():
            self.state = State.MOVING
            self._move()

    def _move(self):
        target = self._next_target()
        if target is None:
            self.state = State.IDLE
            self.direction = Direction.NONE
            return
        # 模拟移动到目标楼层
        self.current_floor = target
        self.up_requests.discard(target)
        self.down_requests.discard(target)
        self.state = State.DOOR_OPEN
        self._open_door()

    def _open_door(self):
        # 开门后自动关门
        self._close_door()

    def _close_door(self):
        if self._has_requests():
            self.state = State.MOVING
            self._move()
        else:
            self.state = State.IDLE
            self.direction = Direction.NONE

    def step(self):
        """手动推进状态机(用于测试)"""
        if self.state == State.DOOR_OPEN:
            self._close_door()
        elif self.state == State.MOVING:
            self._move()
        elif self.state == State.IDLE:
            self._process()

    def get_status(self):
        return {
            "state": self.state.name,
            "floor": self.current_floor,
            "direction": self.direction.name,
            "up_queue": sorted(self.up_requests),
            "down_queue": sorted(self.down_requests)
        }

运行与验证

运行方式:直接执行Python脚本,创建电梯实例并添加请求。

# 测试脚本
elev = ElevatorFSM(total_floors=10)
elev.request(5, Direction.UP)   # 1楼乘客上行去5楼
print(elev.get_status())        # 预期: state=DOOR_OPEN, floor=5, direction=UP
elev.step()                     # 关门
print(elev.get_status())        # 预期: state=IDLE, floor=5, direction=NONE

# 多请求场景
elev.request(3, Direction.DOWN) # 5楼乘客下行去3楼
elev.request(8, Direction.UP)   # 5楼乘客上行去8楼
print(elev.get_status())        # 预期: state=DOOR_OPEN, floor=8, direction=UP
elev.step()
print(elev.get_status())        # 预期: state=DOOR_OPEN, floor=3, direction=DOWN

验证标准:

  • 状态迁移严格遵循FSM定义,无非法状态跳转
  • 同方向请求按顺序响应,方向切换仅在当前方向无请求时发生
  • 所有请求最终被清除,电梯回到IDLE状态
  • 输出状态与手动推演结果一致

关键设计说明

本实现采用双队列(上行/下行请求集合)加待定队列的结构,确保方向优先策略:电梯优先完成当前方向的全部请求,再切换方向。内部目标楼层若与当前方向不一致,先存入待定队列,待方向切换后处理。状态机通过step()方法显式推进,便于测试与调试。实际工程中,_move()应替换为逐层移动的定时逻辑,_open_door()应加入开门延时控制,此处为保持核心逻辑清晰做了简化。