python的工业过程控制场景模拟第九十七篇:AGV交通管制仿真，交叉路口设置通行优先级，危险品运输车辆优先通过。
AGV 交通管制与交叉路口优先级调度仿真 —— 基于 Petri 网与动态优先级
“那年厂里上了 12 台 AGV，平时跑得好好的，一到早晚高峰，两台车在十字路口‘顶牛’，谁也不让谁，最后只能人工遥控拖走。后来我们给调度系统加了Petri 网建模 + 动态优先级仲裁，危险品运输车永远最高级，普通物料车见它就避让，路口从此再没堵死过。”
—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸
一、实际应用场景描述
在大型化工、制药园区，AGV 承担着原料、半成品、危化品的自动化转运任务。多车混行、路径交叉是常态：
┌──────────────────────────────────────────────┐
│ AGV 交通管制与优先级调度系统 │
│ │
│ [上位机调度中枢] │
│ │ 路径规划 / 冲突检测 / 优先级仲裁 │
│ ▼ │
│ ┌────────────────────────────┐ │
│ │ 交通管制引擎 (Traffic Control)│ │
│ │ ┌──────────────────────┐ │ │
│ │ │ 1. 冲突区域检测 │ │ │
│ │ │ (Crossing Zones) │ │ │
│ │ └──────────────────────┘ │ │
│ │ ┌──────────────────────┐ │ │
│ │ │ 2. 动态优先级计算 │ │ │
│ │ │ (危化品 > 批次 > FIFO)│ │
│ │ └──────────────────────┘ │ │
│ │ ┌──────────────────────┐ │ │
│ │ │ 3. 死锁检测与解除 │ │ │
│ │ │ (Petri 网/超时) │ │ │
│ │ └──────────────────────┘ │ │
│ └────────────┬───────────────┘ │
│ │ 通行令牌 (Token) │
│ ┌───────┴───────┐ │
│ ▼ ▼ │
│ ┌─────────┐ ┌─────────┐ │
│ │ AGV-01 │ │ AGV-02 │ │
│ │ (危化品) │ │ (普通物料) │ │
│ │ Prio=0 │ │ Prio=2 │ │
│ └────┬────┘ └────┬────┘ │
│ │ 申请通行 │ 申请通行 │
│ ▼ ▼ │
│ ┌────────────────────────────┐ │
│ │ 交叉路口 (Conflict Zone) │ │
│ │ ┌─────┐ ┌─────┐ ┌─────┐ │ │
│ │ │ZoneA│ │ZoneB│ │ZoneC│ │ │
│ │ │(互斥)│ │(互斥)│ │(互斥)│ │ │
│ │ └──┬──┘ └──┬──┘ └──┬──┘ │ │
│ │ └────┬────┘ │ │ │
│ │ ▼ ▼ │ │
│ │ ┌─────────────────┐ │ │
│ │ │ 仲裁器 (Arbiter) │ │ │
│ │ │ • 比较优先级 │ │ │
│ │ │ • 分配通行权 │ │ │
│ │ │ • 超时释放 │ │ │
│ │ └─────────────────┘ │ │
│ └───────────────────────────┘ │
│ │
│ 核心: Petri网建模 + 动态优先级 + 死锁预防 │
└──────────────────────────────────────────────┘
传统自由竞争 vs 优先级管制
维度 自由竞争（无管制） 优先级管制
路口通行 ❌ 互相等待/死锁 ✅ 有序仲裁
危化品运输 ❌ 无保障 ✅ 最高优先级
吞吐量 ❌ 低（频繁避让） ✅ 高（预测性通行）
安全性 ❌ 风险高 ✅ 隔离保障
二、引入痛点
2.1 现场的真实困境
场景 现场发生了什么 根因
“路口顶牛” “两车对峙半小时” 无仲裁机制
“危化品堵路” “剧毒原料车被普通车挡道” 优先级缺失
“幽灵堵车” “明明没车，却一直等” 死锁未检测
“调度混乱” “12台车互相发指令” 中心化不足
“效率低” “高峰期产能腰斩” 无流量控制
2.2 核心矛盾
多 AGV 系统的本质是“共享资源的并发控制”。 交叉路口是典型的临界区（Critical Section），必须保证互斥访问。单纯的“先到先得”无法满足危化品运输的特殊安全要求，必须引入动态优先级，并且通过 Petri 网 形式化建模，确保无死锁、无饥饿。
2.3 我们要解决什么
用一段精简的 Python 程序，构建一个AGV 交通管制仿真系统，实现：
1. Petri 网建模 —— 将路口抽象为 Place，AGV 为 Token
2. 动态优先级 —— 危化品 > 紧急批次 > 普通 FIFO
3. 死锁检测 —— 超时释放 + 资源预留
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, field
from typing import List, Dict, Optional, Tuple, Set
from enum import Enum, auto
import numpy as np
import matplotlib.pyplot as plt
from collections import deque
import heapq
import 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()
@dataclass
class AGV:
"""AGV 实体 —— 实体对象"""
id: str
agv_type: AGVType
priority_base: int = field(init=False)
position: Tuple[float, float] = (0.0, 0.0)
path: List[Tuple[float, float]] = field(default_factory=list)
state: AGVState = AGVState.IDLE
wait_start_time: float = 0.0
batch_id: Optional[str] = None
def __post_init__(self):
self.priority_base = self.agv_type.value
def calculate_dynamic_priority(self, current_time: float) -> float:
"""
计算动态优先级
公式: P = base_prio * 1000 - wait_time * 10 + random_jitter
"""
wait_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 + jitter
def move_one_step(self, dt: float = 1.0) -> bool:
"""移动一步，返回是否到达目标"""
if not self.path:
return True
target = 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 = target
self.path.pop(0)
return len(self.path) == 0
# 简单移动模型
step = 0.5 * dt
ratio = min(step / dist, 1.0)
self.position = (
self.position[0] + dx * ratio,
self.position[1] + dy * ratio
)
return False
# ============================================================
# 2. 冲突区域与 Petri 网建模
# ============================================================
@dataclass
class ConflictZone:
"""
冲突区域（交叉路口）
对应 Petri 网中的 Place
"""
id: str
center: Tuple[float, float]
radius: float = 2.0
capacity: int = 1 # 最多容纳AGV数（互斥）
occupied_by: Optional[str] = None
waiting_queue: List[str] = field(default_factory=list)
def is_available(self) -> bool:
return self.occupied_by is None
def enter(self, agv_id: str) -> bool:
"""尝试进入区域"""
if self.is_available():
self.occupied_by = agv_id
return True
return False
def exit(self, agv_id: str) -> bool:
"""离开区域"""
if self.occupied_by == agv_id:
self.occupied_by = None
return True
return False
def 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.0
def register_agv(self, agv: AGV) -> None:
"""注册 AGV"""
self.agvs[agv.id] = agv
def 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 conflicts
def arbitrate(self, zone: ConflictZone) -> Optional[str]:
"""
仲裁：决定哪个 AGV 可以通行
返回获得通行权的 AGV ID
"""
if 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_id
return None
def 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_time
if 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 deadlocked
def 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.ERROR
print(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. 遍历所有 AGV
for 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.MOVING
agv.wait_start_time = 0.0
agv.move_one_step(dt)
else:
# 有冲突，进入仲裁流程
zone = conflicts[0] # 简化处理：只处理第一个冲突
if zone.is_available():
# 尝试获取通行权
if zone.enter(agv.id):
agv.state = AGVState.CROSSING
zone.release_access(agv.id)
print(f"✅ AGV {agv.id} 获得 {zone.id} 通行权")
else:
# 加入等待队列
if agv.state != AGVState.WAITING:
agv.state = AGVState.WAITING
agv.wait_start_time = self.current_time
zone.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.WAITING
agv.wait_start_time = self.current_time
zone.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.CROSSING
print(f"🎯 仲裁结果: {zone.id} 分配给 AGV {winner}")
# ============================================================
# 4. Petri 网仿真引擎
# ============================================================
class PetriNetSimulator:
"""
Petri 网仿真引擎 —— 策略模式
用于形式化验证交通管制逻辑
"""
def __init__(self, controller: TrafficController):
self.controller = controller
self.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 True
return False
def _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 False
def _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 False
def 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 = controller
self.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,
fill=False, edgecolor='black', linewidth=2, linestyle='--')
self.ax.add_patch(circle)
self.ax.text(zone.center[0], zone.center[1], zone.id,
ha='center', va='center', fontsize=12, fontweight='bold')
# 显示占用状态
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, ha='center', va='top', fontsize=8, color='red')
# 显示等待队列
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, ha='center', va='bottom', fontsize=8, color='orange')
# 2. 绘制 AGV
for agv in self.controller.agvs.values():
color = self.colors.get(agv.agv_type, 'black')
marker = 'o' if agv.state != AGVState.ERROR else 'x'
size = 100 if agv.agv_type == AGVType.HAZARDOUS else 60
self.ax.scatter(agv.position[0], agv.position[1],
c=color, s=size, marker=marker, edgecolors='black', linewidths=1)
# AGV 标签
label = f"{agv.id}"
if agv.state == AGVState.WAITING:
wait_time = self.controller.current_time - agv.wait_start_time
label += f"(等{wait_time:.0f}s)"
self.ax.text(agv.position[0], agv.position[1] - 0.8, label,
ha='center', va='top', fontsize=8, fontweight='bold')
# 绘制路径
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, '--', color=color, alpha=0.3, linewidth=1)
# 3. 绘制图例
legend_elements = [
plt.Line2D([0], [0], marker='o', color='w', markerfacecolor='red', markersize=10, label='危化品'),
plt.Line2D([0], [0], marker='o', color='w', markerfacecolor='orange', markersize=10, label='紧急'),
plt.Line2D([0], [0], marker='o', color='w', markerfacecolor='blue', markersize=10, label='普通'),
plt.Line2D([0], [0], marker='o', color='w', markerfacecolor='gray', markersize=10, label='维护'),
plt.Line2D([0], [0], marker='x', color='black', markersize=10, label='故障/死锁')
]
self.ax.legend(handles=legend_elements, loc='upper right')
# 4. 设置坐标轴
self.ax.set_xlim(-5, 15)
self.ax.set_ylim(-5, 15)
self.ax.set_aspect('equal')
self.ax.grid(True, alpha=0.3)
self.ax.set_title(f'AGV 交通管制仿真 (t={self.controller.current_time:.1f}s)',
fontsize=14, fontweight='bold')
self.ax.set_xlabel('X 坐标 (m)')
self.ax.set_ylabel('Y 坐标 (m)')
plt.tight_layout()
plt.savefig(save_path, dpi=150, bbox_inches='tight')
plt.close()
print(f"📊 状态可视化已保存至: {save_path}")
# ============================================================
# 6. 仿真演示
# ============================================================
def demo():
"""演示 AGV 交通管制仿真"""
print("=" * 60)
print(" AGV 交通管制与优先级调度仿真")
p
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！