Adapter Programming Guide

This guide is intended for integration developers who need to connect robots, PLCs, or other external devices to the RVS 2.0 vision system. It explains how to use the Adapter Generator to create adapter programs and how to develop custom communication protocols based on the Python Adapter framework.

The Adapter sits between the external device and the vision system. It receives external commands, parses the protocol, calls the vision-system capabilities, and returns status codes, poses, and other results in the format required by the external device.

Note

This guide describes the Adapter implementation in the current repository. The IP addresses, ports, project numbers, and protocol fields in the examples are for illustration only. Replace them with the actual configuration used on site.

Prerequisites

Before reading this article, it is recommended to be familiar with the following:

• Basic syntax of Python 3.10 or later

• Basic concepts of TCP/IP clients and servers

• Basic knowledge of ASCII text protocols or binary protocols

• The JSON data format

• Basic concepts of RVS 2.0 projects, workflows, steps, recipes, and vision results

Note

If you only need to configure routine photo triggering, recipe switching, robot-pose input, and vision-result return, you can directly use the Adapter Generator without writing Python code.

How the Adapter Works

A complete communication cycle executes in the following order:

1. The Adapter establishes a connection as a TCP server or TCP client.

2. The Adapter receives one frame of ASCII text or binary data.

3. The Header Parser parses the frame header and command code, and selects the corresponding Payload Parser.

4. The Payload Parser reads the project number, recipe number, robot pose, or step parameters according to the configuration.

5. The Adapter converts the parsing result into vision-system operations and executes them in sequence through the execution queue.

6. The Response Assembler converts the execution results into external status codes, poses, labels, and custom output.

7. The Adapter sends the response frame to the external device.

Robot / PLC / Host Computer
        │  TCP Request
        ▼
Communication Layer (TCP Server / TCP Client)
        │
        ▼
Header Parser ──Identify command code──► Payload Parser
                                          │
                                          ▼
                                  Operation / Execution Queue
                                          │
                                          ▼
                                  RVS 2.0 Vision System
                                          │
                                          ▼
                                  Response Assembler
        ◄────────────── TCP Response ──────┘

Choose a Development Method

The Adapter provides two development methods.

• Use the Adapter Generator

  Suitable for the following scenarios:

  • Using TCP/IP communication.

  • The protocol is ASCII text or fixed-field binary data.

  • The external command contains a command code, project number, recipe number, robot pose, or vision step parameters.

  • The return content is a status code, vision pose, robot path, label, or other output supported by the generator.

  When the configuration is saved, the generator creates code for protocol parsing, operation invocation, response assembly, and status-code mapping. The generated code is located in the generated/ subdirectory of the Adapter configuration directory.

  See also

  The Adapter Generator provides a graphical configuration interface, allowing you to quickly configure and generate an Adapter program. For detailed usage, see Adapter Generator Guide.

• Custom Python Adapter

  Custom development is recommended in the following cases:

  • The frame header, checksum, escaping, or sticky-packet rules are complex.

  • One request needs to execute a special business flow.

  • Input or output data types not covered by the generator need to be supported.

  • The response frame structure or error handling method needs to be customized.

  • A reusable device protocol package needs to be developed.

Custom Python Adapter

Install the Development Package

Enter the Adapter Python project directory and install in editable mode:

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

The current package requires Python 3.10 or later.

Protocol Package Structure

A custom protocol package must contain at least __init__.py and expose config_class and register(adapter):

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

__init__.py example:

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)

Define the Configuration Class

A custom configuration class inherits from 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"

The common configuration fields are as follows.

Field	Default	Description
transport	tcp	Transport layer type. The current implementation uses TCP.
service_type	server	server or client.
host	0.0.0.0	Listening address for the server, or the target address for the client.
port	50000	TCP port.
mode	ascii	ascii or hex.
byte_order	<	< is little-endian, > is big-endian.
delimiter	,	ASCII field separator.
reconnect_delay	5.0	Client reconnection interval, in seconds.
recv_size	4096	Single receive buffer size.
receive_timeout	None	Receive timeout, in seconds.

