Adapter 编程指南

本指南面向需要将机器人、PLC 或其他外部设备接入 RVS 2.0 视觉系统的集成开发者，介绍如何使用 Adapter 生成器创建适配程序，以及如何基于 Python Adapter 框架开发自定义通信协议。

Adapter 位于外部设备与视觉系统之间，负责接收外部指令、解析协议、调用视觉系统能力，并按照外部设备要求的格式返回状态码、位姿及其他结果。

备注

本指南描述当前仓库中的 Adapter 实现。示例中的 IP 地址、端口、工程编号和协议字段仅用于说明，现场使用时应替换为实际配置。

阅读前提

阅读本文前，建议具备以下基础知识：

• Python 3.10 或以上版本的基础语法

• TCP/IP 客户端与服务端的基本概念

• ASCII 文本协议或二进制协议的基本知识

• JSON 数据格式

• RVS 2.0 工程、工作流、步骤、配方和视觉结果的基本概念

备注

如果只需要配置常规的拍照触发、配方切换、机器人位姿传入和视觉结果返回，可直接使用 Adapter 生成器，无需手写 Python 代码。

Adapter 的工作方式

一次完整的通信按以下顺序执行：

1. Adapter 以 TCP 服务端或 TCP 客户端方式建立连接

2. Adapter 接收一帧 ASCII 文本或二进制数据

3. Header Parser 解析帧头和命令码，并选择对应的 Payload Parser

4. Payload Parser 按配置读取工程编号、配方编号、机器人位姿或步骤参数

5. Adapter 将解析结果转换为视觉系统操作，并通过执行队列依次执行

6. Response Assembler 将执行结果转换为外部状态码、位姿、标签和自定义输出

7. Adapter 将响应帧发送给外部设备

机器人 / PLC / 上位机
        │  TCP 请求
        ▼
通信层（TCP Server / TCP Client）
        │
        ▼
Header Parser ──识别命令码──► Payload Parser
                                      │
                                      ▼
                              Operation / 执行队列
                                      │
                                      ▼
                                RVS 2.0 视觉系统
                                      │
                                      ▼
                              Response Assembler
        ◄────────────── TCP 响应 ──────┘

选择开发方式

Adapter 提供两种开发方式。

• 使用 Adapter 生成器

  适用于以下场景：

  • 使用 TCP/IP 通信

  • 协议为 ASCII 文本或固定字段的二进制数据

  • 外部指令包含命令码、工程编号、配方编号、机器人位姿或视觉步骤参数

  • 返回内容为状态码、视觉位姿、机器人路径、标签或生成器支持的其他输出

  生成器保存配置时会生成协议解析、操作调用、响应组装和状态码映射代码。生成代码位于 Adapter 配置目录的 generated/ 子目录中。

  参见

  Adapter 生成器提供图形化配置界面，可快速完成 Adapter 程序配置与生成。详细使用说明请参阅 Adapter 生成器指南。

• 自定义 Python Adapter

  以下情况建议进行自定义开发：

  • 帧头、校验、转义或粘包规则较为复杂

  • 单个请求需要执行特殊的业务流程

  • 需要支持生成器未覆盖的输入或输出数据类型

  • 需要定制响应帧结构或错误处理方式

  • 需要开发一套可复用的设备协议包

自定义 Python Adapter

安装开发包

进入 Adapter Python 工程目录并以可编辑模式安装：

cd source\tools\communication\python\adapter
python -m pip install -e .

当前包要求 Python 3.10 或以上版本。

协议包结构

一个自定义协议包至少包含 __init__.py，并公开 config_class 和 register(adapter)：

my_robot_protocol/
├── __init__.py
├── config.py
├── header_parser.py
├── payload_parser.py
└── assembler.py

__init__.py 示例：

from .config import MyRobotConfig
from .header_parser import MyHeaderParser
from .payload_parser import build_trigger_parser
from .assembler import MyAssembler

config_class = MyRobotConfig

def register(adapter) -> None:
    header = MyHeaderParser(
        mode=adapter.mode,
        byte_order=adapter.byte_order,
    )
    header.bind("P", build_trigger_parser(), MyAssembler())
    adapter.add_header_parser(header)

定义配置类

自定义配置类继承 ProtocolConfig：

from dataclasses import dataclass

from adapter import ProtocolConfig

@dataclass
class MyRobotConfig(ProtocolConfig):
    host: str = "0.0.0.0"
    port: int = 50000
    mode: str = "ascii"
    delimiter: str = ","
    response_terminator: str = "\r"

公共配置字段如下。

字段	默认值	说明
transport	tcp	传输层类型；当前实现使用 TCP。
service_type	server	server 或 client。
host	0.0.0.0	服务端监听地址或客户端目标地址。
port	50000	TCP 端口。
mode	ascii	ascii 或 hex。
byte_order	<	< 为小端，> 为大端。
delimiter	,	ASCII 字段分隔符。
reconnect_delay	5.0	客户端重连间隔，单位为秒。
recv_size	4096	单次接收缓冲区大小。
receive_timeout	None	接收超时，单位为秒。

