企业官网建设流程全解析

厂区多机器人协同仿真 —— 进料/取样/转运全流程

"那年液化气站6个储罐同时进料，三台机器人各干各的，结果Filler在A1灌满溢出、Sampler还在B3慢慢取样、Transporter空跑两趟。后来我们用角色分工+集中调度+状态机同步，三台机器人像流水线一样接力，储罐填充率从混乱的'有的满溢有的空'变成全部齐刷刷停在95%，质检异常也实时送达实验室。"

—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸

一、实际应用场景描述

在石化仓储、精细化工、粮油储运等场景，多台机器人需要在同一厂区内协同完成进料→取样→转运的连续作业：

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

│ 厂区多机器人协同系统 │

│ │

│ [上位机调度中枢] │

│ │ 任务编排 / 状态监控 / 冲突仲裁 │

│ ▼ │

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

│ │ 任务调度层 │ │

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

│ │ │ 1. 任务池管理 │ │ │

│ │ │ (FIFO+优先级) │ │ │

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

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

│ │ │ 2. 角色匹配 │ │ │

│ │ │ Filler/Sampler/ │ │ │

│ │ │ Transporter │ │ │

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

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

│ │ │ 3. 代价评估 │ │ │

│ │ │ (距离+电量+队列) │ │ │

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

│ │ 任务分配指令 │

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

│ ▼ ▼ │

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

│ │ Filler │ │ Sampler │ │

│ │ 进料机器人│ │ 取样机器人│ │

│ │ 💧 │ │ 🔬 │ │

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

│ ▼ ▼ │

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

│ │Transporter│ │ Lab │ │

│ │ 转运机器人│ │ 实验室 │ │

│ │ 📦 │ │ 🔬 │ │

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

│ │

│ 物理世界: 6储罐 + 进料区 + 实验室 + 出货区 + 充电站 │

│ 核心: 角色分工 + 状态机 + 集中调度 + 故障自愈 │

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

单台万能机器人 vs 多角色协同

维度 单台万能机器人 三角色协同

进料速度 ❌ 忙完进料才能取样 ✅ Filler 专攻进料，并行不误

取样时效 ❌ 取样排队等空 ✅ Sampler 实时送达实验室

转运节拍 ❌ 储罐满溢才想起转运 ✅ Transporter 持续排空，防止溢出

故障影响 ❌ 单点故障全线停 ✅ 一台故障，其余接管

电量管理 ❌ 频繁中断充放电 ✅ 集中调度，错峰充电

二、引入痛点

2.1 现场的真实困境

场景 现场发生了什么 根因

"储罐溢出" "A1满了没人管，溢出30L原料" 无角色分工，机器人串行作业

"样品过期" "取样后3小时才送实验室" 无专用 Sampler，取样被插队

"转运空跑" "Transporter 两次去空罐" 无状态感知，盲目执行

"电量雪崩" "三台同时没电，产线停摆" 无错峰充电调度

"任务互踩" "两台机器人撞在路口" 无冲突仲裁与路径协调

2.2 核心矛盾

厂区物流的本质不是"把东西搬来搬去"，而是"让对的人在对的时间出现在对的地点"。 单台机器人什么都能做但什么都做不快，真正的效率来自角色专业化 + 集中调度 + 状态机驱动的并行流水。

2.3 我们要解决什么

用一段精简的 Python 程序，构建一个 厂区多机器人协同仿真系统，实现：

1. 三种角色 —— Filler（进料）、Sampler（取样）、Transporter（转运）

2. 状态机驱动 —— IDLE→MOVING→WORKING→IDLE 闭环

3. 集中调度 —— 任务池 + 角色匹配 + 代价评估

4. 多阶段任务 —— 进料(装载区→储罐)、取样(储罐→实验室)、转运(储罐→出货区)

5. 故障与充电 —— 随机故障、低电量自动回充

6. 可视化 —— 厂区俯视图、液位变化、电量、温度、任务统计

三、核心逻辑讲解

3.1 理论基础：多变量协同与状态机

本工具基于哈工程《工业过程控制》第七章"多变量系统"和第九章"计算机控制系统"：

① 多机器人协同模型

三台机器人构成并行多变量系统：

\mathbf{x}(t+1) = \mathbf{A}\mathbf{x}(t) + \mathbf{B}\mathbf{u}(t)

