宇树 R1 迎宾机器人（二）：接住相机与语音——GStreamer 视频流 + 双 TTS + 遥控器启停

上一篇把人脸识别在电脑上跑通了。这一篇讲怎么把它接到宇树 R1 身上——拿到 R1 头部相机的画面、让 R1 开口说话、以及用遥控器一键启停。这三件事都通过宇树的 unitree_sdk2py（基于 DDS 通信）来做。

一、拿到 R1 的相机画面：GStreamer 接 UDP 视频流

R1 的头部相机不是普通 USB 摄像头，它把画面编码成 H264、通过 UDP 推流出来。不同端口是不同的图：

用之前要在 R1 的 APP 里开启推流服务、并关掉 video_hub 服务（两者会抢流）。

接收端用 OpenCV + GStreamer 管线。关键是拼出正确的管线字符串：从 UDP 收 RTP 包 → 解 H264 → 转成 BGR → 交给 OpenCV：

import cv2, threading, time

def gst_pipeline(port=5001):
    return (
        f"udpsrc port={port} ! "
        "application/x-rtp,media=video,clock-rate=90000,encoding-name=H264,payload=96 ! "
        "rtph264depay ! avdec_h264 ! videoconvert ! "
        "video/x-raw,format=BGR ! appsink drop=1 sync=false"      # drop=1 只留最新帧，低延迟
    )

cap = cv2.VideoCapture(gst_pipeline(5001), cv2.CAP_GSTREAMER)

两个要点：

• appsink drop=1 sync=false：只保留最新一帧、不按时间戳同步。人脸识别要的是"此刻"的画面，宁可丢旧帧也不要延迟堆积。
• 单开一个线程一直读帧，主循环随取随用。因为如果主循环处理慢了，cap.read() 会积压延迟；用后台线程不停读、只保留最新帧，主循环拿到的永远是最新的：

class R1Camera:
    def __init__(self, port=5001):
        self.cap = cv2.VideoCapture(gst_pipeline(port), cv2.CAP_GSTREAMER)
        self.frame = None; self.lock = threading.Lock(); self.running = True
        threading.Thread(target=self._loop, daemon=True).start()

    def _loop(self):
        while self.running:
            ret, frame = self.cap.read()
            if ret:
                with self.lock:
                    self.frame = frame          # 只存最新帧
            else:
                time.sleep(0.01)

    def read(self):
        with self.lock:
            return (True, self.frame.copy()) if self.frame is not None else (False, None)

我还做了个回退：如果 GStreamer 管线打不开（比如没插到机器人网络里），就自动退回电脑摄像头 cv2.VideoCapture(0)。这样同一套代码在办公室和机器人上都能跑。

二、让 R1 开口说话：原生 TTS + LED

R1 自带语音合成，通过 unitree_sdk2py 的音频客户端调用。先要初始化 DDS 通道（指定机器人所在的网卡接口），再拿到 AudioClient：

from unitree_sdk2py.core.channel import ChannelFactoryInitialize
from unitree_sdk2py.g1.audio.g1_audio_client import AudioClient

ChannelFactoryInitialize(0, "eth10")            # 第二个参数是连机器人的网卡名
client = AudioClient()
client.SetTimeout(10.0)
client.Init()

client.TtsMaker("张三先生你好，欢迎光临", 0)      # 说话
client.LedControl(0, 255, 0)                     # 顺便把灯点成绿色

TtsMaker(文本, 说话人id) 就能让机器人念出来，延迟低、还不依赖网络。LedControl(r,g,b) 控制机器人身上的灯——我用它做状态反馈：待命蓝色、识别成功绿色、停止红色。

不是所有场景都有机器人，所以语音这块我写了三级回退，auto 模式依次尝试：

1. R1 原生 TTS（AudioClient）——机器人上首选；
2. edge-tts（微软在线语音，pip install edge-tts）——电脑上开发用，联网就能出声，声音也自然；
3. 纯打印——什么都没有时，至少把要说的话打到终端，方便调试。

三者对外是同一个 speak(text) 接口，主程序不用关心底下用的是哪个。这种"同一接口、多级回退"的写法，让代码在机器人和电脑之间无缝切换，非常省心。

三、遥控器一键启停：读 R1 的手柄按键

年会现场不可能连着电脑敲命令，得用 R1 的遥控器控制。遥控器的按键状态藏在机器人的底层状态数据里——订阅 DDS 话题 rt/lf/lowstate，里面有个 wireless_remote 字段（40 字节），按键是其中某几个 bit。

我用 F1 启动识别、F2（代码里叫 F3）停止。它们分别是第 2 个字节的第 6、第 7 位：

from unitree_sdk2py.core.channel import ChannelSubscriber, ChannelFactoryInitialize
from unitree_sdk2py.idl.unitree_hg.msg.dds_ import LowState_

ChannelFactoryInitialize(0, "eth10")

low_state = [None]
def handler(msg): low_state[0] = msg
sub = ChannelSubscriber("rt/lf/lowstate", LowState_)
sub.Init(handler, 10)                            # 注册回调，持续接收底层状态

# 解析按键
def read_buttons(remote):
    f1 = (remote[2] >> 6) & 1                    # 第2字节 第6位
    f3 = (remote[2] >> 7) & 1                    # 第2字节 第7位
    return f1, f3

有两个细节必须处理，否则按一下会触发好多次：

• 只在"按下的瞬间"触发（上升沿检测）：记住上一帧的按键状态，只有"上一帧没按、这一帧按下"才算一次有效按键，而不是"只要按着就一直触发"。
• 去抖：两次动作之间隔至少 0.5 秒，避免手一抖连触发。

prev_f1 = 0; last_action = 0
while True:
    if low_state[0] is not None:
        f1, f3 = read_buttons(low_state[0].wireless_remote)
        now = time.time()
        if f1 == 1 and prev_f1 == 0 and (now - last_action) > 0.5:   # 上升沿 + 去抖
            start_recognition()                 # 启动人脸识别子进程
            last_action = now
        prev_f1 = f1
    time.sleep(0.05)

"启动识别"就是用 subprocess.Popen 把上一篇的主程序拉起来、"停止"就是 terminate() 掉。最后把这个控制脚本设成 systemd 服务，机器人开机自动运行，现场只要拿着遥控器按 F1 就开工。

到这里，机器人的"眼睛"（相机）、"嘴"（语音）、"开关"（遥控器）都接好了。下一篇讲"手和脖子"——怎么让 R1 挥手打招呼、空闲时自然地左右张望，以及人形机器人做动作时怎么保证安全。