Fields not declared in the configuration class are stored in the extra dictionary of the configuration object and can be used to hold protocol-private parameters.

Implement the Header Parser

The Header Parser is responsible for validating the frame header, extracting the command code, determining the payload range, and returning the Payload Parser and Response Assembler bound to that command.

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,
        }

Notes during implementation:

• header_len and payload_len are calculated in bytes, not in Python character counts.

• Unrecognized command codes should raise a clear error.

• Do not execute business operations in the Header Parser.

• Binary protocols must unify the byte order, field width, and signedness conventions.

Implement the Payload Parser

CompositePayloadParser can combine multiple field parsers. In text mode, fields are read by the delimiter. In binary mode, fields are read by fixed width.

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=",",
    )

Common field parsers include:

Data Type	Scalar	Array
Signed integer	Int8Field, Int16Field, Int32Field	Corresponding *ArrayField
Unsigned integer	UInt8Field, UInt16Field, UInt32Field	Corresponding *ArrayField
Float	Float32Field, Float64Field	Corresponding *ArrayField
Boolean	BoolField	BoolArrayField
String	StrField	Combine manually according to the protocol

You can also use ConditionalFieldParser for conditional fields, or subclass FieldParser to implement custom data types.

Call Vision-System Operations

The built-in operations of the Adapter are as follows.

Operation	Typical Use
get_devices	Get device information.
get_project_id	Get the project number based on project information.
get_workflow_id	Get the workflow ID based on the project number.
get_display_id	Get the display object ID.
get_solution_info	Get the current solution information.
get_step_info	Get step information.
get_step_property	Read a step property.
update_step_property	Update a step property.
switch_recipe	Switch the project recipe.
set_robot_pose	Write robot joint angles and an optional flange pose.
execute_workflow	Execute the specified workflow.
get_vision_result	Wait for and obtain the vision result.

In an asynchronous Payload Parser, call operations with operation.submit(exec_ctx, params=...). When a subsequent operation depends on the result of a previous step, check status_code and data before continuing.

The recommended photo flow is:

1. Get the workflow ID by the project number.

2. Update step properties as needed.

3. Switch the recipe as needed.

4. Set the robot pose as needed.

5. Execute the workflow.

6. Get the vision result.

7. Return the final execution result to the Assembler.

Implement the Response Assembler

The Response Assembler converts an ExecutionResult into a protocol response. The following example only returns a status code and terminator:

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)

Real projects usually also need to:

• Prioritize checking the first failed status in the sub-results.

• Perform unit conversion and pose conversion according to the configuration.

• Return the pose count, poses, labels, and custom output.

• Return explicit status codes for missing results, timeouts, and exceptions.

• Pack data with a consistent byte order in binary mode.

Start the Custom Protocol Package

Use ProtocolAdapter.start_protocol() to load the protocol package:

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 can be:

• None: use the defaults of the protocol configuration class.

• dict: override specified configuration items.

• a ProtocolConfig instance for the protocol.

Adapter Development API Reference

This chapter only describes the interfaces in the current code that can be used for Adapter development. For routine projects, use the Generator. These interfaces only need to be called directly when developing a custom protocol, extending generation templates, or troubleshooting low-level issues.

Interface Layers

Layer	Main Interfaces	Responsibility
Adapter lifecycle	ProtocolAdapter, ProtocolConfig	Create the communication object, load the protocol, start and stop the service.
Communication layer	TCPServer, TCPClient	Establish the TCP connection and send/receive data.
Frame header parsing	HeaderParser	Identify the protocol and command, and determine the payload range.
Field parsing	PayloadParser, CompositePayloadParser, FieldParser	Convert the payload into named parameters.
Business execution	Operation, ExecutionContext, CommandRequest	Submit the parsing result to the vision system for execution.
Execution result	ExecutionResult, VisionEngineSubErrorCode	Uniformly represent success, failure, and business data.
Response assembly	ResponseAssembler, WriteContext	Encode the execution result as the external device response.

