厂区多机器人协同仿真 —— 进料/取样/转运全流程那年液化气站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}(t1) \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%)──▶ IDLE3.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 个类、三种角色机器人、集中调度、故障注入、可视化。以下为完整版可直接复制运行。detailssummary 完整源码点击展开/折叠/summary厂区多机器人协同仿真 —— 储罐进料/取样/转运全流程参考哈尔滨工程大学《工业过程控制》第七章多变量系统与第九章计算机控制系统from dataclasses import dataclassfrom typing import List, Dict, Optional, Tuplefrom enum import Enum, autofrom abc import ABC, abstractmethodimport numpy as npimport matplotlib.pyplot as pltfrom collections import deque, defaultdictimport time, math, randomfrom datetime import datetime# # 1. 基础数据结构值对象# dataclassclass Pose2D:x: float 0.0; y: float 0.0; theta: float 0.0def distance_to(self, o: Pose2D) - float:return math.sqrt((self.x-o.x)**2 (self.y-o.y)**2)dataclassclass TankState:tank_id: str; capacity: float 1000.0; level: float 0.0temperature: float 25.0; pressure: float 101.3fluid_type: str unknown; quality_ok: bool Truepropertydef fill_ratio(self) - float:return self.level / self.capacity if self.capacity 0 else 0dataclassclass TaskSpec:task_id: str; task_type: TaskTypesource: str; destination: strpayload: 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 fTANK-{row}{j}self.locations[tid] Pose2D(20(j-1)*30, 30i*30)self.tanks[tid] TankState(tank_idtid, capacityrandom.choice([800,1000,1200]),levelrandom.uniform(50,300),temperaturerandom.uniform(20,35),fluid_typefluids[(i*3j-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 fCHARGE-{i1}self.locations[cs] Pose2D(5i*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,colorplt.cm.RdYlGn(t.fill_ratio),alpha0.6, ecblack, lw1.5))ax.text(p.x, p.y, f{tid}\n{t.fill_ratio*100:.0f}%,hacenter, vacenter, fontsize7, fontweightbold)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,s200,ccol,markers,ecblack,lw1.5,zorder5)for cs in self.charging_stations:p self.locations[cs]ax.scatter(p.x,p.y,s100,corange,marker^,ecblack,lw1)ax.set_xlabel(X (m)); ax.set_ylabel(Y (m))ax.set_title(Tank Farm Layout); ax.set_aspect(equal)ax.grid(True, alpha0.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_speed2.0, battery100.0):self.ridrid; self.rolerole; self.poseposeself.max_speedmax_speed; self.batterybatteryself.stateRobotState.IDLE; self.taskNoneself.queue: deque deque()self.work_t0.0; self.phase0self.total_dist0.0; self.done_count0self.fault_prob0.001def move_to(self, tgt: Pose2D, dt: float) - bool:d self.pose.distance_to(tgt); step self.max_speed*dtif d max(step,0.5):self.pose.x,self.pose.y tgt.x,tgt.yself.total_dist d; return Truer step/dself.pose.x (tgt.x-self.pose.x)*rself.pose.y (tgt.y-self.pose.y)*rself.total_dist step; return Falsedef consume_bat(self, dt: float, load1.0):self.battery max(0.0, self.battery - (0.010.005*load)*dt)def check_fault(self) - bool:if random.random() self.fault_prob:self.state RobotState.FAULTprint(f [FAULT] {self.rid} 故障); return Truereturn Falsedef assign(self, task: TaskSpec) - bool:if self.state RobotState.FAULT: return Falseself.queue.append(task); return Truedef update(self, dt: float, ctx: SimContext):if self.state RobotState.CHARGING:self.battery min(100.0, self.battery5.0*dt)if self.battery 90:self.state RobotState.IDLEprint(f [CHARGE] {self.rid} 充电完成)returnif self.battery 15 and self.state ! RobotState.CHARGING:self._go_charge(ctx); returnif self.check_fault(): returnif self.state RobotState.IDLE and self.queue:self.task self.queue.popleft()self.state RobotState.MOVINGself.work_t 0.0; self.phase 0if self.task and self.state ! RobotState.FAULT:if self.execute(dt, ctx):self.done_count 1ctx.scheduler.task_done(self.task)self.task Noneself.state RobotState.IDLEload 1.5 if self.state RobotState.WORKING else 1.0self.consume_bat(dt, load)def _go_charge(self, ctx: SimContext):nearestNone; mdfloat(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 csif nearest:self.task TaskSpec(fCHG-{self.rid},TaskType.CHARGE,self.rid,nearest)self.state RobotState.MOVING; self.phase0print(f [CHARGE] {self.rid} 低电量→{nearest})abstractmethoddef execute(self, dt: float, ctx: SimContext) - bool: passpropertyabstractmethoddef color(self) - str: pass# # 5. 三种角色机器人# class FillerRobot(Robot):进料装载区→储罐持续注入至95%def __init__(self, rid, pose): super().__init__(rid,RobotRole.FILLER,pose,1.8)propertydef color(self): return bluedef execute(self, dt, ctx):t self.taskif t.task_type TaskType.CHARGE:p ctx.map.get_pose(t.destination)if p and self.move_to(p, dt): self.state RobotState.CHARGINGreturn Falseif self.phase 0: # →装载区p ctx.map.get_pose(t.source)if p and self.move_to(p, dt):self.phase1; self.work_t0.0print(f [FILL] {self.rid}: 开始注入 → {t.destination})return Falseif self.phase 1: # →储罐p ctx.map.get_pose(t.destination)if p and self.move_to(p, dt):self.phase2; self.work_t0.0return False# 注入中self.state RobotState.WORKING; self.work_t dttank 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)*dtif tank.fill_ratio 0.95:print(f [FILL] {self.rid}: {t.destination} 已满 ({tank.level:.0f}L))return Trueif self.work_t 25.0:print(f [FILL] {self.rid}: 注入超时); return Truereturn Falseclass SamplerRobot(Robot):取样储罐→取样3s→送实验室def __init__(self, rid, pose): super().__init__(rid,RobotRole.SAMPLER,pose,2.5)propertydef color(self): return purpledef execute(self, dt, ctx):t self.taskif t.task_type TaskType.CHARGE:p ctx.map.get_pose(t.destination)if p and self.move_to(p, dt): self.state RobotState.CHARGINGreturn Falseif self.phase 0: # →储罐p ctx.map.get_pose(t.source)if p and self.move_to(p, dt):self.phase1; self.work_t0.0print(f [SAMP] {self.rid}: 在 {t.source} 取样中...)return Falseif self.phase 1: # 取样中self.state RobotState.WORKING; self.work_t dtif self.work_t 3.0:self.phase2; self.work_t0.0return Falseif 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.05ctx.stats[qc] 1if not tank.quality_ok: ctx.stats[qf] 1st 合格 if tank.quality_ok else 异常print(f [SAMP] {self.rid}: 样品→实验室 [{st}])return Truereturn Falseclass TransporterRobot(Robot):转运储罐→装载→出货区→卸料def __init__(self, rid, pose): super().__init__(rid,RobotRole.TRANSPORTER,pose,2.2)propertydef color(self): return darkgreendef execute(self, dt, ctx):t self.taskif t.task_type TaskType.CHARGE:p ctx.map.get_pose(t.destination)if p and self.move_to(p, dt): self.state RobotState.CHARGINGreturn Falseif self.phase 0: # →储罐p ctx.map.get_pose(t.source)if p and self.move_to(p, dt):self.phase1; self.work_t0.0return Falseif self.phase 1: # 装载self.state RobotState.WORKING; self.work_t dttank ctx.map.tanks.get(t.source)if tank and tank.level 50:load min(t.payload, tank.level, 500)tank.level - loadprint(f [TRANS] {self.rid}: 从 {t.source} 装 {load:.0f}L)else:print(f [TRANS] {self.rid}: {t.source} 液位不足)self.phase2; self.work_t0.0return Falseif self.phase 2: # →出货区p ctx.map.get_pose(t.destination)if p and self.move_to(p, dt):self.phase3; self.work_t0.0return Falseif self.phase 3: # 卸料self.state RobotState.WORKING; self.work_t dtif self.work_t 4.0:print(f [TRANS] {self.rid}: 卸料→{t.destination})return Truereturn 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 0for task in list(self.pending):rr role_map.get(task.task_type)if not rr: continuebestNone; mcfloat(inf)for r in robots.values():if r.role!rr or r.state in (RobotState.FAULT,RobotState.CHARGING):continueif len(r.queue)3: continuecost len(r.queue)*10 (50 if r.battery30 else 0)if cost mc: mccost; bestrif best and best.assign(task):self.pending.remove(task); assigned1return assigneddef stats(self) - Dict:return {pending:len(self.pending),done:len(self.completed)}# # 7. 仿真上下文# class SimContext:def __init__(self, m: TankFarmMap):self.mapm; self.robots: Dict[str,Robot]{}self.schedulerTaskScheduler()self.time0.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.0self._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 roleRobotRole.FILLER elseSamplerRobot if roleRobotRole.SAMPLER elseTransporterRobot)(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解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