跳到主要内容

202605 RCS-Lite 挂纱机器人部署 — 类叉车套壳与机械臂透传

· 阅读需 11 分钟

挂纱机器人是"托盘叉车底盘 + 车载协作机械臂"的复合形态:底盘负责叉取和搬运装有纱管的容器,机械臂负责在挂纱工位完成抓纱、挂纱动作。RCS-Lite 侧没有这种整机的标准车型,落地方案是套壳 + 透传:底盘按"类叉车"客制车接入,复用标准叉车搬运业务;机械臂动作则通过任务模板中的"客制车透传"步骤下发到车端脚本,由脚本与机械臂进行 ModbusTCP 握手。

本文基于现场配置界面与 hfang_load_unload.pyhufang_arm.py 两个车端脚本,整理一套可复用的部署方案。

整机形态与业务流程

挂纱机器人整机

整机由两部分组成:

  • 类叉车底盘:带门架和叉齿的托盘叉车,负责容器的叉取、搬运和放回。
  • 车载机械臂:安装在车体上方的协作机械臂,负责在工位上执行抓纱/挂纱动作,机械臂控制器通过车内网络(ModbusTCP)与车体通信。

一次完整的挂纱任务流程为:

  1. 车辆到储位叉取装有纱管的容器(不放下,载着走)。
  2. 载着容器移动到挂纱工位。
  3. 透传任务通知机械臂"AGV 已到位",机械臂开始抓纱作业。
  4. 机械臂上报"抓料完成",握手结束,任务继续。
  5. 车辆把容器搬回储位。

对应 RCS-Lite 侧,就是一套"不放容器(叉车)→ 客制车透传 → 搬回容器(叉车)"的任务模板。

总体方案

整条链路可以概括为 4 步:

  1. 车端在 MapStudioPro 中配置为客制车"类叉车",绑定取放货脚本 hfang_load_unload(套壳:叉车的移动、抬/降叉走标准业务,动作由客制脚本执行)。
  2. 任务模板中插入"客制车透传"步骤,脚本名称指定 hufang_arm,参数里带上机械臂的 IP、寄存器地址等信息。
  3. 车端脚本 hufang_arm 收到透传参数后,作为 ModbusTCP 客户端连接机械臂,完成"到站"双向握手。
  4. 机械臂抓纱完成、寄存器清零后,脚本返回任务完成,调度继续走搬回容器流程。

RCS-Lite 侧配置

1. 服务能力集开启参数任务上报

在"运营管理 → 服务管理"中编辑机器人控制服务(RCS),进入"Rcs能力集 → 本地配置",开启"amr参数设置信任车上报"。

RCS能力集开启参数任务上报

该开关使平台信任小车上报的能力集参数,客制车型项目必须开启。现场联调时,如果车端脚本收不到透传参数,优先检查这里是否已启用。

2. 容器与货架类型开启盲举

本项目任务模板使用的是"不放容器""搬回容器"业务,上层也不下发容器编号,按《RCS-Lite叉取机器人业务部署操作手册V3.0》的要求,这种场景下容器类型、货架类型都需要把"是否盲举"设置为

RCS-Lite 容器类型配置(是否盲举)

手册中对盲举的说明:

是否盲举:根据业务需求,若现场有不放货架、搬回货架场景,上层又无法下发容器编号的,选择是,其他情况选中否。

容器尺寸配置时还要注意两点:

  1. 叉取机器人叉取货物时,叉齿平行边为容器深(例如托盘长 1200mm、宽 800mm,则深填 800、宽填 1200)。
  2. 空叉高度设置为空叉行走高度,满叉高度设置为满叉行走高度,且满叉高度需大于激光安装高度。

3. 任务模板:搬运加机械臂

新建任务模板"搬运加机械臂",流程编排如下:

顺序任务组/子任务内部步骤
1不放容器(叉车)移动 → 抬/降叉 → 移动
2客制车透传触发车端脚本 hufang_arm
3搬回容器(叉车)移动 → 抬/降叉

搬运加机械臂任务模板(有叉车等待点关闭)

模板开关项的要点:

  • 锁定标志:开启,保证任务执行期间资源锁定。
  • 是否记货架:搬回容器任务组开启,盲举场景下车辆需要自己记住容器/货架位置,搬回时才能放回原储位。
  • 有叉车等待点关闭。关闭后的实际执行流程是:车辆先移动到叉车等待点,在等待点把叉齿调整到取货高度,然后再进入储位取货。调叉动作发生在等待点而不是储位内,进出储位更干脆,也避免在储位里长时间调叉。

搬回容器任务组(是否记货架开启)

4. 客制车透传步骤配置

选中流程中的"客制车透传"步骤,右侧配置:

  • 透传类型:普通透传
  • 脚本名称hufang_arm
  • 参数配置:机械臂通信参数,JSON 格式如下

