Adapter プログラミングガイド

本ガイドは、ロボット、PLC、その他の外部デバイスを RVS 2.0 ビジョンシステムに接続する必要のあるインテグレーション開発者を対象に、Adapter生成ツールを使用してアダプタプログラムを作成する方法と、Python Adapter フレームワークに基づいてカスタム通信プロトコルを開発する方法を説明します。

Adapter は外部デバイスとビジョンシステムの間に位置し、外部コマンドの受信、プロトコルの解析、ビジョンシステム機能の呼び出し、および外部デバイスが要求する形式での状態コード、姿勢、その他の結果の返却を行います。

注釈

本ガイドは、現在のリポジトリ内の Adapter 実装について説明しています。例の中の IP アドレス、ポート、プロジェクト番号、プロトコルフィールドは説明用のみです。現場では実際の設定に置き換えてください。

読むための前提知識

この記事を読む前に、以下の基礎知識があることをお勧めします。

• Python 3.10 以降の基本構文

• TCP/IP クライアントとサーバーの基本概念

• ASCII テキストプロトコルまたはバイナリプロトコルの基本知識

• JSON データ形式

• RVS 2.0 プロジェクト、ワークフロー、ステップ、レシピ、ビジョン結果の基本概念

注釈

通常の撮影トリガー、レシピ切り替え、ロボット姿勢の入力、ビジョン結果の返却だけを設定する必要がある場合は、Python コードを書かずに Adapter生成ツールを直接使用できます。

Adapter の動作方式

1 回の完全な通信は次の順序で実行されます。

1. Adapter は TCP サーバーまたは TCP クライアントとして接続を確立します

2. Adapter は 1 フレームの 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 には 2 つの開発方式があります。

• Adapter生成ツールを使用する

  以下のシーンに適しています。

  • TCP/IP 通信を使用する場合

  • プロトコルが ASCII テキストまたは固定フィールドのバイナリデータの場合

  • 外部コマンドにコマンドコード、プロジェクト番号、レシピ番号、ロボット姿勢、またはビジョンステップパラメータが含まれる場合

  • 返却内容が状態コード、ビジョン姿勢、ロボットパス、ラベル、または生成ツールがサポートするその他の出力の場合

  設定を保存すると、生成ツールはプロトコル解析、操作呼び出し、応答組み立て、状態コードマッピングのコードを生成します。生成コードは Adapter 設定ディレクトリの generated/ サブディレクトリにあります。

  参考

  Adapter生成ツールはグラフィカルな設定画面を提供し、Adapterプログラムの設定と生成をすばやく完了できます。詳細な使用方法は Adapter 生成器ガイド を参照してください。

• カスタム Python Adapter

  以下の場合にカスタム開発をお勧めします。

  • フレームヘッダー、チェックサム、エスケープ、粘着パケットの規則が複雑な場合

  • 1 つのリクエストで特別な業務フローを実行する必要がある場合

  • 生成ツールがカバーしていない入力または出力データ型をサポートする必要がある場合

  • 応答フレーム構造またはエラー処理方法をカスタマイズする必要がある場合

  • 再利用可能なデバイスプロトコルパッケージを開発する必要がある場合

カスタム 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

1 つのプロトコルの 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()	アップストリームサーバーから 1 回データを読み取り。クライアントのみ。
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 16進数テキストを 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	1 回の接続を確立。失敗時はネットワーク例外を発生。
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 個のトークンをスキップし、バイナリモードでは各項目が現在 1 つの 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	1 つのリクエストで複数の操作を実行した場合の元の結果リスト。

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 インターフェースを先に作成してはいけません。

プログラミング規範

プロトコル解析

• 1 つの Header Parser が 1 セットのフレームヘッダー規則を担当

• 1 つのコマンドが 1 つの独立した Payload Parser を使用

• フィールド名は小文字とアンダースコア形式を使用。たとえば project_id

• 解析段階では形式検証とフィールド変換のみを行い、UI に直接アクセスしない