A typical call relationship is as follows:

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

ProtocolAdapter

Definition:

@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

Constructor parameters:

Parameter	Type	Description
transport	str	Transport layer identifier. Currently uses tcp.
service_type	str	server creates a TCP server. client creates a TCP client.
host	str	Listening address for the server, or target address for the client.
port	int	TCP port.
mode	str	ascii or hex.
byte_order_init	str	Byte order for binary fields: < little-endian, > big-endian. An invalid value raises ValueError.
reconnect_delay	float	Retry interval after disconnection in client mode.
recv_size	int	Maximum number of bytes read by the client at one time.
receive_timeout	`float	None`
skip_default_protocols	bool	False automatically registers the built-in vanilla protocol. Custom protocol packages usually set this to True.
communication	`TCPServer	TCPClient

Public methods:

await adapter.start() -> None

Starts the currently bound communication object and executor. In server mode, starts listening. In client mode, connects to the target server and enters the receive loop.

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

If startup fails, the underlying network exception propagates to the caller, for example the OSError corresponding to a port in use.

await adapter.stop() -> None

Stops receiving, closes the client connection and listening socket, and stops the executor. It should be called in a finally block when the application exits.

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

close() is currently equivalent to stop().

adapter.add_header_parser(parser) -> None

Registers a Header Parser for one protocol. Multiple parsers are tried in registration order. The first parser that successfully returns parsing metadata takes effect.

header = MyHeaderParser()
adapter.add_header_parser(header)

adapter.bind_communication(communication) -> None

Replaces the communication object used by the Adapter. The parameter must be a TCPServer or TCPClient, otherwise a TypeError is raised.

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

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

Loads a custom protocol package. The __init__.py of the protocol package must expose:

config_class = MyProtocolConfig

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

Parameters and exceptions:

Item	Description
package_path	The protocol directory containing __init__.py.
config	None, a configuration dictionary, or a ProtocolConfig instance.
start	When False, only creation and registration are completed, and the network is not started.
Return value	The created ProtocolAdapter.
FileNotFoundError	The directory does not exist or __init__.py is missing.
AttributeError	The protocol package does not expose config_class.
TypeError	The config type is wrong.

Client-specific methods:

Method	Description
init_client((host, port))	Creates and binds a TCP client.
set_bind_port(is_bind=True)	Sets whether to bind the local port before the client connects. Client only.
set_recv_size(size)	Modifies the client receive buffer size.
await send(msg)	Sends data to the upstream server. Client only.
await recv()	Reads data once from the upstream server. Client only.
await reconnect_server()	Tries to reconnect to the upstream server.

Calling client-specific methods in server mode raises a RuntimeError.

ProtocolConfig

ProtocolConfig is the common configuration base class of a protocol package:

@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)

When creating a configuration through from_dict(), declared fields are assigned directly, and undeclared fields go into extra:

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

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

Protocol-private configuration that really needs type hints should be declared in the subclass. A small number of extended values that only need to be passed through can be placed in extra.

TCPServer

Constructor:

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

mode can only be ascii or hex:

• ascii: decodes received bytes as UTF-8, and the callback parameter is str.

• hex: converts the received ASCII hexadecimal text to bytes, and the callback parameter is bytes.

Register a callback:

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()

Callback signature:

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

Public methods:

Method	Return Value	Description
on_data(callback)	Original callback	Registers a sync or async data callback. Can also be used as a decorator.
await start()	None	Starts listening.
await stop()	None	Closes all connections and stops listening.

Current limitations:

• The server uses reader.read(4096) each time.

• No automatic frame splitting based on terminator, length field, or fixed frame length.

• Input that fails ASCII decoding is discarded with a warning logged.

• Hex text containing illegal characters is discarded with a warning logged.

TCPClient

Constructor:

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,
)

Main methods:

Method	Return Value	Description
await connect()	None	Establishes one connection. Raises a network exception on failure.
await reconnect_server()	bool	Reconnects with exponential backoff. Returns True on success.
is_connected()	bool	Returns True when the current writer exists and is not closed.
await send(msg)	None	Sends a str or bytes. Raises ConnectionError when not connected.
await recv()	`str	bytes
set_timeout(timeout)	None	Modifies the read timeout. Raises ValueError for non-positive values.
set_recv_size(size)	None	Modifies the read size. Raises ValueError for non-positive values.
set_bind_port(...)	None	Configures the local binding address and port of the client.
await close()	None	Closes the connection.

