PhysicalAIへの第一歩。Unitree GO2 Proでの開発方法を探る！～その３～

2025年9月19日 12:00

最近注目を集める「Physical AI」。
リアルなロボットとAIを組み合わせて、動き・知能・環境応答を統合的に扱うこの領域は、まさに次世代のものづくりのフロンティアです。

このシリーズでは、開発が不可能なAirやProといった低価格モデルで、開発や制御をできる方法がないかを調べていきます。

前回までは、ビジュアルプログラミング、SBUSを使用する方法をご紹介しましたが、今回は本命であるWebRTCを利用した方法をご紹介します。

WebRTC

WebRTCとは

WebRTCは、Webブラウザやモバイルアプリが、特別なプラグインを必要とせず、音声、映像、データをリアルタイムでやり取りできる技術です。端末同士が直接接続するP2P方式により、低遅延で高品質な通信を実現し、Web会議やライブストリーミングなどのリアルタイムコミュニケーションに広く活用されています。

Unitree GO2 × WebRTC?

GO2専用のスマートフォンのアプリケーションは、WebRTCを用いてGO2を通信しています。
このアプリケーション操作可能な本体移動や、専用のアクション、カメラや、LiDARデータの表示、音声再生や、LED点灯などはWebRTC経由で操作されています。

これらの機能をスマートフォンアプリからではなく、外部のPC等から使えばいろいろ操作できるんじゃないか、というのがこのWebRTCを利用した開発の基本です。

WebRTCを利用した開発

ブラウザから操作しよう

go2-webrtcを利用すると、すぐにWebRTCとの接続を体験できます。
GO2のIPアドレスを設定し、接続すればカメラ映像と仮想のコントローラーでGO2を制御可能です。
また、Commandを選ぶこと、ダンスや、握手など専用の動作も実行することができるようになっています

参考）https://github.com/tfoldi/go2-webrtc

動作映像

※操作がシビアなため、壁にぶつかっていますね・・

Pythonから制御しよう

go2_webrtc_connectを使うと、Pythonのプログラムから簡単に接続、制御することができます。

WebRTCの制御は、Go2WebRTCConnectionクラスを使用します。
難しい接続等は不要で、どのような接続方法を選ぶかだけで簡単に接続されます。

// 接続方法は、GO2 APモード、・同一ネットワーク上でのIP指定
// 同一ネットワーク上でのシリアル番号指定、Unitree TRUNサーバを利用したリモート接続
conn = Go2WebRTCConnection(WebRTCConnectionMethod.LocalSTA, ip="192.168.0.12")
await conn.connect()

実行すると接続まで一発。面倒な手続きも不要です。
※スマホや、ほかのPCから接続されているとエラーがでます。複数台接続するには、「Unitree TRUNサーバを利用したリモート接続」が必要そう。

Go2WebRTCConnectionのできること

Go2WebRTCConnectionにかはいくつかのサンプルが用意されており、すぐに動作を確認することができます。

audio（スピーカーや、マイクを利用したサンプル）
data_channel（LiDARステータスの取得や、操作のサンプル）
video/camera_stream（カメラのストリーム表示のサンプル）

ここからいくつかのサンプルと応用例を紹介します。

音声録音（save_audio_to_file）

WebRTCのAudioチャンネルの情報を受信し、録音するサンプルの録音データです。
GO2のマイクは、犬の腰のあたりにあるようで、内部のファンの音によりひノイズがひどい状況でした。
Whisperを利用してSTTもチャレンジしましたが、雑音の影響でまともなテキストになりませんでした・・。
期待してただけに残念です。

# オーディオトラックからの受信を開始
conn.audio.switchAudioChannel(True)

# オーディオトラックへのCallbackを設定
conn.audio.add_track_callback(recv_audio_stream)

# frameにaudioのframe情報が入っている
async def recv_audio_stream(frame):
    audio_data = np.frombuffer(frame.to_ndarray(), dtype=np.int16)

録画音声

LiDARの描画サンプル（plot_lidar_stream）

LiDARからの情報をブロックとして描画するサンプルです。
設定で「Voxel Mesh」、「Point Cloud」どちらの形式のデータも取得可能。

# LiDARの情報受信をONに
conn.datachannel.pub_sub.publish_without_callback("rt/utlidar/switch", "on")

# decoderの設定で Voxel Mesh、Point Cloudの切り替えが可能
conn.datachannel.set_decoder(decoder_type='libvoxel')

# LiDAR情報の受信
conn.datachannel.pub_sub.subscribe(
    "rt/utlidar/voxel_map_compressed",
    lambda message: asyncio.create_task(lidar_callback_task(message))
)

# messageには頂点情報が入る
async def lidar_callback_task(message):
     positions = message["data"]["data"].get("positions", [])

動作デモ

応用事例（Camera＋ yolo）

