diff --git a/apps/hello/src/audio/codec.rs b/apps/hello/src/audio/codec.rs index f7e2646..5367ee2 100644 --- a/apps/hello/src/audio/codec.rs +++ b/apps/hello/src/audio/codec.rs @@ -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 { - 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 { + self.decoder + .decode(opus, out, true) + .context("Opus FEC decoding failed") + } + + /// 丢包补偿(PLC) pub fn decode_loss(&mut self, out: &mut [i16]) -> Result { self.decoder .decode(&[], out, false) diff --git a/apps/hello/src/bin/client.rs b/apps/hello/src/bin/client.rs deleted file mode 100644 index 1fee262..0000000 --- a/apps/hello/src/bin/client.rs +++ /dev/null @@ -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 = 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::(&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 = 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::(&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; - } - } -} diff --git a/apps/hello/src/bin/server.rs b/apps/hello/src/bin/server.rs deleted file mode 100644 index fdf3b15..0000000 --- a/apps/hello/src/bin/server.rs +++ /dev/null @@ -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>> = 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::(&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::(&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>>, -) -> 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>>, - 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(()) -} diff --git a/apps/hello/src/bin/test.rs b/apps/hello/src/bin/test.rs deleted file mode 100644 index 0ee8a4d..0000000 --- a/apps/hello/src/bin/test.rs +++ /dev/null @@ -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)> = 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; - } - } - } -} diff --git a/apps/hello/src/config.rs b/apps/hello/src/config.rs index 8ce1907..1c982ef 100644 --- a/apps/hello/src/config.rs +++ b/apps/hello/src/config.rs @@ -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, } } } diff --git a/apps/hello/src/stereo_core/master.rs b/apps/hello/src/stereo_core/master.rs index a2a02fb..dd8f21c 100644 --- a/apps/hello/src/stereo_core/master.rs +++ b/apps/hello/src/stereo_core/master.rs @@ -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 = 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 读取 diff --git a/apps/hello/src/stereo_core/slave.rs b/apps/hello/src/stereo_core/slave.rs index 158601a..d29cb50 100644 --- a/apps/hello/src/stereo_core/slave.rs +++ b/apps/hello/src/stereo_core/slave.rs @@ -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 {