From a62af4b377643d3b76318ee782763b3833b27456 Mon Sep 17 00:00:00 2001
From: cai <cai@nbcai.cc>
Date: Tue, 07 Jul 2026 15:38:44 +0800
Subject: [PATCH] fix: keep pcm stream continuous
---
src/audio.rs | 270 +++++++++++++++++++++++++++++++++
src/main.rs | 180 ++++++++-------------
2 files changed, 340 insertions(+), 110 deletions(-)
diff --git a/src/audio.rs b/src/audio.rs
index 6bdea19..34415a7 100644
--- a/src/audio.rs
+++ b/src/audio.rs
@@ -181,6 +181,223 @@
))
}
+#[derive(Debug, Clone, Default)]
+pub struct PcmS16leStreamChunkResult {
+ pub frames: Vec<PcmFrame>,
+ pub dropped_tail_bytes: usize,
+ pub buffered_source_bytes: usize,
+ pub buffered_source_samples: usize,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct PcmS16leStreamDebugDump {
+ pub debug_source_path: Option<String>,
+ pub debug_pcm_wav_path: Option<String>,
+ pub debug_pcm_wav_size_bytes: Option<u64>,
+}
+
+#[derive(Debug, Clone)]
+pub struct PcmS16leStreamDecoder {
+ source_sample_rate_hz: u32,
+ source_num_channels: u16,
+ target_sample_rate_hz: u32,
+ target_num_channels: u16,
+ source_byte_buffer: Vec<u8>,
+ source_sample_buffer: Vec<i16>,
+ debug_source_samples: Vec<i16>,
+ debug_target_samples: Vec<i16>,
+}
+
+impl PcmS16leStreamDecoder {
+ pub fn new(
+ source_sample_rate_hz: u32,
+ source_num_channels: u32,
+ target_sample_rate_hz: u32,
+ target_num_channels: u16,
+ ) -> Result<Self> {
+ if source_sample_rate_hz == 0 {
+ return Err(anyhow!("pcm_s16le source sample rate is zero"));
+ }
+ if source_num_channels == 0 {
+ return Err(anyhow!("pcm_s16le source channel count is zero"));
+ }
+ let source_num_channels = u16::try_from(source_num_channels)
+ .map_err(|_| anyhow!("unsupported pcm_s16le channel count {source_num_channels}"))?;
+ Ok(Self {
+ source_sample_rate_hz,
+ source_num_channels,
+ target_sample_rate_hz,
+ target_num_channels,
+ source_byte_buffer: Vec::new(),
+ source_sample_buffer: Vec::new(),
+ debug_source_samples: Vec::new(),
+ debug_target_samples: Vec::new(),
+ })
+ }
+
+ pub fn push_bytes(
+ &mut self,
+ pcm_bytes: &[u8],
+ source_sample_rate_hz: u32,
+ source_num_channels: u32,
+ last: bool,
+ ) -> Result<PcmS16leStreamChunkResult> {
+ let source_num_channels = u16::try_from(source_num_channels)
+ .map_err(|_| anyhow!("unsupported pcm_s16le channel count {source_num_channels}"))?;
+ if source_sample_rate_hz != self.source_sample_rate_hz
+ || source_num_channels != self.source_num_channels
+ {
+ return Err(anyhow!(
+ "pcm_s16le stream format changed from {}Hz/{}ch to {}Hz/{}ch",
+ self.source_sample_rate_hz,
+ self.source_num_channels,
+ source_sample_rate_hz,
+ source_num_channels
+ ));
+ }
+
+ self.source_byte_buffer.extend_from_slice(pcm_bytes);
+ let source_frame_alignment_bytes = usize::from(self.source_num_channels) * 2;
+ let aligned_len = self.source_byte_buffer.len()
+ - self.source_byte_buffer.len() % source_frame_alignment_bytes;
+ if aligned_len > 0 {
+ let aligned_bytes: Vec<u8> = self.source_byte_buffer.drain(..aligned_len).collect();
+ let source_samples: Vec<i16> = aligned_bytes
+ .chunks_exact(2)
+ .map(|chunk| i16::from_le_bytes([chunk[0], chunk[1]]))
+ .collect();
+ self.debug_source_samples.extend_from_slice(&source_samples);
+ let remapped = remap_channels(
+ source_samples,
+ self.source_num_channels,
+ self.target_num_channels,
+ );
+ self.source_sample_buffer.extend(remapped);
+ }
+
+ let dropped_tail_bytes = if last && !self.source_byte_buffer.is_empty() {
+ let dropped = self.source_byte_buffer.len();
+ self.source_byte_buffer.clear();
+ dropped
+ } else {
+ 0
+ };
+
+ let mut frames = self.drain_full_frames();
+ if last {
+ if let Some(frame) = self.drain_final_partial_frame() {
+ frames.push(frame);
+ }
+ }
+
+ Ok(PcmS16leStreamChunkResult {
+ frames,
+ dropped_tail_bytes,
+ buffered_source_bytes: self.source_byte_buffer.len(),
+ buffered_source_samples: self.source_sample_buffer.len(),
+ })
+ }
+
+ pub fn write_debug_dump(
+ &self,
+ dir: &str,
+ call_id: &str,
+ debug_label: &str,
+ ) -> Result<PcmS16leStreamDebugDump> {
+ let base_dir = PathBuf::from(dir);
+ fs::create_dir_all(&base_dir)
+ .with_context(|| format!("failed to create audio debug dump dir {dir}"))?;
+ let safe_call_id = sanitize_file_segment(call_id);
+ let safe_label = sanitize_file_segment(debug_label);
+
+ let source_path = base_dir.join(format!("{safe_call_id}-{safe_label}-source.wav"));
+ write_debug_wav(
+ &source_path,
+ &self.debug_source_samples,
+ self.source_sample_rate_hz,
+ self.source_num_channels,
+ )?;
+
+ let pcm_path = base_dir.join(format!("{safe_call_id}-{safe_label}-target.wav"));
+ write_debug_wav(
+ &pcm_path,
+ &self.debug_target_samples,
+ self.target_sample_rate_hz,
+ self.target_num_channels,
+ )?;
+ let pcm_size = fs::metadata(&pcm_path)
+ .with_context(|| format!("failed to stat audio debug wav {}", pcm_path.display()))?
+ .len();
+
+ Ok(PcmS16leStreamDebugDump {
+ debug_source_path: Some(source_path.to_string_lossy().to_string()),
+ debug_pcm_wav_path: Some(pcm_path.to_string_lossy().to_string()),
+ debug_pcm_wav_size_bytes: Some(pcm_size),
+ })
+ }
+
+ fn drain_full_frames(&mut self) -> Vec<PcmFrame> {
+ let source_samples_per_frame = self.source_samples_per_20ms_frame();
+ let mut frames = Vec::new();
+ while self.source_sample_buffer.len() >= source_samples_per_frame {
+ let source_samples: Vec<i16> = self
+ .source_sample_buffer
+ .drain(..source_samples_per_frame)
+ .collect();
+ frames.push(self.convert_source_samples_to_frame(source_samples, true));
+ }
+ frames
+ }
+
+ fn drain_final_partial_frame(&mut self) -> Option<PcmFrame> {
+ if self.source_sample_buffer.is_empty() {
+ return None;
+ }
+ let source_samples: Vec<i16> = self.source_sample_buffer.drain(..).collect();
+ Some(self.convert_source_samples_to_frame(source_samples, true))
+ }
+
+ fn convert_source_samples_to_frame(
+ &mut self,
+ source_samples: Vec<i16>,
+ pad_to_frame: bool,
+ ) -> PcmFrame {
+ let mut target_samples = resample_linear(
+ source_samples,
+ self.source_sample_rate_hz,
+ self.target_sample_rate_hz,
+ self.target_num_channels,
+ );
+ let target_frame_samples = self.target_samples_per_20ms_frame();
+ if target_samples.len() > target_frame_samples {
+ target_samples.truncate(target_frame_samples);
+ } else if pad_to_frame && target_samples.len() < target_frame_samples {
+ target_samples.resize(target_frame_samples, 0);
+ }
+ self.debug_target_samples.extend_from_slice(&target_samples);
+ PcmFrame::new(
+ target_samples,
+ self.target_sample_rate_hz,
+ u32::from(self.target_num_channels),
+ self.target_samples_per_channel_per_20ms_frame() as u32,
+ )
+ }
+
+ fn source_samples_per_20ms_frame(&self) -> usize {
+ let source_frames =
+ ((f64::from(self.source_sample_rate_hz) / 50.0).round() as usize).max(1);
+ source_frames * usize::from(self.target_num_channels)
+ }
+
+ fn target_samples_per_20ms_frame(&self) -> usize {
+ self.target_samples_per_channel_per_20ms_frame() * usize::from(self.target_num_channels)
+ }
+
+ fn target_samples_per_channel_per_20ms_frame(&self) -> usize {
+ ((self.target_sample_rate_hz / 1000) * 20).max(1) as usize
+ }
+}
+
fn decode_audio_frames(
audio_bytes: &[u8],
source_kind: &'static str,
@@ -713,7 +930,7 @@
#[cfg(test)]
mod tests {
- use super::{load_pre_recorded_frames, pcm_s16le_bytes_to_frames};
+ use super::{PcmS16leStreamDecoder, load_pre_recorded_frames, pcm_s16le_bytes_to_frames};
#[tokio::test]
async fn load_pre_recorded_frames_reads_local_wav() {
@@ -772,4 +989,55 @@
assert_eq!(frames[0].num_channels, 1);
assert_eq!(frames[0].samples_per_channel, 960);
}
+
+ #[test]
+ fn pcm_s16le_stream_decoder_does_not_pad_each_network_chunk() {
+ let mut decoder = PcmS16leStreamDecoder::new(44_100, 1, 48_000, 1).expect("stream decoder");
+ let mut pcm = Vec::new();
+ for index in 0..4410 {
+ let sample = if index % 2 == 0 { 1024_i16 } else { -1024_i16 };
+ pcm.extend_from_slice(&sample.to_le_bytes());
+ }
+
+ let mut frames = Vec::new();
+ let chunk_size = 698 * 2;
+ for (index, chunk) in pcm.chunks(chunk_size).enumerate() {
+ let last = (index + 1) * chunk_size >= pcm.len();
+ let result = decoder
+ .push_bytes(chunk, 44_100, 1, last)
+ .expect("push pcm chunk");
+ frames.extend(result.frames);
+ }
+
+ assert_eq!(5, frames.len());
+ assert_eq!(
+ 4_800,
+ frames
+ .iter()
+ .map(|frame| frame.samples.len())
+ .sum::<usize>()
+ );
+ assert!(frames.iter().all(|frame| frame.sample_rate == 48_000));
+ assert!(frames.iter().all(|frame| frame.num_channels == 1));
+ assert!(frames.iter().all(|frame| frame.samples_per_channel == 960));
+ }
+
+ #[test]
+ fn pcm_s16le_stream_decoder_waits_until_full_20ms_frame() {
+ let mut decoder = PcmS16leStreamDecoder::new(44_100, 1, 48_000, 1).expect("stream decoder");
+ let ten_ms = vec![0_u8; 441 * 2];
+
+ let first = decoder
+ .push_bytes(&ten_ms, 44_100, 1, false)
+ .expect("first chunk");
+ assert!(first.frames.is_empty());
+ assert_eq!(441, first.buffered_source_samples);
+
+ let second = decoder
+ .push_bytes(&ten_ms, 44_100, 1, false)
+ .expect("second chunk");
+ assert_eq!(1, second.frames.len());
+ assert_eq!(960, second.frames[0].samples_per_channel);
+ assert_eq!(0, second.buffered_source_samples);
+ }
}
diff --git a/src/main.rs b/src/main.rs
index 3b18943..64eedd4 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -12,7 +12,7 @@
};
use anyhow::{Context, Result, anyhow};
-use audio::{AudioDiagnostics, PcmFrame, load_pre_recorded_frames};
+use audio::{AudioDiagnostics, load_pre_recorded_frames};
use base64::{Engine as _, engine::general_purpose};
use futures_util::StreamExt;
use libwebrtc::{
@@ -1343,45 +1343,54 @@
.trim()
.to_ascii_lowercase();
let frames = if format == "pcm_s16le" {
- let frame_alignment_bytes = pcm_s16le_frame_alignment_bytes(
- audio_chunk
- .channels
- .unwrap_or(u32::from(TARGET_NUM_CHANNELS)),
- )?;
- let (aligned_payload, dropped_tail_bytes) = take_aligned_pcm_payload(
- &mut state.pcm_audio_buffer,
- &payload,
- frame_alignment_bytes,
- audio_chunk.last.unwrap_or(false),
- );
- if dropped_tail_bytes > 0 {
+ let sample_rate = audio_chunk.sample_rate.unwrap_or(TARGET_SAMPLE_RATE_HZ);
+ let channels = audio_chunk
+ .channels
+ .unwrap_or(u32::from(TARGET_NUM_CHANNELS));
+ if state.pcm_stream_decoder.is_none() {
+ state.pcm_stream_decoder = Some(audio::PcmS16leStreamDecoder::new(
+ sample_rate,
+ channels,
+ TARGET_SAMPLE_RATE_HZ,
+ TARGET_NUM_CHANNELS,
+ )?);
+ }
+ state.pcm_stream_network_chunk_count =
+ state.pcm_stream_network_chunk_count.saturating_add(1);
+ let stream_result = state
+ .pcm_stream_decoder
+ .as_mut()
+ .expect("pcm stream decoder initialized")
+ .push_bytes(
+ &payload,
+ sample_rate,
+ channels,
+ audio_chunk.last.unwrap_or(false),
+ )?;
+ if stream_result.dropped_tail_bytes > 0 {
warn!(
call_id = %call_id,
trace_id = %trace_id,
turn_id = %turn.turn_id,
chunk_seq = audio_chunk.chunk_seq.unwrap_or_default(),
- dropped_tail_bytes,
+ dropped_tail_bytes = stream_result.dropped_tail_bytes,
"runtime helper stream_audio_pcm_unaligned_tail_dropped"
);
}
- if aligned_payload.is_empty() {
+ if stream_result.frames.is_empty() {
warn!(
call_id = %call_id,
trace_id = %trace_id,
turn_id = %turn.turn_id,
chunk_seq = audio_chunk.chunk_seq.unwrap_or_default(),
- buffered_bytes = state.pcm_audio_buffer.len(),
- "runtime helper stream_audio_pcm_waiting_for_sample_boundary"
+ buffered_source_bytes = stream_result.buffered_source_bytes,
+ buffered_source_samples = stream_result.buffered_source_samples,
+ network_chunk_count = state.pcm_stream_network_chunk_count,
+ "runtime helper stream_audio_pcm_waiting_for_20ms_frame"
);
return Ok(0);
}
- pcm_s16le_payload_to_frames(
- &aligned_payload,
- audio_chunk.sample_rate.unwrap_or(TARGET_SAMPLE_RATE_HZ),
- audio_chunk
- .channels
- .unwrap_or(u32::from(TARGET_NUM_CHANNELS)),
- )?
+ stream_result.frames
} else if matches!(format.as_str(), "mp3" | "mpeg" | "wav") {
state.encoded_audio_buffer.extend_from_slice(&payload);
match audio::decode_audio_bytes_to_frames(
@@ -1473,6 +1482,34 @@
sleep_until(pacing_started_at + Duration::from_millis(((index + 1) as u64) * 20)).await;
}
if audio_chunk.last.unwrap_or(false) {
+ let mut debug_source_path = None;
+ let mut debug_pcm_wav_path = None;
+ let mut debug_pcm_wav_size_bytes = None;
+ if format == "pcm_s16le" {
+ if let (Some(debug_dump_dir), Some(decoder)) = (
+ bridge_config.audio_debug_dump_dir.as_deref(),
+ state.pcm_stream_decoder.as_ref(),
+ ) {
+ match decoder.write_debug_dump(
+ debug_dump_dir,
+ call_id,
+ &format!("stream-reply-{}", turn.turn_id),
+ ) {
+ Ok(debug_dump) => {
+ debug_source_path = debug_dump.debug_source_path;
+ debug_pcm_wav_path = debug_dump.debug_pcm_wav_path;
+ debug_pcm_wav_size_bytes = debug_dump.debug_pcm_wav_size_bytes;
+ }
+ Err(error) => warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ turn_id = %turn.turn_id,
+ error = %safe_error(&error.to_string()),
+ "runtime helper stream_audio_pcm_debug_dump_failed"
+ ),
+ }
+ }
+ }
emit_activity(
call_id,
trace_id,
@@ -1485,94 +1522,17 @@
"replyPlaybackMode": state.reply_playback_mode.as_str(),
"format": format.as_str(),
"chunkSeq": audio_chunk.chunk_seq,
+ "networkChunkCount": state.pcm_stream_network_chunk_count,
+ "debugSourcePath": debug_source_path,
+ "debugPcmWavPath": debug_pcm_wav_path,
+ "debugPcmWavSizeBytes": debug_pcm_wav_size_bytes,
+ "sampleRate": audio_chunk.sample_rate,
+ "channels": audio_chunk.channels,
"replyTotalAfterVadEndMs": turn_pipeline_started_at.elapsed().as_millis() as u64,
}),
);
}
Ok(frames.len())
-}
-
-fn pcm_s16le_payload_to_frames(
- payload: &[u8],
- sample_rate: u32,
- channels: u32,
-) -> Result<Vec<PcmFrame>> {
- audio::pcm_s16le_bytes_to_frames(
- payload,
- sample_rate,
- channels,
- TARGET_SAMPLE_RATE_HZ,
- TARGET_NUM_CHANNELS,
- )
-}
-
-fn pcm_s16le_frame_alignment_bytes(channels: u32) -> Result<usize> {
- if channels == 0 {
- return Err(anyhow!("pcm_s16le channel count is zero"));
- }
- let channel_count = usize::try_from(channels)
- .map_err(|_| anyhow!("unsupported pcm_s16le channel count {channels}"))?;
- Ok(2 * channel_count)
-}
-
-fn take_aligned_pcm_payload(
- buffer: &mut Vec<u8>,
- payload: &[u8],
- frame_alignment_bytes: usize,
- last: bool,
-) -> (Vec<u8>, usize) {
- let alignment = frame_alignment_bytes.max(2);
- buffer.extend_from_slice(payload);
- let aligned_len = buffer.len() - buffer.len() % alignment;
- let aligned_payload = if aligned_len == 0 {
- Vec::new()
- } else {
- buffer.drain(..aligned_len).collect()
- };
- let dropped_tail_bytes = if last && !buffer.is_empty() {
- let dropped = buffer.len();
- buffer.clear();
- dropped
- } else {
- 0
- };
- (aligned_payload, dropped_tail_bytes)
-}
-
-#[cfg(test)]
-mod pcm_stream_tests {
- use super::take_aligned_pcm_payload;
-
- #[test]
- fn take_aligned_pcm_payload_buffers_split_sample_bytes() {
- let mut buffer = Vec::new();
-
- let (first, dropped) = take_aligned_pcm_payload(&mut buffer, &[0x01], 2, false);
- assert!(first.is_empty());
- assert_eq!(0, dropped);
- assert_eq!(vec![0x01], buffer);
-
- let (second, dropped) = take_aligned_pcm_payload(&mut buffer, &[0x02, 0x03], 2, false);
- assert_eq!(vec![0x01, 0x02], second);
- assert_eq!(0, dropped);
- assert_eq!(vec![0x03], buffer);
-
- let (third, dropped) = take_aligned_pcm_payload(&mut buffer, &[0x04], 2, true);
- assert_eq!(vec![0x03, 0x04], third);
- assert_eq!(0, dropped);
- assert!(buffer.is_empty());
- }
-
- #[test]
- fn take_aligned_pcm_payload_drops_final_half_sample() {
- let mut buffer = Vec::new();
-
- let (payload, dropped) =
- take_aligned_pcm_payload(&mut buffer, &[0x01, 0x02, 0x03], 2, true);
- assert_eq!(vec![0x01, 0x02], payload);
- assert_eq!(1, dropped);
- assert!(buffer.is_empty());
- }
}
fn trim_ascii_whitespace(value: &[u8]) -> &[u8] {
@@ -1950,7 +1910,8 @@
audio_chunk_count: u64,
device_output_count: u64,
encoded_audio_buffer: Vec<u8>,
- pcm_audio_buffer: Vec<u8>,
+ pcm_stream_decoder: Option<audio::PcmS16leStreamDecoder>,
+ pcm_stream_network_chunk_count: u64,
}
impl Default for RuntimeTurnStreamState {
@@ -1964,7 +1925,8 @@
audio_chunk_count: 0,
device_output_count: 0,
encoded_audio_buffer: Vec::new(),
- pcm_audio_buffer: Vec::new(),
+ pcm_stream_decoder: None,
+ pcm_stream_network_chunk_count: 0,
}
}
}
--
Gitblit v1.9.3