No.8 音频录制与播放 | 众擎开源平台

本章介绍如何使用机器人内置的麦克风与扬声器进行音频录制、回放与系统音量控制。内容从环境准备出发，先给出命令行快速操作，再提供三个完整的 Python 可运行示例（系统音量控制、麦克风录音、音频播放），帮助开发者快速搭建音频应用。

环境说明：目前仅应用计算单元支持音频使用，开发音频相关应用需要在该单元上进行。

1.1 环境准备

1.1.1 连接应用计算单元

请先远程连接到应用计算单元：

Bash

ssh ubuntu@192.168.0.162

1.1.2 安装系统依赖

以下操作需要先通过机器人可视化应用连接网络。

PLAIN

sudo apt-get update
sudo apt-get install -y \
  python3-venv \
  python3-dev \
  portaudio19-dev \
  libsdl2-dev \
  libsdl2-mixer-dev

可选：国内用户可以通过修改 apt 源为国内镜像以提升安装速度。

Bash

# 1. 备份原配置
sudo cp /etc/apt/sources.list /etc/apt/sources.list.bak

# 2. 换成国内源（以清华为例）
sudo tee /etc/apt/sources.list > /dev/null <<'EOF'
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ jammy main restricted universe multiverse
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ jammy-updates main restricted universe multiverse
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ jammy-backports main restricted universe multiverse
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ jammy-security main restricted universe multiverse
EOF

# 注意：如果是ARM架构，应该使用如下源路径
sudo tee /etc/apt/sources.list > /dev/null <<'EOF'
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu-ports/ jammy main restricted universe multiverse
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu-ports/ jammy-updates main restricted universe multiverse
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu-ports/ jammy-backports main restricted universe multiverse
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu-ports/ jammy-security main restricted universe multiverse
EOF

# 3. 更新索引
sudo apt-get update

# 4. 进行依赖包的安装
sudo apt-get install -y \
  python3-venv \
  python3-dev \
  portaudio19-dev \
  libsdl2-dev \
  libsdl2-mixer-dev

1.1.3 创建虚拟环境并安装 Python 依赖

假设脚本都在 /home/ubuntu/audio_example（按你实际路径改）：

PLAIN

cd /home/ubuntu/audio_example

# 在本目录创建名为 .venv 的虚拟环境
python3 -m venv .venv

会生成目录 .venv/。激活虚拟环境并安装依赖：

Bash

# 激活（每次新开终端都要执行一次）
source .venv/bin/activate

# 确认当前用的是 venv 里的 pip
which pip
# 应类似：.../audio_example/.venv/bin/python

# 升级 pip（可选但建议）
pip install -U pip setuptools wheel

# 安装本目录脚本需要的包
pip install numpy pyaudio pygame

# 可选：国内用户可以通过临时指定镜像源的方式提升安装速度
pip install numpy pyaudio pygame -i https://pypi.tuna.tsinghua.edu.cn/simple

# 如果使用workspace的仓库，可以使用requirements.txt进行安装
pip install -r requirements.txt
# 可选：国内用户可以通过临时指定镜像源的方式提升安装速度
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

1.2 命令行快速操作

机器人内置麦克风和扬声器可以通过 Linux 命令行工具快速使用，无需编写代码即可完成录音、回放与音量调节。

1.2.1 录制音频

Bash

arecord -f cd test.wav

1.2.2 回放录制的音频

Bash

aplay test.wav

1.2.3 控制系统音量

PLAIN

# 设置系统音量为50%
pactl set-sink-volume @DEFAULT_SINK@ 50%

1.3 使用 Python 开发音频应用

在命令行快速验证之后，如果需要将音频能力集成到自己的应用中，可以使用下面的 Python 示例。包含三个独立脚本：系统音量控制、麦克风录音、音频播放，均可直接复制运行。

1.3.1 系统音量控制

通过 pactl 命令封装，支持查询和设置系统主音量（0-100）。

文件名：system_volume_example.py

