企业官网建设流程全解析

AGV与过程控制系统联动：储罐低位自动调度原料输送

“反应釜还差3吨原料，但中控室没人注意到储罐液位已经报警。等到工艺员发现时，生产线已经停了20分钟。”

—— 基于哈尔滨工程大学《工业过程控制》第九章“计算机集成过程控制”与第五章“前馈-反馈控制”思想

一、实际应用场景描述

在化工、制药、食品、锂电等流程工业中，原料输送是典型的间歇式批量控制（Batch Control）场景：

┌──────────────────────────────────────────────────────┐

│ AGV与过程控制系统联动架构 │

│ │

│ [上层MES/ERP系统] │

│ • 生产排程 • 原料需求计算 • 批次管理 │

│ │ │

│ ▼ │

│ ┌──────────────────────────────────────────┐ │

│ │ 过程控制系统 (DCS/PLC) │ │

│ │ ┌────────────────────────────────────┐ │ │

│ │ │ 储罐液位监控 (LT-101) │ │ │

│ │ │ • 实时采集 • 滤波 • 报警判断 │ │ │

│ │ └──────────────┬─────────────────────┘ │ │

│ │ │ LL(低低液位)报警 │ │

│ │ ▼ │ │

│ │ ┌────────────────────────────────────┐ │ │

│ │ │ 联锁逻辑 (Interlock Logic) │ │ │

│ │ │ • 液位<LL → 触发AGV调度请求 │ │ │

│ │ │ • 检查阀门状态 • 安全联锁 │ │ │

│ │ └──────────────┬─────────────────────┘ │ │

│ └──────────────────┼───────────────────────┘ │

│ │ Modbus/TCP, OPC UA │

│ ▼ │

│ ┌──────────────────────────────────────────┐ │

│ │ AGV调度管理系统 (上位机) │ │

│ │ ┌────────────┐ ┌────────────┐ │ │

│ │ │ 任务生成器 │ │ 路径规划器 │ │ │

│ │ │ • 解析调度请求 │ │ │ │

│ │ │ • 生成运输任务 │ │ │ │

│ │ └────────────┘ └────────────┘ │ │

│ │ │ │ │ │

│ │ ▼ ▼ │ │

│ │ ┌────────────────────────────────────┐ │ │

│ │ │ 交通管制与避障 (Traffic Control) │ │ │

│ │ │ • 区域锁定 • 路口仲裁 • 防碰撞 │ │ │

│ │ └──────────────┬─────────────────────┘ │ │

│ └──────────────────┼───────────────────────┘ │

│ │ MQTT/WebSocket │

│ ▼ │

│ ┌──────────────────────────────────────────┐ │

│ │ 现场AGV执行层 │ │

│ │ • 激光导航 • 二维码定位 • 避障雷达 │ │

│ │ • 举升机构 • 称重传感器 • 料位检测 │ │

│ │ • 对接储罐进料阀 • 对接原料仓出料阀 │ │

│ └──────────────────────────────────────────┘ │

│ │

│ 典型控制流程: │

│ 1. 储罐液位持续下降 → 2. 触发LL报警 │

│ 3. DCS发送调度请求 → 4. AGV系统生成任务 │

│ 5. AGV前往原料仓 → 6. 装载原料（称重校验） │

│ 7. AGV前往储罐区 → 8. 对接进料阀（位置+信号） │

│ 9. 开启进料阀 → 10. 输送完成（重量/液位确认） │

│ 11. AGV返回待机位 → 12. DCS确认原料到位 │

└──────────────────────────────────────────────────────┘

典型应用场景

场景 原料类型 储罐容量 联动痛点

精细化工 溶剂/助剂 5~20m³ 多品种切换频繁，人工调度易出错

原料药 反应中间体 2~10m³ GMP要求严格，需完整批次追溯

锂电池 NMP溶剂 10~50m³ 易燃易爆，需本质安全设计

食品添加剂 糖浆/香精 5~30m³ 卫生级要求，CIP清洗联动

涂料 树脂/溶剂 20~100m³ 多储罐协同，配方管理复杂

二、引入痛点

2.1 现场的真实困境

场景 现场反馈 根因

“断料了才知道” “液位报警响了10分钟，工艺员才看到，反应釜已经干了” 无自动调度，依赖人工巡检

“叫车没人理” “打电话叫AGV，调度室说车在充电，等了半小时才来” 无统一调度，资源分配混乱

“送错料了” “AGV把A原料送到了B储罐，整批产品报废” 无联锁校验，人工确认不可靠

“送了但没到位” “AGV显示任务完成，但储罐还是空的，原来对接没到位” 无位置闭环确认，信号不可靠