其中状态向量 \mathbf{x} = [x_f, y_f, x_s, y_s, x_t, y_t]^T 分别对应三台机器人的位置。

② 任务分配优化

调度器解决分配问题：

\min \sum_{i,j} c_{ij} \cdot a_{ij}

约束：

- 每个任务分配给唯一角色匹配的机器人

- 队列长度上限： q_i \le 3

- 电量约束： E_i \ge E_{min}

③ 状态机模型

每台机器人遵循统一状态机：

IDLE ──分配任务──▶ MOVING ──到达──▶ WORKING ──完成──▶ IDLE

▲ │

└──────────── 任务完成 ──────────────────────────┘

│

▼ (电量<15%)

CHARGING ──充满(>90%)──▶ IDLE

3.2 系统架构总览

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

│ 厂区地图/MAP │

│ 6储罐+功能区域 │

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

│ 地理信息

┌─────────▼─────────┐

│ 任务生成器 │

│ • 初始任务批 │

│ • 动态订单注入 │

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

│ 任务池

┌─────────▼─────────┐

│ 集中调度器 │

│ • 角色匹配 │

│ • 代价评估 │

│ • 队列管理 │

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

│ 分配指令

┌─────────▼─────────┐

│ 机器人状态机 │

│ IDLE→MOVE→WORK │

│ CHARGE/Fault处理 │

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

│ 位姿更新

▼

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

│ 物理世界 │

│ 储罐液位变化 │

│ 电量消耗模型 │

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

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

4.1 类结构总览

类名 职责 设计模式

"Pose2D" 二维位姿（值对象） 值对象

"TankState" 储罐状态（dataclass） 值对象

"TaskSpec" 任务规格 值对象

"Robot"（抽象） 机器人基类 模板方法

"FillerRobot" 进料机器人 继承

"SamplerRobot" 取样机器人 继承

"TransporterRobot" 转运机器人 继承

"TaskScheduler" 集中调度器 中介者

"SimContext" 仿真上下文 上下文对象

"TankFarmSimulator" 仿真引擎（聚合根） 聚合根

"VizEngine" 可视化引擎 封装

4.2 核心代码（完整可运行）

完整源码约 577 行，包含 12 个类、三种角色机器人、集中调度、故障注入、可视化。

以下为完整版，可直接复制运行。

<details><summary>🔧 完整源码（点击展开/折叠）</summary>

"""

厂区多机器人协同仿真 —— 储罐进料/取样/转运全流程

参考哈尔滨工程大学《工业过程控制》第七章"多变量系统"与第九章"计算机控制系统"

"""

from dataclasses import dataclass

from typing import List, Dict, Optional, Tuple

from enum import Enum, auto

from abc import ABC, abstractmethod

import numpy as np

import matplotlib.pyplot as plt

from collections import deque, defaultdict

import time, math, random

from datetime import datetime

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

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

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

@dataclass

class Pose2D:

x: float = 0.0; y: float = 0.0; theta: float = 0.0

def distance_to(self, o: 'Pose2D') -> float:

return math.sqrt((self.x-o.x)**2 + (self.y-o.y)**2)

@dataclass

class TankState:

tank_id: str; capacity: float = 1000.0; level: float = 0.0

temperature: float = 25.0; pressure: float = 101.3

fluid_type: str = "unknown"; quality_ok: bool = True

@property

def fill_ratio(self) -> float:

return self.level / self.capacity if self.capacity > 0 else 0

@dataclass

class TaskSpec:

task_id: str; task_type: 'TaskType'

source: str; destination: str

payload: float = 0.0; priority: int = 1

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

# 2. 枚举

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

class TaskType(Enum):

LOAD = "进料"; UNLOAD = "转运"; SAMPLE = "取样"; CHARGE = "充电"

class RobotState(Enum):

IDLE = "空闲"; MOVING = "移动"; WORKING = "工作"

CHARGING = "充电"; FAULT = "故障"

class RobotRole(Enum):

FILLER = "进料机器人"; SAMPLER = "取样机器人"; TRANSPORTER = "转运机器人"

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

# 3. 厂区地图

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

class TankFarmMap:

def __init__(self):

self.locations: Dict[str, Pose2D] = {}

self.tanks: Dict[str, TankState] = {}

self.charging_stations: List[str] = []

self._init()

def _init(self):

fluids = ['原油','柴油','汽油','化工中间体','溶剂','润滑油']