实现说明：脚本封装两个核心函数，通过调用系统 pactl 命令完成音量控制（完整源码见随附的 system_volume_example.py 示例文件）。

函数	功能	实现要点
`get_system_volume()`	获取当前默认音量百分比	执行 `pactl get-sink-volume` 查询默认输出设备，从输出文本中提取第一个百分比值并返回整数
`set_system_volume(volume)`	设置系统主音量（0-100）	执行 `pactl set-sink-volume`，将指定百分比写入默认输出设备

命令行参数（基于 argparse）：

参数	说明
`volume`（可选位置参数）	目标音量，0-100 的整数；不传且未指定 -g 时，默认打印当前音量
`-g` / `--get`	仅获取当前系统主音量

底层命令（脚本内部实际调用的系统命令，可直接在终端使用）：

Bash

# 查询默认输出设备当前音量
pactl get-sink-volume @DEFAULT_SINK@

# 设置系统主音量为 50%
pactl set-sink-volume @DEFAULT_SINK@ 50%

命令行运行：

Bash

# 查询系统音量
python3 system_volume_example.py

# 设置系统音量，有效值0-100，下方将设置为50
python3 system_volume_example.py 50

1.3.2 麦克风录音（PyAudio 异步采集）

基于 PyAudio 实现异步麦克风录音，录制结束自动生成 WAV 音频文件。示例包含完整的 Recorder 类，支持异步回调采集、队列缓存、浮点归一化、声道提取和 WAV 文件保存。

核心配置参数：

配置项	默认值	说明
`recorder_device_index`	None	麦克风设备编号，None 使用系统默认
`recorder_audio_rate`	16000	采样率 16000 Hz（语音识别常用）
`recorder_chunk_size`	1024	单次采集帧数
`recorder_channels`	1	硬件总声道数
`recorder_pickup_channels`	[0]	选取的声道索引列表

Recorder 类主要方法：

方法	说明
`stream_callback`	音频缓冲区满时自动触发的后台回调（独立子线程），进行 int16→float32 转换和声道提取
`get()`	主线程调用，从队列获取一帧音频浮点数据（最多等待 0.5 秒）
`clear_queue()`	清空队列所有历史缓存音频
`stop_recording()`	停止音频采集流
`close()`	释放所有音频硬件资源

WAV 文件保存： 录音完成后，通过 save_wav() 函数将浮点音频数组还原为 16 位整型并写入 WAV 文件（16 位音频固定占 2 字节）。

文件名：record_example.py

Python

"""
完整音频采集+录音保存示例
基于PyAudio实现异步麦克风录音，录制结束自动生成wav音频文件
"""
import argparse
import queue
import numpy as np
import pyaudio
import wave
import time

