オープンソースのロボットアームであるOpenArmの開発をお手伝いしている須藤です。OpenArmのデータ収集データフローと推論データフローはdoraを使っています。（もう半年くらい前になりますが）doraベースのデータフローの初期設計をしたのでなにを考えて設計したかを記録しておきます。
データフロー
データ収集や推論などといったタスクは単一のプログラムとしても実現できますが、データフローとしても表現できます。
たとえば、ロボットを遠隔操作して学習用の教師データを集めるデータ収集タスクは次のように分解できます。
- 人が遠隔操作用デバイスを動かして動作命令を出力
- 1.の情報を受け取ってロボットを動かす
- ロボットの今の状態を出力
- 1.と3.の情報を記録
人→遠隔操作用デバイス→ロボット→記
ーーーー→録
データフロー内では、矢印の両端にいるもの（「人」や「遠隔操作用デバイス」）がノードで行動主体になります。矢印がデータフロー内でのデータの移動になります。
専用のプログラムでこのようなタスクを実現することもできます。実際、OpenArmのデータ収集処理・推論処理の初期実装ではそうなっていましたし、フィジカルAIの便利ツールLeRobotでもそうなっています。（lerobot-recordというツールが提供されています。）
専用のプログラムで実現するメリットは、自由度が高く、柔軟に処理を実現できることです。一方、変更するにはプログラムを変更する必要がでてきて、専用のプログラム固有の知識が要求されることがデメリットです。もちろん、よく変更したくなる箇所はコマンドラインオプションや設定ファイルでカスタマイズできるようにすることでこのデメリットを軽減できます。
一方、このような処理を一般的なデータフローとみなして、汎用的なデータフローツール上で実現する方法もありえます。この方法のメリットは、汎用的なデータフローツールの知識を再利用できることおよび汎用的な処理をせずにすむことです。デメリットは汎用的なデータフローツールの制限に設計が左右されることです。
たとえば、カメラで撮った画像も学習データとして記録したくなったとしましょう。
人→遠隔操作用デバイス→ロボット→記
ーーーーー→
カメラ→録
専用のプログラムの場合は、カメラから画像を取り出す機能を実装して組み込みます。
データフローツール上で実現する場合は、すでにカメラから画像を出力するノードがあれば、それを再利用してデータフロー内に組み込むだけです。プログラムを書く必要はなく、データフロー定義を変更するだけです。
OpenArmは汎用的なデータフローツールとしてdoraを採用しました。これは、doraのコミュニティが大きくなりつつあるためコントリビューターが増えやすそうだったからです。もちろん、doraの制限により実現したいことが実現できなくなることは避けたいので、そのあたりは事前に検証しました。
dora
doraはDataflow-Oriented Robotic Architectureの略でデータフローをいい感じに実現するためのもろもろをやってくれるミドルウェアです。今、Webサイトを見たら1.0がリリースされそうな雰囲気を醸し出しています。ざっと見た感じ、ノード間の通信方法も増えていそうで、serviceとactionは便利そうだなぁという気がします。
当時はdora 0.4.0でtopicしかなかったので、これベースで検証・設計しました。
一番気にしたのは性能でした。
doraは速さをウリの一つにしているプロダクトです。ROS 2のPythonバインディングとdoraのPythonバインディングを比べるとめっちゃ速いと主張しています。（ただし、doraもROS 2もどちらもC++から使うと同じくらいの速さというデータも出していました。）これはデータ交換フォーマットとしてApache Arrowを使っているからです。Apache Arrowは私も開発に参加しているすごく得意なやつなのですが、データ交換コストを下げることを重視したデータフォーマットです。たしかに、画像データのように大きなデータをやりとりする場合は性能に効いてきそうです。
性能といってもいろいろな指標がありますが、特にレイテンシーを気にしました。できるだけ小さな遅延でデータ収集・推論を実行したかったからです。遅延が大きいと学習結果や推論結果に悪い影響がでる可能性が高いからです。
次の2つのケースを測定しました。
- 小さなデータを1msec/10msec/17msec（約60Hz）/100msecでそれぞれ3ノードで送り続けたときのレイテンシー
- 画像相当のデータ（640x480x3（RGB））を1msec/10msec/17msec（約60fpz）/100msecでそれぞれ3ノードで送り続けたときのレイテンシー
小さなデータのときは、10msec以上なら大丈夫そうでした。1msecのときは最大で数msecのjitterが発生していたのできびしそうでした。
画像相当データのときは、100msecなら大丈夫だけど17msec（約60fps）ならきびしそうでした。17msecのときは5-8msec程度のjitterが発生していました。
画像相当データのときは気になる結果ですが、60fpsで画像を送らないだろう（当時も今も30fpsで送っている）から大丈夫だろうと判断しました。
一応、実際のカメラでも検証しました。既存のopencv-video-captureノードを4つつなげて、4台のカメラから同時に30fpsで画像を送りました。JPEGへの変換処理をしても数msec程度のjitterでした。なお、USB 3ではなくUSB 2.0にすると10fpsになりました。（帯域がボトルネックになった。）
ということで、性能面が大丈夫そうなので、doraを採用することにしました。次は、どのような単位でノードを作るか、どのようなデータフローにするかを設計しないといけません。
doraデータフローの設計
前述の通り、dora 1.0ではservice/actionという通信方法が増えそうですが、当時はtopicしか通信方法がなかったので、その前提で設計しました。topicはようはpub/subです。あるノードが出力したデータを0個以上のノードが受信できます。
最初はループが発生しないように設計できないかを考えていました。ループというのは、次のような状態です。
ノード1→ノード2
↑ ↓
ノード4←ノード3
自分の入力が巡り巡ってまた自分に返ってきています。
ループを避けたかったのは、ループがあると自律的に終了することが面倒になったり、うっかり終了しないデータフローになりやすいと考えていたからです。データ収集や推論タスクは一通りの処理が終わったら完了するタスクです。doraのデータフローはCtrl-cなどで手動で終了できますが、タスクが完了したら自動でデータフローも終了するようにしたかったのです。これは、OpenArmを使ったAutoEval実装でも使いたい性質でした。
しかし、ループを許容し、最終的にはループがあっても自律的に終了する設計にしました。ループがないとやりたいことを実現できなかったのです。
たとえば、推論データフローは次のようになります。
OpenArmー現在位置→推論モデル（ポリシーサーバー）
←推論位置ー
OpenArmの現在の情報を元に次にどのように動けばよいかを推論して、その推論結果の通りにOpenArmを動かして、また動いたあとの情報を元に推論して、ということをタスクを完了するまで繰り返します。このような場合はやっぱりループしないとだめだなぁと思ったので、ループを許容するようにしました。
ループがあるといつまでも止まらないので、どこかに止める仕組みをいれないといけません。doraのデータフローはノードがすべて止まると止まります。つまり、各ノードがすべて止まる仕組みをいれればよいということです。
各ノードは自分でプロセスを終了するか入力がすべてなくなると止まります。
たとえば、次のループを考えます。
ノード1→ノード2
↑ ↓
ノード4←ノード3
このとき、ノード2が自分で終了したとします。するとこうなります。
ノード1→
↑
ノード4←ノード3
ノード3の入力がなくなりました。すると、ノード3も終了します。（doraがノード3に終了するための通知を送ります。）
ノード1→
↑
ノード4
次はノード4の入力がなくなったので、ノード4も終了します。
ノード1→
ノード1の入力もなくなったので、ノード1も終了します。これですべてのノードが終了したのでデータフローも終了します。
入力がなくなるとノードが終了するとはこういうことです。ただ、実際のデータフローではノードが提供する入力だけではなく、doraが提供する入力も使います。よく使うのが、dora/timer/${UNIT}/${VALUE}という入力です。たとえば、dora/timer/millis/33だと33msecごとに（約30Hzの）入力が発生します。そのため、実際は次のようなデータフローになっています。
dora/timer/millis/33→ノード1→ノード2→...
この場合、dora/timer/millis/33がなくなることはないので、ノード1が自分で終了しないとデータフローは終了しません。しかし、各ノードにどういうタイミングで自分から終了するべきかを実装するのはイヤだなぁと思いました。自分が持っている情報だけでは終了するべきかどうかを判断できなそうだからです。だからといって、無闇に入力を増やすのもイヤです。
ということで、dora-openarm-quitterノードを発明しました。このノードはcommandという入力と任意の入力を受け取ります。commandの中身がquitなら自分で終了します。それ以外の入力はそのまま出力します。
どう使うかというと、dora/timerをラップして使います。別途、なにか終了条件を知っているノードを作り、そのノードは終了を検知したらquitを送ります。これと、dora/timerをquitterの入力にし、各ノードはdora/timerを直接使うのではなく、quitter経由で使います。次のような感じです。
コントローラー→quitter→ノード1→ノード2→...
dora/timer→
コントローラーが終了を検出すると、quitコマンドを送ってから自分から終了します。
(quit)→quitter→ノード1→ノード2→...
dora/timer→
quitterはquitを受け取ったので終了します。
ノード1→ノード2→...
ノード1は入力がなくなったので終了し、次にノード2も終了します。
これで、各ノードで終了条件を実装しなくてもよくなりました。データフロー内でdora/timerを直接使う代わりにquitter経由で使うだけで、各ノードはdoraから終了通知をもらうまで動き続けるだけでよくなりました。
データフローの設計はこんな感じです。あとはノードの設計です。
doraノードの設計
doraノードは再利用可能な単位を意識して設計しています。
たとえば、OpenArmを制御（動かしたり現在情報取得したり）するノードはデータ収集でも使いたいですし、推論でも使いたいです。データ処理固有あるいは収集に固有の処理は別のノードに分けます。
データ収集処理でも、遠隔操作用デバイスが変わることもあります。たとえば、現在はOpenArm KERとMeta Quest 3というVRヘッドセットを使えます。（参考：Meta Quest 3を使う方法）
どちらの場合でも同じOpenArmを制御するノードを使えるようになっています。
具体的にどのようにノードを切り分けているかはdora-openarm-data-collectionとdora-openarm-evaluationを見てください。
各ノードごとにリポジトリーを分けているのはそっちの方がコントリビュートしやすいのではないかと思っているからです。一方、ノードが分かれていて全体像が把握しにくくなっていそうだなぁとは思っています。そのために、↑のようにsubmoduleで関連リポジトリーをまとめたリポジトリーも作っていますが、これで十分ケアできている。。。？
まとめ
OpenArmのデータ収集データフローと推論データフローではdoraを使っていますが、なにを考えて設計したのかをまとめました。
UIはどうしてこんな設計（FastAPIでWeb UIを提供するdoraノードとして実装）にしたのか他にも細かい話はあるのですが、全体的にはこんなことを考えて設計しました。
聞きたいことがある人はGitHub DiscussionsかDiscordで私に声をかけて聞いてください。どちらでも@kouで私にメンションできます。