The wait time of reconnect_server() doubles from reconnect_base_delay and does not exceed reconnect_max_delay. reconnect_max_retries=0 means unlimited retries.

HeaderParser

A custom Header Parser must implement:

class HeaderParser(ABC):
    default_failure_assembler = None

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

On success, parse() must return at least payload_parser. Common return fields:

Key	Type	Required	Description
payload_parser	PayloadParser	Yes	The payload parser used by the current command.
assembler	ResponseAssembler	No	The response assembler used by the current command.
header_len	int	No	The byte offset of the payload start. Default is 0.
payload_len	int	No	The payload length. Used when the frame suffix needs to be excluded.
command_code	str	No	The command code, for logging or downstream parsers.
Custom keys	Any	No	Passed to the Payload Parser as metadata.

Matching rules:

• Return a dictionary when the current frame belongs to this protocol.

• Raise ValueError when the current frame does not belong to this protocol.

• Among multiple Header Parsers, the first one that succeeds takes effect.

• If none match, the dispatch layer tries to assemble an error response with default_failure_assembler.

Minimal implementation:

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 and CompositePayloadParser

The underlying interface:

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

It is recommended to use CompositePayloadParser: first generate named parameters through field parsers, then enter the asynchronous business function:

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

Asynchronous function signature:

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

After field parsing completes:

• exec_ctx.parsed stores the parsing result.

• each parsed field is also written to metadata for subsequent conditional parsers.

• a ValueError is raised when the payload has unconsumed non-empty fields or bytes.

• when fields are insufficient, the underlying data-stream read function raises an exception.

Example:

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=",",
)

Field Parsing Interfaces

Scalar field construction is unified as:

FieldType(field_name: str)

Array field construction is unified as:

ArrayFieldType(field_name: str, count: int)

Python Type	Scalar Field	Array Field	Binary Width
int	Int8Field	Int8ArrayField	1 byte/value
int	UInt8Field	UInt8ArrayField	1 byte/value
int	Int16Field	Int16ArrayField	2 bytes/value
int	UInt16Field	UInt16ArrayField	2 bytes/value
int	Int32Field	Int32ArrayField	4 bytes/value
int	UInt32Field	UInt32ArrayField	4 bytes/value
bool	BoolField	BoolArrayField	Per underlying bool format
float	Float32Field	Float32ArrayField	4 bytes/value
float	Float64Field	Float64ArrayField	8 bytes/value
str	StrField	No built-in string array field	Used in text mode

Other field interfaces:

Gap(count=1)

Skips reserved fields without writing them to the result. In text mode, skips count tokens. In binary mode, each item is currently skipped as one uint32.

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

LeafFieldParser(field_name, reader_fn)

Injects a simple custom read function:

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

ConditionalFieldParser(predicate, true_parser, false_parser=None)

Decides whether to read a field based on already-parsed metadata:

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

Fields that the condition depends on must be placed before the conditional field.

ExecutionContext

Core definition:

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

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

run() will:

1. Hand the command to the executor.

2. Wait for the result of the single command.

3. Write the command name to result.data["_cmd"].

