随笔十六、音频采集、UDP发送

此功能是远程对讲的一部分，由泰山派实时采集语音，然后UDP发送到远端。泰山派硬件使用RK809-5管理内核电源（PMIC），此IC同时具备音频编解码器（CODEC）功能，接口I2S1。

现在到处都是大模型，编写程序就变得简单多了，向大模型提出需求，程序就写好了。当然，需要修改是难免的，但少了许多码字时间。

为了提升效率，降低CPU占用，需要合理安排对底层接口的频繁调用，一些参数还是需要进一步磨合调试。泰山派这部分程序使用2个线程，分别处理音频的读取和UDP发送。主要是考虑UDP一帧数据控制在512字节，不用拆包。而声卡音频采集一个周期数据太少会浪费CPU。数据交换使用FIFO，参考linux内核函数处理。ALSA库。

#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <alsa/asoundlib.h>
#include <kfifo.h>


#define DEST_IP             "192.168.1.44"
#define DEST_PORT           9999

#define UDP_SIZE            512

#define SAMPLE_RATE         44100
#define CHANNELS            2
#define FORMAT              SND_PCM_FORMAT_S16_LE
#define FRAMES_PER_PERIOD   1024
#define PERIODS             2

#define FRAME_SIZE          (CHANNELS*snd_pcm_format_physical_width(FORMAT)/8)
#define BUFF_NUM            (1<<2)
#define BUFF_SIZE           (FRAMES_PER_PERIOD*FRAME_SIZE)

#if 0
// 定义WAV文件头结构体
typedef struct {
  char chunk_id[4];         // "RIFF"
  int chunk_size;           // 文件总大小 - 8
  char format[4];           // "WAVE"
  char subchunk1_id[4];     // "fmt "
  int subchunk1_size;       // 16 for PCM
  short audio_format;       // 1 for PCM
  short num_channels;       // 声道数
  int sample_rate;          // 采样率
  int byte_rate;            // 每秒的字节数
  short block_align;        // 每个样本的字节数
  short bits_per_sample;    // 每个样本的位数
  char subchunk2_id[4];     // "data"
  int subchunk2_size;       // 音频数据大小
} WavHeader;
#endif

void* udpPut(void* arg);    // 线程函数，发送数据
void* pcmGet(void* arg);    // 线程函数，采集数据

Kfifo_st* fifo;

// 主程序
int main()
{
  pthread_t tid[2];

  // 申请FIFO空间
  if ((fifo = Kfifo_Init(BUFF_SIZE * BUFF_NUM)) == NULL) {
    fprintf(stderr, "failed to allocate fifo memory\n");
    exit(1);
  }

  fprintf(stdout, "PCM: %dHz, %dch, %dbits\n", SAMPLE_RATE, CHANNELS, snd_pcm_format_physical_width(FORMAT));

  pthread_create(&tid[0], NULL, pcmGet, NULL);
  pthread_create(&tid[1], NULL, udpPut, NULL);

  pthread_join(tid[0], NULL);
  pthread_join(tid[1], NULL);

  Kfifo_Free(fifo);

  return 0;
}