未在配置类中声明的字段会存入配置对象的 extra 字典，可用于保存协议私有参数。

实现 Header Parser

Header Parser 的职责是验证帧头、提取命令码、确定 payload 范围，并返回该命令绑定的 Payload Parser 和 Response Assembler。

from adapter import HeaderParser

class MyHeaderParser(HeaderParser):
    def __init__(self, delimiter=",", **_):
        self.delimiter = delimiter
        self._bindings = {}

    def bind(self, command_code, parser, assembler):
        self._bindings[command_code] = (parser, assembler)

    def parse(self, raw: bytes) -> dict:
        text = raw.decode("utf-8").rstrip("\r\n")
        command_code, separator, _ = text.partition(self.delimiter)
        if command_code not in self._bindings:
            raise ValueError(f"Unknown command code: {command_code!r}")

        parser, assembler = self._bindings[command_code]
        header_len = len((command_code + separator).encode("utf-8"))
        return {
            "payload_parser": parser,
            "assembler": assembler,
            "header_len": header_len,
            "payload_len": len(raw) - header_len,
            "command_code": command_code,
        }

实现时应注意：

• header_len 和 payload_len 按字节计算，而不是按 Python 字符数计算

• 未识别的命令码应明确报错

• 不要在 Header Parser 中执行业务操作

• 二进制协议必须统一端序、字段宽度和有无符号约定

实现 Payload Parser

CompositePayloadParser 可以组合多个字段解析器。文本模式下按分隔符读取，二进制模式下按固定宽度读取。

from adapter import CompositePayloadParser, Float32ArrayField, UInt32Field

async def parse_trigger(payload, metadata, exec_ctx):
    project_id = exec_ctx.parsed["project_id"]
    joint_angles = exec_ctx.parsed.get("joint_angles")

    from adapter.operations import get_workflow_id, set_robot_pose
    from adapter.operations import execute_workflow, get_vision_result

    workflow = await get_workflow_id.submit(
        exec_ctx, params={"project_id": project_id}
    )
    if int(workflow.status_code) != 0:
        return workflow

    flow_id = workflow.data["workflow_id"]
    if joint_angles:
        await set_robot_pose.submit(
            exec_ctx,
            params={
                "project_id": project_id,
                "joint_angles": joint_angles,
            },
        )

    await execute_workflow.submit(exec_ctx, params={"flow_id": flow_id})
    return await get_vision_result.submit(
        exec_ctx,
        params={"flow_id": flow_id, "timeout": 30000},
    )

def build_trigger_parser():
    return CompositePayloadParser(
        "trigger",
        async_parse_fn=parse_trigger,
        parsers=[
            UInt32Field("project_id"),
            Float32ArrayField("joint_angles", 6),
        ],
        mode="text",
        delimiter=",",
    )

常用字段解析器包括：

数据类型	标量	数组
有符号整数	Int8Field、Int16Field、Int32Field	对应的 *ArrayField
无符号整数	UInt8Field、UInt16Field、UInt32Field	对应的 *ArrayField
浮点数	Float32Field、Float64Field	对应的 *ArrayField
布尔值	BoolField	BoolArrayField
字符串	StrField	按协议自行组合

还可以使用 ConditionalFieldParser 实现条件字段，或继承 FieldParser 实现自定义数据类型。

调用视觉系统操作

Adapter 内置操作如下。

操作	典型用途
get_devices	获取设备信息。
get_project_id	根据工程信息获取工程编号。
get_workflow_id	根据工程编号获取工作流 ID。
get_display_id	获取显示对象 ID。
get_solution_info	获取当前解决方案信息。
get_step_info	获取步骤信息。
get_step_property	读取步骤属性。
update_step_property	更新步骤属性。
switch_recipe	切换工程配方。
set_robot_pose	写入机器人关节角和可选法兰位姿。
execute_workflow	执行指定工作流。
get_vision_result	等待并获取视觉结果。

在异步 Payload Parser 中使用 operation.submit(exec_ctx, params=...) 调用操作。后续操作依赖前一步结果时，应检查 status_code 和 data 后再继续。

推荐的拍照流程为：

1. 通过工程编号获取工作流 ID

2. 根据需要更新步骤属性

3. 根据需要切换配方

4. 根据需要设置机器人位姿

5. 执行工作流

6. 获取视觉结果

7. 返回最终执行结果给 Assembler

实现 Response Assembler

Response Assembler 将 ExecutionResult 转换为协议响应。以下示例只返回状态码和结束符：

from adapter import ResponseAssembler