• 長さ不足、エンコーディングエラー、未知のコマンドコード、不正な列挙値に明確な例外を発生

• 1 回の recv 呼び出しが必ず完全な業務メッセージを返すとは想定しないでください。TCP はバイトストリームプロトコルであり、メッセージは分割または結合される可能性があります。

業務実行

• Operation と実行コンテキストを介してビジョンシステムを呼び出し、実行キューを迂回しない

• 前提操作の状態を厳密に確認し、失敗時はその結果を直ちに返す

• 時間がかかる可能性のあるビジョン結果取得に妥当なタイムアウトを設定

• プロトコルが明示的に許可する場合にのみ複数のビジョン操作を並列実行

• ログにコマンドコード、プロジェクト番号、トレース 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 は各生メッセージにトレース 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. 工場PCと外部デバイスが相互に ping できるかどうか

5. Windows ファイアウォールがそのポートを許可しているかどうか

6. 外部デバイスが Adapter が動作する NIC の IP を設定しているかどうか（0.0.0.0 ではない）

Adapter がデータを受信したが未知のコマンドコードと表示される

• コマンドコードの大文字小文字を確認

• 前文字が除去されていないか確認

• ASCII 区切り文字が一致しているか確認

• バイナリコマンドコードの幅とエンディアンを確認

• 生成ツールのプレビューと実際に送信したバイトを照合

フィールド全体がずれる

• フィールド位置が 1 から始まっているか確認

• 設定していないが外部が送信する予約フィールドがないか確認

• 配列フィールドの長さを確認

• ASCII メッセージに余分な空フィールドがないか確認

• バイナリフィールドのバイト幅を確認

姿勢の数値は正しいが方向が誤っている

• 長さの単位と回転の単位を確認

• オイラー角の軸順序を確認

• 固定軸と回転軸の定義を確認

• 入力が関節角、フランジ姿勢、物体姿勢のどれかを確認

• ロボット座標変換を 2 回実行していないか確認

生成ファイルを変更しても反映されない、または元に戻される

generated/ は生成ツールの出力ディレクトリで、設定を保存すると再生成されます。Adapter生成ツールで設定を変更するか、生成ツールのテンプレートを変更して再生成してください。デバイス専用コードは独立したプロトコルパッケージに配置してください。

成功状態コードが返るが姿勢がない

• 「姿勢データを返す」が有効かどうか確認

• ビジョンポイントデータとロボットパスのどちらを選択したか確認

• ワークフローの出力に姿勢が含まれているか確認

• ビジョン結果の取得がワークフロー実行後に行われるか確認

• 固定返却数が実際の結果数と一致するかどうかを確認

• 応答解析プログラムが姿勢数フィールドを正しく読み取るか確認

ステップパラメータの設定に失敗する

• ビジョンプロジェクトとワークフローがロードされているか確認

• ステッププロパティを選択し直し、無効化されたステップ ID やプロパティパスを使用しない

• 外部フィールドの型と長さを確認

• 列挙値が許可範囲内か確認

• 独立ステップパラメータ指令と撮影指令が異なる正しいコマンドコードを使用しているか確認

納品チェックリスト

Adapter を現場に納品する前に、少なくとも以下の項目を確認してください。

• Adapter 設定名、IP、ポート、サービス種類が正しい

• 外部デバイスと工場PCのネットワークが疎通している

• 入力と出力のプロトコルに双方確認済みのフィールド表がある

• 区切り文字、終端文字、フレーム前文字、フレーム後文字が一致している

• バイナリプロトコルのエンディアンとフィールド幅が一致している

• 長さの単位、回転の単位、オイラー角形式が一致している

• プロジェクト番号、レシピ番号、ステップパラメータのマッピングが正しい

• 成功とすべての失敗状態コードが検証済み

• 結果なし、タイムアウト、切断、再接続のシーンがテスト済み

• 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/：通信、プロトコル解析、生成ツール、実行フローのテスト