for i, row in enumerate(['A','B']):

for j in range(1, 4):

tid = f"TANK-{row}{j}"

self.locations[tid] = Pose2D(20+(j-1)*30, 30+i*30)

self.tanks[tid] = TankState(

tank_id=tid, capacity=random.choice([800,1000,1200]),

level=random.uniform(50,300),

temperature=random.uniform(20,35),

fluid_type=fluids[(i*3+j-1) % len(fluids)])

self.locations['LOAD-ZONE'] = Pose2D(5, 45)

self.locations['LAB'] = Pose2D(95, 45)

self.locations['SHIP-ZONE'] = Pose2D(50, 90)

for i in range(3):

cs = f"CHARGE-{i+1}"

self.locations[cs] = Pose2D(5+i*45, 5)

self.charging_stations.append(cs)

def get_pose(self, lid: str) -> Optional[Pose2D]:

return self.locations.get(lid)

def draw(self, ax):

for tid, t in self.tanks.items():

p = self.locations[tid]

ax.add_patch(plt.Circle((p.x,p.y), 8,

color=plt.cm.RdYlGn(t.fill_ratio),

alpha=0.6, ec='black', lw=1.5))

ax.text(p.x, p.y, f'{tid}\n{t.fill_ratio*100:.0f}%',

ha='center', va='center', fontsize=7, fontweight='bold')

zones = {'LOAD-ZONE':('进料区','green'),

'LAB':('实验室','purple'),'SHIP-ZONE':('出货区','blue')}

for zid,(lab,col) in zones.items():

p = self.locations[zid]

ax.scatter(p.x,p.y,s=200,c=col,marker='s',ec='black',lw=1.5,zorder=5)

for cs in self.charging_stations:

p = self.locations[cs]

ax.scatter(p.x,p.y,s=100,c='orange',marker='^',ec='black',lw=1)

ax.set_xlabel('X (m)'); ax.set_ylabel('Y (m)')

ax.set_title('Tank Farm Layout'); ax.set_aspect('equal')

ax.grid(True, alpha=0.3); ax.set_xlim(-5,105); ax.set_ylim(-5,100)

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

# 4. 机器人基类

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

class Robot(ABC):

def __init__(self, rid: str, role: RobotRole, pose: Pose2D,

max_speed=2.0, battery=100.0):

self.rid=rid; self.role=role; self.pose=pose

self.max_speed=max_speed; self.battery=battery

self.state=RobotState.IDLE; self.task=None

self.queue: deque = deque()

self.work_t=0.0; self.phase=0

self.total_dist=0.0; self.done_count=0

self.fault_prob=0.001

def move_to(self, tgt: Pose2D, dt: float) -> bool:

d = self.pose.distance_to(tgt); step = self.max_speed*dt

if d < max(step,0.5):

self.pose.x,self.pose.y = tgt.x,tgt.y

self.total_dist += d; return True

r = step/d

self.pose.x += (tgt.x-self.pose.x)*r

self.pose.y += (tgt.y-self.pose.y)*r

self.total_dist += step; return False

def consume_bat(self, dt: float, load=1.0):

self.battery = max(0.0, self.battery - (0.01+0.005*load)*dt)

def check_fault(self) -> bool:

if random.random() < self.fault_prob:

self.state = RobotState.FAULT

print(f" [FAULT] {self.rid} 故障！"); return True

return False

def assign(self, task: TaskSpec) -> bool:

if self.state == RobotState.FAULT: return False

self.queue.append(task); return True

def update(self, dt: float, ctx: 'SimContext'):

if self.state == RobotState.CHARGING:

self.battery = min(100.0, self.battery+5.0*dt)

if self.battery > 90:

self.state = RobotState.IDLE

print(f" [CHARGE] {self.rid} 充电完成")

return

if self.battery < 15 and self.state != RobotState.CHARGING:

self._go_charge(ctx); return

if self.check_fault(): return

if self.state == RobotState.IDLE and self.queue:

self.task = self.queue.popleft()

self.state = RobotState.MOVING

self.work_t = 0.0; self.phase = 0

if self.task and self.state != RobotState.FAULT:

if self.execute(dt, ctx):

self.done_count += 1

ctx.scheduler.task_done(self.task)

self.task = None

self.state = RobotState.IDLE

load = 1.5 if self.state == RobotState.WORKING else 1.0

