ARTICLE DETAIL

资讯详情

深耕编程入门与网站建设的一线实战洞察。

电梯调度教学系统:基于进程模型与状态机的Python实现

电梯调度教学系统:基于进程模型与状态机的Python实现 简介本资源是一份面向计算机专业本科生与Python初学者的课程设计实践项目聚焦电梯系统进程建模与调度算法实现解决多楼层、多请求场景下的实时响应与资源协调问题。压缩包共36个文件含3个核心Python源码myElevator.py、dispatch.py、myElevatorInterface.py、1份完整Word版设计方案报告、1份README说明文档、18张UI界面资源图含按钮状态、电梯运行图标、楼层状态图等及1个PyQt5应用图标整体大小19.32MB结构清晰便于理解MVC架构下GUI与逻辑层的协同机制。已有668人学习下载读者可直接运行PyQt5图形界面观察电梯响应队列、楼层请求调度过程并结合设计报告深入掌握线程同步threading、事件驱动交互及状态机建模等关键知识点是巩固Python基础、提升工程实践能力的典型教学案例。1. 这不是模拟器而是一套可调试、可扩展的电梯调度教学系统你见过用 Python 写出真实响应按钮、状态切换、多楼层并发请求的电梯逻辑吗不是画个动画完事而是让myElevator.py真正维护一个带就绪队列、运行态、等待态的进程模型不是静态演示而是通过dispatch.py实现 FCFS先来先服务非抢占式调度策略——进程一旦获得 CPU即电梯轿厢控制权就持续执行到目标楼层停稳、开关门完成才释放。它不依赖硬件但严格复现了单处理器系统中「就绪队列→调度器→运行态」的完整生命周期。课程设计者能直接修改dispatch.py中的select_next_request()函数替换为 SSTF 或 SCAN 算法初学者可逐行打断点观察threading.Event如何协调开门/关门/移动三阶段状态同步PyQt5 界面层与业务逻辑完全解耦所有 UI 按钮点击最终都转化为ElevatorSystem.submit_request(floor, direction)调用。它面向的是需要理解「进程状态转换」「调度策略落地」「线程安全边界」这三层抽象的真实教学场景而非仅展示视觉效果。2. 从进程建模到状态机驱动myElevator.py的核心实现逻辑2.1 为什么用「进程」而非「对象」建模电梯行为在操作系统课程语境下“电梯进程”不是指 OS 层面的fork()进程而是对电梯任务的进程抽象建模每个上下行请求被封装为一个具备NEW → READY → RUNNING → WAITING → TERMINATED全生命周期的逻辑实体。myElevator.py中的ElevatorRequest类明确包含floor目标楼层、direction方向UP/DOWN、state当前状态、arrival_time提交时间等字段。关键在于state字段的流转受控于调度器和电梯物理状态——例如当电梯正在上行至 8 楼时新提交的 3 楼下行请求只能进入READY队列不能直接触发移动。这种设计强制学生思考调度决策必须基于当前系统状态电梯位置、方向、载客状态与请求队列的实时关系而非简单排序。提示ElevatorRequest.state的枚举值定义在myElevatorInterface.py中共 5 种状态。TERMINATED并非表示销毁对象而是标记该请求已服务完毕后续不再参与调度计算。2.2 状态机驱动的电梯控制器ElevatorController类解析ElevatorController是整个系统的核心协调者其run()方法在一个独立线程中持续轮询执行。它不使用time.sleep()硬阻塞而是通过threading.Condition实现精准唤醒# myElevator.py 第 127 行起 def run(self): while self.running: with self.condition: # 等待有新请求或当前任务完成 self.condition.wait_for(lambda: self.has_pending_requests() or not self.is_moving()) if not self.running: break self._process_next_request()_process_next_request()是状态跃迁的中枢若电梯静止且队列非空 → 调用dispatch.select_next_request()获取下一个服务目标若目标楼层 当前楼层 → 设置self.direction UP启动上行动画若到达目标楼层 → 触发开门事件self.door_open_event.set()并等待DOOR_OPEN_DURATION秒后自动关门关门完成 → 将该请求状态设为TERMINATED从队列移除。这个循环清晰体现了「调度器输出决策 → 控制器执行动作 → 动作完成反馈状态 → 触发下一轮调度」的闭环。2.3 就绪队列的 FCFS 实现与线程安全保障dispatch.py中的FCFSDispatcher类管理就绪队列其add_request()和select_next_request()方法均需保证线程安全# dispatch.py 第 42 行 class FCFSDispatcher: def __init__(self): self.ready_queue deque() # 使用 collections.deque 保证 O(1) 头部弹出 self.lock threading.Lock() # 显式锁保护队列读写 def add_request(self, request): with self.lock: self.ready_queue.append(request) request.state ElevatorState.READY def select_next_request(self, current_floor, current_direction): with self.lock: for req in self.ready_queue: # FCFS 不做优先级判断只检查是否可达同向或静止时允许反向 if (current_direction req.direction or current_direction STOP) and req.state ElevatorState.READY: req.state ElevatorState.RUNNING self.ready_queue.remove(req) return req return None注意select_next_request()中的current_direction STOP分支当电梯静止时允许立即响应任意方向请求这是 FCFS 在电梯场景下的合理变体。若此处误写为req.direction current_direction将导致静止时无法响应反向请求——这是学生调试中最常卡住的逻辑坑。3. PyQt5 界面与业务逻辑解耦信号-槽机制的实战应用3.1 UI 文件生成与资源加载流程项目中的.png图片如doorup.png,up_hover.png并非硬编码路径而是通过 Qt Designer 生成的resources.qrc文件统一管理。构建步骤如下# 在项目根目录执行需已安装 pyqt5-tools pyside2-rcc resources.qrc -o resources_rc.py # 注意本项目实际用 PyQt5但命令兼容 # 或使用 PyQt5 自带工具推荐 pyrcc5 resources.qrc -o resources_rc.pyresources_rc.py被myElevatorInterface.py导入所有图片资源通过:/images/doorup.png这类 Qt 资源路径引用。这样做的好处是打包成单文件时如用 PyInstaller图片自动嵌入无需额外指定--add-data。3.2 按钮事件如何触发底层调度UI 层的ElevatorWindow类myElevatorInterface.py不直接调用ElevatorController而是通过自定义信号解耦# myElevatorInterface.py 第 89 行 class ElevatorWindow(QMainWindow): request_submitted pyqtSignal(int, str) # 发射楼层号、方向字符串 def __init__(self): super().__init__() self.setupUi(self) # 绑定所有楼层按钮 for floor in range(1, 12): btn_up getattr(self, fbtn_up_{floor}) btn_down getattr(self, fbtn_down_{floor}) btn_up.clicked.connect(lambda ffloor: self._emit_request(f, UP)) btn_down.clicked.connect(lambda ffloor: self._emit_request(f, DOWN)) def _emit_request(self, floor, direction): self.request_submitted.emit(floor, direction) # 仅发射信号不处理业务业务逻辑层在初始化时连接该信号# main.py 第 36 行 elevator_sys ElevatorSystem() window ElevatorWindow() window.request_submitted.connect(elevator_sys.submit_request) # 真正的调度入口这种模式让学生清晰看到UI 是输入源submit_request()是唯一入口所有调度决策都在dispatch.py和myElevator.py中完成。若学生想测试 SCAN 算法只需修改dispatch.py无需碰 UI 代码。3.3 状态同步如何让界面实时反映电梯物理状态电梯的实时位置、方向、门状态由ElevatorController通过pyqtSignal主动推送# myElevator.py 第 205 行 class ElevatorController(QObject): position_updated pyqtSignal(int) # 当前楼层 direction_updated pyqtSignal(str) # UP/DOWN/STOP door_state_updated pyqtSignal(bool) # True开启False关闭 def _move_to_floor(self, target_floor): # ... 移动逻辑 for floor in range(self.current_floor, target_floor 1): self.current_floor floor self.position_updated.emit(floor) # 每层都发射 time.sleep(0.5) # 模拟移动耗时UI 层监听这些信号并更新控件# myElevatorInterface.py 第 152 行 self.controller.position_updated.connect(self.update_floor_display) self.controller.direction_updated.connect(self.update_direction_indicator) self.controller.door_state_updated.connect(self.update_door_animation)update_door_animation()方法根据布尔值切换doorup.png/doordown.png图片并控制QPropertyAnimation动画时长。这种「信号驱动 UI 更新」的方式避免了轮询controller.current_floor带来的性能浪费也符合 Qt 最佳实践。4. 调度策略替换实战从 FCFS 到 SCAN 算法的三步改造4.1 理解 SCAN 算法在电梯场景的特殊性FCFS 是“谁先来谁先服务”而 SCAN又称电梯算法要求电梯沿一个方向移动服务途中所有请求到达端点后反转方向。关键约束是电梯当前移动方向决定了哪些请求可被立即服务。例如电梯正上行至 10 楼此时收到 5 楼下行请求该请求必须等待电梯到达 10 楼、转向下行后才能处理。这比单纯按距离排序更贴近物理现实。4.2 修改dispatch.py实现 SCAN 调度器新建scan_dispatcher.py继承BaseDispatcher# scan_dispatcher.py from collections import deque from myElevatorInterface import ElevatorState class SCANDispatcher: def __init__(self): self.up_requests [] # 存储上行请求floor current self.down_requests [] # 存储下行请求floor current self.lock threading.Lock() def add_request(self, request): with self.lock: if request.direction UP: self.up_requests.append(request) self.up_requests.sort(keylambda x: x.floor) # 升序2→5→8 else: self.down_requests.append(request) self.down_requests.sort(keylambda x: x.floor, reverseTrue) # 降序9→6→3 def select_next_request(self, current_floor, current_direction): with self.lock: if current_direction UP: # 优先服务上行队列中 current_floor 的请求 for req in self.up_requests[:]: if req.floor current_floor and req.state ElevatorState.READY: req.state ElevatorState.RUNNING self.up_requests.remove(req) return req # 上行队列空或无可达请求转向下行 current_direction DOWN for req in self.down_requests[:]: if req.state ElevatorState.READY: req.state ElevatorState.RUNNING self.down_requests.remove(req) return req elif current_direction DOWN: for req in self.down_requests[:]: if req.floor current_floor and req.state ElevatorState.READY: req.state ElevatorState.RUNNING self.down_requests.remove(req) return req current_direction UP for req in self.up_requests[:]: if req.state ElevatorState.READY: req.state ElevatorState.RUNNING self.up_requests.remove(req) return req return None注意select_next_request()返回None时ElevatorController.run()会继续等待不会报错。这是设计容错。4.3 在main.py中切换调度器原main.py中第 32 行# 原 FCFS dispatcher FCFSDispatcher()改为# 启用 SCAN from scan_dispatcher import SCANDispatcher dispatcher SCANDispatcher()同时需在ElevatorSystem.__init__()中传入dispatcher实例并确保ElevatorController初始化时接收该实例。此改动仅涉及 3 个文件、不足 10 行代码却完整替换了核心调度逻辑。5. 调试与验证用日志和断点定位典型调度异常5.1 启用详细日志追踪请求生命周期项目默认日志级别为WARNING需手动提升至DEBUG以观察每一步状态变化。在main.py开头添加import logging logging.basicConfig( levellogging.DEBUG, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[logging.StreamHandler()] ) logger logging.getLogger(__name__)然后在myElevator.py的ElevatorRequest.__init__()和状态变更处插入日志# myElevator.py 第 65 行 def __init__(self, floor, direction): self.floor floor self.direction direction self.state ElevatorState.NEW self.arrival_time time.time() logger.debug(fNew request: floor{floor}, dir{direction}, id{id(self)}) # 在 _process_next_request() 中 logger.info(fSelected request to floor {req.floor} ({req.direction}), fqueue size now {len(self.dispatcher.ready_queue)})运行后终端将输出类似2023-07-15 14:22:33,102 - __main__ - DEBUG - New request: floor5, dirUP, id140234567890123 2023-07-15 14:22:33,105 - __main__ - INFO - Selected request to floor 5 (UP), queue size now 0通过时间戳和 ID 可确认请求是否重复提交、是否被错误移除、是否存在状态滞留如长期卡在READY。5.2 验证 FCFS 调度正确性的关键测试用例设计以下测试序列在 UI 中依次点击按下 3 楼UP按钮请求 A立即按下 8 楼DOWN按钮请求 B电梯开始上行途中在 5 楼暂停因无请求继续上行到达 8 楼后应先服务请求 B8 楼 DOWN再转向下行服务请求 A3 楼若实际行为是到达 8 楼后直接下行至 3 楼跳过 8 楼开门则说明select_next_request()未正确识别current_direction STOP时的反向请求。此时需检查dispatch.py中 FCFS 的条件分支是否遗漏or current_direction STOP。5.3 监控线程状态避免资源竞争当频繁点击按钮时可能出现RuntimeError: deque mutated during iteration。这是因为FCFSDispatcher.select_next_request()在遍历ready_queue时另一线程UI调用了add_request()修改了队列。解决方案是在select_next_request()中使用快照# dispatch.py 修正版 def select_next_request(self, current_floor, current_direction): with self.lock: # 创建队列快照避免遍历时被修改 snapshot list(self.ready_queue) for req in snapshot: if (current_direction req.direction or current_direction STOP) and req.state ElevatorState.READY: req.state ElevatorState.RUNNING try: self.ready_queue.remove(req) # 移除时仍需锁保护 except ValueError: pass # 已被其他线程移除忽略 return req return None此修改确保即使在高并发点击下调度器也能稳定工作这是课程设计中必须覆盖的健壮性要点。本文还有配套的精品资源点击获取
返回列表