python的工业过程控制场景模拟第九十七篇:AGV交通管制仿真，交叉路口设置通行优先级，危险品运输车辆优先通过。-北京尧图网络

AGV 交通管制与交叉路口优先级调度仿真 —— 基于 Petri 网与动态优先级“那年厂里上了 12 台 AGV平时跑得好好的一到早晚高峰两台车在十字路口‘顶牛’谁也不让谁最后只能人工遥控拖走。后来我们给调度系统加了Petri 网建模动态优先级仲裁危险品运输车永远最高级普通物料车见它就避让路口从此再没堵死过。”—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸一、实际应用场景描述在大型化工、制药园区AGV 承担着原料、半成品、危化品的自动化转运任务。多车混行、路径交叉是常态┌──────────────────────────────────────────────┐│ AGV 交通管制与优先级调度系统 ││ ││ [上位机调度中枢] ││ │ 路径规划 / 冲突检测 / 优先级仲裁 ││ ▼ ││ ┌────────────────────────────┐ ││ │ 交通管制引擎 (Traffic Control)│ ││ │ ┌──────────────────────┐ │ ││ │ │ 1. 冲突区域检测 │ │ ││ │ │ (Crossing Zones) │ │ ││ │ └──────────────────────┘ │ ││ │ ┌──────────────────────┐ │ ││ │ │ 2. 动态优先级计算 │ │ ││ │ │ (危化品批次 FIFO)│ ││ │ └──────────────────────┘ │ ││ │ ┌──────────────────────┐ │ ││ │ │ 3. 死锁检测与解除 │ │ ││ │ │ (Petri 网/超时) │ │ ││ │ └──────────────────────┘ │ ││ └────────────┬───────────────┘ ││ │ 通行令牌 (Token) ││ ┌───────┴───────┐ ││ ▼ ▼ ││ ┌─────────┐ ┌─────────┐ ││ │ AGV-01 │ │ AGV-02 │ ││ │ (危化品) │ │ (普通物料) │ ││ │ Prio0 │ │ Prio2 │ ││ └────┬────┘ └────┬────┘ ││ │ 申请通行 │ 申请通行 ││ ▼ ▼ ││ ┌────────────────────────────┐ ││ │ 交叉路口 (Conflict Zone) │ ││ │ ┌─────┐ ┌─────┐ ┌─────┐ │ ││ │ │ZoneA│ │ZoneB│ │ZoneC│ │ ││ │ │(互斥)│ │(互斥)│ │(互斥)│ │ ││ │ └──┬──┘ └──┬──┘ └──┬──┘ │ ││ │ └────┬────┘ │ │ ││ │ ▼ ▼ │ ││ │ ┌─────────────────┐ │ ││ │ │ 仲裁器 (Arbiter) │ │ ││ │ │ • 比较优先级 │ │ ││ │ │ • 分配通行权 │ │ ││ │ │ • 超时释放 │ │ ││ │ └─────────────────┘ │ ││ └───────────────────────────┘ ││ ││ 核心: Petri网建模动态优先级死锁预防 │└──────────────────────────────────────────────┘传统自由竞争 vs 优先级管制维度自由竞争无管制优先级管制路口通行 ❌ 互相等待/死锁 ✅ 有序仲裁危化品运输 ❌ 无保障 ✅ 最高优先级吞吐量 ❌ 低频繁避让 ✅ 高预测性通行安全性 ❌ 风险高 ✅ 隔离保障二、引入痛点2.1 现场的真实困境场景现场发生了什么根因“路口顶牛” “两车对峙半小时” 无仲裁机制“危化品堵路” “剧毒原料车被普通车挡道” 优先级缺失“幽灵堵车” “明明没车却一直等” 死锁未检测“调度混乱” “12台车互相发指令” 中心化不足“效率低” “高峰期产能腰斩” 无流量控制2.2 核心矛盾多 AGV 系统的本质是“共享资源的并发控制”。交叉路口是典型的临界区Critical Section必须保证互斥访问。单纯的“先到先得”无法满足危化品运输的特殊安全要求必须引入动态优先级并且通过 Petri 网形式化建模确保无死锁、无饥饿。2.3 我们要解决什么用一段精简的 Python 程序构建一个AGV 交通管制仿真系统实现1. Petri 网建模 —— 将路口抽象为 PlaceAGV 为 Token2. 动态优先级 —— 危化品紧急批次普通 FIFO3. 死锁检测 —— 超时释放资源预留4. 可视化 —— 展示路口占用、队列、优先级决策三、核心逻辑讲解3.1 理论基础Petri 网与交通管制本工具基于哈工程《工业过程控制》第十五章“离散事件系统”和第八章“同步控制”① 基本 Petri 网定义PN (P, T, F, W, M_0)- P 库所Place代表路口、缓冲区- T 变迁Transition代表 AGV 通行请求- F 流关系- W 权重- M 标识Marking代表资源占用状态② 交叉路口的 Petri 网模型┌─────────┐│ 空闲(1) │ ← 有1个Token表示路口空闲└────┬────┘│ t_request (AGV申请)┌────▼────┐│ 仲裁中(0) │ ← 无Token等待仲裁└────┬────┘│ t_grant (授权通行)┌────▼────┐│ 占用(1) │ ← AGV占用路口└────┬────┘│ t_release (释放)▼┌─────────┐│ 空闲(1) │└─────────┘③ 优先级仲裁规则\text{Priority}(AGV_i) f(\text{Type}, \text{Batch}, \text{WaitTime})- Type: 危化品(0) 紧急物料(1) 普通(2)- Batch: 同一批次内 FIFO- WaitTime: 等待时间越长优先级微调防饥饿3.2 管制架构┌─────────────┐│ AGV集群 ││ (多智能体) │└──────┬──────┘│ 通行请求┌─────────▼─────────┐│ 交通管制器 ││ ┌─────────────┐ ││ │ 1. 冲突检测 │ ││ │ (空间重叠) │ ││ └─────────────┘ ││ ┌─────────────┐ ││ │ 2. 优先级排序 │ ││ │ (动态计算) │ ││ └─────────────┘ ││ ┌─────────────┐ ││ │ 3. Petri网演化 │ ││ │ (状态迁移) │ ││ └─────────────┘ ││ ┌─────────────┐ ││ │ 4. 死锁检测 │ ││ │ (超时/回退) │ ││ └─────────────┘ │└─────────┬─────────┘│ 授权/拒绝▼┌─────────────┐│ 交叉路口 ││ (互斥资源) │└─────────────┘四、代码讲解面向对象设计4.1 类结构总览类名职责设计模式AGVType AGV 类型枚举枚举AGVState AGV 状态枚举枚举AGV AGV 实体dataclass 实体对象ConflictZone 冲突区域Petri 网 Place 观察者模式TrafficController 交通控制器仲裁器中介者模式PetriNetSimulator Petri 网仿真引擎策略模式VisualizationEngine 可视化引擎封装4.2 核心代码实现from dataclasses import dataclass, fieldfrom typing import List, Dict, Optional, Tuple, Setfrom enum import Enum, autoimport numpy as npimport matplotlib.pyplot as pltfrom collections import dequeimport heapqimport time# # 1. 基础枚举与数据结构# class AGVType(Enum):AGV 类型 —— 决定基础优先级HAZARDOUS 0 # 危化品运输 (最高)URGENT 1 # 紧急批次NORMAL 2 # 普通物料MAINTENANCE 3 # 维护车辆 (最低)class AGVState(Enum):AGV 运行状态IDLE auto()MOVING auto()WAITING auto() # 在路口等待CROSSING auto() # 正在通过路口ERROR auto()dataclassclass AGV:AGV 实体 —— 实体对象id: stragv_type: AGVTypepriority_base: int field(initFalse)position: Tuple[float, float] (0.0, 0.0)path: List[Tuple[float, float]] field(default_factorylist)state: AGVState AGVState.IDLEwait_start_time: float 0.0batch_id: Optional[str] Nonedef __post_init__(self):self.priority_base self.agv_type.valuedef calculate_dynamic_priority(self, current_time: float) - float:计算动态优先级公式: P base_prio * 1000 - wait_time * 10 random_jitterwait_time current_time - self.wait_start_time if self.wait_start_time 0 else 0# 危化品绝对优先等待时间提供微调以防饥饿jitter np.random.uniform(-0.1, 0.1)return self.priority_base * 1000 - wait_time * 10 jitterdef move_one_step(self, dt: float 1.0) - bool:移动一步返回是否到达目标if not self.path:return Truetarget self.path[0]dx target[0] - self.position[0]dy target[1] - self.position[1]dist np.sqrt(dx*dx dy*dy)if dist 0.1: # 到达节点self.position targetself.path.pop(0)return len(self.path) 0# 简单移动模型step 0.5 * dtratio min(step / dist, 1.0)self.position (self.position[0] dx * ratio,self.position[1] dy * ratio)return False# # 2. 冲突区域与 Petri 网建模# dataclassclass ConflictZone:冲突区域交叉路口对应 Petri 网中的 Placeid: strcenter: Tuple[float, float]radius: float 2.0capacity: int 1 # 最多容纳AGV数互斥occupied_by: Optional[str] Nonewaiting_queue: List[str] field(default_factorylist)def is_available(self) - bool:return self.occupied_by is Nonedef enter(self, agv_id: str) - bool:尝试进入区域if self.is_available():self.occupied_by agv_idreturn Truereturn Falsedef exit(self, agv_id: str) - bool:离开区域if self.occupied_by agv_id:self.occupied_by Nonereturn Truereturn Falsedef request_access(self, agv_id: str) - None:加入等待队列if agv_id not in self.waiting_queue:self.waiting_queue.append(agv_id)def release_access(self, agv_id: str) - None:从等待队列移除if agv_id in self.waiting_queue:self.waiting_queue.remove(agv_id)# # 3. 交通控制器仲裁器# class TrafficController:AGV 交通控制器 —— 中介者模式负责冲突检测、优先级仲裁、死锁预防def __init__(self, conflict_zones: List[ConflictZone]):self.zones: Dict[str, ConflictZone] {z.id: z for z in conflict_zones}self.agvs: Dict[str, AGV] {}self.deadlock_timeout: float 30.0 # 死锁检测超时秒self.current_time: float 0.0def register_agv(self, agv: AGV) - None:注册 AGVself.agvs[agv.id] agvdef detect_conflicts(self, agv: AGV) - List[ConflictZone]:检测 AGV 路径上的冲突区域conflicts []for zone in self.zones.values():# 简单圆形碰撞检测dist np.sqrt((agv.position[0] - zone.center[0])**2 (agv.position[1] - zone.center[1])**2)if dist zone.radius 1.0: # 1.0 为 AGV 半径conflicts.append(zone)return conflictsdef arbitrate(self, zone: ConflictZone) - Optional[str]:仲裁决定哪个 AGV 可以通行返回获得通行权的 AGV IDif zone.is_available():# 从等待队列中选择优先级最高的candidates []for agv_id in zone.waiting_queue:agv self.agvs.get(agv_id)if agv:priority agv.calculate_dynamic_priority(self.current_time)heapq.heappush(candidates, (priority, agv_id))if candidates:_, winner_id heapq.heappop(candidates)# 重新构建等待队列排除获胜者zone.waiting_queue [aid for _, aid in candidates]return winner_idreturn Nonedef check_deadlock(self) - List[str]:死锁检测检查是否有 AGV 等待超时返回需要强制释放的 AGV ID 列表deadlocked []for agv_id, agv in self.agvs.items():if agv.state AGVState.WAITING:wait_time self.current_time - agv.wait_start_timeif wait_time self.deadlock_timeout:# 危化品特殊处理永不强制释放而是释放其他车辆if agv.agv_type ! AGVType.HAZARDOUS:deadlocked.append(agv_id)else:# 强制释放所有非危化品车辆for other_id, other_agv in self.agvs.items():if other_id ! agv_id and other_agv.state AGVState.WAITING:deadlocked.append(other_id)break # 一次只处理一个危化品死锁return deadlockeddef resolve_deadlock(self, agv_id: str) - None:解决死锁强制释放资源agv self.agvs.get(agv_id)if agv:# 从所有冲突区域移除for zone in self.zones.values():zone.exit(agv_id)zone.release_access(agv_id)agv.state AGVState.ERRORprint(f 死锁解除: AGV {agv_id} 被强制重置)def step(self, dt: float) - None:执行一个控制周期self.current_time dt# 1. 检查死锁deadlocked self.check_deadlock()for agv_id in deadlocked:self.resolve_deadlock(agv_id)# 2. 遍历所有 AGVfor agv in self.agvs.values():if agv.state AGVState.ERROR:continue# 检测冲突conflicts self.detect_conflicts(agv)if not conflicts:# 无冲突正常移动if agv.state AGVState.WAITING:agv.state AGVState.MOVINGagv.wait_start_time 0.0agv.move_one_step(dt)else:# 有冲突进入仲裁流程zone conflicts[0] # 简化处理只处理第一个冲突if zone.is_available():# 尝试获取通行权if zone.enter(agv.id):agv.state AGVState.CROSSINGzone.release_access(agv.id)print(f✅ AGV {agv.id} 获得 {zone.id} 通行权)else:# 加入等待队列if agv.state ! AGVState.WAITING:agv.state AGVState.WAITINGagv.wait_start_time self.current_timezone.request_access(agv.id)print(f⏳ AGV {agv.id} 在 {zone.id} 等待 (优先级:{agv.calculate_dynamic_priority(self.current_time):.1f}))else:# 区域被占用加入等待队列if agv.state ! AGVState.WAITING:agv.state AGVState.WAITINGagv.wait_start_time self.current_timezone.request_access(agv.id)occupier self.agvs.get(zone.occupied_by)occupier_info f被{zone.occupied_by}({occupier.agv_type.name})占用 if occupier else 被占用print(f⏳ AGV {agv.id} 在 {zone.id} 等待 ({occupier_info}))# 3. 仲裁所有冲突区域for zone in self.zones.values():if zone.is_available() and zone.waiting_queue:winner self.arbitrate(zone)if winner:agv self.agvs[winner]zone.enter(winner)agv.state AGVState.CROSSINGprint(f 仲裁结果: {zone.id} 分配给 AGV {winner})# # 4. Petri 网仿真引擎# class PetriNetSimulator:Petri 网仿真引擎 —— 策略模式用于形式化验证交通管制逻辑def __init__(self, controller: TrafficController):self.controller controllerself.transitions {request: self._transition_request,grant: self._transition_grant,release: self._transition_release}def _transition_request(self, agv_id: str, zone_id: str) - bool:变迁AGV 申请进入路口zone self.controller.zones.get(zone_id)agv self.controller.agvs.get(agv_id)if zone and agv:zone.request_access(agv_id)return Truereturn Falsedef _transition_grant(self, agv_id: str, zone_id: str) - bool:变迁授权 AGV 进入路口zone self.controller.zones.get(zone_id)if zone and zone.is_available():return zone.enter(agv_id)return Falsedef _transition_release(self, agv_id: str, zone_id: str) - bool:变迁AGV 离开路口zone self.controller.zones.get(zone_id)return zone.exit(agv_id) if zone else Falsedef fire_transition(self, transition_name: str, *args) - bool:触发变迁if transition_name in self.transitions:return self.transitions[transition_name](*args)return False# # 5. 可视化引擎# class VisualizationEngine:可视化引擎 —— 封装def __init__(self, controller: TrafficController):self.controller controllerself.fig, self.ax plt.subplots(figsize(10, 10))self.colors {AGVType.HAZARDOUS: red,AGVType.URGENT: orange,AGVType.NORMAL: blue,AGVType.MAINTENANCE: gray}def draw(self, save_path: str agv_traffic_control.png) - None:绘制当前状态self.ax.clear()# 1. 绘制冲突区域路口for zone in self.controller.zones.values():circle plt.Circle(zone.center, zone.radius,fillFalse, edgecolorblack, linewidth2, linestyle--)self.ax.add_patch(circle)self.ax.text(zone.center[0], zone.center[1], zone.id,hacenter, vacenter, fontsize12, fontweightbold)# 显示占用状态if zone.occupied_by:occupier self.controller.agvs.get(zone.occupied_by)status f占用:{zone.occupied_by}if occupier:status f({occupier.agv_type.name[0]})self.ax.text(zone.center[0], zone.center[1] - zone.radius - 0.5,status, hacenter, vatop, fontsize8, colorred)# 显示等待队列if zone.waiting_queue:queue_text 等待: ,.join(zone.waiting_queue[:3])if len(zone.waiting_queue) 3:queue_text ...self.ax.text(zone.center[0], zone.center[1] zone.radius 0.5,queue_text, hacenter, vabottom, fontsize8, colororange)# 2. 绘制 AGVfor agv in self.controller.agvs.values():color self.colors.get(agv.agv_type, black)marker o if agv.state ! AGVState.ERROR else xsize 100 if agv.agv_type AGVType.HAZARDOUS else 60self.ax.scatter(agv.position[0], agv.position[1],ccolor, ssize, markermarker, edgecolorsblack, linewidths1)# AGV 标签label f{agv.id}if agv.state AGVState.WAITING:wait_time self.controller.current_time - agv.wait_start_timelabel f(等{wait_time:.0f}s)self.ax.text(agv.position[0], agv.position[1] - 0.8, label,hacenter, vatop, fontsize8, fontweightbold)# 绘制路径if agv.path:path_x [agv.position[0]] [p[0] for p in agv.path]path_y [agv.position[1]] [p[1] for p in agv.path]self.ax.plot(path_x, path_y, --, colorcolor, alpha0.3, linewidth1)# 3. 绘制图例legend_elements [plt.Line2D([0], [0], markero, colorw, markerfacecolorred, markersize10, label危化品),plt.Line2D([0], [0], markero, colorw, markerfacecolororange, markersize10, label紧急),plt.Line2D([0], [0], markero, colorw, markerfacecolorblue, markersize10, label普通),plt.Line2D([0], [0], markero, colorw, markerfacecolorgray, markersize10, label维护),plt.Line2D([0], [0], markerx, colorblack, markersize10, label故障/死锁)]self.ax.legend(handleslegend_elements, locupper right)# 4. 设置坐标轴self.ax.set_xlim(-5, 15)self.ax.set_ylim(-5, 15)self.ax.set_aspect(equal)self.ax.grid(True, alpha0.3)self.ax.set_title(fAGV 交通管制仿真 (t{self.controller.current_time:.1f}s),fontsize14, fontweightbold)self.ax.set_xlabel(X 坐标 (m))self.ax.set_ylabel(Y 坐标 (m))plt.tight_layout()plt.savefig(save_path, dpi150, bbox_inchestight)plt.close()print(f 状态可视化已保存至: {save_path})# # 6. 仿真演示# def demo():演示 AGV 交通管制仿真print( * 60)print( AGV 交通管制与优先级调度仿真)p利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛

python的工业过程控制场景模拟第九十七篇:AGV交通管制仿真，交叉路口设置通行优先级，危险品运输车辆优先通过。

相关推荐

2024年淮南招聘网站建设全流程深度解析与企业转型实战指南

解决xactengine3_7.dll丢失的5种安全修复方法

Python周刊2026W32 | PEP 842导出、PR-AF登顶、NumPy线程扩展、PyCon Korea 2026

更多资讯

从零构建Coding Agent：Deuz SDK写代码、跑测试、自愈循环完整教程

class-validator 验证的5个开关：如何让错误只出现在该出现的地方

March7thAssistant 上手指南：3 步跑通星穹铁道日常自动化

Boxer CLI参数速查表：从阈值到精度，所有命令行选项一次讲透

REA：本地 AI 逆向工程指南，用 1 条命令看清竞品应用的内部实现

Kimi-K3 上手指南：从零到首次推理的极简路径

今日更新

AI辅助技术栈评估：提升研发决策效率60%

从生成Token到直接决策：Jev如何重塑AI Agent工程实践

32G统一内存跑大模型：内存账、TPS实测与端云分流

本周精选

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？