“送多了溢出来” “液位计卡了，AGV一直送，原料从储罐顶溢出来” 无多重校验，单点故障风险

“批次追溯不了” “审计来了要查这批原料谁送的、什么时候送的，查不到” 无电子化记录，纸质记录易丢失

2.2 核心矛盾

AGV与过程控制系统联动的核心矛盾是“过程控制的连续性要求”与“AGV调度的离散事件特性”之间的时空失配。

过程控制是连续时间系统，而AGV调度是离散事件系统，两者需要通过事件驱动架构和状态同步机制实现协同。

2.3 三种解决思路

思路 核心思想 优势 劣势

轮询查询 DCS定时查询AGV状态 实现简单 响应延迟大，资源浪费

中断触发 液位报警触发AGV调度 响应快 需处理并发冲突

前馈-反馈 预测需求+实际校验 精度高，鲁棒性强 建模复杂，需历史数据

三、核心逻辑讲解

3.1 理论基础：前馈-反馈复合控制

本工具基于哈工程《工业过程控制》第五章“前馈控制”与第九章“计算机集成过程控制”：

① 前馈控制（Feedforward Control）

G_{ff}(s) = -\frac{G_{d}(s)}{G_{p}(s)}

其中：

- G_{d}(s) ：干扰通道传递函数（原料消耗）

- G_{p}(s) ：控制通道传递函数（AGV输送）

物理意义：根据原料消耗速率提前预测储罐液位变化，在液位到达低位前就调度AGV。

② 反馈控制（Feedback Control）

u(t) = K_p e(t) + K_i \int e(t)dt + K_d \frac{de(t)}{dt}

其中 e(t) = H_{set} - H(t) （液位设定值-实际值）

③ 前馈-反馈复合控制

┌─────────────────────────────────────────────────────┐

│ 前馈-反馈复合控制结构 │

│ │

│ 原料消耗干扰 d(t) ──→ [Gd(s)] ──┐ │

│ ↓ │ │

│ 设定值 Hsp ──→ [+] → [Gc(s)] → [+] → u(t) → [Gp(s)] → 储罐液位 H(t) │

│ ↑ (PID) ↑ (AGV输送) │

│ │ │ │

│ └── 反馈 ──────────┘ │

│ │

│ 前馈补偿器 Gff(s) = -Gd(s)/Gp(s) │

│ • 根据消耗速率预测液位下降 │

│ • 提前调度AGV，避免液位过低 │

│ • 反馈PID消除余差，保证最终精度 │

└─────────────────────────────────────────────────────┘

3.2 事件驱动架构（EDA）

AGV调度事件流:

┌─────────────────────────────────────────────────────┐

│ Level 0: 物理事件 │

│ • 液位传感器信号变化 │

│ • AGV到位传感器触发 │

│ • 阀门开关状态反馈 │

├─────────────────────────────────────────────────────┤

│ Level 1: 检测与报警 │

│ • 液位低(L)报警 → 触发前馈调度 │

│ • 液位低低(LL)报警 → 触发紧急调度 │

│ • 阀门故障 → 触发安全联锁 │

├─────────────────────────────────────────────────────┤

│ Level 2: 逻辑处理 │

│ • 联锁条件检查 (AND/OR逻辑) │

│ • 调度请求生成 (含原料类型/数量/目标) │

│ • 安全条件验证 (无人在罐区/无冲突任务) │

├─────────────────────────────────────────────────────┤

│ Level 3: 调度执行 │

│ • AGV资源分配 (最近空闲AGV) │

│ • 路径规划 (避障+最短路径) │

│ • 任务下发 (含对接坐标/动作序列) │

├─────────────────────────────────────────────────────┤

│ Level 4: 执行监控 │

│ • 位置闭环确认 (激光雷达+二维码) │

│ • 重量校验 (AGV称重 vs 储罐液位) │

│ • 超时监控 (单步动作超时报警) │

├─────────────────────────────────────────────────────┤

│ Level 5: 完成确认 │

│ • 输送量确认 (液位增量+重量减量) │

│ • 批次记录生成 (时间戳+操作人+设备ID) │

│ • DCS反馈 (原料到位信号) │

└─────────────────────────────────────────────────────┘

3.3 状态机设计（AGV任务生命周期）

AGV任务状态机:

┌──────────┐ 调度请求 ┌──────────┐ 资源分配 ┌──────────┐

│ IDLE │ ─────────→ │ PENDING │ ─────────→ │ DISPATCHED│

│ (空闲) │ │ (等待) │ │ (已派车) │

└──────────┘ └──────────┘ └─────┬────┘

│ 路径规划完成

