flowcat_core/audio/
mixer.rs

1// SPDX-License-Identifier: Apache-2.0
2//
3//! Output-leg audio mixers.
4//!
5//! Mirrors pipecat's `pipecat/audio/mixers/`: a mixer blends the bot's
6//! TTS/realtime output with a secondary source (hold music, soundboard) before it
7//! reaches the transport-out leg. This module defines:
8//! - [`AudioMixer`] — the pure-DSP mixer trait (`mix(audio) -> audio`), the Rust
9//!   analogue of pipecat's `BaseAudioMixer`;
10//! - [`SilenceMixer`] — a pass-through (pipecat `SilenceAudioMixer`), always
11//!   available; used to keep continuous output streaming;
12//! - [`SoundfileMixer`] — mixes a pre-loaded mono PCM source at a configurable
13//!   volume, looping (pipecat `SoundfileMixer`, but taking decoded PCM in-memory
14//!   rather than pulling a `soundfile`/codec dep into core);
15//! - [`MixerProcessor`] — a [`FrameProcessor`] that mixes `OutputAudio`/`TtsAudio`
16//!   and is toggled live by [`Frame::MixerEnable`].
17
18use async_trait::async_trait;
19
20use crate::error::Result;
21use crate::processor::frame::{AudioFrame, Frame, StartParams};
22use crate::processor::{Envelope, FrameProcessor, Link, ProcessorSetup};
23
24/// A pure-DSP audio mixer over 16-bit mono PCM. Mirrors pipecat `BaseAudioMixer`:
25/// `start(sample_rate)` initializes the mixer for the output rate, `mix(audio)`
26/// blends one chunk of transport audio with the mixer's own source.
27///
28/// Returns a chunk of the same length as the input (per-sample blend). Must be
29/// deterministic (fixture-testable) and non-blocking.
30pub trait AudioMixer: Send {
31    /// A stable name for tracing/observers.
32    fn name(&self) -> &str;
33    /// Initialize for the output `sample_rate` (Hz). Default: no-op.
34    fn start(&mut self, _sample_rate: u32) -> Result<()> {
35        Ok(())
36    }
37    /// Blend the mixer's source into one chunk of transport `pcm`, returning the
38    /// mixed chunk (same length).
39    fn mix(&mut self, pcm: &[i16]) -> Vec<i16>;
40}
41
42/// A pass-through mixer: returns transport audio unchanged. Mirrors pipecat
43/// `SilenceAudioMixer` — used so the output transport can stream continuously
44/// (mixing silence) even when no TTS audio is flowing. Always available (no dep).
45#[derive(Debug, Default, Clone)]
46pub struct SilenceMixer;
47
48impl AudioMixer for SilenceMixer {
49    fn name(&self) -> &str {
50        "SilenceMixer"
51    }
52    fn mix(&mut self, pcm: &[i16]) -> Vec<i16> {
53        pcm.to_vec()
54    }
55}
56
57/// Mixes a pre-loaded mono PCM source (hold music / soundboard) into the bot
58/// audio at a configurable `volume`, looping the source. Mirrors pipecat
59/// `SoundfileMixer` but takes **decoded** PCM in-memory (the caller decodes the
60/// file with whatever codec it likes) so core pulls no `soundfile`/audio-decode
61/// dependency.
62///
63/// The source must already be at the output sample rate (set via [`Self::start`]);
64/// a rate mismatch is logged and mixing is disabled, mirroring pipecat.
65#[derive(Debug, Clone)]
66pub struct SoundfileMixer {
67    source: Vec<i16>,
68    source_rate: u32,
69    pos: usize,
70    volume: f32,
71    looping: bool,
72    sample_rate: u32,
73    ready: bool,
74}
75
76impl SoundfileMixer {
77    /// Build a mixer from decoded mono `source` PCM at `source_rate`, blended at
78    /// `volume` (0.0–1.0), looping when `looping`.
79    pub fn new(source: Vec<i16>, source_rate: u32, volume: f32, looping: bool) -> Self {
80        Self {
81            source,
82            source_rate,
83            pos: 0,
84            volume: volume.clamp(0.0, 1.0),
85            looping,
86            sample_rate: source_rate,
87            ready: false,
88        }
89    }
90
91    /// Update the mixing volume (0.0–1.0) at runtime.
92    pub fn set_volume(&mut self, volume: f32) {
93        self.volume = volume.clamp(0.0, 1.0);
94    }
95}
96
97impl AudioMixer for SoundfileMixer {
98    fn name(&self) -> &str {
99        "SoundfileMixer"
100    }
101
102    fn start(&mut self, sample_rate: u32) -> Result<()> {
103        self.sample_rate = sample_rate;
104        // The source must match the output rate (pipecat logs + disables on
105        // mismatch rather than resampling the loop on the hot path).
106        self.ready = self.source_rate == sample_rate && !self.source.is_empty();
107        if !self.ready && !self.source.is_empty() {
108            tracing::warn!(
109                "SoundfileMixer: source rate {} != output rate {} — mixing disabled",
110                self.source_rate,
111                sample_rate
112            );
113        }
114        Ok(())
115    }
116
117    fn mix(&mut self, pcm: &[i16]) -> Vec<i16> {
118        if !self.ready || self.volume == 0.0 {
119            return pcm.to_vec();
120        }
121        let mut out = Vec::with_capacity(pcm.len());
122        for &s in pcm {
123            if self.pos >= self.source.len() {
124                if self.looping {
125                    self.pos = 0;
126                } else {
127                    out.push(s);
128                    continue;
129                }
130            }
131            let mixed = s as f32 + self.source[self.pos] as f32 * self.volume;
132            out.push(mixed.round().clamp(i16::MIN as f32, i16::MAX as f32) as i16);
133            self.pos += 1;
134        }
135        out
136    }
137}
138
139/// A [`FrameProcessor`] that mixes a secondary [`AudioMixer`] source into the
140/// bot's output audio.
141///
142/// On [`Frame::OutputAudio`]/[`Frame::TtsAudio`] it blends the mixer source and
143/// forwards the mixed audio; on [`Frame::MixerEnable`] it toggles mixing live
144/// (pipecat's `MixerEnableFrame`). Sits **after** TTS / before the transport-out
145/// sink. All other frames pass through.
146pub struct MixerProcessor<M: AudioMixer> {
147    mixer: M,
148    enabled: bool,
149    sample_rate: u32,
150}
151
152impl<M: AudioMixer> MixerProcessor<M> {
153    /// Wrap `mixer`, enabled by default.
154    pub fn new(mixer: M) -> Self {
155        Self {
156            mixer,
157            enabled: true,
158            sample_rate: 24_000,
159        }
160    }
161
162    /// Wrap `mixer`, initially disabled (pass-through until a
163    /// [`Frame::MixerEnable`]`(true)` arrives).
164    pub fn disabled(mixer: M) -> Self {
165        Self {
166            mixer,
167            enabled: false,
168            sample_rate: 24_000,
169        }
170    }
171}
172
173#[async_trait]
174impl<M: AudioMixer + 'static> FrameProcessor for MixerProcessor<M> {
175    fn name(&self) -> &str {
176        self.mixer.name()
177    }
178
179    async fn start(&mut self, _setup: &ProcessorSetup, params: &StartParams) -> Result<()> {
180        self.sample_rate = params.audio_out_sample_rate;
181        self.mixer.start(self.sample_rate)
182    }
183
184    async fn process_frame(&mut self, env: Envelope, link: &Link) -> Result<()> {
185        match &env.frame {
186            Frame::MixerEnable(on) => {
187                self.enabled = *on;
188                link.push(env.meta, env.frame, env.direction).await;
189                return Ok(());
190            }
191            Frame::OutputAudio(audio) if self.enabled => {
192                let mixed = self.mixer.mix(&audio.pcm);
193                let out = AudioFrame::mono(mixed, audio.sample_rate);
194                link.push_down(Frame::OutputAudio(std::sync::Arc::new(out)))
195                    .await;
196                return Ok(());
197            }
198            Frame::TtsAudio { audio, context_id } if self.enabled => {
199                let mixed = self.mixer.mix(&audio.pcm);
200                let out = AudioFrame::mono(mixed, audio.sample_rate);
201                link.push_down(Frame::TtsAudio {
202                    audio: std::sync::Arc::new(out),
203                    context_id: context_id.clone(),
204                })
205                .await;
206                return Ok(());
207            }
208            _ => {}
209        }
210        link.push(env.meta, env.frame, env.direction).await;
211        Ok(())
212    }
213}
214
215#[cfg(test)]
216mod tests {
217    use super::*;
218    use crate::observer::{FrameEvent, FrameObserver};
219    use crate::pipeline::{Pipeline, PipelineTask, PipelineTaskParams};
220    use std::sync::{Arc, Mutex};
221
222    #[test]
223    fn silence_mixer_passes_through() {
224        let mut m = SilenceMixer;
225        let pcm = vec![1i16, 2, 3];
226        assert_eq!(m.mix(&pcm), pcm);
227    }
228
229    #[test]
230    fn soundfile_mixer_blends_at_volume() {
231        // Source is constant 1000; volume 0.5 => +500 per sample.
232        let mut m = SoundfileMixer::new(vec![1000i16; 4], 24_000, 0.5, true);
233        m.start(24_000).unwrap();
234        let out = m.mix(&[2000i16; 4]);
235        assert_eq!(out, vec![2500i16; 4], "2000 + 1000*0.5 = 2500");
236    }
237
238    #[test]
239    fn soundfile_mixer_loops_source() {
240        // Source shorter than the chunk: it must loop.
241        let mut m = SoundfileMixer::new(vec![100i16, 200], 8_000, 1.0, true);
242        m.start(8_000).unwrap();
243        let out = m.mix(&[0i16; 4]);
244        // 0+100, 0+200, loop -> 0+100, 0+200
245        assert_eq!(out, vec![100, 200, 100, 200]);
246    }
247
248    #[test]
249    fn soundfile_mixer_clamps_on_overflow() {
250        let mut m = SoundfileMixer::new(vec![30000i16; 2], 8_000, 1.0, true);
251        m.start(8_000).unwrap();
252        let out = m.mix(&[30000i16; 2]);
253        assert!(
254            out.iter().all(|&s| s == i16::MAX),
255            "60000 saturates to i16::MAX"
256        );
257    }
258
259    #[test]
260    fn soundfile_mixer_rate_mismatch_disables_mixing() {
261        let mut m = SoundfileMixer::new(vec![1000i16; 4], 8_000, 0.5, true);
262        m.start(24_000).unwrap(); // mismatch
263                                  // Mixing disabled -> pass-through.
264        assert_eq!(m.mix(&[2000i16; 4]), vec![2000i16; 4]);
265    }
266
267    // Records the first sample of each OutputAudio frame.
268    #[derive(Default)]
269    struct OutTap {
270        first_samples: Mutex<Vec<i16>>,
271    }
272    #[async_trait]
273    impl FrameObserver for OutTap {
274        async fn on_process(&self, e: &FrameEvent<'_>) {
275            if let Frame::OutputAudio(a) = e.frame {
276                if let Some(&s) = a.pcm.first() {
277                    self.first_samples.lock().unwrap().push(s);
278                }
279            }
280        }
281    }
282
283    #[tokio::test]
284    async fn mixer_processor_mixes_output_audio() {
285        let mixer = SoundfileMixer::new(vec![1000i16; 8], 24_000, 1.0, true);
286        let proc = MixerProcessor::new(mixer);
287        let pipeline = Pipeline::new(vec![Box::new(proc)]);
288        let tap = Arc::new(OutTap::default());
289        let task = PipelineTask::new(
290            pipeline,
291            PipelineTaskParams::default(),
292            vec![tap.clone() as Arc<dyn FrameObserver>],
293        );
294
295        task.queue_frame(Frame::OutputAudio(Arc::new(AudioFrame::mono(
296            vec![2000i16; 4],
297            24_000,
298        ))))
299        .await;
300        task.stop_when_done().await;
301        tokio::time::timeout(std::time::Duration::from_secs(5), task.run())
302            .await
303            .expect("timed out")
304            .expect("run ok");
305
306        let firsts = tap.first_samples.lock().unwrap().clone();
307        assert!(
308            firsts.contains(&3000),
309            "mixed OutputAudio (2000+1000) must appear; saw {firsts:?}"
310        );
311    }
312
313    #[tokio::test]
314    async fn mixer_processor_toggle_bypasses_when_disabled() {
315        let mixer = SoundfileMixer::new(vec![1000i16; 8], 24_000, 1.0, true);
316        let proc = MixerProcessor::new(mixer);
317        let pipeline = Pipeline::new(vec![Box::new(proc)]);
318        let tap = Arc::new(OutTap::default());
319        let task = PipelineTask::new(
320            pipeline,
321            PipelineTaskParams::default(),
322            vec![tap.clone() as Arc<dyn FrameObserver>],
323        );
324
325        task.queue_frame(Frame::MixerEnable(false)).await;
326        task.queue_frame(Frame::OutputAudio(Arc::new(AudioFrame::mono(
327            vec![2000i16; 4],
328            24_000,
329        ))))
330        .await;
331        task.stop_when_done().await;
332        tokio::time::timeout(std::time::Duration::from_secs(5), task.run())
333            .await
334            .expect("timed out")
335            .expect("run ok");
336
337        let firsts = tap.first_samples.lock().unwrap().clone();
338        assert!(
339            firsts.iter().all(|&s| s == 2000),
340            "disabled mixer must pass audio through unchanged; saw {firsts:?}"
341        );
342    }
343}
flowcat_core/audio/mixer.rs

flowcat_core/audio/
mixer.rs
