feat(audio): 调试录音前后对照 + AEC 两种实现可选 + 扬声器回环参考

openless-all/app/crates/openless-core/src/audio.rs

@@ -16,6 +16,141 @@ pub struct AudioProcessingConfig {
     pub dsp: crate::audio_dsp::AudioDspConfig,
 }
 
+/// 扬声器回环（far-end）参考音频的交接队列。
+///
+/// AEC 需要「扬声器此刻在放什么」。宿主在一条独立的采集流里拿到输出音频后，
+/// 按自己的采样率塞进这里；DSP 线程在处理每一帧之前把它取走、重采样到 48 kHz
+/// 单声道后交给 MicYou。队列有上限：AEC 只关心最近的参考，堆积只会拖慢。
+#[derive(Clone, Default)]
+pub struct FarEndQueue {
+    inner: std::sync::Arc<std::sync::Mutex<std::collections::VecDeque<FarEndFrame>>>,
+}
+
+#[derive(Debug)]
+struct FarEndFrame {
+    rate: u32,
+    channels: usize,
+    samples: Vec<f32>,
+}
+
+/// 参考队列最多保留的样本数（约 2 秒 @48 kHz）。超出时丢弃最旧的。
+const FAR_END_QUEUE_LIMIT: usize = 96_000;
+
+impl FarEndQueue {
+    /// 宿主采集线程调用。非有限值与空帧直接丢弃。
+    pub fn push(&self, samples: &[f32], channels: usize, rate: u32) {
+        if samples.is_empty() || channels == 0 || rate == 0 {
+            return;
+        }
+        let clean: Vec<f32> = samples
+            .iter()
+            .map(|sample| if sample.is_finite() { sample.clamp(-1.0, 1.0) } else { 0.0 })
+            .collect();
+        let mut queue = self.inner.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        queue.push_back(FarEndFrame {
+            rate,
+            channels,
+            samples: clean,
+        });
+        let mut total: usize = queue.iter().map(|frame| frame.samples.len()).sum();
+        while total > FAR_END_QUEUE_LIMIT {
+            let Some(dropped) = queue.pop_front() else { break };
+            total -= dropped.samples.len();
+        }
+    }
+
+    /// DSP 线程调用：取走全部待处理参考帧（保持到达顺序）。
+    fn drain(&self) -> Vec<FarEndFrame> {
+        let mut queue = self.inner.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        queue.drain(..).collect()
+    }
+
+    fn is_empty(&self) -> bool {
+        self.inner
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .is_empty()
+    }
+}
+
+const ECHO_FILTER_TAPS: usize = 1_024;
+const ECHO_ADAPTATION_RATE: f32 = 0.12;
+const ECHO_DOUBLE_TALK_RATIO: f32 = 2.5;
+const ECHO_MIN_REFERENCE_POWER: f32 = 0.000_001;
+
+/// 轻量回声消除：有界归一化 LMS（沿用 #1228 的实现）。
+///
+/// 学的是「扬声器 → 麦克风」这条声学路径；近端能量明显强于估计回声（双讲）时
+/// 暂停自适应，避免把说话人自己的声音学掉。单核开销约 3%，弱机可用。
+#[derive(Debug, Clone)]
+struct LightEchoCanceller {
+    coefficients: Vec<f32>,
+    history: Vec<f32>,
+    history_cursor: usize,
+}
+
+impl Default for LightEchoCanceller {
+    fn default() -> Self {
+        Self {
+            coefficients: vec![0.0; ECHO_FILTER_TAPS],
+            history: vec![0.0; ECHO_FILTER_TAPS],
+            history_cursor: 0,
+        }
+    }
+}
+
+impl LightEchoCanceller {
+    fn reset(&mut self) {
+        self.coefficients.fill(0.0);
+        self.history.fill(0.0);
+        self.history_cursor = 0;
+    }
+
+    fn process(&mut self, microphone: &[f32], reference: &[f32]) -> Vec<f32> {
+        if microphone.is_empty() || reference.is_empty() {
+            return microphone.to_vec();
+        }
+        let mut output = Vec::with_capacity(microphone.len());
+        for (index, &sample) in microphone.iter().enumerate() {
+            // 块边界处重采样可能少一个样本：保留麦克风样本、等下一块重建对齐，
+            // 不要把上一个参考值铺满，那样会教坏滤波器。
+            let Some(&reference_sample) = reference.get(index) else {
+                output.push(sample);
+                continue;
+            };
+            self.history[self.history_cursor] = reference_sample;
+            let mut estimate = 0.0;
+            let mut reference_power = 0.0;
+            for tap in 0..ECHO_FILTER_TAPS {
+                let history_index =
+                    (self.history_cursor + ECHO_FILTER_TAPS - tap) % ECHO_FILTER_TAPS;
+                let value = self.history[history_index];
+                estimate += self.coefficients[tap] * value;
+                reference_power += value * value;
+            }
+            let error = sample - estimate;
+            let near_power = sample * sample;
+            let should_adapt = reference_power > ECHO_MIN_REFERENCE_POWER
+                && (estimate.abs() < 0.01
+                    || estimate * estimate * ECHO_DOUBLE_TALK_RATIO >= near_power
+                    || near_power < ECHO_MIN_REFERENCE_POWER);
+            if should_adapt {
+                let normalization = reference_power + 0.000_01;
+                for tap in 0..ECHO_FILTER_TAPS {
+                    let history_index =
+                        (self.history_cursor + ECHO_FILTER_TAPS - tap) % ECHO_FILTER_TAPS;
+                    self.coefficients[tap] +=
+                        ECHO_ADAPTATION_RATE * error * self.history[history_index] / normalization;
+                    self.coefficients[tap] = self.coefficients[tap].clamp(-2.0, 2.0);
+                }
+            }
+            output.push(if error.is_finite() { error } else { 0.0 });
+            self.history_cursor = (self.history_cursor + 1) % ECHO_FILTER_TAPS;
+        }
+        output
+    }
+}
+
 /// Shared processor for native and external microphone hosts.
 /// Enhancement starts with the session's settings and may be disabled under
 /// load. Hosts must deliver every returned tail before feeding more input or