┌──────────┐ 对接完成 ┌──────────┐ 输送完成 ┌─────▼────┐

│ COMPLETED│ ←──────── │ DOCKING │ ←──────── │ LOADING │

│ (完成) │ │ (对接中) │ │ (输送中) │

└──────────┘ └──────────┘ └─────┬────┘

│ 开始移动

┌──────────┐ 异常恢复 ┌──────────┐ 任务取消 ┌─────▼────┐

│ FAILED │ ←──────── │ ERROR │ ←──────── │ MOVING_TO │

│ (失败) │ │ (异常) │ │ (前往中) │

└──────────┘ └──────────┘ └──────────┘

关键状态转换条件:

• IDLE → PENDING: 收到DCS调度请求 + 联锁条件满足

• PENDING → DISPATCHED: AGV资源分配成功

• DISPATCHED → MOVING_TO: 路径规划完成 + 安全条件OK

• MOVING_TO → DOCKING: AGV到位 + 位置精度<±5mm

• DOCKING → LOADING: 对接成功 + 阀门状态确认

• LOADING → COMPLETED: 输送完成 + 液位/重量校验OK

• 任意状态 → ERROR: 超时/传感器故障/安全联锁触发

• ERROR → PENDING: 人工确认 + 故障复位

四、代码讲解（面向对象设计）

4.1 类结构总览

类名 职责 设计模式

"TankLevel" 储罐液位状态（dataclass） 值对象

"AGVStatus" AGV状态（dataclass） 值对象

"TransportTask" 运输任务（dataclass） 值对象

"AlarmLevel" 报警等级枚举 枚举类型

"TaskStatus" 任务状态枚举 枚举类型

"TankController" 储罐控制器（前馈-反馈） 领域模型

"AGVManager" AGV调度管理器 策略模式

"InterlockLogic" 联锁逻辑校验 模板方法

"EventDispatcher" 事件分发器 观察者模式

"BatchRecorder" 批次记录器 单例模式

4.2 核心代码（精简版，CSDN友好）

完整源码约 520 行，单文件，包含 9 个核心类 + 完整联动逻辑。

以下为关键类摘录，完整代码见附件。

"""

AGV与过程控制系统联动：储罐低位自动调度原料输送

参考哈尔滨工程大学《工业过程控制》第五章"前馈控制"与第九章"计算机集成过程控制"

"""

from dataclasses import dataclass, field

from typing import List, Optional, Dict, Callable, Deque

from enum import Enum, auto

import time

import threading

import numpy as np

from collections import deque

import logging

from abc import ABC, abstractmethod

# ============================================================

# 1. 基础数据结构（枚举与值对象）

# ============================================================

class AlarmLevel(Enum):

"""报警等级"""

NORMAL = 0 # 正常

LOW = 1 # 低液位

LOW_LOW = 2 # 低低液位（紧急）

HIGH = 3 # 高液位

HIGH_HIGH = 4 # 高高液位（溢流）

class TaskStatus(Enum):

"""任务状态"""

IDLE = "空闲"

PENDING = "等待调度"

DISPATCHED = "已派车"

MOVING_TO_SOURCE = "前往原料仓"

LOADING = "装载原料"

MOVING_TO_TARGET = "前往储罐"

UNLOADING = "卸载原料"

DOCKING = "对接中"

COMPLETED = "完成"

FAILED = "失败"

CANCELLED = "取消"

ERROR = "异常"

@dataclass

class TankLevel:

"""储罐液位状态"""

tank_id: str

current_level: float # 当前液位 (m)

capacity: float # 储罐容量 (m³)

low_limit: float # 低液位报警值 (m)

low_low_limit: float # 低低液位报警值 (m)

high_limit: float # 高液位报警值 (m)

max_level: float # 最高液位 (m)

consumption_rate: float # 原料消耗速率 (m³/h)

last_update: float = field(default_factory=time.time)

@property

def level_percent(self) -> float:

"""液位百分比"""

return (self.current_level / self.max_level) * 100

@property

def alarm_status(self) -> AlarmLevel:

"""当前报警状态"""

if self.current_level <= self.low_low_limit:

return AlarmLevel.LOW_LOW

elif self.current_level <= self.low_limit:

return AlarmLevel.LOW

elif self.current_level >= self.high_limit:

return AlarmLevel.HIGH

elif self.current_level >= self.max_level * 0.95:

return AlarmLevel.HIGH_HIGH

return AlarmLevel.NORMAL

def predict_time_to_low(self) -> float:

"""预测到达低液位的时间（小时）"""

if self.consumption_rate <= 0:

return float('inf')

remaining_volume = self.current_level - self.low_limit

