aria_kit/synthesis/
playback.rs

1//! Playing synthesized speech on the default output device.
2//!
3//! The audio callback reads buffered samples, converts their rate, and fills
4//! underruns with silence. Cancellation and device-drain waits run on the caller's thread.
5
6use std::sync::mpsc::{SyncSender, sync_channel};
7use std::time::Duration;
8
9use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
10use cpal::{FromSample, SizedSample};
11
12use super::{PlaybackError, SpeechStream, TryRead};
13use crate::cancel::CancelToken;
14
15/// How often the playing thread looks at the cancel token.
16const CANCEL_POLL_INTERVAL: Duration = Duration::from_millis(50);
17
18/// Grace period for the device to play buffered audio, including Bluetooth latency.
19const DRAIN_GRACE: Duration = Duration::from_millis(600);
20
21/// Checks for an output device at construction and resolves it again for each utterance.
22pub struct SpeakerOutput;
23
24impl SpeakerOutput {
25    pub fn new() -> Result<Self, PlaybackError> {
26        default_output_device()?;
27        Ok(Self)
28    }
29
30    /// Calls `started` when the first source sample reaches the output callback.
31    /// Waits a drain grace period after the final sample. Cancellation drops the stream.
32    pub fn play_with(
33        &self,
34        speech: SpeechStream,
35        cancel: &CancelToken,
36        started: impl FnOnce(),
37    ) -> Result<(), PlaybackError> {
38        // Devices come and go, so each utterance resolves its own.
39        let device = default_output_device()?;
40        let config = device
41            .default_output_config()
42            .map_err(|source| PlaybackError::NoUsableConfig(Box::new(source)))?;
43
44        let (outcomes, outcome) = sync_channel(2);
45        let feed = Feed::new(speech, config.sample_rate(), outcomes.clone());
46        let stream = match config.sample_format() {
47            cpal::SampleFormat::F32 => open_stream::<f32>(&device, config.into(), feed, outcomes),
48            cpal::SampleFormat::I16 => open_stream::<i16>(&device, config.into(), feed, outcomes),
49            cpal::SampleFormat::U16 => open_stream::<u16>(&device, config.into(), feed, outcomes),
50            other => Err(PlaybackError::UnsupportedFormat(other.to_string())),
51        }?;
52        stream
53            .play()
54            .map_err(|source| PlaybackError::Stream(Box::new(source)))?;
55
56        let mut started = Some(started);
57        loop {
58            match outcome.recv_timeout(CANCEL_POLL_INTERVAL) {
59                Ok(PlaybackEnd::Started) => {
60                    if let Some(started) = started.take() {
61                        started();
62                    }
63                }
64                Ok(PlaybackEnd::Drained) => {
65                    drain(cancel);
66                    return Ok(());
67                }
68                Ok(PlaybackEnd::DeviceLost) => return Err(PlaybackError::DeviceLost),
69                Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
70                    if cancel.is_cancelled() {
71                        // Dropping the stream on return is the stop.
72                        return Err(PlaybackError::Cancelled);
73                    }
74                }
75                Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
76                    return Err(PlaybackError::DeviceLost);
77                }
78            }
79        }
80    }
81}
82
83/// Lets the device play out what it already holds. A stop cuts the wait
84/// short: the caller drops the stream next, and silence is what it asked for.
85fn drain(cancel: &CancelToken) {
86    let until = std::time::Instant::now() + DRAIN_GRACE;
87    while std::time::Instant::now() < until {
88        if cancel.is_cancelled() {
89            return;
90        }
91        std::thread::sleep(CANCEL_POLL_INTERVAL);
92    }
93}
94
95fn default_output_device() -> Result<cpal::Device, PlaybackError> {
96    cpal::default_host()
97        .default_output_device()
98        .ok_or(PlaybackError::NoOutputDevice)
99}
100
101fn open_stream<T>(
102    device: &cpal::Device,
103    config: cpal::StreamConfig,
104    mut feed: Feed,
105    outcomes: SyncSender<PlaybackEnd>,
106) -> Result<cpal::Stream, PlaybackError>
107where
108    T: SizedSample + FromSample<f32>,
109{
110    let channels = usize::from(config.channels);
111    device
112        .build_output_stream(
113            config,
114            move |data: &mut [T], _info: &cpal::OutputCallbackInfo| {
115                feed.fill(data, channels);
116            },
117            move |error| {
118                tracing::warn!(%error, "the output stream failed");
119                let _listener_gone = outcomes.try_send(PlaybackEnd::DeviceLost);
120            },
121            None,
122        )
123        .map_err(|source| PlaybackError::Stream(Box::new(source)))
124}
125
126enum PlaybackEnd {
127    Started,
128    /// Every source sample was handed to the device.
129    Drained,
130    DeviceLost,
131}
132
133/// The callback's side of the utterance: pulls source samples, converts the
134/// rate linearly, and duplicates mono across the device's channels.
135struct Feed {
136    speech: SpeechStream,
137    /// Source samples per output frame.
138    step: f64,
139    /// Fractional read position into the source, relative to `pending[0]`.
140    position: f64,
141    pending: Vec<f32>,
142    ended: bool,
143    reported: bool,
144    started: bool,
145    outcomes: SyncSender<PlaybackEnd>,
146}
147
148impl Feed {
149    fn new(speech: SpeechStream, device_rate: u32, outcomes: SyncSender<PlaybackEnd>) -> Self {
150        let step = f64::from(speech.sample_rate()) / f64::from(device_rate);
151        Self {
152            speech,
153            step,
154            position: 0.0,
155            pending: Vec::new(),
156            ended: false,
157            reported: false,
158            started: false,
159            outcomes,
160        }
161    }
162
163    fn fill<T>(&mut self, data: &mut [T], channels: usize)
164    where
165        T: SizedSample + FromSample<f32>,
166    {
167        for frame in data.chunks_mut(channels) {
168            let value = self.next_sample().unwrap_or(0.0);
169            let converted = T::from_sample(value);
170            for slot in frame {
171                *slot = converted;
172            }
173        }
174        if self.ended && self.consumed() && !self.reported {
175            self.reported = true;
176            let _listener_gone = self.outcomes.try_send(PlaybackEnd::Drained);
177        }
178    }
179
180    /// The next output-rate sample, or `None` while starved or after the end.
181    fn next_sample(&mut self) -> Option<f32> {
182        let wanted = self.position as usize + 2;
183        while !self.ended && self.pending.len() < wanted {
184            let mut chunk = [0.0; 512];
185            match self.speech.try_read(&mut chunk) {
186                TryRead::Samples(count) => self.pending.extend_from_slice(&chunk[..count]),
187                // An underrun plays as silence; the source position holds so
188                // nothing is skipped when samples catch up.
189                TryRead::Starved => return None,
190                TryRead::Ended => self.ended = true,
191            }
192        }
193        let index = self.position as usize;
194        let first = *self.pending.get(index)?;
195        let second = self.pending.get(index + 1).copied().unwrap_or(first);
196        let fraction = (self.position - index as f64) as f32;
197        self.position += self.step;
198
199        // What interpolation no longer reaches is dropped, keeping the
200        // buffer a chunk long rather than the whole utterance.
201        let consumed = self.position as usize;
202        if consumed > 0 {
203            self.pending.drain(..consumed.min(self.pending.len()));
204            self.position -= consumed as f64;
205        }
206        if !self.started {
207            self.started = true;
208            let _listener_gone = self.outcomes.try_send(PlaybackEnd::Started);
209        }
210        Some(first + (second - first) * fraction)
211    }
212
213    fn consumed(&self) -> bool {
214        self.pending.len() <= 1
215    }
216}
217
218#[cfg(test)]
219mod tests {
220    use std::sync::mpsc::TryRecvError;
221
222    use super::*;
223    use crate::synthesis::stream;
224
225    fn feed(source_rate: u32, device_rate: u32, samples: &[f32]) -> Feed {
226        let (mut writer, speech) = stream(source_rate);
227        writer.write(samples).expect("the reader is listening");
228        writer.finish();
229        let (outcomes, _receiver) = sync_channel(2);
230        Feed::new(speech, device_rate, outcomes)
231    }
232
233    #[test]
234    fn equal_rates_pass_samples_through() {
235        let mut feed = feed(24_000, 24_000, &[0.1, 0.2, 0.3]);
236        let mut data = [9.9f32; 4];
237
238        feed.fill(&mut data, 1);
239
240        assert_eq!(data[..3], [0.1, 0.2, 0.3]);
241        assert_eq!(data[3], 0.0, "past the end plays silence");
242    }
243
244    #[test]
245    fn a_mono_source_fills_every_device_channel() {
246        let mut feed = feed(24_000, 24_000, &[0.5]);
247        let mut data = [9.9f32; 2];
248
249        feed.fill(&mut data, 2);
250
251        assert_eq!(data, [0.5, 0.5]);
252    }
253
254    #[test]
255    fn upsampling_interpolates_between_source_samples() {
256        let mut feed = feed(24_000, 48_000, &[0.0, 1.0]);
257        let mut data = [9.9f32; 4];
258
259        feed.fill(&mut data, 1);
260
261        assert_eq!(data[..2], [0.0, 0.5]);
262    }
263
264    #[test]
265    fn the_drained_report_fires_once_everything_played() {
266        let (mut writer, speech) = stream(24_000);
267        writer.write(&[0.1, 0.2]).expect("the reader is listening");
268        writer.finish();
269        let (outcomes, ended) = sync_channel(2);
270        let mut feed = Feed::new(speech, 24_000, outcomes);
271        let mut data = [0.0f32; 8];
272
273        feed.fill(&mut data, 1);
274
275        assert!(matches!(ended.try_recv(), Ok(PlaybackEnd::Started)));
276        assert!(matches!(ended.try_recv(), Ok(PlaybackEnd::Drained)));
277        assert!(matches!(ended.try_recv(), Err(TryRecvError::Empty)));
278    }
279
280    #[test]
281    fn a_starved_stream_plays_silence_and_resumes_where_it_left_off() {
282        let (mut writer, speech) = stream(24_000);
283        let (outcomes, _receiver) = sync_channel(2);
284        let mut feed = Feed::new(speech, 24_000, outcomes);
285
286        let mut data = [9.9f32; 2];
287        feed.fill(&mut data, 1);
288        assert_eq!(data, [0.0, 0.0], "an underrun plays silence");
289        assert!(matches!(_receiver.try_recv(), Err(TryRecvError::Empty)));
290
291        writer.write(&[0.7]).expect("the reader is listening");
292        writer.finish();
293        let mut data = [9.9f32; 1];
294        feed.fill(&mut data, 1);
295        assert_eq!(data, [0.7], "no sample was skipped over the underrun");
296        assert!(matches!(_receiver.try_recv(), Ok(PlaybackEnd::Started)));
297    }
298}
aria_kit/synthesis/playback.rs

aria_kit/synthesis/
playback.rs
