pipecrab_audio/
resampler.rs

1//! Backend-independent audio resampling and its pipeline stage adapter.
2
3use std::fmt;
4use std::sync::Mutex;
5
6use pipecrab_core::{
7    AudioChunk, AudioFormat, DataFrame, Decision, Direction, Processor, SystemFrame,
8};
9use pipecrab_runtime::{MaybeSend, Outbound, Stage, StageError, maybe_async_trait};
10
11use crate::rubato_sinc::RubatoSincResampler;
12
13/// A synchronous, stateful audio-to-audio format converter.
14///
15/// Implementations own their streaming state. One input chunk may produce one
16/// output chunk or no output yet when the implementation must buffer more
17/// samples. Every produced chunk must use [`output_format`](Self::output_format).
18///
19/// [`reset`](Self::reset) is a non-blocking, infallible control call. It must
20/// discard buffered input and filter history so audio processed afterward is
21/// independent of audio processed before the reset.
22pub trait Resampler: MaybeSend {
23    /// The fixed format attached to every output chunk.
24    fn output_format(&self) -> AudioFormat;
25
26    /// Convert one input chunk, or return `None` while buffering it.
27    fn resample(&mut self, input: &AudioChunk) -> Result<Option<AudioChunk>, ResamplerError>;
28
29    /// Clear all streaming state.
30    fn reset(&mut self);
31}
32
33/// Adapts any [`Resampler`] into a pipeline [`Stage`].
34///
35/// `ResamplerStage<R>` consumes each [`DataFrame::Audio`] and emits the chunk
36/// returned by `R`. A buffered input produces no frame. Non-audio data and all
37/// system frames pass through unchanged; [`SystemFrame::Interrupt`] also resets
38/// the backend before it is forwarded.
39///
40/// `R` defaults to [`RubatoSincResampler`], so
41/// [`ResamplerStage::new`](ResamplerStage::new) is the ergonomic production
42/// constructor. Use [`with_resampler`](Self::with_resampler) to inject another
43/// implementation.
44///
45/// # Orchestrator occupancy
46///
47/// Resampling happens synchronously in [`Processor::decide_data`]. This keeps
48/// all mutable DSP state in the non-cancellable half of a stage, but it also
49/// means one conversion occupies the orchestrator until it finishes. The
50/// default backend uses bounded internal blocks and reusable buffers, and the
51/// crate benchmark asserts an upper bound relative to audio duration.
52/// TODO: Add an asynchronous resampler adapter backed by a shared worker so DSP
53/// does not occupy the orchestrator thread.
54pub struct ResamplerStage<R: Resampler = RubatoSincResampler> {
55    output_format: AudioFormat,
56    // A backend only needs Send, while Stage requires Sync because perform
57    // borrows &self. decide_data has exclusive access and uses get_mut, so the
58    // synchronous hot path never acquires the mutex.
59    resampler: Mutex<R>,
60}
61
62impl ResamplerStage<RubatoSincResampler> {
63    /// Create a stage using the bundled windowed-sinc implementation.
64    pub fn new(output_format: AudioFormat) -> Result<Self, ResamplerError> {
65        Self::with_resampler(RubatoSincResampler::new(output_format)?)
66    }
67}
68
69impl<R: Resampler> ResamplerStage<R> {
70    /// Create a stage using `resampler` as its audio conversion backend.
71    pub fn with_resampler(resampler: R) -> Result<Self, ResamplerError> {
72        let output_format = resampler.output_format();
73        validate_format(output_format)?;
74        Ok(Self {
75            output_format,
76            resampler: Mutex::new(resampler),
77        })
78    }
79
80    /// The fixed format attached to every output chunk.
81    pub fn output_format(&self) -> AudioFormat {
82        self.output_format
83    }
84}
85
86/// A conversion command emitted by [`ResamplerStage`].
87///
88/// Its payload is private because only the stage constructs and interprets it.
89pub struct ResamplerEffect(Result<AudioChunk, ResamplerError>);
90
91impl<R: Resampler> Processor for ResamplerStage<R> {
92    type Effect = ResamplerEffect;
93
94    fn decide_data(&mut self, frame: &DataFrame) -> Decision<Self::Effect> {
95        let DataFrame::Audio(chunk) = frame else {
96            return Decision::forward();
97        };
98        let result = self
99            .resampler
100            .get_mut()
101            .expect("resampler mutex poisoned")
102            .resample(chunk);
103        match result {
104            Ok(Some(chunk)) => Decision::drop().emit(ResamplerEffect(Ok(chunk))),
105            Ok(None) => Decision::drop(),
106            Err(error) => Decision::drop().emit(ResamplerEffect(Err(error))),
107        }
108    }
109
110    fn decide_system(
111        &mut self,
112        _direction: Direction,
113        frame: &SystemFrame,
114    ) -> Decision<Self::Effect> {
115        if matches!(frame, SystemFrame::Interrupt) {
116            self.resampler
117                .get_mut()
118                .expect("resampler mutex poisoned")
119                .reset();
120        }
121        Decision::forward()
122    }
123}
124
125maybe_async_trait! {
126    impl<R: Resampler> Stage for ResamplerStage<R> {
127        async fn perform(
128            &self,
129            ResamplerEffect(result): ResamplerEffect,
130            out: &Outbound,
131        ) -> Result<(), StageError> {
132            let chunk = result.map_err(|error| StageError::fatal(error.to_string()))?;
133            let _ = out.send_data(DataFrame::Audio(chunk)).await;
134            Ok(())
135        }
136    }
137}
138
139/// Why an audio chunk could not be converted by a [`Resampler`].
140#[derive(Clone, Debug, PartialEq, Eq)]
141pub enum ResamplerError {
142    /// A sample rate or channel count was zero.
143    InvalidFormat {
144        /// The unusable format.
145        format: AudioFormat,
146    },
147    /// An interleaved buffer did not contain whole audio frames.
148    MisalignedSamples {
149        /// Number of samples in the buffer.
150        samples: usize,
151        /// Number of interleaved channels declared by the chunk.
152        channels: u16,
153    },
154    /// A resampling implementation rejected an operation.
155    Resampling(String),
156}
157
158impl fmt::Display for ResamplerError {
159    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
160        match self {
161            Self::InvalidFormat { format } => write!(
162                formatter,
163                "invalid audio format: sample rate and channels must be non-zero, got {} Hz/{} ch",
164                format.sample_rate, format.channels
165            ),
166            Self::MisalignedSamples { samples, channels } => write!(
167                formatter,
168                "audio buffer has {samples} samples, which is not divisible by {channels} channels"
169            ),
170            Self::Resampling(message) => write!(formatter, "audio resampling failed: {message}"),
171        }
172    }
173}
174
175impl std::error::Error for ResamplerError {}
176
177pub(crate) fn validate_format(format: AudioFormat) -> Result<(), ResamplerError> {
178    if format.sample_rate == 0 || format.channels == 0 {
179        Err(ResamplerError::InvalidFormat { format })
180    } else {
181        Ok(())
182    }
183}
184
185pub(crate) fn validate_chunk(chunk: &AudioChunk) -> Result<(), ResamplerError> {
186    validate_format(chunk.format)?;
187    if chunk.samples.len() % usize::from(chunk.format.channels) != 0 {
188        return Err(ResamplerError::MisalignedSamples {
189            samples: chunk.samples.len(),
190            channels: chunk.format.channels,
191        });
192    }
193    Ok(())
194}
195
196#[cfg(test)]
197mod tests {
198    use std::sync::Arc;
199    use std::sync::atomic::{AtomicUsize, Ordering};
200
201    use pipecrab_core::Disposition;
202
203    use super::*;
204
205    struct Gain {
206        output_format: AudioFormat,
207        resets: Arc<AtomicUsize>,
208    }
209
210    impl Resampler for Gain {
211        fn output_format(&self) -> AudioFormat {
212            self.output_format
213        }
214
215        fn resample(&mut self, input: &AudioChunk) -> Result<Option<AudioChunk>, ResamplerError> {
216            let samples: Arc<[f32]> = input.samples.iter().map(|sample| sample * 2.0).collect();
217            Ok(Some(AudioChunk::new(samples, self.output_format)))
218        }
219
220        fn reset(&mut self) {
221            self.resets.fetch_add(1, Ordering::Relaxed);
222        }
223    }
224
225    #[test]
226    fn custom_resampler_drives_the_generic_stage() {
227        let output_format = AudioFormat::new(24_000, 1);
228        let resets = Arc::new(AtomicUsize::new(0));
229        let mut stage = ResamplerStage::with_resampler(Gain {
230            output_format,
231            resets: resets.clone(),
232        })
233        .unwrap();
234        let frame = DataFrame::Audio(AudioChunk::new(
235            Arc::from([0.25, -0.5]),
236            AudioFormat::new(48_000, 1),
237        ));
238
239        let decision = stage.decide_data(&frame);
240        assert_eq!(decision.disposition, Disposition::Drop);
241        let chunk = decision.effects.into_iter().next().unwrap().0.unwrap();
242        assert_eq!(chunk.format, output_format);
243        assert_eq!(&*chunk.samples, &[0.5, -1.0]);
244
245        stage.decide_system(Direction::Down, &SystemFrame::Interrupt);
246        assert_eq!(resets.load(Ordering::Relaxed), 1);
247    }
248
249    #[test]
250    fn rejects_a_custom_resampler_with_an_invalid_output_format() {
251        let result = ResamplerStage::with_resampler(Gain {
252            output_format: AudioFormat::new(0, 1),
253            resets: Arc::new(AtomicUsize::new(0)),
254        });
255        assert!(matches!(result, Err(ResamplerError::InvalidFormat { .. })));
256    }
257}
pipecrab_audio/resampler.rs

pipecrab_audio/
resampler.rs