return remaining_volume / self.consumption_rate

@dataclass

class AGVStatus:

"""AGV状态"""

agv_id: str

position: List[float] = field(default_factory=lambda: [0, 0])

battery_level: float = 100.0

is_available: bool = True

current_task_id: Optional[str] = None

payload_weight: float = 0.0 # 当前载重 (kg)

max_payload: float = 1000.0 # 最大载重 (kg)

location_id: str = "charging_station"

@property

def is_busy(self) -> bool:

return self.current_task_id is not None

@dataclass

class TransportTask:

"""运输任务"""

task_id: str

source_id: str # 原料仓ID

target_tank_id: str # 目标储罐ID

material_type: str # 原料类型

required_amount: float # 需求数量 (kg)

priority: int = 1 # 优先级（1-5，5最高）

status: TaskStatus = TaskStatus.IDLE

created_time: float = field(default_factory=time.time)

start_time: Optional[float] = None

completion_time: Optional[float] = None

retry_count: int = 0

max_retries: int = 3

context: Dict = field(default_factory=dict)

# ============================================================

# 2. 储罐控制器（前馈-反馈复合控制）

# ============================================================

class TankController:

"""

储罐控制器 —— 前馈-反馈复合控制

基于哈工程《工业过程控制》第五章"前馈控制"思想

控制策略:

1. 前馈: 根据消耗速率预测液位，提前调度AGV

2. 反馈: PID控制液位，消除余差

3. 联锁: 液位低低时强制调度，保障安全

"""

def __init__(self, tank_config: TankLevel):

self.tank = tank_config

self.pid_params = {'Kp': 0.8, 'Ki': 0.1, 'Kd': 0.05}

self.pid_state = {'integral': 0.0, 'prev_error': 0.0}

self.feedforward_gain = 1.2 # 前馈补偿增益

self.setpoint = tank_config.capacity * 0.6 # 设定值60%液位

self.last_control_time = time.time()

# 调度请求回调

self.dispatch_callback: Optional[Callable[[str, float], None]] = None

# 历史数据

self.level_history: Deque[float] = deque(maxlen=100)

self.dispatch_history: List[Dict] = []

logging.info(f"[Tank] 储罐 {tank_config.tank_id} 控制器初始化完成")

def update_level(self, new_level: float, consumption_rate: Optional[float] = None):

"""更新液位（模拟传感器采集）"""

self.tank.current_level = new_level

if consumption_rate is not None:

self.tank.consumption_rate = consumption_rate

self.tank.last_update = time.time()

self.level_history.append(new_level)

# 检查报警

self._check_alarms()

def _check_alarms(self):

"""检查液位报警并触发调度"""

alarm = self.tank.alarm_status

if alarm == AlarmLevel.LOW_LOW:

logging.critical(f"[Tank] {self.tank.tank_id} 低低液位报警! "

f"当前: {self.tank.current_level:.2f}m")

# 紧急调度AGV

if self.dispatch_callback:

required = self._calculate_required_amount()

self.dispatch_callback(self.tank.tank_id, required)

self.dispatch_history.append({

'time': time.time(),

'type': 'LL_ALARM',

'level': self.tank.current_level,

'required': required

})

elif alarm == AlarmLevel.LOW:

logging.warning(f"[Tank] {self.tank.tank_id} 低液位报警! "

f"当前: {self.tank.current_level:.2f}m")

# 预测性调度（前馈）

time_to_low = self.tank.predict_time_to_low()

if time_to_low < 0.5: # 30分钟内到达低低液位

logging.warning(f"[Tank] 预计 {time_to_low*60:.1f} 分钟后到达低低液位")

if self.dispatch_callback:

required = self._calculate_required_amount()

self.dispatch_callback(self.tank.tank_id, required)

def _calculate_required_amount(self) -> float:

"""计算所需原料量（考虑安全余量）"""

# 目标液位：设定值 + 安全余量（10%）

target_level = self.setpoint * 1.1

required_volume = target_level - self.tank.current_level

required_volume = max(0, required_volume)

# 转换为重量（假设密度1000kg/m³）

density = 1000.0

required_weight = required_volume * density

# 考虑AGV载重限制（分批输送）

max_agv_load = 800.0 # kg

if required_weight > max_agv_load:

required_weight = max_agv_load

return required_weight

def feedforward_control(self) -> float:

"""

前馈控制：根据消耗速率预测需求

前馈补偿器: G_ff = -Gd/Gp

简化: u_ff = K_ff * d(t)

其中 d(t) 是消耗速率

"""

if self.tank.consumption_rate <= 0:

return 0.0

