gemini_adk_fluent_rs/voice/
devices.rs

1//! The `cpal` device adapter behind [`Talk`] — default microphone in, default
2//! speakers out, barge-in flush, drain signaling. Feature `voice-io`.
3
4use std::collections::VecDeque;
5use std::sync::atomic::{AtomicBool, Ordering};
6use std::sync::{Arc, Mutex};
7use std::time::Duration;
8
9use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
10use gemini_adk_rs::live::LiveHandle;
11use tokio::sync::mpsc;
12
13use super::{Playback, downmix, pump};
14
15/// Errors from the device layer.
16#[derive(Debug, thiserror::Error)]
17pub enum VoiceIoError {
18    /// No default input (microphone) device.
19    #[error("no default input device")]
20    NoInputDevice,
21    /// No default output (speaker) device.
22    #[error("no default output device")]
23    NoOutputDevice,
24    /// The audio backend refused a stream.
25    #[error("audio backend: {0}")]
26    Backend(String),
27}
28
29/// Run a full-duplex voice conversation on the system's default audio
30/// devices.
31///
32/// `talk()` bridges the default microphone and speakers into the session via
33/// [`pump`](super::pump): capture is down-mixed to mono and resampled to the
34/// session's input rate; model speech is resampled to the device rate and
35/// buffered; an interruption flushes the buffer instantly (barge-in); the
36/// session's voice reactor is told when playback drains. Returns when the
37/// session ends or on Ctrl-C.
38#[allow(async_fn_in_trait)]
39pub trait Talk {
40    /// See the trait docs. The five-line voice app:
41    ///
42    /// ```ignore
43    /// // `ignore`: opens the default microphone and speakers (feature `voice-io`).
44    /// Live::builder()
45    ///     .instruction("You are a helpful concierge.")
46    ///     .greeting("Greet the caller.")
47    ///     .connect_from_env().await?
48    ///     .talk().await?;
49    /// ```
50    async fn talk(&self) -> Result<(), VoiceIoError>;
51}
52
53impl Talk for LiveHandle {
54    async fn talk(&self) -> Result<(), VoiceIoError> {
55        let host = cpal::default_host();
56        let input = host
57            .default_input_device()
58            .ok_or(VoiceIoError::NoInputDevice)?;
59        let output = host
60            .default_output_device()
61            .ok_or(VoiceIoError::NoOutputDevice)?;
62        let input_config = input
63            .default_input_config()
64            .map_err(|e| VoiceIoError::Backend(e.to_string()))?;
65        let output_config = output
66            .default_output_config()
67            .map_err(|e| VoiceIoError::Backend(e.to_string()))?;
68
69        let mic_hz = input_config.sample_rate().0;
70        let mic_channels = input_config.channels();
71        let speaker_hz = output_config.sample_rate().0;
72        let speaker_channels = output_config.channels();
73
74        // Microphone → pump. Bounded; a saturated channel drops the frame
75        // (mic loss under pressure beats blocking the audio callback).
76        let (mic_tx, mic_rx) = mpsc::channel::<Vec<i16>>(64);
77        let capture = move |mono: Vec<i16>| {
78            let _ = mic_tx.try_send(mono);
79        };
80        let input_stream = build_input_stream(&input, &input_config, mic_channels, capture)?;
81
82        // Pump → speaker ring. The cpal output callback drains the ring;
83        // `Flush` clears it — barge-in silences the device within one buffer.
84        let ring: Arc<Mutex<VecDeque<i16>>> = Arc::new(Mutex::new(VecDeque::new()));
85        let was_playing = Arc::new(AtomicBool::new(false));
86        let (spk_tx, mut spk_rx) = mpsc::channel::<Playback>(64);
87        {
88            let ring = ring.clone();
89            let was_playing = was_playing.clone();
90            tokio::spawn(async move {
91                while let Some(playback) = spk_rx.recv().await {
92                    let mut ring = ring.lock().expect("playback ring poisoned");
93                    match playback {
94                        Playback::Chunk(samples) => {
95                            ring.extend(samples);
96                            was_playing.store(true, Ordering::Relaxed);
97                        }
98                        Playback::Flush => ring.clear(),
99                    }
100                }
101            });
102        }
103        let output_stream =
104            build_output_stream(&output, &output_config, speaker_channels, ring.clone())?;
105
106        input_stream
107            .play()
108            .map_err(|e| VoiceIoError::Backend(e.to_string()))?;
109        output_stream
110            .play()
111            .map_err(|e| VoiceIoError::Backend(e.to_string()))?;
112
113        let running = pump(self, mic_rx, mic_hz, spk_tx, speaker_hz);
114
115        // Tell the voice reactor when the speaker goes quiet, so prompt
116        // gating and barge-in accounting see real playback state.
117        let drain_handle = self.clone();
118        let drain_task = tokio::spawn(async move {
119            let mut interval = tokio::time::interval(Duration::from_millis(50));
120            loop {
121                interval.tick().await;
122                let empty = ring.lock().expect("playback ring poisoned").is_empty();
123                if empty && was_playing.swap(false, Ordering::Relaxed) {
124                    let _ = drain_handle.playback_drained().await;
125                }
126            }
127        });
128
129        // Converse until the session ends or the user hits Ctrl-C.
130        tokio::select! {
131            _ = running.join() => {}
132            _ = tokio::signal::ctrl_c() => {
133                let _ = self.disconnect().await;
134            }
135        }
136        drain_task.abort();
137        drop(input_stream);
138        drop(output_stream);
139        Ok(())
140    }
141}
142
143fn build_input_stream(
144    device: &cpal::Device,
145    config: &cpal::SupportedStreamConfig,
146    channels: u16,
147    on_mono: impl Fn(Vec<i16>) + Send + 'static,
148) -> Result<cpal::Stream, VoiceIoError> {
149    let stream_config: cpal::StreamConfig = config.config();
150    let err = |e: cpal::BuildStreamError| VoiceIoError::Backend(e.to_string());
151    let stream = match config.sample_format() {
152        cpal::SampleFormat::I16 => device
153            .build_input_stream(
154                &stream_config,
155                move |data: &[i16], _| on_mono(downmix(data, channels)),
156                |e| tracing::warn!("input stream error: {e}"),
157                None,
158            )
159            .map_err(err)?,
160        cpal::SampleFormat::F32 => device
161            .build_input_stream(
162                &stream_config,
163                move |data: &[f32], _| {
164                    let pcm: Vec<i16> = data
165                        .iter()
166                        .map(|&s| (s.clamp(-1.0, 1.0) * i16::MAX as f32) as i16)
167                        .collect();
168                    on_mono(downmix(&pcm, channels));
169                },
170                |e| tracing::warn!("input stream error: {e}"),
171                None,
172            )
173            .map_err(err)?,
174        other => {
175            return Err(VoiceIoError::Backend(format!(
176                "unsupported input sample format {other:?}"
177            )));
178        }
179    };
180    Ok(stream)
181}
182
183fn build_output_stream(
184    device: &cpal::Device,
185    config: &cpal::SupportedStreamConfig,
186    channels: u16,
187    ring: Arc<Mutex<VecDeque<i16>>>,
188) -> Result<cpal::Stream, VoiceIoError> {
189    let stream_config: cpal::StreamConfig = config.config();
190    let err = |e: cpal::BuildStreamError| VoiceIoError::Backend(e.to_string());
191    let channels = channels as usize;
192    let stream = match config.sample_format() {
193        cpal::SampleFormat::I16 => device
194            .build_output_stream(
195                &stream_config,
196                move |data: &mut [i16], _| {
197                    let mut ring = ring.lock().expect("playback ring poisoned");
198                    for frame in data.chunks_mut(channels) {
199                        let sample = ring.pop_front().unwrap_or(0);
200                        frame.fill(sample);
201                    }
202                },
203                |e| tracing::warn!("output stream error: {e}"),
204                None,
205            )
206            .map_err(err)?,
207        cpal::SampleFormat::F32 => device
208            .build_output_stream(
209                &stream_config,
210                move |data: &mut [f32], _| {
211                    let mut ring = ring.lock().expect("playback ring poisoned");
212                    for frame in data.chunks_mut(channels) {
213                        let sample = ring.pop_front().unwrap_or(0) as f32 / i16::MAX as f32;
214                        frame.fill(sample);
215                    }
216                },
217                |e| tracing::warn!("output stream error: {e}"),
218                None,
219            )
220            .map_err(err)?,
221        other => {
222            return Err(VoiceIoError::Backend(format!(
223                "unsupported output sample format {other:?}"
224            )));
225        }
226    };
227    Ok(stream)
228}
gemini_adk_fluent_rs/voice/devices.rs

gemini_adk_fluent_rs/voice/
devices.rs
