Hass/open-xiaoai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: 整理 AudioConfig

Browse Source
This commit is contained in:
Del Wang
2025-12-31 22:53:33 +08:00
committed by Del
parent 4bce9e8d23
commit 316f61d0bb
7 changed files with 94 additions and 377 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/hello/src/audio/codec.rs
+33 -8
View File
@@ -1,4 +1,4 @@
use crate::config::AudioConfig;
use crate::config::{AudioConfig, AudioScene};
use anyhow::{Context, Result};
use opus::{Application, Bitrate, Channels, Decoder, Encoder};
@@ -9,14 +9,32 @@ pub struct OpusCodec {
impl OpusCodec {
pub fn new(config: &AudioConfig) -> Result<Self> {
let channels = if config.channels == 1 {
Channels::Mono
} else {
Channels::Stereo
let channels = match config.channels {
1 => Channels::Mono,
2 => Channels::Stereo,
_ => return Err(anyhow::anyhow!("Invalid channels: {}", config.channels)),
};
let mut encoder = Encoder::new(config.sample_rate, channels, Application::Audio)
let mode = match config.audio_scene {
AudioScene::Music => Application::Audio,
AudioScene::Voice => Application::Voip,
};
let bitrate = match config.bitrate {
-1 => Bitrate::Max,
0 => Bitrate::Auto,
_ => Bitrate::Bits(config.bitrate),
};
let mut encoder = Encoder::new(config.sample_rate, channels, mode)
.context("Failed to create Opus encoder")?;
encoder.set_bitrate(Bitrate::Bits(config.bitrate))?;
encoder.set_bitrate(bitrate)?;
if config.vbr {
encoder.set_vbr(true)?;
}
if config.fec {
encoder.set_inband_fec(true)?; // 内联前向纠错
encoder.set_packet_loss_perc(20)?; // 预期丢包率20%
}
let decoder =
Decoder::new(config.sample_rate, channels).context("Failed to create Opus decoder")?;
@@ -36,7 +54,14 @@ impl OpusCodec {
.context("Opus decoding failed")
}
// Packet Loss Concealment
/// 前向纠错(FEC)
pub fn decode_fec(&mut self, opus: &[u8], out: &mut [i16]) -> Result<usize> {
self.decoder
.decode(opus, out, true)
.context("Opus FEC decoding failed")
}
/// 丢包补偿(PLC)
pub fn decode_loss(&mut self, out: &mut [i16]) -> Result<usize> {
self.decoder
.decode(&[], out, false)
apps/hello/src/bin/client.rs
-123
View File
@@ -1,123 +0,0 @@
#![cfg(target_os = "linux")]
use anyhow::Result;
use hello::audio::{AudioPlayer, OpusCodec};
use hello::config::AudioConfig;
use hello::net::{AudioPacket, ChannelRole, ControlPacket, Discovery};
use hello::sync::{ClockSync, now_us};
use std::collections::VecDeque;
use std::env;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
#[tokio::main]
async fn main() -> Result<()> {
let args: Vec<String> = env::args().collect();
if args.len() < 2 {
eprintln!("Usage: {} [left|right]", args[0]);
return Ok(());
}
let role = if args[1] == "left" {
ChannelRole::Left
} else {
ChannelRole::Right
};
let config = AudioConfig {
sample_rate: 48000,
channels: 1,
frame_size: 960,
bitrate: 32000,
..AudioConfig::default()
};
println!("Searching for server...");
let server_addr = Discovery::client_discover_server().await?;
let mut stream = TcpStream::connect(server_addr).await?;
// 1. Identify
stream
.write_all(&postcard::to_allocvec(&ControlPacket::ClientIdentify {
role: role.clone(),
})?)
.await?;
// 2. Sync
let mut clock = ClockSync::new();
let t1 = now_us();
stream
.write_all(&postcard::to_allocvec(&ControlPacket::Ping {
client_ts: t1,
})?)
.await?;
let mut buf = [0u8; 1024];
let n = stream.read(&mut buf).await?;
if let Ok(ControlPacket::Pong {
client_ts,
server_ts,
..
}) = postcard::from_bytes::<ControlPacket>(&buf[..n])
{
let t4 = now_us();
clock.update(client_ts, server_ts, t4);
println!(
"Clock synced. Offset: {}us, RTT: {}us",
clock.offset,
t4 - t1
);
}
let player = AudioPlayer::new(&config)?;
let mut codec = OpusCodec::new(&config)?;
let mut jitter_buffer: VecDeque<AudioPacket> = VecDeque::new();
let mut pcm_buf = vec![0i16; config.frame_size];
let (tx, mut rx) = tokio::sync::mpsc::channel(100);
tokio::spawn(async move {
loop {
let mut s_buf = [0u8; 4];
if stream.read_exact(&mut s_buf).await.is_err() {
break;
}
let size = u32::from_be_bytes(s_buf) as usize;
let mut data = vec![0u8; size];
if stream.read_exact(&mut data).await.is_err() {
break;
}
if let Ok(p) = postcard::from_bytes::<AudioPacket>(&data) {
let _ = tx.send(p).await;
}
}
});
println!("Streaming started. Playing channel {:?}", role);
loop {
while let Ok(p) = rx.try_recv() {
jitter_buffer.push_back(p);
}
if let Some(p) = jitter_buffer.front() {
let target_client_time = clock.to_client_time(p.timestamp);
let now = now_us();
if now >= target_client_time {
let packet = jitter_buffer.pop_front().unwrap();
let len = codec.decode(&packet.data, &mut pcm_buf)?;
player.write(&pcm_buf[..len])?;
} else if target_client_time - now > 500_000 {
// Too far in the future, maybe clock jumped?
jitter_buffer.pop_front();
} else {
// Wait until it's time
let wait = (target_client_time - now) as u64;
if wait > 1000 {
tokio::time::sleep(Duration::from_micros(wait)).await;
}
}
} else {
tokio::time::sleep(Duration::from_millis(5)).await;
}
}
}
apps/hello/src/bin/server.rs
-155
View File
@@ -1,155 +0,0 @@
use anyhow::Result;
use hello::audio::{AudioReader, OpusCodec};
use hello::config::AudioConfig;
use hello::net::{AudioPacket, ChannelRole, ControlPacket, Discovery, SERVER_PORT};
use hello::sync::now_us;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::Mutex;
#[tokio::main]
async fn main() -> Result<()> {
let config = AudioConfig {
sample_rate: 48000,
channels: 1,
frame_size: 960, // 20ms at 48kHz
bitrate: 32000, // 32kbps
..AudioConfig::default()
};
println!("Starting Stereo Server on port {}", SERVER_PORT);
Discovery::start_server_beacon().await?;
let listener = TcpListener::bind(format!("0.0.0.0:{}", SERVER_PORT)).await?;
let clients: Arc<Mutex<HashMap<ChannelRole, TcpStream>>> = Arc::new(Mutex::new(HashMap::new()));
let clients_clone = Arc::clone(&clients);
tokio::spawn(async move {
while let Ok((mut stream, addr)) = listener.accept().await {
let mut buf = [0u8; 1024];
if let Ok(n) = stream.read(&mut buf).await {
if let Ok(packet) = postcard::from_bytes::<ControlPacket>(&buf[..n]) {
match packet {
ControlPacket::ClientIdentify { role } => {
println!("Client {} identified as {:?}", addr, role);
// Simple handshake for sync
let mut sync_buf = [0u8; 1024];
if let Ok(sn) = stream.read(&mut sync_buf).await {
if let Ok(ControlPacket::Ping { client_ts }) =
postcard::from_bytes::<ControlPacket>(&sync_buf[..sn])
{
let pong = ControlPacket::Pong {
client_ts,
server_ts: now_us(),
};
let pong_msg = postcard::to_allocvec(&pong).unwrap();
let _ = stream.write_all(&pong_msg).await;
clients_clone.lock().await.insert(role, stream);
}
}
}
_ => {}
}
}
}
}
});
println!("Waiting for Left and Right clients...");
loop {
if clients.lock().await.len() >= 2 {
break;
}
tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
}
println!("Both clients connected. Starting stream in 1s...");
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
stream_audio("test.mp3", &config, clients).await?;
Ok(())
}
async fn stream_audio(
path: &str,
config: &AudioConfig,
clients: Arc<Mutex<HashMap<ChannelRole, TcpStream>>>,
) -> Result<()> {
let mut reader = AudioReader::new(path)?;
let mut left_codec = OpusCodec::new(config)?;
let mut right_codec = OpusCodec::new(config)?;
let start_time = now_us() + 500_000;
let mut current_frame_ts = start_time;
let frame_duration_us =
(config.frame_size as f64 / config.sample_rate as f64 * 1_000_000.0) as u128;
while let Some((left_pcm, right_pcm)) = reader.read_chunk(config.frame_size)? {
if left_pcm.len() < config.frame_size {
break;
}
send_packets(
&clients,
&mut left_codec,
&mut right_codec,
&left_pcm,
&right_pcm,
current_frame_ts,
)
.await?;
current_frame_ts += frame_duration_us;
pace(current_frame_ts).await;
}
Ok(())
}
async fn send_packets(
clients: &Arc<Mutex<HashMap<ChannelRole, TcpStream>>>,
left_codec: &mut OpusCodec,
right_codec: &mut OpusCodec,
left_pcm: &[i16],
right_pcm: &[i16],
ts: u128,
) -> Result<()> {
let mut left_opus = vec![0u8; 1275];
let mut right_opus = vec![0u8; 1275];
let l_len = left_codec.encode(left_pcm, &mut left_opus)?;
let r_len = right_codec.encode(right_pcm, &mut right_opus)?;
let l_p = postcard::to_allocvec(&AudioPacket {
timestamp: ts,
data: left_opus[..l_len].to_vec(),
})?;
let r_p = postcard::to_allocvec(&AudioPacket {
timestamp: ts,
data: right_opus[..r_len].to_vec(),
})?;
let mut lock = clients.lock().await;
if let Some(s) = lock.get_mut(&ChannelRole::Left) {
let _ = send_with_size(s, &l_p).await;
}
if let Some(s) = lock.get_mut(&ChannelRole::Right) {
let _ = send_with_size(s, &r_p).await;
}
Ok(())
}
async fn pace(current_frame_ts: u128) {
let now = now_us();
if current_frame_ts > now + 1_000_000 {
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
}
}
async fn send_with_size(stream: &mut TcpStream, data: &[u8]) -> Result<()> {
stream.write_all(&(data.len() as u32).to_be_bytes()).await?;
stream.write_all(data).await?;
Ok(())
}
apps/hello/src/bin/test.rs
-62
View File
@@ -1,62 +0,0 @@
#![cfg(target_os = "linux")]
use anyhow::Result;
use hello::audio::{AudioPlayer, AudioRecorder, OpusCodec};
use hello::config::AudioConfig;
use std::collections::VecDeque;
use std::time::{Duration, Instant};
fn main() -> Result<()> {
let config = AudioConfig::default();
println!("Starting Audio Modular Demo with 1s delay");
println!(
"Config: capture={}, playback={}, rate={}, channels={}, frame_size={}",
config.capture_device,
config.playback_device,
config.sample_rate,
config.channels,
config.frame_size
);
let recorder = AudioRecorder::new(&config)?;
let player = AudioPlayer::new(&config)?;
let mut codec = OpusCodec::new(&config)?;
// 使用队列存储编码后的 Opus 数据块及录制时间
let mut delay_queue: VecDeque<(Instant, Vec<u8>)> = VecDeque::new();
let delay_duration = Duration::from_secs(1);
let mut pcm_in = vec![0i16; config.frame_size * config.channels as usize];
let mut opus_buf = vec![0u8; 1024];
println!("Recording... Delaying playback by 1s. Press Ctrl+C to stop.");
loop {
// 1. 录制原始 PCM
let read = recorder.read(&mut pcm_in)?;
if read == config.frame_size {
// 2. 编码并存入队列
let opus_len = codec.encode(&pcm_in, &mut opus_buf)?;
let opus_data = opus_buf[..opus_len].to_vec();
delay_queue.push_back((Instant::now(), opus_data));
}
// 3. 处理延迟播放：检查队列头部数据是否已等待超过 1s
while let Some((timestamp, _)) = delay_queue.front() {
if timestamp.elapsed() >= delay_duration {
let (_, opus_data) = delay_queue.pop_front().unwrap();
let mut pcm_out = vec![0i16; config.frame_size * config.channels as usize];
let decoded_len = codec.decode(&opus_data, &mut pcm_out)?;
if decoded_len == config.frame_size {
// 4. 播放还原后的 PCM
player.write(&pcm_out)?;
}
} else {
// 头部数据还未到 1s，由于队列是按时间排序的，后面的肯定也没到
break;
}
}
}
}
apps/hello/src/config.rs
+42 -7
View File
@@ -1,26 +1,61 @@
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum AudioScene {
Music,
Voice,
}
#[derive(Debug, Clone)]
pub struct AudioConfig {
// ALSA 设备参数，用于录音和播放
pub capture_device: String,
pub playback_device: String,
pub sample_rate: u32,
pub channels: u16,
pub frame_size: usize,
pub frame_size: usize, // 帧大小，单位为采样点
// Opus 编解码参数，用于音频传输
pub audio_scene: AudioScene,
pub bitrate: i32,
pub vbr: bool, // 是否启用 VBR（动态比特率）
pub fec: bool, // 是否启用 FEC（内联前向纠错）
}
impl AudioConfig {
pub fn music() -> Self {
Self {
audio_scene: AudioScene::Music,
sample_rate: 48_000, // 48kHz
channels: 2,
frame_size: 960, // 20ms at 48kHz
bitrate: 320_000, // 320 kbps
..Default::default()
}
}
pub fn voice() -> Self {
Self {
audio_scene: AudioScene::Voice,
sample_rate: 16_000, // 16kHz
channels: 1,
frame_size: 320, // 20ms at 16kHz
bitrate: 32_000, // 32 kbps
..Default::default()
}
}
}
impl Default for AudioConfig {
fn default() -> Self {
Self {
// 根据 apps/runtime/squashfs-root/etc/asound.conf 分析：
// 录音设备对应 pcm.Capture -> hw:0,3 (S32_LE, 48000Hz)
// 播放设备对应 pcm.dmixer -> hw:0,2 (S16_LE, 48000Hz)
// 使用 "default" 或 "plug" 设备可以由 ALSA 自动处理采样率和格式转换
audio_scene: AudioScene::Voice,
capture_device: "plug:Capture".to_string(),
playback_device: "default".to_string(),
sample_rate: 16000, // Opus 常用 16k，ALSA plug 会自动从 48k 转换
sample_rate: 16_000,
channels: 1,
frame_size: 320, // 20ms at 16kHz
bitrate: 16000,
bitrate: 32_000,
vbr: false,
fec: false,
}
}
}
apps/hello/src/stereo_core/master.rs
+7 -15
View File
@@ -70,14 +70,12 @@ pub async fn run_master(master_role: ChannelRole) -> Result<()> {
});
// 4. 音频处理主循环
let config = AudioConfig {
sample_rate: 48000,
channels: 2,
frame_size: 960,
bitrate: 64000,
..AudioConfig::default()
let config = AudioConfig::music();
let encode_config = AudioConfig {
channels: 1,
vbr: true,
..AudioConfig::music()
};
let mut current_player_channels = 0;
let mut player: Option<AudioPlayer> = None;
@@ -108,14 +106,8 @@ pub async fn run_master(master_role: ChannelRole) -> Result<()> {
};
// 每个新流开始时，重置编码器状态以避免残留音频导致爆音
let mut left_encoder = OpusCodec::new(&AudioConfig {
channels: 1,
..config.clone()
})?;
let mut right_encoder = OpusCodec::new(&AudioConfig {
channels: 1,
..config.clone()
})?;
let mut left_encoder = OpusCodec::new(&encode_config)?;
let mut right_encoder = OpusCodec::new(&encode_config)?;
loop {
// 从 FIFO 读取
apps/hello/src/stereo_core/slave.rs
+12 -7
View File
@@ -58,11 +58,8 @@ async fn handle_connection(role: ChannelRole) -> Result<()> {
// 5. 初始化音频与同步组件
let config = AudioConfig {
sample_rate: 48000,
channels: 1,
frame_size: 960,
bitrate: 64000,
..AudioConfig::default()
..AudioConfig::music()
};
let player = AudioPlayer::new(&config)?;
let mut codec = OpusCodec::new(&config)?;
@@ -164,10 +161,17 @@ async fn handle_connection(role: ChannelRole) -> Result<()> {
let current_server_time = clock.lock().await.to_server_time(now);
if let Some((seq, data)) = jitter.pop_frame(current_server_time) {
// 丢失处理 (PLC)
if let Some(last) = last_seq {
if seq > last + 1 {
for _ in 0..(seq - last - 1) {
let loss_count = seq.wrapping_sub(last) as i32 - 1;
if loss_count > 0 {
// 1. 优先尝试 FEC 恢复最近丢失的那一帧
// Opus 的 FEC 数据存储在当前包(data)中，用于恢复“前一帧”
if let Ok(len) = codec.decode_fec(&data, &mut pcm_buf) {
let _ = player.write(&pcm_buf[..len]);
}
// 2. 如果丢包超过 1 帧，剩下的帧只能靠丢包补偿(PLC)
for _ in 0..(loss_count - 1) {
if let Ok(len) = codec.decode_loss(&mut pcm_buf) {
let _ = player.write(&pcm_buf[..len]);
}
@@ -176,6 +180,7 @@ async fn handle_connection(role: ChannelRole) -> Result<()> {
}
last_seq = Some(seq);
// 3. 正常解码当前帧
let len = codec.decode(&data, &mut pcm_buf)?;
player.write(&pcm_buf[..len])?;
} else {