客制车透传步骤绑定 hufang_arm 脚本

{
"name": "hufang_arm",
"params": {
"info": [
{ "key": "timeout", "value": 0 },
{ "key": "robot_reg", "value": 12574 },
{ "key": "agv_reg", "value": 12572 },
{ "key": "task_type", "value": "arrive" },
{ "key": "unit_id", "value": 1 },
{ "key": "device_port", "value": 502 },
{ "key": "device_ip", "value": "192.168.2.20" }
]
}
}

各参数含义:

参数示例值说明
device_ip192.168.2.20机械臂控制器 IP(车内网络)
device_port502ModbusTCP 端口
unit_id1从站地址
task_typearrive任务类型:arrive(到站)/ cancel(取消)
agv_reg12572AGV 写入寄存器地址
robot_reg12574机械臂写入寄存器地址
timeout0握手等待超时(秒),0 表示无限等待

参数通过透传任务原样到达车端脚本,脚本内不写死任何 IP 和寄存器地址,后续换臂、换寄存器只需要改模板参数。

车端配置(MapStudioPro)

在 MapStudioPro 的"基础配置 → 客制车参数"(专家模式)中完成车型与脚本绑定:

MapStudioPro 客制车参数配置

配置项说明
自运行脚本开关开启允许车端运行客制脚本
客制车类型类叉车套壳为标准叉车业务
取放货脚本hfang_load_unload抬/降叉动作的执行脚本
第三方机构通信脚本hufang_arm透传任务触发的机械臂握手脚本
叉车最高举升高度250 mm与机构行程一致
叉车最低举升高度185 mm与脚本中的安全行走高度一致

车端脚本设计

取放货脚本 hfang_load_unload

该脚本承接标准叉车业务的"抬/降叉"步骤,通过 act_type 参数分发 4 种动作:

act_type动作流程
1取货叉子到目标高度 → 推出门架 → 抬升固定高度 → 收回门架 → 下降到安全位置
2放货叉子抬升到目标高度+固定高度 → 推出门架 → 下降固定高度 → 收回门架 → 下降到安全位置
3空车出储位叉齿下放到空车移动高度
4载货出储位叉齿下放到载货行走高度

核心代码(保留主干逻辑):

from hfang_fork import hfang_fork
from hfang_mast import hfang_mast

ACT_LOAD = 1 # 取货
ACT_UNLOAD = 2 # 放货
ACT_EMPTY_EXIT = 3 # 空车出储位
ACT_LOADED_EXIT = 4 # 载货出储位

SAFE_HEIGHT = 185 # 空车安全移动叉齿高度


class Business(TemplateBusiness):
def __init__(self, robot: SecDevInterface, params):
super().__init__()
self.fork = hfang_fork(robot)
self.mast = hfang_mast(robot)
self.robot_operator = ParamOperator(robot, __file__, params)

self.act_type = self.robot_operator.get_value_in_task_data("act_type")
self.lift_height = self.robot_operator.get_value_in_task_data("lift_height")
self.final_height = self.robot_operator.get_value_in_task_data("final_height")
self.pod_type = self.robot_operator.get_value_in_task_data("pod_type")

def _load_steps(self, robot, elevated_height):
return [
("叉子到目标高度", lambda: self.fork.lift(robot, self.final_height)),
("推出门架", lambda: self.mast.extend(robot)),
("抬升固定高度", lambda: self.fork.lift(robot, elevated_height)),
("收回门架", lambda: self.mast.retract(robot)),
("下降到安全位置", lambda: self.fork.lift(robot, SAFE_HEIGHT)),
]

def execute(self, robot: SecDevInterface, args):
elevated_height = self.final_height + self.lift_height
if self.act_type == ACT_LOAD:
steps = self._load_steps(robot, elevated_height)
# act_type 2/3/4 同理分发 ...

for i, (step_name, step_func) in enumerate(steps):
try:
robot.log(f"步骤 {i+1}: {step_name}")
step_func()
except Exception as e:
self.fork.set_module_alarm(robot, 5, 1, f"第{i+1}步({step_name})执行失败")
return BusinessStatus.FAILED

# 取货/放货完成后更新动作状态,供调度判断车辆载货情况
if self.act_type == ACT_LOAD:
Status(robot, "").set_action_status({"pod_type": self.pod_type, "lift_status": 1})
elif self.act_type == ACT_UNLOAD:
Status(robot, "").set_action_status({"pod_type": 0, "lift_status": 0})

return BusinessStatus.FINISHED