4. Append the result to results.

5. Return that result for the next step.

When the execution queue returns no result within 30 seconds, a VISIONFLOW_OUTPUT_TIMEOUT result is created. When the executor raises an exception, it is converted to SERVICE_HANDLER_EXCEPTION.

The caller should always check the status first:

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 and ExecutionResult

Command definition:

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

Field	Description
msg_type	The operation name, also the execution dispatch key.
params	The Python parameter dictionary passed to the operation.
raw	Optional raw payload, used for logging and tracing.
executor	Optional direct execution function. Usually auto-generated by Operation.

Result definition:

@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)

Field	Description
status_code	Internal status code. 0 means success.
data	Business data returned by the operation.
error_message	Human-readable error message suitable for logging and diagnostics.
duration_ms	Execution time of a single operation.
sub_results	The original result list when a request executes multiple operations.

The Assembler should not only check the top-level status. For multi-operation flows, check sub_results and map the first failed result to an external status code.

Operation

Definition:

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

Common methods:

Method	Description
await operation.submit(exec_ctx, params)	Submits and waits for the result. Use when subsequent steps depend on this result.
operation.submit_async(exec_ctx, params)	Creates an async task without waiting for the result. Only for operations with truly no sequential dependency.
operation.params.<name>	Gets the parameter key name to avoid repeating hard-coded strings in multiple parsers.

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

Built-in operation interfaces:

Operation	Parameters	Main Data on Success
get_workflow_id	project_id: int	{"workflow_id": int}
set_robot_pose	project_id: int, joint_angles: list, optional flange_pose	No business data
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 (milliseconds)	Vision output dictionary
get_step_property	flow_id: int, step_id: int, property_path: str	Property data dictionary
update_step_property	flow_id: int, step_id: int, property_path: str, json_value: str	No business data or underlying error data
get_solution_info	None	Current solution information
get_project_id	display_project_id: int	Project ID data. Non-dict results are wrapped as {"project_id": value}.
get_step_info	flow_id: int, step_id: int	Step information. Non-dict results are wrapped as {"value": value}.
get_devices	None	Project device information. Non-dict results are wrapped as {"devices": value}.
get_display_id	project_id: int	Display project ID. Non-dict results are wrapped as {"display_project_id": value}.

set_robot_pose Parameter Shapes

Joint angles only:

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

Joint angles and flange pose:

{
    "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],
    },
}

The pose received by an Operation should already be converted in units and rotation expression according to the internal API requirements. The Generator calls conversion tools to complete this step. Handwritten protocols need to perform the conversion themselves.

update_step_property Parameter Shape

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

json_value is a string, not a Python dictionary. When the property path, type, or value is invalid, the underlying sub-error code is preserved in ExecutionResult.status_code.

ResponseAssembler

A custom response assembler implements at least:

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

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

Return values:

• Return bytes: send this response.

• Return None: handle silently without sending a response.

• assemble_multi() calls assemble() by default. Override it when sub_results needs separate handling.

Recommended structure:

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 provides a simple “internal code to external code” dictionary implementation, suitable for prototypes and simple comma protocols. Production projects usually need to subclass ResponseAssembler and explicitly control the pose, label, prefix/suffix, and failure-response formats.

WriteContext

WriteContext is used to write fields in text or binary mode:

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()

The text result is:

1,12.346,box_A

Binary mode:

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

Available write methods:

Method	Description
write_int8 / write_uint8	Writes an 8-bit integer.
write_int16 / write_uint16	Writes a 16-bit integer.
write_int32 / write_uint32	Writes a 32-bit integer.
write_float32 / write_float64	Writes a float.
write_str	Writes a string. Writes UTF-8 bytes in binary mode.
write_bytes	Writes raw bytes.
to_bytes	Returns the final byte string.

WriteContext does not automatically add the \r, \n, CRC, or length fields required by the business protocol. These are added explicitly by the Assembler.