# 前馈输出 = 增益 × 消耗速率

ff_output = self.feedforward_gain * self.tank.consumption_rate

# 转换为AGV调度需求（kg/h → kg）

ff_demand = ff_output * 0.1 # 假设调度周期为0.1h (6分钟)

return max(0, ff_demand)

def feedback_control(self) -> float:

"""

反馈控制：PID调节液位

u(t) = Kp*e + Ki*∫e*dt + Kd*de/dt

"""

current_time = time.time()

dt = current_time - self.last_control_time

if dt <= 0:

dt = 0.1

# 计算误差

error = self.setpoint - self.tank.current_level

# 比例项

p_term = self.pid_params['Kp'] * error

# 积分项（带抗饱和）

self.pid_state['integral'] += error * dt

if self.pid_state['integral'] > 100:

self.pid_state['integral'] = 100

elif self.pid_state['integral'] < -100:

self.pid_state['integral'] = -100

i_term = self.pid_params['Ki'] * self.pid_state['integral']

# 微分项

d_term = self.pid_params['Kd'] * (error - self.pid_state['prev_error']) / dt

self.pid_state['prev_error'] = error

# PID总输出

pid_output = p_term + i_term + d_term

self.last_control_time = current_time

return max(0, pid_output)

def composite_control(self) -> float:

"""前馈-反馈复合控制"""

ff = self.feedforward_control()

fb = self.feedback_control()

total = ff + fb

logging.debug(f"[Tank] 复合控制: 前馈={ff:.2f}, 反馈={fb:.2f}, 总={total:.2f}")

return total

# ============================================================

# 3. AGV调度管理器（策略模式）

# ============================================================

class AGVManager:

"""

AGV调度管理器

负责AGV资源分配、路径规划、任务调度

"""

def __init__(self):

self.agvs: Dict[str, AGVStatus] = {}

self.tasks: Dict[str, TransportTask] = {}

self.task_queue: Deque[TransportTask] = deque()

self.lock = threading.Lock()

# 站点坐标（简化）

self.locations = {

'charging_station': [0, 0],

'raw_material_A': [10, 0],

'raw_material_B': [10, 10],

'tank_T001': [30, 5],

'tank_T002': [30, 15],

'weighing_station': [20, 5]

}

# 调度线程

self.scheduler_thread = threading.Thread(

target=self._scheduler_loop, daemon=True

)

self.scheduler_thread.start()

logging.info("[AGV] AGV调度管理器启动")

def register_agv(self, agv: AGVStatus):

"""注册AGV"""

with self.lock:

self.agvs[agv.agv_id] = agv

logging.info(f"[AGV] 注册AGV: {agv.agv_id}")

def create_transport_task(self, tank_id: str, required_amount: float,

material_type: str = "A") -> str:

"""创建运输任务"""

task_id = f"TASK_{int(time.time())}_{tank_id}"

# 确定原料仓（简化逻辑）

source_id = f"raw_material_{material_type}"

task = TransportTask(

task_id=task_id,

source_id=source_id,

target_tank_id=tank_id,

material_type=material_type,

required_amount=required_amount,

priority=5 if required_amount > 800 else 3 # 大需求量高优先级

)

with self.lock:

self.tasks[task_id] = task

self.task_queue.append(task)

logging.info(f"[AGV] 创建任务: {task_id}, 储罐: {tank_id}, "

f"需求量: {required_amount:.1f}kg")

return task_id

def _scheduler_loop(self):

"""调度器主循环"""

while True:

with self.lock:

if self.task_queue:

# 按优先级排序

sorted_tasks = sorted(

self.task_queue,

key=lambda t: (-t.priority, t.created_time)

)

for task in sorted_tasks:

if self._dispatch_task(task):

self.task_queue.remove(task)

break

time.sleep(0.5) # 调度周期

def _dispatch_task(self, task: TransportTask) -> bool:

"""调度任务到AGV"""

# 查找可用AGV

available_agvs = [

agv for agv in self.agvs.values()

if agv.is_available and agv.battery_level > 20

]

if not available_agvs:

logging.warning(f"[AGV] 无可用AGV，任务 {task.task_id} 等待")

return False

# 选择最近AGV（简化：选择第一个）

selected_agv = available_agvs[0]

# 分配任务

selected_agv.is_available = False

selected_agv.current_task_id = task.task_id

task.status = TaskStatus.DISPATCHED

task.start_time = time.time()

logging.info(f"[AGV] 任务 {task.task_id} 分配给 AGV {selected_agv.agv_id}")

# 启动任务执行线程

threading.Thread(

target=self._execute_task,

args=(selected_agv, task),

daemon=True

).start()

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！