class Recorder:
    def __init__(self, config):
        """Initialize audio recorder
        Args:
            config (dict): configuration dictionary
        """
        # 保存配置参数
        self.config = config
        # 线程安全队列：缓存后台采集的音频帧
        self.recorder_queue = queue.Queue()
        # 录音设备编号，选择指定麦克风
        self.recorder_device_index = config.get("recorder_device_index", None)
        # 音频采样率，默认16000Hz（语音识别常用）
        self.recorder_rate = config.get("recorder_audio_rate", 16000)
        # 单次采集音频帧数
        self.recorder_chunk_size = config.get("recorder_chunk_size", 1024)
        # 麦克风硬件总声道数
        self.recorder_channels = config.get("recorder_channels", 1)
        # 需要提取的声道索引列表
        self.recorder_pickup_channels = config["recorder_pickup_channels"]
        # 播放状态标记：播放音频时置True，屏蔽麦克风收音防啸叫
        self.is_playing = False

        # 初始化PyAudio音频实例
        self.p = pyaudio.PyAudio()
        # 打开麦克风输入流，注册异步回调采集
        self.stream = self.p.open(
            format=pyaudio.paInt16,
            channels=self.recorder_channels,
            rate=self.recorder_rate,
            input=True,
            input_device_index=self.recorder_device_index,
            frames_per_buffer=self.recorder_chunk_size,
            stream_callback=self.stream_callback,
        )

    def stream_callback(self, in_data, frame_count, time_info, status):
        """音频缓冲区满自动触发的后台回调函数（独立子线程运行）"""
        # 如果正在播放声音，直接跳过录音处理，防止啸叫回声
        if self.is_playing:
            return (None, pyaudio.paContinue)

        try:
            # 原始二进制字节流转16位整型数组
            audio_data = np.frombuffer(in_data, dtype=np.int16)
            # 重塑维度：(单次帧数, 总声道数)
            audio_reshaped = audio_data.reshape(
                self.recorder_chunk_size, self.recorder_channels
            )
            # 归一化：int16范围[-32768,32767] 转为浮点[-1.0, 1.0]
            audio_float = audio_reshaped.astype(np.float32) / 32768.0
            # 提取配置中指定的声道数据
            audio_picked = audio_float[:, self.recorder_pickup_channels]
            # 非阻塞写入队列，队列满则丢弃当前帧，避免声卡阻塞
            self.recorder_queue.put(audio_picked, block=False)
        except queue.Full:
            # 队列满直接丢弃，不抛出异常
            pass

        # 告知声卡继续采集下一段音频
        return (None, pyaudio.paContinue)

    def get(self):
        """主线程调用：从队列获取一帧音频浮点数据"""
        try:
            # 最多等待0.5秒获取音频
            data = self.recorder_queue.get(timeout=0.5)
            return data
        except queue.Empty:
            # 无音频返回静音0数组兜底
            return np.zeros(self.recorder_chunk_size, dtype=np.float32)

    def clear_queue(self):
        """清空队列所有历史缓存音频"""
        while not self.recorder_queue.empty():
            try:
                self.recorder_queue.get_nowait()
            except queue.Empty:
                break

    def stop_recording(self):
        """停止音频采集流"""
        self.stream.stop_stream()
        self.stream.close()

    def close(self):
        """释放所有音频硬件资源"""
        self.stop_recording()
        self.p.terminate()

