run_robot_audio.py

#!/usr/bin/env python3

import os
import signal
import sys
import threading

# The G1 source build lives in this interpreter's own environment.
os.environ["CYCLONEDDS_HOME"] = sys.prefix

from unitree_sdk2py.core.channel import ChannelFactoryInitialize

from robogesture.config import DDS_DOMAIN, ROBOT_AUDIO_SINK, ROBOT_DDS_INTERFACE
from robogesture.robot_audio import PulsePlayer, RobotAudioService


def main():
    ChannelFactoryInitialize(DDS_DOMAIN, ROBOT_DDS_INTERFACE)
    service = RobotAudioService(PulsePlayer(ROBOT_AUDIO_SINK))
    stopped = threading.Event()
    signal.signal(signal.SIGINT, lambda *_: stopped.set())
    signal.signal(signal.SIGTERM, lambda *_: stopped.set())
    try:
        service.start()
        print(
            "[机器人音频] RoboGesture 已就绪：eth2 -> {}。".format(ROBOT_AUDIO_SINK),
            flush=True,
        )
        stopped.wait()
    finally:
        service.close()


if __name__ == "__main__":
    main()