カメラの映像を表示するdisplay_video_channelのサンプルをもとに、物体検出を追加したものです。
通信や受信環境(RTX3060使用）に依存すると思いますが、カメラの受信速度は15FPSで、物体検出を行うと13~14FPSくらいになっています。

# ビデオチャンネルからの受信を開始
conn.video.switchVideoChannel(True)

# ビデオチャンネルからのCallbackを設定
conn.video.add_track_callback(recv_camera_stream)

# MediaStremTrackから映像データを取得する 
async def recv_camera_stream(track: MediaStreamTrack):
    while True:
        frame = await track.recv()
        # Convert the frame to a NumPy array
        img = frame.to_ndarray(format="bgr24")

動作デモ

応用事例（Camera＋状態表示＋回転制御）

GO2の回転＋状態表示＋カメラ映像の複合サンプルです。
制御は、sportmode、状態の取得は、lowstateを参考にしています。
状態は、IMUや、各サーボの動作状況、足の圧力など情報が1秒ごとに取得可能です。
1秒ごとだとIMUの動作を活用するのは厳しそうな感じですね。。

# 状態を取得するcallbackを設定
conn.datachannel.pub_sub.subscribe(RTC_TOPIC['LOW_STATE'], lowstate_callback)

# 状態を取得する
def lowstate_callback(message):
    message = message['data']
    imu_state = message['imu_state']['rpy']
    motor_state = message['motor_state']
    bms_state = message['bms_state']
    foot_force = message['foot_force']
    temperature_ntc1 = message['temperature_ntc1 ']
    power_v = message['power_v']

制御は、normalであることが必要です。
まずGO2の状態確認し、normalモードでない場合は、normalモードに変更します。
移動は、x, y, zのパラメータを指定することで制御可能です。
例えば、"x":0.3の場合は、1秒間に0.3m進みます。
このデモでは回転のみを行っているのでZに回転角度を指定しています。
動作コマンドを送信すると、停止コマンド（すべてが0の状態）を送るまで動作し続けるので注意してください

# GO2の現在の状態を取得する
response = await conn.datachannel.pub_sub.publish_request_new(
    RTC_TOPIC["MOTION_SWITCHER"], 
        {"api_id": 1001}
    )

if response['data']['header']['status']['code'] == 0:
    data = json.loads(response['data']['data'])
    current_motion_switcher_mode = data['name']

# 状態がnormal出ない場合は、normal状態に変更する（これをしないと動作しない）
if current_motion_switcher_mode != "normal":    
    await conn.datachannel.pub_sub.publish_request_new(
        RTC_TOPIC["MOTION_SWITCHER"], 
            {
                "api_id": 1002,
                "parameter": {"name": "normal"}
            }
        )                

# 回転動作を指定する
await conn.datachannel.pub_sub.publish_request_new(
    RTC_TOPIC["SPORT_MOD"], 
        {
            "api_id": SPORT_CMD["Move"],
            "parameter": {
                "x": 0.0,    # 前後移動なし
                "y": 0.0,    # 左右移動なし  
                "z": 0.5     # 右回転（正の値で時計回り）
            }
        }
    )

# 1.5秒間、上記の動作をさせる
await asyncio.sleep(1.5)

# すべてを0にすることで止まる（これがないと動き続ける）
await conn.datachannel.pub_sub.publish_request_new(
    RTC_TOPIC["SPORT_MOD"], 
    {
        "api_id": SPORT_CMD["Move"],
        "parameter": {"x": 0.0, "y": 0.0, "z": 0.0}
    }
)

動作デモ

まとめ

WebRTCを利用すると今までできなかったGO2からの入力が取れることがわかりました。
しかし、IMUなどのセンシングデータは取得速度が遅いため、状態把握に使うには厳しい気がします。
また、音声に関してもノイズがひどく、利用するには工夫が必要になるかもしれません。
今後、GO2でPhysicalAIを実現するには、カメラ映像やLiDARを中心にし、IMUなどは補助的に活用するか、別のデバイスからの情報をとれるようにすることになると思います。

著者 : 田中雅也 (痛快株式会社)
痛快師として日夜よくわからない活動をしています。

痛快技術株式会社について

痛快技術株式会社　：　https://www.tsu-x.com/

当社は、技術の力で社会の問題を解決することを目的として設立されました。私たちは、技術の楽しさを広め、社会のニーズに合わせた製品を開発し、幸せな未来を実現することを使命としています。

PhysicalAIへの第一歩。Unitree GO2 Proでの開発方法を探る！～その３～

WebRTC

WebRTCとは

Unitree GO2 × WebRTC?

WebRTCを利用した開発

ブラウザから操作しよう

Pythonから制御しよう

Go2WebRTCConnectionのできること

音声録音（save_audio_to_file）

LiDARの描画サンプル（plot_lidar_stream）

応用事例（Camera＋ yolo）

応用事例（Camera＋状態表示＋回転制御）

まとめ

痛快技術株式会社について

いいなと思ったら応援しよう！