class MyAssembler(ResponseAssembler):
    STATUS_MAP = {
        0: 1,
    }

    def assemble(self, result) -> bytes:
        internal_code = int(result.status_code)
        external_code = self.STATUS_MAP.get(internal_code, 9)
        return f"{external_code}\r".encode("utf-8")

    def assemble_multi(self, result) -> bytes:
        return self.assemble(result)

实际项目通常还需要：

• 优先检查子操作中的第一个失败状态

• 根据配置进行单位转换和姿态转换

• 返回位姿数量、位姿、标签和自定义输出

• 对缺失结果、超时和异常返回明确状态码

• 在二进制模式下使用一致的端序打包数据

启动自定义协议包

使用 ProtocolAdapter.start_protocol() 加载协议包：

import asyncio

from adapter import ProtocolAdapter

async def main():
    adapter = await ProtocolAdapter.start_protocol(
        "my_robot_protocol",
        config={
            "service_type": "server",
            "host": "0.0.0.0",
            "port": 50000,
            "mode": "ascii",
            "delimiter": ",",
        },
        start=False,
    )

    await adapter.start()
    try:
        await asyncio.Event().wait()
    finally:
        await adapter.stop()

asyncio.run(main())

config 可以是：

• None：使用协议配置类默认值

• dict：覆盖指定配置项

• 协议对应的 ProtocolConfig 实例

Adapter 开发接口参考

本指南只说明当前代码中可用于 Adapter 开发的接口。常规项目使用生成器即可；只有开发自定义协议、扩展生成模板或排查底层问题时，才需要直接调用这些接口。

接口分层

层级	主要接口	职责
Adapter 生命周期	ProtocolAdapter、ProtocolConfig	创建通信对象、加载协议、启动和停止服务。
通信层	TCPServer、TCPClient	建立 TCP 连接并收发数据。
帧头解析	HeaderParser	识别协议和命令，确定 payload 范围。
字段解析	PayloadParser、CompositePayloadParser、FieldParser	将 payload 转换成命名参数。
业务执行	Operation、ExecutionContext、CommandRequest	将解析结果提交给视觉系统执行。
执行结果	ExecutionResult、VisionEngineSubErrorCode	统一表示成功、失败和业务数据。
响应组装	ResponseAssembler、WriteContext	将执行结果编码为外部设备响应。

典型调用关系如下：

ProtocolAdapter
 ├─ TCPServer / TCPClient
 ├─ HeaderParser.parse(raw)
 │    └─ 返回 PayloadParser + ResponseAssembler + header_len
 ├─ PayloadParser.parse(payload, metadata)
 │    └─ ExecutionContext.run(CommandRequest)
 │         └─ Operation handler
 └─ ResponseAssembler.assemble(ExecutionResult)

ProtocolAdapter

定义：

@dataclass
class ProtocolAdapter:
    transport: str = "tcp"
    service_type: str = "server"
    host: str = "0.0.0.0"
    port: int = 8888
    mode: str = "ascii"
    byte_order_init: str = "<"
    reconnect_delay: float = 5.0
    recv_size: int = 4096
    receive_timeout: float | None = None
    skip_default_protocols: bool = False
    communication: TCPServer | TCPClient | None = None

构造参数：