self.consume_bat(dt, load)

def _go_charge(self, ctx: 'SimContext'):

nearest=None; md=float('inf')

for cs in ctx.map.charging_stations:

p = ctx.map.get_pose(cs)

if p and self.pose.distance_to(p) < md:

md = self.pose.distance_to(p); nearest = cs

if nearest:

self.task = TaskSpec(f"CHG-{self.rid}",TaskType.CHARGE,self.rid,nearest)

self.state = RobotState.MOVING; self.phase=0

print(f" [CHARGE] {self.rid} 低电量→{nearest}")

@abstractmethod

def execute(self, dt: float, ctx: 'SimContext') -> bool: pass

@property

@abstractmethod

def color(self) -> str: pass

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

# 5. 三种角色机器人

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

class FillerRobot(Robot):

"""进料：装载区→储罐，持续注入至95%"""

def __init__(self, rid, pose): super().__init__(rid,RobotRole.FILLER,pose,1.8)

@property

def color(self): return 'blue'

def execute(self, dt, ctx):

t = self.task

if t.task_type == TaskType.CHARGE:

p = ctx.map.get_pose(t.destination)

if p and self.move_to(p, dt): self.state = RobotState.CHARGING

return False

if self.phase == 0: # →装载区

p = ctx.map.get_pose(t.source)

if p and self.move_to(p, dt):

self.phase=1; self.work_t=0.0

print(f" [FILL] {self.rid}: 开始注入 → {t.destination}")

return False

if self.phase == 1: # →储罐

p = ctx.map.get_pose(t.destination)

if p and self.move_to(p, dt):

self.phase=2; self.work_t=0.0

return False

# 注入中

self.state = RobotState.WORKING; self.work_t += dt

tank = ctx.map.tanks.get(t.destination)

if tank:

tank.level = min(tank.capacity, tank.level + 50.0*dt)

tank.temperature += random.uniform(-0.05,0.1)*dt

if tank.fill_ratio >= 0.95:

print(f" [FILL] {self.rid}: {t.destination} 已满 ({tank.level:.0f}L)")

return True

if self.work_t > 25.0:

print(f" [FILL] {self.rid}: 注入超时"); return True

return False

class SamplerRobot(Robot):

"""取样：储罐→取样3s→送实验室"""

def __init__(self, rid, pose): super().__init__(rid,RobotRole.SAMPLER,pose,2.5)

@property

def color(self): return 'purple'

def execute(self, dt, ctx):

t = self.task

if t.task_type == TaskType.CHARGE:

p = ctx.map.get_pose(t.destination)

if p and self.move_to(p, dt): self.state = RobotState.CHARGING

return False

if self.phase == 0: # →储罐

p = ctx.map.get_pose(t.source)

if p and self.move_to(p, dt):

self.phase=1; self.work_t=0.0

print(f" [SAMP] {self.rid}: 在 {t.source} 取样中...")

return False

if self.phase == 1: # 取样中

self.state = RobotState.WORKING; self.work_t += dt

if self.work_t >= 3.0:

self.phase=2; self.work_t=0.0

return False

if self.phase == 2: # →实验室

p = ctx.map.get_pose(t.destination)

if p and self.move_to(p, dt):

tank = ctx.map.tanks.get(t.source)

if tank:

tank.quality_ok = random.random() > 0.05

ctx.stats['qc'] += 1

if not tank.quality_ok: ctx.stats['qf'] += 1

st = "合格" if tank.quality_ok else "异常"

print(f" [SAMP] {self.rid}: 样品→实验室 [{st}]")

return True

return False

class TransporterRobot(Robot):

"""转运：储罐→装载→出货区→卸料"""

def __init__(self, rid, pose): super().__init__(rid,RobotRole.TRANSPORTER,pose,2.2)

@property

def color(self): return 'darkgreen'

def execute(self, dt, ctx):

t = self.task

if t.task_type == TaskType.CHARGE:

p = ctx.map.get_pose(t.destination)

if p and self.move_to(p, dt): self.state = RobotState.CHARGING

return False

if self.phase == 0: # →储罐

p = ctx.map.get_pose(t.source)

if p and self.move_to(p, dt):

self.phase=1; self.work_t=0.0

return False

if self.phase == 1: # 装载

self.state = RobotState.WORKING; self.work_t += dt