// 采集音频数据
void* pcmGet(void* arg)
{
  int err;
  snd_pcm_t* handle;
  snd_pcm_hw_params_t* params;
  unsigned int rate = SAMPLE_RATE;
  snd_pcm_uframes_t frames = FRAMES_PER_PERIOD;

  // 申请一个采样周期缓冲区
  int size = BUFF_SIZE;
  char* buffer = (char*)malloc(size);
  if (buffer == NULL) {
    fprintf(stderr, "failed to allocate buffer memory\n");
    exit(1);
  }

#if 0
  // 打开文件进行写入
  FILE* file;
  file = fopen("talk_out.wav", "wb");
  if (file == NULL) {
    fprintf(stderr, "failed to open file for writing\n");
    free(buffer);
    exit(1);
  }

  // 初始化WAV文件头
  WavHeader header;
  int sample_counter = 0;
  int total_samples = 500;
  int data_size = total_samples * BUFF_SIZE;
  int file_size = data_size + sizeof(WavHeader);

  // 填充WAV文件头
  memcpy(header.chunk_id, "RIFF", 4);
  header.chunk_size = file_size - 8;
  memcpy(header.format, "WAVE", 4);
  memcpy(header.subchunk1_id, "fmt ", 4);
  header.subchunk1_size = 16;
  header.audio_format = 1;
  header.num_channels = CHANNELS;
  header.sample_rate = SAMPLE_RATE;
  header.byte_rate = SAMPLE_RATE * FRAME_SIZE;
  header.block_align = FRAME_SIZE;
  header.bits_per_sample = snd_pcm_format_physical_width(FORMAT);
  memcpy(header.subchunk2_id, "data", 4);
  header.subchunk2_size = data_size;

  // 写入WAV文件头
  fwrite(&header, sizeof(WavHeader), 1, file);
#endif

  // 打开音频捕获设备
  if ((err = snd_pcm_open(&handle, "default", SND_PCM_STREAM_CAPTURE, 0)) < 0) {
    fprintf(stderr, "snd_pcm_open: %s\n", snd_strerror(err));
    exit(1);
  }

  // 分配硬件参数结构
  if ((err = snd_pcm_hw_params_malloc(&params)) < 0) {
    fprintf(stderr, "snd_pcm_hw_params_malloc: %s\n", snd_strerror(err));
    exit(1);
  }

  // 初始化硬件参数结构
  if ((err = snd_pcm_hw_params_any(handle, params)) < 0) {
    fprintf(stderr, "snd_pcm_hw_params_any: %s\n", snd_strerror(err));
    exit(1);
  }

  // 设置访问类型为交错模式
  if ((err = snd_pcm_hw_params_set_access(handle, params, SND_PCM_ACCESS_RW_INTERLEAVED)) < 0) {
    fprintf(stderr, "snd_pcm_hw_params_set_access: %s\n", snd_strerror(err));
    exit(1);
  }

  // 设置采样格式
  if ((err = snd_pcm_hw_params_set_format(handle, params, FORMAT)) < 0) {
    fprintf(stderr, "snd_pcm_hw_params_set_format: %s\n", snd_strerror(err));
    exit(1);
  }

  // 设置声道数
  if ((err = snd_pcm_hw_params_set_channels(handle, params, CHANNELS)) < 0) {
    fprintf(stderr, "snd_pcm_hw_params_set_channels: %s\n", snd_strerror(err));
    exit(1);
  }

  // 设置采样率
  if ((err = snd_pcm_hw_params_set_rate_near(handle, params, &rate, 0)) < 0) {
    fprintf(stderr, "snd_pcm_hw_params_set_rate_near: %s\n", snd_strerror(err));
    exit(1);
  }

  // 设置周期数
  if ((err = snd_pcm_hw_params_set_periods(handle, params, PERIODS, 0)) < 0) {
    fprintf(stderr, "snd_pcm_hw_params_set_periods: %s\n", snd_strerror(err));
    exit(1);
  }

  // 设置周期大小
  if ((err = snd_pcm_hw_params_set_period_size_near(handle, params, &frames, 0)) < 0) {
    fprintf(stderr, "snd_pcm_hw_params_set_period_size_near: %s\n", snd_strerror(err));
    exit(1);
  }

  // 将硬件参数应用到设备上
  if ((err = snd_pcm_hw_params(handle, params)) < 0) {
    fprintf(stderr, "snd_pcm_hw_params: %s\n", snd_strerror(err));
    exit(1);
  }

  // 释放硬件参数结构
  snd_pcm_hw_params_free(params);

  // 准备音频捕获
  if ((err = snd_pcm_prepare(handle)) < 0) {
    fprintf(stderr, "snd_pcm_prepare: %s\n", snd_strerror(err));
    exit(1);
  }

  // 采样循环
  for (;;)
  {
    // 读取一个周期数据
    err = snd_pcm_readi(handle, buffer, frames);
    if (err != frames) {
      if (err == -EPIPE) {
        snd_pcm_prepare(handle);
      }
      else {
        fprintf(stderr, "error reading audio data: %s\n", snd_strerror(err));
      }
      continue;;
    }

    // 填充数据到FIFO
    Kfifo_In(fifo, buffer, size, 1);
    usleep(1000);

#if 0
    if (++sample_counter == total_samples) {
      fclose(file);
      printf("ok\n");
    }
    else {
      fwrite(buffer, 1, size, file);
    }
#endif
  }

  free(buffer);
  snd_pcm_drain(handle);
  snd_pcm_close(handle);
}