参数	类型	说明
transport	str	传输层标识，当前使用 tcp。
service_type	str	server 创建 TCP 服务端；client 创建 TCP 客户端。
host	str	服务端监听地址或客户端目标地址。
port	int	TCP 端口。
mode	str	ascii 或 hex。
byte_order_init	str	二进制字段端序：< 小端，> 大端。非法值抛出 ValueError。
reconnect_delay	float	客户端模式断线后的重试间隔。
recv_size	int	客户端单次读取的最大字节数。
receive_timeout	`float	None`
skip_default_protocols	bool	False 时自动注册内置的默认（vanilla）协议；自定义协议包通常设为 True。
communication	`TCPServer	TCPClient

公共方法：

await adapter.start() -> None

启动当前绑定的通信对象和执行器。服务端模式下开始监听；客户端模式下连接目标服务端并进入接收循环。

adapter = ProtocolAdapter(host="0.0.0.0", port=50000)
await adapter.start()

启动失败时，底层网络异常会传播给调用方，例如端口被占用时会抛出 OSError。

await adapter.stop() -> None

停止接收、关闭客户端连接和监听套接字，并停止执行器。应用退出时应放在 finally 中调用。

try:
    await adapter.start()
finally:
    await adapter.stop()

close() 当前等价于 stop()。

adapter.add_header_parser(parser) -> None

注册一套协议的 Header Parser。多个 Parser 按注册顺序尝试，首个成功返回解析元数据的 Parser 生效。

header = MyHeaderParser()
adapter.add_header_parser(header)

adapter.bind_communication(communication) -> None

替换 Adapter 使用的通信对象。参数必须是 TCPServer 或 TCPClient，否则抛出 TypeError。

server = TCPServer("0.0.0.0", 50000, "ascii")
adapter.bind_communication(server)

await ProtocolAdapter.start_protocol(package_path, config=None, *, start=True)

加载自定义协议包。协议包的 __init__.py 必须公开：

config_class = MyProtocolConfig

def register(adapter) -> None:
    ...

参数和异常：

项目	说明
package_path	包含 __init__.py 的协议目录。
config	None、配置字典或 ProtocolConfig 实例。
start	为 False 时只完成创建和注册，不启动网络。
返回值	已创建的 ProtocolAdapter。
FileNotFoundError	目录不存在或缺少 __init__.py。
AttributeError	协议包未公开 config_class。
TypeError	config 类型错误。

客户端专用方法：

方法	说明
init_client((host, port))	创建并绑定 TCP 客户端。
set_bind_port(is_bind=True)	设置客户端连接前是否绑定本地端口。仅客户端可用。
set_recv_size(size)	修改客户端读取缓冲区大小。
await send(msg)	向上游服务端发送数据。仅客户端可用。
await recv()	从上游服务端读取一次数据。仅客户端可用。
await reconnect_server()	尝试重新连接上游服务端。

在服务端模式调用客户端专用方法会抛出 RuntimeError。

ProtocolConfig

ProtocolConfig 是协议包的公共配置基类：

@dataclass
class ProtocolConfig:
    transport: str = "tcp"
    service_type: str = "server"
    host: str = "0.0.0.0"
    port: int = 50000
    mode: str = "ascii"
    byte_order: str = "<"
    delimiter: str = ","
    reconnect_delay: float = 5.0
    recv_size: int = 4096
    receive_timeout: float | None = None
    extra: dict[str, Any] = field(default_factory=dict)

通过 from_dict() 创建配置时，已声明字段直接赋值，未声明字段进入 extra：

config = MyRobotConfig.from_dict({
    "port": 50000,
    "checksum": "crc16",
})

assert config.port == 50000
assert config.extra["checksum"] == "crc16"

协议私有且确实需要类型提示的配置，应在子类中声明；只需透传的少量扩展值可以放在 extra 中。

TCPServer

构造函数：

TCPServer(
    host: str = "0.0.0.0",
    port: int = 8888,
    mode: str = "ascii",
)

mode 只能是 ascii 或 hex：

• ascii：收到字节后按 UTF-8 解码，回调参数为 str

• hex：收到 ASCII 十六进制文本后转换成 bytes，回调参数为 bytes

注册回调：

server = TCPServer("0.0.0.0", 50000, "ascii")

@server.on_data
async def on_data(data, addr, writer):
    print(addr, data)
    writer.write(b"1\r")
    await writer.drain()

await server.start()

回调签名：

(data: str | bytes,
 addr: tuple[str, int],
 writer: asyncio.StreamWriter) -> None | Awaitable

公共方法：

方法	返回值	说明
on_data(callback)	原回调	注册同步或异步数据回调，也可作为装饰器。
await start()	None	开始监听。
await stop()	None	关闭全部连接并停止监听。

当前限制：

• 服务端每次使用 reader.read(4096)

• 没有根据结束符、长度字段或固定帧长自动分帧

• ASCII 解码失败的输入会被丢弃，并记录警告日志

• hex 文本包含非法字符时会被丢弃，并记录警告日志

TCPClient

构造函数：

TCPClient(
    host: str,
    port: int,
    mode: str = "ascii",
    *,
    recv_size: int = 4096,
    timeout: float | None = None,
    reconnect_max_retries: int = 10,
    reconnect_base_delay: float = 1.0,
    reconnect_max_delay: float = 60.0,
    on_reconnect: Callable[[], None] | None = None,
)

主要方法：

方法	返回值	说明
await connect()	None	建立一次连接。失败时抛出网络异常。
await reconnect_server()	bool	按指数退避重连；成功返回 True。
is_connected()	bool	当前 writer 存在且未关闭时返回 True。
await send(msg)	None	发送 str 或 bytes。未连接时抛出 ConnectionError。
await recv()	`str	bytes
set_timeout(timeout)	None	修改读取超时，非正数抛出 ValueError。
set_recv_size(size)	None	修改读取大小，非正数抛出 ValueError。
set_bind_port(...)	None	配置客户端本地绑定地址和端口。
await close()	None	关闭连接。

reconnect_server() 的等待时间从 reconnect_base_delay 开始倍增，最大不超过 reconnect_max_delay。reconnect_max_retries=0 表示不限次数。

HeaderParser

自定义 Header Parser 必须实现：

class HeaderParser(ABC):
    default_failure_assembler = None

    @abstractmethod
    def parse(self, raw: bytes) -> dict:
        ...