Internal Error Codes

All Operations first return a unified internal error code, and the Assembler then maps it to the on-site protocol status code.

Range	Category	Examples
0	Success	NONE
2000~2099	Common execution errors	Invalid parameter, IPC unavailable, invalid license, service exception
2100~2199	Solution and project	Project does not exist, project has no workflow
2200~2299	Step properties	Workflow or step does not exist, invalid property path, deserialization failure
2300~2399	Workflow and vision results	Workflow execution failed, result not ready, result timeout, no output
2400~2499	Recipes	Recipe manager unavailable, recipe does not exist, empty list
2500~2549	Robot pose	Wrong joint count, invalid quaternion, invalid pose format
2550~2599	Devices	Device manager unavailable

Common precise values:

Enum	Value
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

Do not hard-code internal error codes in robot programs. Robot programs only recognize the external status codes agreed in the protocol of both parties. The mapping should be centrally maintained in status_code_map.py or a custom Assembler.

Why No RobotService Interface Is Provided

The current Adapter has no separate RobotService abstraction layer and does not directly provide robot motion, IO control, or robot status query interfaces. Robot-related capability is currently limited to passing the robot pose at photo time into the vision project through set_robot_pose.

Therefore, developers should not assume the following capabilities exist in protocol code:

• Controlling robot joint or linear motion.

• Reading real-time robot status.

• Setting robot digital or analog outputs.

• Managing robot programs or tasks.

If the product does add these capabilities in the future, add the corresponding Operation based on the actual runtime API, and supplement the parameters, return values, error codes, and tests. Do not first create an empty RobotService interface with no implementation.

Coding Guidelines

Protocol Parsing

• One Header Parser is responsible for one set of frame-header rules.

• One command uses one independent Payload Parser.

• Field names use lowercase with underscores, for example project_id.

• The parsing stage only performs format validation and field conversion, and does not directly access the UI.

• Raise clear exceptions for insufficient length, encoding errors, unknown command codes, and invalid enum values.

• Do not assume that one recv call necessarily returns a complete business message. TCP is a byte-stream protocol; a message may be split or merged.

Business Execution

• Call the vision system through Operation and the execution context, avoiding bypassing the execution queue.

• Strictly check the status of prerequisite operations, and return that result immediately on failure.

• Set a reasonable timeout for potentially time-consuming vision-result retrieval.

• Execute multiple vision operations in parallel only when the protocol explicitly allows it.

• Record the command code, project number, and trace ID in logs, but do not record passwords or sensitive data.

Response Assembly

• Maintain the mapping between external status codes and internal error codes centrally.

• Unify float precision, separators, and terminators in ASCII responses.

• Unify byte order and field width in binary responses.

• Both success and failure responses should satisfy the same framing rules.

• Use different status codes for “no result” and “communication timeout” so they can be distinguished on site.

Generated Code

• Do not directly modify the generated/ directory.

• Complete requirements expressible by configuration through the Generator.

• Extend general generation capabilities through template extensions.

• Put the special logic of a single device into an independent protocol package.

• Regenerate after modifying templates, and verify the configuration and generated results together.

Debugging and Testing

Enable Logging

When running standalone, use the standard logging configuration:

import logging

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

Pay attention to the following log categories:

• adapter.comm: connection, sending/receiving, and disconnect-reconnect.

• adapter.protocol: frame-header matching, field parsing, and response assembly.

• adapter.exec: command queueing, execution status, and timeouts.

The Adapter generates a trace ID for each raw message. You can use it to correlate logs from the communication, protocol, and execution stages of the same request.

Attach the VS Code Debugger

The Generator creates .vscode/launch.json in the Adapter configuration directory. After starting RVS 2.0 and the Adapter, you can select the generated attach-debug configuration in VS Code to connect to the Python debug port.

Run Unit Tests

In the Adapter Python project directory, run:

python -m pytest tests -v