@@ -24,6 +159,13 @@ pub struct AudioProcessor {
     config: AudioProcessingConfig,
     normalizer: PcmNormalizer,
     enhanced: Option<EnhancedStream>,
+    /// AEC 的扬声器参考输入（宿主在独立的回环采集流里喂）。
+    far_end: Option<FarEndQueue>,
+    /// 轻量 NLMS 抵消器（`AecMode::Light`）。
+    light_aec: Option<LightEchoCanceller>,
+    /// 轻量路径把参考重采样到输出采样率（16 kHz 单声道）。
+    light_reference_resampler: Option<crate::audio_dsp::SpeechResampler>,
+    light_reference_rate: Option<u32>,
     finished: bool,
 }
 
@@ -41,10 +183,61 @@ impl AudioProcessor {
             config,
             normalizer: Default::default(),
             enhanced: None,
+            far_end: None,
+            light_aec: matches!(config.dsp.aec_mode, crate::audio_dsp::AecMode::Light)
+                .then(LightEchoCanceller::default),
+            light_reference_resampler: None,
+            light_reference_rate: None,
             finished: false,
         }
     }
 
+    /// 把参考队列里的帧转成「与麦克风同一采样率」的单声道，供轻量抵消器使用。
+    ///
+    /// NLMS 的抽头是按样本对齐的，采样率不一致就完全对不上 —— 早先版本把参考重
+    /// 采样到 16 kHz 却拿 48 kHz 的麦克风样本做抵消，实测 ERLE ≈ 0（等于没做）。
+    fn light_reference_at(
+        &mut self,
+        queue: &FarEndQueue,
+        rate: u32,
+    ) -> Option<Vec<f32>> {
+        if queue.is_empty() {
+            return None;
+        }
+        let frames = queue.drain();
+        if frames.is_empty() {
+            return None;
+        }
+        if self.light_reference_rate != Some(rate) {
+            self.light_reference_rate = Some(rate);
+            self.light_reference_resampler =
+                Some(crate::audio_dsp::SpeechResampler::new(rate, rate));
+        }
+        let mut reference = Vec::new();
+        for frame in frames {
+            let mono = downmix_to_mono(&frame.samples, frame.channels);
+            if let Some(resampler) = self.light_reference_resampler.as_mut() {
+                reference.extend(resampler.push(&mono, false));
+            }
+        }
+        (!reference.is_empty()).then_some(reference)
+    }
+
+    /// 会话结束时清掉抵消器状态，避免上一场的声学路径带到下一场。
+    pub fn reset_echo_cancellers(&mut self) {
+        if let Some(canceller) = self.light_aec.as_mut() {
+            canceller.reset();
+        }
+        self.light_reference_resampler = None;
+        self.light_reference_rate = None;
+    }
+
+    /// 接上扬声器回环参考队列。只有 AEC 打开且宿主真的采到了输出音频时才有意义；
+    /// 没接或队列为空时 AEC 按「无参考」处理（MicYou 内部会退化）。
+    pub fn attach_far_end(&mut self, queue: FarEndQueue) {
+        self.far_end = Some(queue);
+    }
+
     pub fn reset(&mut self) {
         *self = Self::new(self.config);
     }