tank = ctx.map.tanks.get(t.source)

if tank and tank.level > 50:

load = min(t.payload, tank.level, 500)

tank.level -= load

print(f" [TRANS] {self.rid}: 从 {t.source} 装 {load:.0f}L")

else:

print(f" [TRANS] {self.rid}: {t.source} 液位不足")

self.phase=2; self.work_t=0.0

return False

if self.phase == 2: # →出货区

p = ctx.map.get_pose(t.destination)

if p and self.move_to(p, dt):

self.phase=3; self.work_t=0.0

return False

if self.phase == 3: # 卸料

self.state = RobotState.WORKING; self.work_t += dt

if self.work_t >= 4.0:

print(f" [TRANS] {self.rid}: 卸料→{t.destination}")

return True

return False

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

# 6. 任务调度器

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

class TaskScheduler:

def __init__(self):

self.pending: deque = deque()

self.completed: List[TaskSpec] = []

def submit(self, t: TaskSpec): self.pending.append(t)

def task_done(self, t: TaskSpec): self.completed.append(t)

def dispatch(self, robots: Dict[str, Robot]) -> int:

role_map = {TaskType.LOAD:RobotRole.FILLER,

TaskType.SAMPLE:RobotRole.SAMPLER,

TaskType.UNLOAD:RobotRole.TRANSPORTER}

assigned = 0

for task in list(self.pending):

rr = role_map.get(task.task_type)

if not rr: continue

best=None; mc=float('inf')

for r in robots.values():

if r.role!=rr or r.state in (RobotState.FAULT,RobotState.CHARGING):

continue

if len(r.queue)>=3: continue

cost = len(r.queue)*10 + (50 if r.battery<30 else 0)

if cost < mc: mc=cost; best=r

if best and best.assign(task):

self.pending.remove(task); assigned+=1

return assigned

def stats(self) -> Dict:

return {'pending':len(self.pending),'done':len(self.completed)}

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

# 7. 仿真上下文

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

class SimContext:

def __init__(self, m: TankFarmMap):

self.map=m; self.robots: Dict[str,Robot]={}

self.scheduler=TaskScheduler()

self.time=0.0; self.events: List[str]=[]

self.history: Dict[str,List[Tuple[float,float]]] = defaultdict(list)

self.stats: Dict[str,int] = {'qc':0,'qf':0}

def add(self, r: Robot): self.robots[r.rid]=r

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

# 8. 仿真引擎（聚合根）

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

class TankFarmSimulator:

def __init__(self):

self.map = TankFarmMap()

self.ctx = SimContext(self.map)

self.dt = 0.5; self.max_time = 150.0

self._init_robots(); self._init_tasks()

print("="*60)

print(" 厂区多机器人协同仿真系统")

print(" 进料 → 取样 → 转运 全流程")

print(" 基于哈尔滨工程大学《工业过程控制》")

print("="*60)

self._print_layout()

def _init_robots(self):

cfgs = [('FILL-01',RobotRole.FILLER,Pose2D(5,10)),

('SAMP-01',RobotRole.SAMPLER,Pose2D(95,80)),

('TRANS-01',RobotRole.TRANSPORTER,Pose2D(50,10))]

for rid,role,pose in cfgs:

bot = (FillerRobot if role==RobotRole.FILLER else

SamplerRobot if role==RobotRole.SAMPLER else

TransporterRobot)(rid, pose)

self.ctx.add(bot)

print(f"\n[ROBOTS] {len(self.ctx.robots)} 台:")

for r in self.ctx.robots.values():

print(f" {r.rid}: {r.role.value} @({r.pose.x:.0f},{r.pose.y:.0f})")

def _init_tasks(self):

tasks = [

TaskSpec('T001',TaskType.LOAD,'LOAD-ZONE','TANK-A1',200,2),

TaskSpec('T002',TaskType.LOAD,'LOAD-ZONE','TANK-A2',300,1),

TaskSpec('T003',TaskType.LOAD,'LOAD-ZONE','TANK-B1',150,3),

TaskSpec('T004',TaskType.SAMPLE,'TANK-A1','LAB',0,2),

TaskSpec('T005',TaskType.SAMPLE,'TANK-B2','LAB',0,1),

TaskSpec('T006',TaskType.SAMPLE,'TANK-A3','LAB',0,3),

TaskSpec('T007',TaskType.UNL

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