parse() 成功时至少返回 payload_parser。常用返回字段：

键	类型	必需	说明
payload_parser	PayloadParser	是	当前命令使用的 payload 解析器。
assembler	ResponseAssembler	否	当前命令使用的响应组装器。
header_len	int	否	payload 起点的字节偏移，默认 0。
payload_len	int	否	payload 长度；需要排除帧后缀时使用。
command_code	str	否	命令码，供日志或下游解析器使用。
自定义键	任意	否	会继续作为 metadata 传给 Payload Parser。

匹配规则：

• 当前帧属于本协议时返回字典

• 当前帧不属于本协议时抛出 ValueError

• 多个 Header Parser 中首个成功者生效

• 如果全部不匹配，调度层尝试使用 default_failure_assembler 组装错误响应

最小实现：

class CommandHeader(HeaderParser):
    def __init__(self):
        self.bindings = {}

    def bind(self, code, parser, assembler):
        self.bindings[code] = parser, assembler

    def parse(self, raw: bytes) -> dict:
        code, separator, _ = raw.partition(b",")
        key = code.decode("ascii")
        if key not in self.bindings:
            raise ValueError(f"unknown command: {key}")

        parser, assembler = self.bindings[key]
        return {
            "payload_parser": parser,
            "assembler": assembler,
            "header_len": len(code) + len(separator),
            "command_code": key,
        }

PayloadParser 与 CompositePayloadParser

底层接口：

class PayloadParser(ABC):
    @abstractmethod
    def parse(
        self,
        payload: bytes,
        metadata: dict,
    ) -> CommandRequest | list[CommandRequest]:
        ...

推荐使用 CompositePayloadParser，先通过字段解析器生成命名参数，再进入异步业务函数：

CompositePayloadParser(
    name: str,
    async_parse_fn: Callable,
    *,
    parsers: list[FieldParser] | None = None,
    mode: str = "text",
    delimiter: str = ",",
)

异步函数签名：

async def parse_command(
    payload: bytes,
    metadata: dict,
    exec_ctx: ExecutionContext,
):
    ...

字段解析完成后：

• exec_ctx.parsed 保存解析结果

• 每个已解析字段也写入 metadata，供后续条件解析器使用

• payload 中仍存在未消费的非空字段或字节时，会抛出 ValueError

• 字段不足时，由底层的数据流读取函数抛出异常

示例：

async def run(payload, metadata, exec_ctx):
    params = exec_ctx.parsed
    result = await get_workflow_id.submit(
        exec_ctx,
        params={"project_id": params["project_id"]},
    )
    return result

parser = CompositePayloadParser(
    "run",
    async_parse_fn=run,
    parsers=[
        UInt32Field("project_id"),
        Float32ArrayField("joint_angles", 6),
    ],
    mode="text",
    delimiter=",",
)

字段解析接口

标量字段构造方式统一为：

FieldType(field_name: str)

数组字段构造方式统一为：

ArrayFieldType(field_name: str, count: int)

Python 类型	标量字段	数组字段	二进制宽度
int	Int8Field	Int8ArrayField	1 字节/值
int	UInt8Field	UInt8ArrayField	1 字节/值
int	Int16Field	Int16ArrayField	2 字节/值
int	UInt16Field	UInt16ArrayField	2 字节/值
int	Int32Field	Int32ArrayField	4 字节/值
int	UInt32Field	UInt32ArrayField	4 字节/值
bool	BoolField	BoolArrayField	按底层 bool 格式
float	Float32Field	Float32ArrayField	4 字节/值
float	Float64Field	Float64ArrayField	8 字节/值
str	StrField	无内置字符串数组字段	文本模式使用

其他字段接口：

Gap(count=1)

跳过保留字段而不写入结果。文本模式跳过 count 个 token；二进制模式当前每项按一个 uint32 跳过。

parsers = [
    UInt32Field("project_id"),
    Gap(1),
    UInt32Field("recipe_id"),
]

LeafFieldParser(field_name, reader_fn)

注入简单的自定义读取函数：

temperature = LeafFieldParser(
    "temperature",
    lambda ctx: ctx.read_int16() / 10.0,
)

ConditionalFieldParser(predicate, true_parser, false_parser=None)

根据已解析 metadata 决定是否读取字段：

pose = ConditionalFieldParser(
    lambda meta: meta.get("has_pose") == 1,
    Float32ArrayField("pose", 6),
)

条件判断所依赖的字段必须位于条件字段之前。

ExecutionContext

核心定义：

class ExecutionContext:
    results: list[ExecutionResult]
    parsed: dict | None

    async def run(self, cmd: CommandRequest) -> ExecutionResult:
        ...

run() 会：

1. 将命令交给执行器

2. 等待单条命令结果

3. 将命令名写入 result.data["_cmd"]

4. 将结果追加到 results

5. 返回该结果供下一步使用