@@ -71,7 +264,24 @@ impl AudioProcessor {
         if self.finished || input.is_empty() || channels == 0 || rate == 0 {
             return None;
         }
-        if !self.config.enhancement_enabled {
+        // 轻量 AEC 在主链之前做：抵消后的信号再进降噪/增益，顺序与 MicYou 的
+        // 链路一致（AEC 必须拿到尚未被其它处理改过的信号）。
+        let mut cancelled: Option<Vec<f32>> = None;
+        if self.light_aec.is_some() {
+            let reference = self
+                .far_end
+                .clone()
+                .and_then(|queue| self.light_reference_at(&queue, rate));
+            if let (Some(reference), Some(canceller)) = (reference, self.light_aec.as_mut()) {
+                cancelled = Some(canceller.process(input, &reference));
+            }
+        }
+        let input = cancelled.as_deref().unwrap_or(input);
+
+        // 模型 AEC（`AecMode::Model`）走 MicYou 链路；增强关掉时也要能单独跑。
+        let aec_standalone = matches!(self.config.dsp.aec_mode, crate::audio_dsp::AecMode::Model)
+            && self.far_end.is_some();
+        if !self.config.enhancement_enabled && !aec_standalone {
             return self.normalizer.process(input, channels, rate);
         }
         let stream = self
@@ -92,6 +302,11 @@ impl AudioProcessor {
         let mono = downmix_to_mono(&stream.interleaved[..complete], channels);
         stream.interleaved.drain(..complete);
         stream.input_frames += mono.len() as u64;
+        // 参考音频必须早于本帧进入 AEC：MicYou 用最近的历史做自适应滤波。
+        if let Some(far_end) = self.far_end.as_ref().filter(|queue| !queue.is_empty()) {
+            let frames = far_end.drain();
+            stream.push_far_end(frames);
+        }
         let working = stream.input_resampler.push(&mono, false);
         let processed = stream.frames(&working, false);
         pcm_chunk(&processed)
@@ -123,6 +338,10 @@ struct EnhancedStream {
     model_delay: usize,
     input_frames: u64,
     emitted_frames: u64,
+    /// 参考音频的重采样器，按输入采样率惰性重建（回环流通常固定，重建仅一次）。
+    far_end_rate: Option<u32>,
+    far_end_resampler: Option<crate::audio_dsp::SpeechResampler>,
+    far_end_pending: Vec<f32>,
 }
 
 impl EnhancedStream {
@@ -143,7 +362,32 @@ impl EnhancedStream {
             model_delay: config.model_delay(),
             input_frames: 0,
             emitted_frames: 0,
+            far_end_rate: None,
+            far_end_resampler: None,
+            far_end_pending: Vec::new(),
+        }
+    }
+
+    /// 把宿主采到的回环参考帧转成 48 kHz 单声道后交给 AEC。
+    fn push_far_end(&mut self, frames: Vec<FarEndFrame>) {
+        for frame in frames {
+            let mono = downmix_to_mono(&frame.samples, frame.channels);
+            if self.far_end_rate != Some(frame.rate) {
+                self.far_end_rate = Some(frame.rate);
+                self.far_end_resampler =
+                    Some(crate::audio_dsp::SpeechResampler::new(frame.rate, 48_000));
+                self.far_end_pending.clear();
+            }
+            if let Some(resampler) = self.far_end_resampler.as_mut() {
+                self.far_end_pending
+                    .extend(resampler.push(&mono, false));
+            }
+        }
+        if self.far_end_pending.is_empty() {
+            return;
         }
+        let reference = std::mem::take(&mut self.far_end_pending);
+        self.dsp.set_far_end_audio(&reference);
     }
 
     fn frames(&mut self, input: &[f32], finish: bool) -> Vec<f32> {
@@ -354,6 +598,17 @@ fn downmix_to_mono(interleaved: &[f32], channels: usize) -> Vec<f32> {
         .collect()
 }
 
+/// f32 交织采样 → 小端 i16 字节流，供宿主把「处理前」的原始 PCM 也写成 WAV
+/// （调试时与处理后那份做 A/B 对照）。量化规则与 [`AudioProcessor`] 内部一致。
+pub fn pcm_f32_to_i16_le(samples: &[f32]) -> Vec<u8> {
+    let mut bytes = Vec::with_capacity(samples.len() * 2);
+    for sample in samples {
+        let quantized = (sample.clamp(-1.0, 1.0) * i16::MAX as f32) as i16;
+        bytes.extend_from_slice(&quantized.to_le_bytes());
+    }
+    bytes
+}
+
 fn quantize_to_i16_le(samples: &[f32]) -> (Vec<u8>, f32) {
     let mut bytes = Vec::with_capacity(samples.len() * 2);
     let mut square_sum = 0.0_f64;