Guidesに戻る
ロボット API 統合ガイド: ロボットをソフトウェアスタックに接続する
ロボットハードウェアをAPI,Webサービス,エンタープライズソフトウェアと統合する方法 <unk> REST API,WebSocket,ROS2橋,クラウド接続
[←ガイド]
システム統合ガイド 機械機器のハードウェアを企業ソフトウェアに接続するエンジニアのための API 設計,リアルタイム通信,クラウド接続パターンを含む.
統合アーキテクチャオプション
プログラミングの方法について,この4つの選択肢は,遅延,結合,生態系互換性との間の異なる妥協をします.
| Pattern | レイテンシ | Coupling | Best For |
|---|---|---|---|
| Direct SDK | <5ms | Tight — language/platform specific | Safety-critical control loops, high-frequency motion |
| REST API | 50–200ms | Loose — language agnostic | Task dispatch, status queries, configuration |
| WebSocket | 5–50ms | Medium — long-lived connection required | Real-time telemetry, servo commands, streaming |
| ROS2 bridge | 10–100ms | Loose — uses ROS2 ecosystem | ROS2-compatible software, visualization, multi-robot |
ほとんどの生産統合は組み合わせを使用します.タスクディスペッチャとステータスクエリ (遅延が許容される場合) にREST API,リアルタイムテレメトリストリーミングのためにWebSocket,ロボットのオンボードコンピュータで実行されている低レベルの制御ループにのみ直接SDKを使用します.ネットワーク上で直接SDKを決して暴露しないでください.これはすべての安全抽象を回避します.
ロボット制御のための REST API デザイン
設計が良くできたロボット REST API は,ロボットを資源と任務として一流のオブジェクトとして扱います.ほとんどの使用例をカバーする最小限のAPI表面は以下の通りです:
- POST /tasks
新しいタスクを提出する. 身体: T1 . 返却: T2 . - GET /tasks/{id}
作業状態を取得. 現在の状態 (キュー /実行 /成功 /失敗),実行メトリック,失敗の場合エラー詳細を返します. - GET /robots/{id}/status
ロボットの健康を把握する:関節温度,バッテリー,エラーコード,現在のタスク,ポーズ. - DELETE /tasks/{id}
列または実行中のタスクをキャンセル. 成功的にキャンセルされた場合200を返します. 409 タスクがキャンセルできない状態 (例えば,中挿入) であった場合,衝突. - 認証: マシン対マシン統合 (単純で監査可能な) に APIキーを使用するか,複数のレンタ展開のために OAuth2クライアント認証流を使用します.すべての要求に対して,キーを
T3 ヘッダーで入力してください. - 速度の制限: APIゲートウェイ層で強制される 100 リクエスト/秒キー速度の制限を適用する.
T4 ヘッダーで429 過剰なリクエストを返します.ほとんどの統合はこの制限に近づかない.
WebSocket 実装
WebSocket は,リアルタイムロボット制御とテレメトリストリーミングに必要な二方向性,低遅延のチャンネルを提供します. テレメトリの量によってコマンド遅延が影響されないようにするために2つの別の WebSocket チャンネルが推奨されます.
- ** コマンドストリーム (ロボット ← クライアント):** クライアントは,50
100Hz**で T5 \フォーマット JSON コマンドを送信します. ロボットのオンボードコントローラーは10 ms以内に各コマンドを認識するか,安全なストップ状態に入らなければなりません. コマンドストリームに JSON の代わりにバイナリーエンコーディング (MessagePack または CBOR) を使用します 3 5×より小さな有用荷,30 50%の遅延. - テレメトリストリーム (ロボット → クライアント): ロボットは,コイントステート,エンドエフェクターポーズ,カメラメタデータ,エラーコードを 10 Hzで送信する.監視とビジュアライゼーションには10 Hz が十分である.より高い速度ではバイナリーコーディングと専用ネットワーク帯域幅が必要である.
- 心拍数と再接続: クライアントは1秒ごとにピンフレームを送信しなければならない.ロボットが3秒間にわたってピンを受け取らない場合は,安全なストップ状態に入ります. 臨時ネットワーク中断を処理するために,クライアントに指数値的なバックオフ再接続論理 (1s,2s,4s,8sで再接続) を実装します.
- メッセージのシーケンス: 各メッセージに単調に増加するシーケンス番号を入れます.受信者は落としたメッセージを検出し,次のメッセージをインターポラするか待つかを決定することができます.
ROS2 ウェブブリッジ
既存のROS2インフラストラクチャを持つチームでは,JSONエンコーディングを使用して WebSocket上でROS2トピック,サービス,アクションを公開します.これは,WebダッシュボードやROS以外のアプリケーションがROS2グラフで相互作用し,ROS2を実行することなく,ROS2自身と相互作用することができます.
- ローズブリッジ_サーバー:**
T6 でインストールします. 既定的にポート9090で WebSocket サーバを実行します. クライアントはシンプルな JSON プロトコルを使用して ROS2 テーマに接続し,サブスクリプション/公開します. - roslibjs: ローズブリッジのJavaScriptクライアントライブラリ. ウェブダッシュボードは,
T7 , T8 ,カスタムトピックをサブスクリプションし,ブラウザからROS2サービスを呼び出すことができます. - JSON コードアップ: rosbridgeは JSON コードアップを用い,ROS2のネイティブ CDRバイナリー コードアップより5
10x大きい.高周波トピック (>10 Hz 大きなメッセージ) では,ブリッジする前に専用バイナリー橋やトピックのフィルタリングを検討してください. - セキュリティ注意: rosbridge はデフォルトで認証はありません.常に [艦隊管理ガイド] (T20
) で説明されている WireGuard VPN の裏で実行してください.
企業統合のパターン
機械を企業システム (WMS, ERP, MES) に接続するには,常にオンである企業システムと時々オフラインであるロボット間の信頼性の不一致に対処するパターンが必要です.
- タスク送送用のメッセージキュー (KafkaまたはRabbitMQ): WMSは,カフカトピックへのピックオーダーを公開します.ロボット艦隊マネージャーは,トピックから消費し,利用可能なロボットにタスクを送信します.タスクが到着するときにロボットがオフラインであれば,タスクはロボットが利用できるまで,キューに留まります.これはWMSをロボット利用可能性から分離します.
- ** ERP/WMS ウェブフック 注文イベント:** 新しい注文が準備ができるときに WMS を,ロボットプラットフォーム API にウェブフック を投稿するように設定します.ロボットプラットフォームはウェブフック を処理し,タスクを送信します.偽装の注文を予防するためにウェブフック サインタグチェック (HMAC-SHA256) を実装します.
- タイムシリーズテレメトリデータベース (InfluxDB): 長期分析とコンプライアンスレポートのために,ロボットテレメトリをInfluxDBに保存する.InfluxDBのデータ保存ポリシーを使用して,1分間の総数を2年間保持しながら,7日後に自動的に原始100Hzデータを終了します.
安全実施
- すべてのネットワーク通信のためのTLS 1.3: すべてのREST,WebSocket,およびロズブリッジトラフィックはTLS 1.3を使用する必要があります. 逆代理 (Nginxまたは Caddy) を
T9 とHSTSヘッダーで設定します.クラウドサービスに接続するロボットがサーバー証明書を検証する必要があります. - ロボットクライアントのための証明書ピン: ロボット搭載ソフトウェアは,期待されるサーバー証明書の指紋をピンし,予期しない証明書への接続を拒絶する必要があります.これは,不正なAPが存在する工場/倉庫ネットワーク環境で中人攻撃を防ぐ.
- 敏感な環境のためのVPN: 高いセキュリティ環境 (医薬品,防衛,金融) の部署では,WireGuard VPNトンネルを通じたロボットクラウド通信をすべてルートします.ロボットのクラウド接続は,複数の個別的にセキュアな接続ではなく,単一の認証された暗号化されたトンネルです.
- 監査ログ: すべての API コールを:タイムスタンプ,APIキーID (キー自身),エンドポイント,応答状態,およびリクエストIPでログインします. once-write store (AWS CloudTrailまたは GCP Cloud Audit Logs) で監査ログを保存します.これはSOC 2のコンプライアンスおよびインシデント調査のために必要です.
テスト と 嘲笑
- 開発のための偽ロボットサーバー: リアルなシミュレーションデータで完全なロボット API に対応する偽ロボット HTTP/WebSocket サーバを構築する.開発者は物理的なロボットアクセスを必要とせずに偽ロボットに対して統合を構築し,テストすることができます.偽物は決定的な統合テストのために録音されたロボットセッションを再再生する必要があります.
- シミュレーションベースの統合テスト: 完全なスタック (WMS →ロボットプラットフォーム →ロボット →テレメトリ →ダッシュボード) のエンドツインテストでは,ガゼボまたはアイザックシムでシミュレーションされたロボットを使用します. これらのテストは,ユニットテストが欠けているすべてのプルリクエストとキャッチ統合レグレーションでICで実行されます.
- 契約テスト: パクトまたは類似の契約テストフレームワークを使用して,ロボットAPIの生産者 (あなたのロボットプラットフォーム) とAPIの消費者が (WMS統合,ダッシュボード) がAPIのスキーマについて合意していることを確認します.これは統合まで変更が検出されないようにします.
統合建築図
生産倉庫部署の典型的なエンドツーエンドアーキテクチャ:
- Layer 1
ロボットハードウェア: ロボット腕 + ROS2を実行する機内コンピュータ,500 HzでSDK制御ループを直接操作する,ローカル・セーフティウォッチドッグ. - **Layer 2
ロボットプラットフォーム (本地またはクラウド):**タスク管理のための REST API,リアルタイムテレメトリのための WebSocket サーバー, RCSV プラットフォーム 艦隊ダッシュボードおよびポリシー展開のための. - 3層
企業統合: WMSの注文イベントを消費し,ロボットプラットフォームに送信する Kafkaメッセージバス; テレメトリストレージのための InfluxDB;オペレーションチームのためのGrafanaダッシュボード. - Layer 4
企業システム: WMS (倉庫管理), ERP (庫存,金融),MES (製造業実行システム)すべてはLayer 3のみで直接コミュニケーションをとるが,ロボットと直接コミュニケーションをとることはありません.
OpenArm 1 ROS2インターフェース: 具体的なテーマ
[OpenArm 1]
T10 --センサー_msgs/JointState 50 Hz. 7フィールド: 6 つの関節 + 1 つのグリッパーアパチュア. 状態インターフェース:位置,速度,力. T11 --軌跡_msgs/JointTrajectory. 位置モードは,ros2_control JointTrajectoryController を通して最大50Hz に コマンドする. T12 -- ジオメトリ_msgs/Wrench100Hzでスタンプ (F/Tセンサーが接続されたとき). T13 -- std_msgs/Float64. グリッパー開口コマンドは0.0 (閉) から1.0 (完全に開). T14 -- 診断_msgs/DiagnosticArray. エンジン温度,エラーコード,ファームウェアバージョン. 1Hzで公開されています. - ROS2サービス:
T15 --位置,速度,阻害制御モードの間のスイッチ. - ROS2アクション:
T16 -- 目標最終効果ポーズを受け取り,MoveIt2計画パイプラインを通じて実行します.
Python クライアント例: REST API 統合
作業を送信し,完成を監視し,テレメトリを回収する最小 Python クライアント:
import requests, time
API_URL = "https://platform.roboticscenter.ai/api/v1"
HEADERS = {"Authorization": "Bearer YOUR_API_KEY"}
# 1. Submit a task
task = requests.post(f"{API_URL}/tasks", json={
"task_type": "pick_and_place",
"robot_id": "arm-001",
"parameters": {
"pick_pose": [0.3, 0.1, 0.05, 0, 0, 0, 1],
"place_pose": [0.3, -0.2, 0.05, 0, 0, 0, 1]
},
"priority": 1
}, headers=HEADERS).json()
task_id = task["task_id"]
print(f"Task submitted: {task_id}")
# 2. Poll for completion
while True:
status = requests.get(
f"{API_URL}/tasks/{task_id}", headers=HEADERS
).json()
if status["status"] in ("succeeded", "failed"):
break
time.sleep(1.0)
print(f"Result: {status['status']}")
# 3. Get robot telemetry
health = requests.get(
f"{API_URL}/robots/arm-001/status", headers=HEADERS
).json()
print(f"Joint temps: {health['joint_temperatures']}")
print(f"Battery: {health['battery_pct']}%")
誤り処理と再試策
- Idempotency: すべての POST/tasks リクエストにはクライアント生成の
T17 (UUID) が含まれます.サーバーは同じキーが再送信された場合,既存のタスクを返します. - ** 倍率バックオフで復旧する:** 5xx サーバーエラーとネットワークタイムアウトの場合,最大4回の復旧による 1s, 2s, 4s, 8s間隔で復旧する. 4xx クライアントエラーの場合,復旧しないでください - 要求を修正する.
- WebSocket再接続: 遠隔測定 WebSocketが断つとき,クライアントは: (1) 1秒待ち,2秒再接続を試み, (3) すべてのトピックを再登録, (4) REST バックフィールエンドポイント
T18 を通じて,失われた遠隔測定の最後の10秒を要求しなければならない. - ** サーキットブレーカーパターンは:** 連続して5回のAPI通話でロボットがエラーを返せば,60秒間サーキットブレーカーを開く.この期間中に,そのロボットへのすべての要求はキャッシュ状態に戻ります.60秒後に,健康チェックの要求を1回送信します.成功した場合,ブレーカーを閉じて通常の操作を再開します.
関連ガイド
- ROS2とMoveIt2統合 --腕制御のための低レベルのROS2統合
- [リモート・フリット管理]
T24 ) -- 艦隊規模でのAPI統合と監視 - [政策展開生産]
T25 ) - 政策展開と監視のためのAPIパターン - [倉庫部署チェックリスト]
T26 ) --WMSと企業統合計画 - [ロボット安全リスク評価]
T27 ) - APIエンドポイントのセキュリティ要件
RCSVとの作業
RCSVプラットフォームは,ロボット制御,艦隊管理,企業統合のための生産レベルのAPIを提供します.
- データプラットフォーム -- RESTとWebSocket APIはタスクディスペッシング,テレメトリ,および艦隊管理のための
- データ収集サービス --APIによるデータ収集とプログラム作業の提出と配信
- Hardware Store -- OpenArm 1 プレビルドROS2ドライバーとAPI統合パッケージを持つ船舶
- [私たちと連絡してください]
T32 ) - 部署のための統合アーキテクチャレビューを要請する
RCSVプラットフォームにより早く統合する
RCSVプラットフォームは,ロボット部署のための,先構築された RESTと WebSocket API,艦隊管理,および企業統合コネクタを提供しています.
[プラットフォームを探求]