A custom protocol should at least cover the following scenarios:

• Parsing and response for every valid command.

• Unknown command codes.

• Missing fields or wrong field lengths.

• Invalid numeric values and invalid encodings.

• Vision operation success, failure, and timeout.

• Pose counts of 0, 1, and multiple.

• TCP packet splitting, sticky packets, and connection interruption.

• Big-endian and little-endian binary examples.

Troubleshooting

The External Device Cannot Connect to the Adapter

Check in order:

1. Whether the Adapter service type is server.

2. Whether the listening IP and port are correct.

3. Whether the port is occupied by another process.

4. Whether the industrial PC and the external device can ping each other.

5. Whether the Windows firewall allows this port.

6. Whether the external device is configured with the IP of the NIC where the Adapter runs, not 0.0.0.0.

The Adapter Receives Data but Reports an Unknown Command Code

• Check the case of the command code.

• Check whether the prefix has been removed.

• Check whether the ASCII separator is consistent.

• Check the width and byte order of the binary command code.

• Compare the actual sent bytes with the Generator preview.

Fields Are Globally Misaligned

• Check whether the field positions start from 1.

• Check whether there are reserved fields that are not configured but still sent by the external device.

• Check the array field length.

• Check whether the ASCII message has extra empty fields.

• Check the byte width of binary fields.

The Pose Values Are Correct but the Orientation Is Wrong

• Check the length unit and rotation unit.

• Check the Euler-angle axis order.

• Check the static-axis and rotated-axis definitions.

• Check whether the input is joint angles, a flange pose, or an object pose.

• Check whether the robot coordinate conversion has been performed twice.

Modifying Generated Files Has No Effect or They Are Restored

generated/ is the output directory of the Generator and is regenerated when the configuration is saved. Modify the configuration in the Adapter Generator, or modify the Generator template and regenerate. Device-specific code should be placed in an independent protocol package.

A Success Status Code Is Returned but No Pose

• Check whether “Return Pose Data” is enabled.

• Check whether Vision Point Data or Robot Path is selected.

• Check whether the workflow output contains poses.

• Check whether the vision result is obtained after the workflow is executed.

• Check whether the fixed return quantity matches the actual result quantity.

• Check whether the response parsing program reads the pose-count field correctly.

Setting Step Parameters Fails

• Check whether the vision project and workflow are loaded.

• Re-select the step property to avoid using an invalidated step ID or property path.

• Check the external field type and length.

• Check whether the enum value is within the allowed range.

• Check whether the independent step-parameter command and the photo command use different and correct command codes.

Delivery Checklist

Before delivering the Adapter to the site, confirm at least the following items:

• The Adapter configuration name, IP, port, and service type are correct.

• The external device and the industrial PC are networked.

• The input and output protocols have field tables confirmed by both parties.

• The separator, terminator, frame prefix, and frame suffix are consistent.

• The byte order and field width of the binary protocol are consistent.

• The length unit, rotation unit, and Euler-angle format are consistent.

• The project number, recipe number, and step-parameter mapping are correct.

• All success and failure status codes have been verified.

• The no-result, timeout, disconnection, and reconnection scenarios have been tested.

• The Adapter configuration and custom protocol code have been backed up.

• The on-site robot or PLC program version has been recorded.

Related Source Code

The main implementations involved in this article are located at:

• src/adapter/protocol_adapter.py: Adapter lifecycle, communication binding, and protocol package loading.

• src/adapter/protocol/: frame header parsing, field parsing, dispatch, and response assembly.

• src/adapter/operations/: vision-system operations.

• src/adapter/exec/: command queue, executor, and state machine.

• src/adapter/communication/: TCP client, server, and data stream.

• src/adapter/configurational_adapter/: configuration-based Adapter and code generator.

• src/adapter/vanilla/: built-in protocol examples.

• tests/: tests for communication, protocol parsing, generator, and execution flow.