// 发送音频数据
void* udpPut(void* arg)
{
  // UDP一帧大小控制在512字节
  char buff[UDP_SIZE] = { 0 };

  int sockfd;
  if ((sockfd = socket(AF_INET, SOCK_DGRAM, 0)) < 0) {
    perror("socket creation failed");
    exit(1);
  }

  // struct sockaddr_in local_addr;
  // socklen_t addrlen = sizeof(local_addr);
  // bzero(&local_addr, sizeof(local_addr));
  // local_addr.sin_family = AF_INET;
  // local_addr.sin_addr.s_addr = htonl(INADDR_ANY);
  // local_addr.sin_port = htons(9999);

  // // 绑定套接字到地址
  // if (bind(sockfd, (struct sockaddr*)&local_addr, sizeof(local_addr)) < 0) {
  //   perror("bind");
  //   close(sockfd);
  //   exit(1);
  // }

  // // 获取本地地址信息
  // if (getsockname(sockfd, (struct sockaddr*)&local_addr, &addrlen) < 0) {
  //   perror("getsockname failed");
  //   close(sockfd);
  //   exit(1);
  // }
  // // 提取本地端口号
  // unsigned short local_port = ntohs(local_addr.sin_port);
  // printf("UDP发送端口号: %hu\n", local_port);

  struct sockaddr_in dest;
  bzero(&dest, sizeof(dest));
  dest.sin_family = AF_INET;
  dest.sin_port = htons(DEST_PORT);
  dest.sin_addr.s_addr = inet_addr(DEST_IP);

  const struct sockaddr* remote_addr = (struct sockaddr*)(&dest);

  for (;;)
  {
    if (Kfifo_Len(fifo) >= UDP_SIZE) {
      // FIFO存在数据则读取
      Kfifo_Out(fifo, buff, UDP_SIZE);

      // 发送到远端
      sendto(sockfd, buff, sizeof(buff), 0, remote_addr, sizeof(struct sockaddr));
    }

    usleep(1000);
  }
}

远端接收就用电脑来实现，尝试用Rust写个程序（当然也是让大模型帮着写啦）

use std::net::UdpSocket;
use rodio::{OutputStream, Sink};
use std::io;
use crossterm::{
    event::{self, Event, KeyCode},
    terminal::{disable_raw_mode, enable_raw_mode},
};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

fn main() -> io::Result<()> {
    // 创建UDP套接字并绑定到本地地址和端口
    let socket = UdpSocket::bind("0.0.0.0:9999")?;
    println!("等待来自UDP的PCM (i16) 数据...");

    // 创建输出流和Sink用于播放声音
    let (_stream, stream_handle) = OutputStream::try_default().unwrap();
    let sink = Sink::try_new(&stream_handle).unwrap();

    // 设置为非阻塞模式以便可以立即开始播放收到的数据
    sink.set_volume(1.0); // 可选：设置音量

    // 准备接收缓冲区
    let mut buf = vec![0u8; 4096]; // 一般4096足够小以保持低延迟，同时足够大以减少系统调用次数

    // 原子布尔值用于控制循环
    let running = Arc::new(AtomicBool::new(true));
    let r = running.clone();

    // 启用原始模式以捕获键盘事件
    enable_raw_mode().expect("无法启用raw mode");

    // 在另一个线程中监听键盘事件
    std::thread::spawn(move || {
        while r.load(Ordering::Relaxed) {
            if event::poll(std::time::Duration::from_millis(100)).unwrap() {
                if let Event::Key(key_event) = event::read().unwrap() {
                    match key_event.code {
                        KeyCode::Esc | KeyCode::Char(_) => {
                            println!("检测到按键，程序即将退出...");
                            r.store(false, Ordering::Relaxed);
                            break;
                        },
                        _ => {}
                    }
                }
            }
        }

        // 程序结束时恢复终端模式
        disable_raw_mode().expect("无法禁用raw mode");
    });

    loop {
        if !running.load(Ordering::Relaxed) {
            break;
        }

        // 接收UDP数据报文
        match socket.recv_from(&mut buf) {
            Ok((size, addr)) => {
                //println!("从 {} 收到了 {} 字节的数据", addr, size);

                if size % 2 != 0 {
                    eprintln!("收到的数据大小不是16位样本的整数倍，可能有误！");
                    continue;
                }

                // 将字节数组转换为i16切片，假设是小端字节序
                let samples = unsafe {
                    std::slice::from_raw_parts(
                        buf.as_ptr() as *const i16,
                        size / 2
                    )
                };

                // 将i16样本添加到播放队列中
                let sample_buffer = rodio::buffer::SamplesBuffer::new(
                    2, // 两声道
                    44100, // 采样率为44100Hz
                    samples.to_vec(), // 复制数据
                );
                sink.append(sample_buffer);
            }
            Err(e) => eprintln!("遇到错误: {}", e),
        }
    }

    // 清理工作
    drop(sink);
    drop(stream_handle);
    println!("程序已退出。");

    Ok(())
}

cargo.toml

[package]
name = "demo"
version = "0.1.0"
edition = "2021"

[dependencies]
rodio = { version = "0.20.1"}
tokio = { version = "1.25.0", features = ["full"] }
crossterm = "0.28.1"

效果还行