python的工业过程控制场景模拟第八十六篇:厂区巡检机器人任务调度，优先前往报警设备点位开展巡检工作。
厂区巡检机器人任务调度与报警优先响应仿真 —— 基于事件驱动与优先级抢占的OOP实战
“去年夏天，厂区凌晨三点，B区压力变送器悄悄触发了低压报警，但巡检机器人还在按部就班地沿着预定路线扫二维码、打卡、抄表。等它绕完一圈走到B区时，已经过去了28分钟，管线早已泄漏，险些酿成停产事故。后来我们在调度系统里加了事件驱动+优先级抢占机制，只要现场有报警信号，机器人立刻中断当前任务，原地转向，以最短路径冲向报警点位。现在，报警响应时间从半小时缩短到了90秒以内。”
—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸
一、实际应用场景描述
在现代化工厂、变电站、油气站场中，巡检机器人已成为保障安全生产的重要一环。它们替代人工完成表计读取、红外测温、气体检测等任务。然而，传统的巡检机器人大多采用固定路径、顺序执行的模式，缺乏对突发状况的敏捷响应能力。
┌──────────────────────────────────────────────┐
│ 厂区巡检机器人任务调度系统 │
│ │
│ [SCADA/DCS 报警系统] │
│ │ 报警信号 (Modbus/TCP) │
│ ▼ │
│ ┌────────────────────────────┐ │
│ │ 任务调度与事件管理器 │ │
│ │ ┌──────────────────────┐ │ │
│ │ │ 1. 事件监听与解析 │ │ │
│ │ │ (报警/定时/手动) │ │ │
│ │ └──────────────────────┘ │ │
│ │ ┌──────────────────────┐ │ │
│ │ │ 2. 优先级仲裁 │ │ │
│ │ │ (抢占/非抢占) │ │ │
│ │ └──────────────────────┘ │ │
│ │ ┌──────────────────────┐ │ │
│ │ │ 3. 任务队列管理 │ │ │
│ │ │ (Heap/PriorityQ) │ │ │
│ │ └──────────────────────┘ │ │
│ │ ┌──────────────────────┐ │ │
│ │ │ 4. 路径重规划 │ │ │
│ │ │ (Dijkstra/A*) │ │ │
│ │ └──────────────────────┘ │ │
│ └────────────┬───────────────┘ │
│ │ 调度指令 │
│ ┌───────┴───────┐ │
│ ▼ ▼ │
│ ┌─────────┐ ┌─────────┐ │
│ │ 巡检机器人 │ │ 充电桩/待机点 │ │
│ │ (Robot) │ │ (Dock) │ │
│ └────┬────┘ └─────────┘ │
│ │ │
│ ▼ │
│ ┌────────────────────────────┐ │
│ │ 厂区拓扑地图 │ │
│ │ ┌───┐ ┌───┐ ┌───┐ │ │
│ │ │A区├──►B区├──►C区├──... │ ← 报警点B │
│ │ └─┬─┘ └─┬─┘ └───┘ │ │
│ │ └──────┴──────── │ │
│ └───────────────────────────┘ │
│ │
│ 核心: 事件驱动 + 优先级抢占 + 最短路径 │
└──────────────────────────────────────────────┘
传统顺序巡检 vs 报警优先调度
维度 传统顺序巡检 报警优先调度
报警响应 ❌ 滞后（等一圈） ✅ 实时抢占
任务灵活性 ❌ 僵化 ✅ 动态插队
安全等级 ❌ 低 ✅ 高
路径效率 ❌ 冗余 ✅ 最优路径
适用场景 日常例行巡检 应急+日常融合
二、引入痛点
2.1 现场的真实困境
场景 现场发生了什么 根因
“报警没人理” “机器人还在慢慢走” 无事件驱动机制
“跑断腿也来不及” “报警点在最远端” 固定路径无重规划
“误报打断工作” “频繁假报警导致任务混乱” 无报警滤波/确认
“任务丢失” “抢占后原任务忘了” 无任务挂起/恢复
“撞车了” “两台机器人抢一个报警点” 无冲突仲裁
2.2 核心矛盾
安全生产要求“毫秒级响应”，而传统巡检机器人的“分钟级遍历”无法满足这一需求。 这本质上是一个多任务实时调度问题。我们需要引入事件驱动架构（Event-Driven Architecture）和优先级抢占式调度（Preemptive Priority Scheduling），使得高优先级的报警任务能够中断低优先级的常规任务，并通过最短路径算法快速抵达现场。
2.3 我们要解决什么
用一段精简的 Python 程序，构建一个厂区巡检机器人任务调度仿真系统，实现：
1. 事件驱动模型 —— 报警、定时、手动任务的统一抽象
2. 优先级抢占机制 —— 高优先级任务中断低优先级任务
3. 任务挂起与恢复 —— 被中断的任务不丢失
4. 最短路径规划 —— 基于厂区拓扑的快速寻路
5. 调度可视化 —— 展示任务队列、机器人轨迹和抢占过程
三、核心逻辑讲解
3.1 理论基础：实时调度与事件驱动
本工具基于哈工程《工业过程控制》第十五章“计算机控制系统”和实时系统调度理论：
① 事件驱动模型（Event-Driven Model）
系统不再依赖轮询，而是等待事件发生。事件分为：
- 异步事件：报警触发、人工急停
- 同步事件：定时任务、周期巡检
② 优先级抢占调度（Priority Preemptive Scheduling）
任务优先级定义为：
P_{Alarm} > P_{Manual} > P_{Scheduled}
调度规则：
- 若新任务 T_{new} 的优先级高于当前任务 T_{curr} ，则 T_{curr} 被挂起（Preempt）
- T_{new} 获得CPU（机器人控制权）
- 当 T_{new} 完成后，恢复 T_{curr}
③ 最短路径算法（Dijkstra/A*）
在厂区拓扑图中寻找从起点 S 到报警点 G 的最短路径：
Path = \arg\min_{p \in Paths(S, G)} \sum_{e \in p} w(e)
其中 w(e) 为路径权重（距离/时间）。
四、代码讲解（面向对象设计）
4.1 类结构总览
类名 职责 设计模式
"TaskType" 任务类型枚举 枚举
"TaskPriority" 任务优先级枚举 枚举
"InspectionTask" 巡检任务（dataclass） 命令模式
"PlantMap" 厂区拓扑地图 封装
"EventDispatcher" 事件分发器 观察者模式
"PriorityScheduler" 优先级调度器 策略模式
"PathPlanner" 路径规划器 策略模式
"RobotAgent" 巡检机器人代理 状态模式
"SimulationClock" 仿真时钟 单例模式
"VisualizationEngine" 可视化引擎 封装
"DispatchSystem" 调度系统编排器（聚合根） 聚合根
4.2 数据模型层
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple, Set, Callable
from enum import Enum, auto
import heapq
import math
from datetime import datetime
class TaskType(Enum):
"""任务类型"""
SCHEDULED = auto() # 计划巡检
ALARM = auto() # 报警响应
MANUAL = auto() # 人工指令
CHARGING = auto() # 充电
class TaskPriority(Enum):
"""任务优先级（数值越小优先级越高）"""
CRITICAL = 1 # 紧急报警
HIGH = 2 # 重要报警/人工
NORMAL = 3 # 计划巡检
LOW = 4 # 充电/自检
@dataclass(frozen=True)
class InspectionTask:
"""巡检任务 —— 命令模式"""
task_id: str
task_type: TaskType
priority: TaskPriority
target_node: str # 目标点位ID
source_node: str = "ANY" # 来源点位（用于路径规划）
description: str = ""
creation_time: float = field(default_factory=lambda: datetime.now().timestamp())
deadline: Optional[float] = None
is_preemptible: bool = True # 是否可被抢占
def __lt__(self, other):
# 优先队列比较：优先级高的（数值小）排前面
if self.priority.value == other.priority.value:
return self.creation_time < other.creation_time
return self.priority.value < other.priority.value
4.3 厂区拓扑地图
class PlantMap:
"""
厂区拓扑地图
使用邻接表存储图结构
"""
def __init__(self):
self.nodes: Dict[str, Tuple[float, float]] = {} # node_id -> (x, y)
self.edges: Dict[str, Dict[str, float]] = {} # node_id -> {neighbor: cost}
self._initialize_default_map()
def _initialize_default_map(self):
"""初始化默认厂区地图"""
# 定义点位 (ID: (x, y))
self.nodes = {
"DOCK": (0, 0), # 充电桩/待机点
"GATE": (2, 0), # 大门
"A1": (4, 1), # A区测点1
"A2": (4, -1), # A区测点2
"B1": (6, 1), # B区测点1（报警高发区）
"B2": (6, -1), # B区测点2
"C1": (8, 1), # C区测点1
"C2": (8, -1), # C区测点2
"CONTROL": (10, 0) # 中控室
}
# 定义连接（双向）
connections = [
("DOCK", "GATE", 2.0),
("GATE", "A1", 2.5), ("GATE", "A2", 2.5),
("A1", "B1", 2.0), ("A2", "B2", 2.0),
("B1", "C1", 2.0), ("B2", "C2", 2.0),
("C1", "CONTROL", 2.0), ("C2", "CONTROL", 2.0),
# 跨区捷径（用于报警快速响应）
("GATE", "B1", 4.5), ("GATE", "B2", 4.5)
]
for n1, n2, cost in connections:
self._add_edge(n1, n2, cost)
self._add_edge(n2, n1, cost)
def _add_edge(self, node1: str, node2: str, cost: float):
if node1 not in self.edges:
self.edges[node1] = {}
self.edges[node1][node2] = cost
def get_shortest_path(self, start: str, goal: str) -> Tuple[List[str], float]:
"""
Dijkstra 最短路径算法
Returns:
(路径节点列表, 总代价)
"""
if start not in self.nodes or goal not in self.nodes:
return [], float('inf')
# 优先队列: (cost, node, path)
pq = [(0.0, start, [start])]
visited = set()
while pq:
cost, node, path = heapq.heappop(pq)
if node in visited:
continue
visited.add(node)
if node == goal:
return path, cost
for neighbor, edge_cost in self.edges.get(node, {}).items():
if neighbor not in visited:
heapq.heappush(
pq,
(cost + edge_cost, neighbor, path + [neighbor])
)
return [], float('inf')
def get_coordinates(self, node_id: str) -> Tuple[float, float]:
return self.nodes.get(node_id, (0.0, 0.0))
4.4 事件分发器（观察者模式）
class EventDispatcher:
"""
事件分发器 —— 观察者模式
负责接收报警信号并通知调度器
"""
def __init__(self):
self.listeners: List[Callable[[InspectionTask], None]] = []
self.event_log: List[Dict] = []
def add_listener(self, listener: Callable[[InspectionTask], None]):
"""注册事件监听器（通常是调度器）"""
self.listeners.append(listener)
def dispatch_alarm(self, alarm_info: Dict):
"""
分发报警事件
Args:
alarm_info: {
'device_id': 'B1_PT101',
'location': 'B1',
'level': 'CRITICAL',
'description': '低压报警'
}
"""
priority_map = {
"CRITICAL": TaskPriority.CRITICAL,
"HIGH": TaskPriority.HIGH,
"MEDIUM": TaskPriority.NORMAL
}
location = alarm_info.get('location', 'UNKNOWN')
target_node = self._map_location_to_node(location)
task = InspectionTask(
task_id=f"ALARM_{datetime.now().strftime('%H%M%S')}_{location}",
task_type=TaskType.ALARM,
priority=priority_map.get(alarm_info.get('level', 'MEDIUM'), TaskPriority.NORMAL),
target_node=target_node,
description=f"报警: {alarm_info.get('description', '')} @ {location}",
deadline=datetime.now().timestamp() + 300 # 5分钟内必须响应
)
self.event_log.append({
"timestamp": datetime.now().timestamp(),
"event": "ALARM",
"task": task
})
print(f"🚨 收到报警: {task.description} (优先级: {task.priority.name})")
# 通知所有监听器
for listener in self.listeners:
listener(task)
def _map_location_to_node(self, location: str) -> str:
"""将位置描述映射为地图节点"""
mapping = {
"A1": "A1", "A2": "A2",
"B1": "B1", "B2": "B2",
"C1": "C1", "C2": "C2",
"CONTROL": "CONTROL",
"DOCK": "DOCK"
}
return mapping.get(location.upper(), "GATE")
4.5 优先级调度器（策略模式）
class PriorityScheduler:
"""
优先级调度器 —— 策略模式
实现抢占式优先级调度
"""
def __init__(self):
self.task_queue: List[InspectionTask] = [] # 优先队列
self.current_task: Optional[InspectionTask] = None
self.preempted_tasks: List[InspectionTask] = [] # 被抢占的任务栈
self.completed_tasks: List[InspectionTask] = []
self.scheduler_log: List[Dict] = []
def submit_task(self, task: InspectionTask) -> bool:
"""
提交任务到调度器
Returns:
True if task was accepted/executed immediately
"""
# 检查是否需要抢占
if self.current_task and task.priority.value < self.current_task.priority.value:
if self.current_task.is_preemptible:
print(f"⚡ 任务抢占: {self.current_task.task_id} 被 {task.task_id} 抢占")
self._preempt_current_task()
self._schedule_task(task)
return True
else:
print(f"⚠️ 任务 {task.task_id} 无法抢占不可中断任务 {self.current_task.task_id}")
# 无需抢占，加入队列
heapq.heappush(self.task_queue, task)
self.scheduler_log.append({
"action": "ENQUEUE",
"task_id": task.task_id,
"queue_size": len(self.task_queue)
})
return False
def _preempt_current_task(self):
"""抢占当前任务"""
if self.current_task:
# 保存任务状态（简化：只保存任务本身）
self.preempted_tasks.append(self.current_task)
self.scheduler_log.append({
"action": "PREEMPT",
"task_id": self.current_task.task_id
})
self.current_task = None
def _schedule_task(self, task: InspectionTask):
"""调度任务执行"""
self.current_task = task
self.scheduler_log.append({
"action": "DISPATCH",
"task_id": task.task_id
})
print(f"▶️ 开始执行任务: {task.task_id} - {task.description}")
def complete_current_task(self):
"""标记当前任务完成"""
if self.current_task:
self.completed_tasks.append(self.current_task)
self.scheduler_log.append({
"action": "COMPLETE",
"task_id": self.current_task.task_id
})
print(f"✅ 任务完成: {self.current_task.task_id}")
self.current_task = None
# 尝试恢复被抢占的任务
self._resume_preempted_task()
def _resume_preempted_task(self):
"""恢复被抢占的任务"""
if self.preempted_tasks and not self.current_task:
# 恢复最近被抢占的任务
task = self.preempted_tasks.pop()
heapq.heappush(self.task_queue, task)
print(f"♻️ 恢复被抢占任务: {task.task_id}")
def get_next_task(self) -> Optional[InspectionTask]:
"""获取下一个待执行任务"""
if not self.current_task and self.task_queue:
self.current_task = heapq.heappop(self.task_queue)
self.scheduler_log.append({
"action": "DISPATCH_FROM_QUEUE",
"task_id": self.current_task.task_id
})
print(f"▶️ 开始执行任务: {self.current_task.task_id}")
return self.current_task
return self.current_task
def get_schedule_status(self) -> Dict:
"""获取调度状态"""
return {
"current_task": self.current_task.task_id if self.current_task else None,
"queue_length": len(self.task_queue),
"preempted_count": len(self.preempted_tasks),
"completed_count": len(self.completed_tasks)
}
4.6 路径规划器（策略模式）
class PathPlanner:
"""
路径规划器 —— 策略模式
封装路径规划算法
"""
def __init__(self, plant_map: PlantMap):
self.map = plant_map
def plan_emergency_path(self, current_pos: str, target_pos: str) -> List[str]:
"""
规划紧急路径（优先选择捷径）
Args:
current_pos: 当前位置
target_pos: 目标位置
Returns:
路径节点列表
"""
# 直接使用Dijkstra找最短路径
path, cost = self.map.get_shortest_path(current_pos, target_pos)
print(f"🗺️ 紧急路径规划: {current_pos} -> {target_pos}, 代价: {cost:.1f}")
return path
def plan_regular_path(self, start: str, waypoints: List[str]) -> List[str]:
"""
规划常规巡检路径（遍历所有点位）
Args:
start: 起点
waypoints: 需要访问的点位列表
Returns:
完整路径
"""
full_path = [start]
current = start
for wp in waypoints:
segment, _ = self.map.get_shortest_path(current, wp)
if segment:
full_path.extend(segment[1:]) # 避免重复起点
current = wp
return full_path
4.7 巡检机器人代理（状态模式）
class RobotAgent:
"""
巡检机器人代理 —— 状态模式
状态流转: IDLE -> MOVING -> INSPECTING -> RETURNING -> ERROR
"""
def __init__(self, robot_id: str, start_node: str = "DOCK"):
self.robot_id = robot_id
self.current_node = start_node
self.state = "IDLE"
self.position: Tuple[float, float] = (0, 0)
self.current_path: List[str] = []
self.path_index = 0
self.current_task: Optional[InspectionTask] = None
self.trajectory: List[Tuple[float, float]] = []
self.speed = 1.0 # 单位: 地图单位/秒
self.is_charging = False
# 依赖注入
self.map: Optional[PlantMap] = None
self.planner: Optional[PathPlanner] = None
def set_dependencies(self, plant_map: PlantMap, planner: PathPlanner):
self.map = plant_map
self.planner = planner
self.position = self.map.get_coordinates(self.current_node)
def assign_task(self, task: InspectionTask):
"""分配任务给机器人"""
self.current_task = task
self.state = "MOVING"
if task.task_type == TaskType.ALARM:
# 报警任务：紧急路径
self.current_path = self.planner.plan_emergency_path(
self.current_node, task.target_node
)
else:
# 常规任务：可能需要规划完整路线
self.current_path = self.planner.plan_emergency_path(
self.current_node, task.target_node
)
self.path_index = 0
print(f"🤖 机器人 {self.robot_id} 接受任务: {task.task_id}")
def update(self, dt: float) -> bool:
"""
更新机器人状态
Returns:
True if task completed in this step
"""
if self.state == "IDLE":
return False
if self.state == "MOVING" and self.current_path:
if self.path_index < len(self.current_path):
target_node = self.current_path[self.path_index]
target_pos = self.map.get_coordinates(target_node)
# 计算移动
dx = target_pos[0] - self.position[0]
dy = target_pos[1] - self.position[1]
dist = math.sqrt(dx*dx + dy*dy)
if dist < 0.05: # 到达节点
self.position = target_pos
self.current_node = target_node
self.path_index += 1
self.trajectory.append(self.position)
if self.path_index >= len(self.current_path):
self.state = "INSPECTING"
print(f"📍 机器人 {self.robot_id} 到达 {target_node}")
else:
# 向目标移动
step = min(self.speed * dt, dist)
self.position = (
self.position[0] + dx/dist * step,
self.position[1] + dy/dist * step
)
self.trajectory.append(self.position)
else:
self.state = "INSPECTING"
elif self.state == "INSPECTING":
# 模拟巡检动作
print(f"🔍 机器人 {self.robot_id} 正在 {self.current_node} 执行巡检...")
self.state = "RETURNING" if self.current_task and self.current_task.task_type == TaskType.ALARM else "IDLE"
return True # 任务完成
elif self.state == "RETURNING":
# 报警任务完成后返回待机点
if self.current_node != "DOCK":
self.current_path = self.planner.plan_emergency_path(
self.current_node, "DOCK"
)
self.path_index = 0
self.state = "MOVING"
else:
self.state = "IDLE"
self.current_task = None
return False
def preempt_task(self):
"""抢占当前任务"""
if self.current_task:
print(f"⚡ 机器人 {self.robot_id} 任务被抢占: {self.current_task.task_id}")
self.state = "IDLE"
self.current_path = []
self.current_task = None
def get_status(self) -> Dict:
return {
"robot_id": self.robot_id,
"state": self.state,
"current_node": self.current_node,
"position": self.position,
"task_id": self.current_task.task_id if self.current_task else None,
"path_progress": f"{self.path_index}/{len(self.current_path)}" if self.current_path else "N/A"
}
4.8 仿真时钟（单例模式）
class SimulationClock:
"""仿真时钟 —— 单例模式"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance.current_time = 0.0
return cls._instance
def tick(self, dt: float):
self.current_time +
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
1
0- 0
已为社区贡献7条内容
所有评论(0)