执行队列 30 秒内没有返回结果时，会创建 VISIONFLOW_OUTPUT_TIMEOUT 结果。执行器抛出异常时，会转换为 SERVICE_HANDLER_EXCEPTION。

调用方应始终先检查状态：

result = await get_workflow_id.submit(
    exec_ctx,
    params={"project_id": project_id},
)

if int(result.status_code) != 0:
    return result
if not result.data or "workflow_id" not in result.data:
    return result

flow_id = result.data["workflow_id"]

CommandRequest 与 ExecutionResult

命令定义：

@dataclass
class CommandRequest:
    msg_type: str
    params: dict
    raw: bytes = b""
    executor: Callable | None = None

字段	说明
msg_type	操作名称，也是执行分发键。
params	传给操作的 Python 参数字典。
raw	可选的原始 payload，用于日志和追踪。
executor	可选直接执行函数；通常由 Operation 自动生成。

结果定义：

@dataclass
class ExecutionResult:
    status_code: int
    data: dict | None = None
    error_message: str | None = None
    duration_ms: float = 0.0
    sub_results: list[ExecutionResult] = field(default_factory=list)

字段	说明
status_code	内部状态码；0 表示成功。
data	操作返回的业务数据。
error_message	适合日志和诊断的人类可读错误信息。
duration_ms	单次操作执行耗时。
sub_results	一个请求执行多条操作时的原始结果列表。

Assembler 不应只检查顶层状态。多操作流程应检查 sub_results，并将第一个失败的结果映射为外部状态码。

Operation

定义：

Operation(
    name: str,
    handler: Callable[[Any, dict], ExecutionResult],
    /,
    **param_names: str,
)

常用方法：

方法	说明
await operation.submit(exec_ctx, params)	提交并等待结果。后续步骤依赖该结果时使用。
operation.submit_async(exec_ctx, params)	创建异步任务，不等待结果。仅用于确实没有顺序依赖的操作。
operation.params.<name>	获取参数键名，避免在多个 Parser 中重复硬编码字符串。

result = await execute_workflow.submit(
    exec_ctx,
    params={execute_workflow.params.flow_id: flow_id},
)

当前内置操作接口：

Operation	参数	成功时主要数据
get_workflow_id	project_id: int	{"workflow_id": int}
set_robot_pose	project_id: int、joint_angles: list、可选 flange_pose	无业务数据
switch_recipe	project_id: int、recipe_id: int	{"recipe_id": int}
execute_workflow	flow_id: int	{"flow_id": int, "status": int}
get_vision_result	flow_id: int、timeout: int（毫秒）	视觉输出字典
get_step_property	flow_id: int、step_id: int、property_path: str	属性数据字典
update_step_property	flow_id: int、step_id: int、property_path: str、json_value: str	无业务数据或底层错误数据
get_solution_info	无	当前解决方案信息
get_project_id	display_project_id: int	工程 ID 数据；非字典结果包装为 {"project_id": value}。
get_step_info	flow_id: int、step_id: int	步骤信息；非字典结果包装为 {"value": value}。
get_devices	无	工程设备信息；非字典结果包装为 {"devices": value}。
get_display_id	project_id: int	显示工程 ID；非字典结果包装为 {"display_project_id": value}。

set_robot_pose 参数形状

仅关节角：

{
    "project_id": 1,
    "joint_angles": [0, 10, 20, 30, 40, 50],
}

关节角和法兰位姿：

{
    "project_id": 1,
    "joint_angles": [0, 10, 20, 30, 40, 50],
    "flange_pose": {
        "position": [0.100, 0.200, 0.300],
        "rotation": [0.0, 0.0, 0.0, 1.0],
    },
}

Operation 接收的位姿应已经按内部 API 要求完成单位和旋转表达转换。生成器会调用转换工具完成该步骤；手写协议需要自行转换。

update_step_property 参数形状

{
    "flow_id": 10,
    "step_id": 24,
    "property_path": "matcher/minScore",
    "json_value": '{"type":"double","value":0.85}',
}

json_value 是字符串，不是 Python 字典。属性路径、类型或值无效时，底层子错误码会被保留到 ExecutionResult.status_code。

ResponseAssembler

自定义响应组装器至少实现：

class ResponseAssembler(ABC):
    @abstractmethod
    def assemble(self, result: ExecutionResult) -> bytes | None:
        ...

    def assemble_multi(self, result: ExecutionResult) -> bytes | None:
        return self.assemble(result)

返回值：

• 返回 bytes：发送该响应

• 返回 None：静默处理，不发送响应

• assemble_multi() 默认调用 assemble()，需要单独处理 sub_results 时重写

推荐结构：