三个值得注意的设计点:

  1. 步骤表驱动:每种动作是一个"(名称, 函数)"列表,逐步执行、逐步打日志,失败时能直接定位到第几步、哪个动作。
  2. 安全高度兜底:取放货完成后统一回到 SAFE_HEIGHT = 185,与 MapStudioPro 里的"叉车最低举升高度"保持一致,保证行走姿态安全。
  3. 动作状态回写:取货/放货完成后通过 set_action_status 更新 pod_typelift_status,调度侧据此判断车辆当前是否载货。

机械臂通信脚本 hufang_arm

透传任务触发该脚本后,车端作为 ModbusTCP 客户端连接机械臂(机械臂作为从站),通过两个寄存器做双向双握手:

寄存器写入方说明
agv_reg(12572)AGVAGV 状态:到位/确认/取消
robot_reg(12574)机械臂机械臂状态:到位确认/抓料完成

寄存器值约定:

含义
0通信完成/清零
1AGV 到位
2机器人抓料完成
3任务取消

到站(arrive)握手时序为:

  1. AGV 写 agv_reg = 1(到位)。
  2. 等机械臂写 robot_reg = 1(确认到位),AGV 清零 agv_reg
  3. 机械臂执行抓纱,完成后写 robot_reg = 2,AGV 立即写 agv_reg = 2 确认。
  4. 等机械臂清零 robot_reg,AGV 清零 agv_reg,握手结束,返回任务完成。

核心代码(保留主干逻辑):

from pymodbus.client import ModbusTcpClient
from pymodbus.exceptions import ModbusException

VAL_CLEAR = 0 # 通信完成/清零
VAL_ARRIVE = 1 # AGV 到位
VAL_GRAB_DONE = 2 # 机器人抓料完成
VAL_CANCEL = 3 # 任务取消


class Business(TemplateBusiness):
"""通过 ModbusTCP 与车上机械臂进行双向双握手通信。"""

def _do_arrive(self, robot, client, agv_reg, robot_reg, unit_id, timeout):
# 1. AGV 到位
self._write_reg(robot, client, agv_reg, VAL_ARRIVE, unit_id)
# 2. 等机械臂确认到位,之后清零
if not self._wait_reg(robot, client, robot_reg, VAL_ARRIVE, unit_id, timeout):
return BusinessStatus.FAILED
self._write_reg(robot, client, agv_reg, VAL_CLEAR, unit_id)
# 3. 等机械臂抓料完成,立即确认
if not self._wait_reg(robot, client, robot_reg, VAL_GRAB_DONE, unit_id, timeout):
return BusinessStatus.FAILED
self._write_reg(robot, client, agv_reg, VAL_GRAB_DONE, unit_id)
# 4. 等机械臂清零,之后 AGV 清零
if not self._wait_reg(robot, client, robot_reg, VAL_CLEAR, unit_id, timeout):
return BusinessStatus.FAILED
self._write_reg(robot, client, agv_reg, VAL_CLEAR, unit_id)
return BusinessStatus.FINISHED

def execute(self, robot: SecDevInterface, args):
client = None
try:
task_params = args if args is not None else self.params
robot_operator = ParamOperator(robot, __file__, task_params)

device_ip = str(robot_operator.get_value_in_task_data("device_ip")).strip()
device_port = int(robot_operator.get_value_in_task_data("device_port"))
unit_id = int(robot_operator.get_value_in_task_data("unit_id"))
task_type = str(robot_operator.get_value_in_task_data("task_type")).strip()
agv_reg = int(robot_operator.get_value_in_task_data("agv_reg"))
robot_reg = int(robot_operator.get_value_in_task_data("robot_reg"))
timeout = int(robot_operator.get_value_in_task_data("timeout"))

client = ModbusTcpClient(host=device_ip, port=device_port, timeout=3)
if not client.connect():
return BusinessStatus.FAILED

if task_type == "arrive":
return self._do_arrive(robot, client, agv_reg, robot_reg, unit_id, timeout)
elif task_type == "cancel":
return self._do_cancel(robot, client, agv_reg, robot_reg, unit_id, timeout)
return BusinessStatus.FAILED
except ModbusException as exc:
robot.log(f"发生 Modbus 协议异常: {exc}")
return BusinessStatus.FAILED
finally:
if client is not None:
client.close()

几个设计要点:

  1. 双向双握手:每一步都要求对方确认后再清零,AGV 和机械臂任何一侧异常都不会造成"单边继续",联调时通过寄存器当前值就能判断卡在哪一步。
  2. 参数全部来自透传:IP、端口、寄存器地址、任务类型都从透传参数解析,脚本对现场零硬编码。
  3. 取消流程独立task_type = cancel 时走独立的取消握手(写 3、等确认、清零),任务被取消时机械臂能安全退出,不会停在作业中间态。
  4. 超时可配timeout = 0 表示无限等待,适合机械臂作业时长不固定的挂纱场景;如需兜底,可在模板参数里直接改超时时间。