def save_wav(audio_frames, sample_rate, channels, save_path):
    """
    将采集的浮点音频数组保存为wav文件
    :param audio_frames: 所有音频帧列表
    :param sample_rate: 采样率
    :param channels: 保存的声道数量
    :param save_path: 音频保存路径
    """
    # 拼接所有音频片段
    all_audio = np.concatenate(audio_frames, axis=0)
    # 浮点数据还原为原始16位整型
    audio_int16 = (all_audio * 32768).astype(np.int16)

    # 创建wav文件写入对象
    wf = wave.open(save_path, 'wb')
    wf.setnchannels(channels)
    wf.setsampwidth(2)  # 16位音频固定占2字节
    wf.setframerate(sample_rate)
    # 写入二进制音频数据
    wf.writeframes(audio_int16.tobytes())
    wf.close()
    print(f"✅ 录音保存成功，文件路径：{save_path}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="麦克风录音并保存为 wav")
    parser.add_argument(
        "-o", "--output",
        default="./my_record.wav",
        help="录音保存文件路径（默认: ./my_record.wav）",
    )
    parser.add_argument(
        "-d", "--duration",
        type=float,
        default=5,
        help="录制时长，单位秒（默认: 5）",
    )
    args = parser.parse_args()

    # ===================== 录音配置 =====================
    recorder_config = {
        "recorder_device_index": None,        # 麦克风设备编号，None使用系统默认麦克风
        "recorder_audio_rate": 16000,          # 采样率16000Hz
        "recorder_chunk_size": 1024,           # 单次采集帧数
        "recorder_channels": 1,                # 硬件总声道数：单声道麦克风
        "recorder_pickup_channels": [0]        # 选取第0声道录音
    }
    record_seconds = args.duration
    save_file_path = args.output

    # 1. 初始化录音器
    recorder = Recorder(recorder_config)
    print(f"🎙️ 开始录音，将录制{record_seconds}秒...")

    # 缓存所有采集到的音频帧
    audio_buffer = []
    start_time = time.time()

    # 2. 循环采集音频
    while time.time() - start_time < record_seconds:
        frame_data = recorder.get()
        audio_buffer.append(frame_data)

    # 3. 录音结束，释放硬件资源
    recorder.close()
    print("录音结束，正在保存音频...")

    # 4. 保存音频到本地wav文件
    save_wav(
        audio_frames=audio_buffer,
        sample_rate=recorder_config["recorder_audio_rate"],
        channels=len(recorder_config["recorder_pickup_channels"]),
        save_path=save_file_path
    )

命令行运行：

Bash

# -o 表示指定输出的录音文件名称 -d表示录音的时长，单位：秒
python3 record_example.py -o my_record.wav -d 10

执行上述脚本，将开始 10 秒的音频录制，并保存一份 my_record.wav 文件在执行目录下。

1.3.3 播放指定音频

基于 pygame.mixer 实现本地音频文件播放，支持加载、播放、暂停、恢复、音量调节、停止等交互控制。

文件名：play_example.py

Python

import argparse

from pygame import mixer

def main():
    parser = argparse.ArgumentParser(description="播放本地音频文件")
    parser.add_argument("audio_file", help="要播放的音频文件路径")
    args = parser.parse_args()
    audio_file = args.audio_file

    # 1. 初始化 mixer 模块
    mixer.init()

    try:
        # 2. 加载音频文件
        mixer.music.load(audio_file)
        print(f"成功加载音频文件: {audio_file}")
    except Exception as e:
        print(f"加载音频失败，请检查文件路径: {e}")
        return

    # 默认音量 (范围 0.0 到 1.0)
    current_volume = 0.5
    mixer.music.set_volume(current_volume)

    # 3. 开始播放
    mixer.music.play()
    print("\n--- 音频播放控制指令 ---")
    print(" [p] 暂停 (Pause)")
    print(" [r] 恢复播放 (Resume)")
    print(" [+] 音量增加 (+10%)")
    print(" [-] 音量减少 (-10%)")
    print(" [s] 停止播放 (Stop)")
    print(" [q] 退出程序")

    # 交互控制循环
    while True:
        cmd = input("\n请输入指令: ").strip().lower()

        if cmd == "p":
            mixer.music.pause()
            print("⏸️  已暂停")

        elif cmd == "r":
            mixer.music.unpause()
            print("▶️  已恢复播放")

        elif cmd == "+":
            current_volume = min(1.0, current_volume + 0.1)
            mixer.music.set_volume(current_volume)
            print(f"🔊 音量调高: {int(current_volume * 100)}%")

        elif cmd == "-":
            current_volume = max(0.0, current_volume - 0.1)
            mixer.music.set_volume(current_volume)
            print(f"🔉 音量调低: {int(current_volume * 100)}%")

        elif cmd == "s":
            mixer.music.stop()
            print("⏹️  已停止播放")

        elif cmd == "q":
            mixer.music.stop()
            mixer.quit()
            print("👋 已退出程序")
            break

        else:
            print("⚠️ 未知指令，请输入 p, r, +, -, s 或 q")

if __name__ == "__main__":
    main()

命令行运行：

Bash

python3 play_example.py mp3_example.mp3

1.4 小结

本章覆盖了机器人音频开发的完整链路：

环境准备：SSH 连接、系统依赖安装、虚拟环境与 Python 包安装；
命令行快速操作：arecord 录音、aplay 回放、pactl 音量控制，无需写代码即可验证硬件；
Python 完整示例：三个独立可运行的脚本——系统音量控制、PyAudio 异步麦克风录音、pygame 音频播放，可直接集成到上层应用中。

如需了解更多 ROS2 接口示例（运动控制、LED 控制等），可参考 GitHub 示例代码目录。