class RobotAssembler(ResponseAssembler):
    STATUS_MAP = {
        0: 1,
        2005: 2,
        2103: 3,
        2306: 5,
        2307: 8,
    }

    def assemble(self, result):
        failed = next(
            (item for item in result.sub_results if int(item.status_code) != 0),
            None,
        )
        effective = failed or result
        status = self.STATUS_MAP.get(int(effective.status_code), 9)

        if status != 1:
            return f"{status}\r".encode("ascii")

        return b"1\r"

MapBasedAssembler 提供一个简单的“内部码到外部码”字典实现，适合原型和简单逗号协议。正式项目通常需要继承 ResponseAssembler，明确控制位姿、标签、前后缀和失败响应格式。

WriteContext

WriteContext 用于按文本或二进制模式写入字段：

text = WriteContext.text(delimiter=",", decimal_place=3)
text.write_uint32(1)
text.write_float64(12.34567)
text.write_str("box_A")
response = text.to_bytes()

文本结果为：

1,12.346,box_A

二进制模式：

binary = WriteContext.binary(byte_order="<")
binary.write_uint32(1)
binary.write_float32(12.5)
response = binary.to_bytes()

可用写入方法：

方法	说明
write_int8 / write_uint8	写入 8 位整数。
write_int16 / write_uint16	写入 16 位整数。
write_int32 / write_uint32	写入 32 位整数。
write_float32 / write_float64	写入浮点数。
write_str	写入字符串；二进制模式下写 UTF-8 字节。
write_bytes	写入原始字节。
to_bytes	返回最终字节串。

WriteContext 不会自动添加业务协议要求的 \r、\n、CRC 或长度字段，这些内容由 Assembler 显式添加。

内部错误码

所有 Operation 先返回统一内部错误码，再由 Assembler 映射成现场协议状态码。

范围	类别	示例
0	成功	NONE
2000～2099	公共执行错误	参数非法、IPC 不可用、许可证无效、服务异常
2100～2199	解决方案和工程	工程不存在、工程没有工作流
2200～2299	步骤属性	工作流或步骤不存在、属性路径无效、反序列化失败
2300～2399	工作流和视觉结果	工作流执行失败、结果未就绪、结果超时、无输出
2400～2499	配方	配方管理器不可用、配方不存在、列表为空
2500～2549	机器人位姿	关节数错误、四元数无效、位姿格式无效
2550～2599	设备	设备管理器不可用

常用精确值：

枚举	值
NONE	0
REQUEST_ARGUMENT_INVALID	2005
PROJECT_ID_NOT_FOUND	2103
WORKFLOW_NOT_FOUND	2202
STEP_NOT_FOUND	2203
PROPERTY_PATH_EMPTY_OR_INVALID	2204
WORKFLOW_EXECUTE_FAILED	2304
VISIONFLOW_OUTPUT_NOT_READY	2306
VISIONFLOW_OUTPUT_TIMEOUT	2307
VISIONFLOW_COMPLETED_BUT_NO_OUTPUT	2308
RECIPE_NOT_FOUND	2402
ROBOT_POSE_JPS_COUNT_INVALID	2501
ROBOT_POSE_QUATERNION_INVALID	2502

不要把内部错误码直接写死到机器人程序中。机器人程序只识别双方协议约定的外部状态码，映射关系集中放在 status_code_map.py 或自定义 Assembler 中。

不提供 RobotService 接口的原因

当前 Adapter 没有独立的 RobotService 抽象层，也不直接提供机器人运动、IO 控制或机器人状态查询接口。机器人相关能力目前仅限于通过 set_robot_pose 向视觉工程传入拍照时的机器人位姿。

因此开发者不应在协议代码中假设存在以下能力：

• 控制机器人关节或直线运动

• 读取机器人实时状态

• 设置机器人数字量或模拟量

• 管理机器人程序或任务

如果未来产品确实需要增加这些能力，应基于实际运行时 API 新增对应的 Operation，并补充参数、返回值、错误码和测试；不应先创建一个没有实现的空 RobotService 接口。

编程规范

协议解析

• 一个 Header Parser 负责一套帧头规则

• 一个命令使用一个独立 Payload Parser

• 字段名使用小写下划线格式，例如 project_id

• 解析阶段只做格式校验和字段转换，不直接访问 UI

• 对长度不足、编码错误、未知命令码和非法枚举值等情况，应抛出明确的异常

• 不要假设一次 recv 调用必然返回一条完整的业务消息（TCP 为字节流，消息可能被拆分或合并）

业务执行

• 通过 Operation 和执行上下文调用视觉系统，避免绕过执行队列

• 严格检查前置操作的状态，失败时立即返回该结果

• 对可能耗时的视觉结果获取设置合理超时

• 只有在协议明确允许时才并行执行多个视觉操作

• 日志中记录命令码、工程编号和 trace ID，不记录密码或敏感数据

响应组装

• 外部状态码与内部错误码的映射集中维护

• ASCII 响应统一浮点精度、分隔符和结束符

• 二进制响应统一端序和字段宽度

• 成功和失败响应都应满足相同的分帧规则

• 无结果与通信超时应使用不同状态码，便于现场排查

生成代码

• 不直接修改 generated/ 目录

• 配置可表达的需求通过生成器完成

• 通用生成能力通过模板扩展

• 单一设备的特殊逻辑放入独立协议包

• 修改模板后重新生成，并将配置与生成结果一同验证

调试与测试

启用日志

独立运行时可使用标准日志配置：

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
)

重点关注以下日志分类：

• adapter.comm：连接、收发和断线重连

• adapter.protocol：帧头匹配、字段解析和响应组装

• adapter.exec：命令排队、执行状态和超时

Adapter 为每条原始消息生成 trace ID，可用它串联同一请求在通信、协议和执行阶段的日志。

使用 VS Code 附加调试

生成器会在 Adapter 配置目录创建 .vscode/launch.json。启动 RVS 2.0 和 Adapter 后，可在 VS Code 中选择已生成的附加调试配置，连接 Python 调试端口。

运行单元测试

在 Adapter Python 工程目录执行：

python -m pytest tests -v

自定义协议至少应覆盖以下场景：

• 每个合法命令的解析与响应

• 未知命令码

• 缺少字段或字段长度错误

• 非法数值和非法编码

• 视觉操作成功、失败和超时

• 位姿数量为 0、1 和多个

• TCP 分包、粘包和连接中断

• 大端与小端二进制样例

常见问题

外部设备无法连接 Adapter

请依次检查：

1. Adapter 的服务类型是否为服务端

2. 监听 IP 和端口是否正确

3. 端口是否被其他进程占用

4. 工控机与外部设备能否互相 ping 通

5. Windows 防火墙是否放行该端口

6. 外部设备配置的 IP 是否为 Adapter 所在网卡的 IP，而不是 0.0.0.0

Adapter 收到数据但提示未知命令码

• 检查命令码大小写

• 检查帧前缀是否正确去除

• 检查 ASCII 分隔符是否一致

• 检查二进制命令码的宽度和端序

• 对照生成器预览，检查实际发送的字节

字段整体错位

• 检查字段位置是否从 1 开始

• 检查是否存在未配置但外部仍然发送的保留字段

• 检查数组字段长度

• 检查 ASCII 报文是否多了空字段

• 检查二进制字段的字节宽度

位姿数值正确但方向错误

• 检查长度单位和旋转单位

• 检查欧拉角轴顺序

• 检查静态轴与旋转轴定义

• 检查输入的是关节角、法兰位姿还是物体位姿

• 检查是否重复执行了机器人坐标转换

修改生成文件后没有生效或被还原

generated/ 是生成器的输出目录，保存配置时会重新生成。请在 Adapter 生成器中修改配置，或修改生成器模板后重新生成。设备专用代码应放在独立协议包中。

返回成功状态码，但没有位姿

• 检查是否启用了“返回位姿数据”

• 检查选择的是视觉点数据还是机器人路径

• 检查工作流输出是否包含位姿

• 检查获取视觉结果是否发生在执行工作流之后

• 检查固定返回数量与实际结果数量是否匹配

• 检查响应解析程序是否正确读取位姿数量字段

设置步骤参数失败

• 检查视觉工程和工作流是否已加载

• 重新选择步骤属性，避免使用已失效的步骤 ID 或属性路径

• 检查外部字段类型和长度

• 检查枚举值是否在允许范围内

• 检查独立步骤参数命令与拍照命令是否使用了不同且正确的命令码

交付检查清单

在将 Adapter 交付到现场前，至少应确认以下项目：

• Adapter 配置名称、IP、端口和服务类型正确

• 外部设备与工控机网络互通

• 输入和输出协议已有双方确认的字段表

• 分隔符、结束符、帧前缀和帧后缀一致

• 二进制协议端序和字段宽度一致

• 长度单位、旋转单位和欧拉角格式一致

• 工程编号、配方编号和步骤参数映射正确

• 成功及全部失败状态码已验证

• 无结果、超时、断线和重连场景已测试

• Adapter 配置和自定义协议代码已备份

• 现场机器人或 PLC 程序版本已记录

相关源码

本文涉及的主要实现位于：

• src/adapter/protocol_adapter.py：Adapter 生命周期、通信绑定和协议包加载

• src/adapter/protocol/：帧头解析、字段解析、调度与响应组装

• src/adapter/operations/：视觉系统操作

• src/adapter/exec/：命令队列、执行器和状态机

• src/adapter/communication/：TCP 客户端、服务端和数据流

• src/adapter/configurational_adapter/：配置式 Adapter 与代码生成器

• src/adapter/vanilla/：内置协议示例

• tests/：通信、协议解析、生成器和执行流程测试