active_call/synthesis/
supertonic.rs

1use crate::offline::get_offline_models;
2use crate::synthesis::{SynthesisClient, SynthesisEvent, SynthesisOption, SynthesisType};
3use anyhow::{Result, anyhow};
4use async_trait::async_trait;
5use audio_codec::BoxedResampler;
6use bytes::Bytes;
7use futures::stream::BoxStream;
8use tokio::sync::mpsc;
9use tokio_stream::wrappers::UnboundedReceiverStream;
10use tokio_util::sync::CancellationToken;
11use tracing::{debug, warn};
12
13pub struct SupertonicTtsClient {
14    voice_style: String,
15    speed: f32,
16    target_rate: i32,
17    tx: Option<mpsc::UnboundedSender<(Option<usize>, Result<SynthesisEvent>)>>,
18    token: CancellationToken,
19}
20
21impl SupertonicTtsClient {
22    pub fn create(_streaming: bool, option: &SynthesisOption) -> Result<Box<dyn SynthesisClient>> {
23        let voice_style = option.speaker.clone().unwrap_or_else(|| "M1".to_string());
24        let speed = option.speed.unwrap_or(1.0);
25        let target_rate = option.samplerate.unwrap_or(16000);
26
27        Ok(Box::new(Self {
28            voice_style,
29            speed,
30            target_rate,
31            tx: None,
32            token: CancellationToken::new(),
33        }))
34    }
35
36    fn ensure_models_initialized() -> Result<()> {
37        if get_offline_models().is_none() {
38            anyhow::bail!(
39                "Offline models not initialized. Please call init_offline_models() first."
40            );
41        }
42        Ok(())
43    }
44
45    async fn synthesize_text(&self, text: String, cmd_seq: Option<usize>) -> Result<()> {
46        let Some(tx) = self.tx.as_ref() else {
47            return Ok(());
48        };
49
50        let models =
51            get_offline_models().ok_or_else(|| anyhow!("offline models not initialized"))?;
52
53        let voice_style = self.voice_style.clone();
54        let speed = self.speed;
55        let target_rate = self.target_rate;
56        let tx_clone = tx.clone();
57        // Supertonic: en is hardcoded for now, or detect from text?
58        let language = "en".to_string();
59
60        let tts_arc = models.get_supertonic().await?;
61
62        // Run synthesis in blocking task
63        tokio::task::spawn_blocking(move || {
64            // Use try_write to avoid potential panics or deadlocks;
65            // blocking_write can panic/deadlock when the lock is held by an async task.
66            let mut guard = match tts_arc.try_write() {
67                Ok(g) => g,
68                Err(_) => {
69                    warn!("Supertonic TTS write lock unavailable, skipping synthesis");
70                    let _ = tx_clone.send((cmd_seq, Err(anyhow!("TTS write lock unavailable"))));
71                    return;
72                }
73            };
74
75            if let Some(tts) = guard.as_mut() {
76                debug!(
77                    text = %text,
78                    voice = %voice_style,
79                    speed = speed,
80                    target_rate = target_rate,
81                    "Calling Supertonic TTS synthesis"
82                );
83
84                match tts.synthesize(&text, &language, Some(&voice_style), Some(speed)) {
85                    Ok(samples) => {
86                        if !samples.is_empty() {
87                            let mut samples_i16: Vec<i16> = samples
88                                .iter()
89                                .map(|&s| (s * 32768.0).max(-32768.0).min(32767.0) as i16)
90                                .collect();
91
92                            // Resample if needed
93                            if tts.sample_rate() != target_rate {
94                                let mut resampler = BoxedResampler::new(
95                                    tts.sample_rate() as usize,
96                                    target_rate as usize,
97                                )
98                                .expect("invalid sample rate");
99                                samples_i16 = resampler.resample(&samples_i16);
100                            }
101
102                            // Convert i16 samples to PCM bytes
103                            let mut bytes = Vec::with_capacity(samples_i16.len() * 2);
104                            for s in samples_i16 {
105                                bytes.extend_from_slice(&s.to_le_bytes());
106                            }
107
108                            // Send AudioChunk
109                            let _ = tx_clone.send((
110                                cmd_seq,
111                                Ok(SynthesisEvent::AudioChunk(Bytes::from(bytes))),
112                            ));
113
114                            // Send Finished
115                            let _ = tx_clone.send((cmd_seq, Ok(SynthesisEvent::Finished)));
116                        } else {
117                            warn!("Supertonic produced empty audio");
118                            let _ = tx_clone.send((cmd_seq, Ok(SynthesisEvent::Finished)));
119                        }
120                    }
121                    Err(e) => {
122                        warn!(error = %e, "Supertonic inference failed");
123                        let _ = tx_clone.send((cmd_seq, Err(anyhow!("Synthesis failed: {}", e))));
124                    }
125                }
126            } else {
127                warn!("Supertonic TTS not initialized");
128                let _ = tx_clone.send((cmd_seq, Err(anyhow!("TTS not initialized"))));
129            }
130        })
131        .await
132        .map_err(|e| anyhow!("task join error: {}", e))?;
133
134        Ok(())
135    }
136}
137
138#[async_trait]
139impl SynthesisClient for SupertonicTtsClient {
140    fn provider(&self) -> SynthesisType {
141        SynthesisType::Supertonic
142    }
143
144    async fn start(
145        &mut self,
146    ) -> Result<BoxStream<'static, (Option<usize>, Result<SynthesisEvent>)>> {
147        Self::ensure_models_initialized()?;
148
149        let (tx, rx) = mpsc::unbounded_channel();
150        self.tx = Some(tx);
151
152        // Initialize TTS if needed
153        let models =
154            get_offline_models().ok_or_else(|| anyhow!("offline models not initialized"))?;
155        models.init_supertonic().await?;
156
157        debug!(
158            "SupertonicTtsClient started with voice: {}",
159            self.voice_style
160        );
161
162        Ok(Box::pin(UnboundedReceiverStream::new(rx)))
163    }
164
165    async fn synthesize(
166        &mut self,
167        text: &str,
168        cmd_seq: Option<usize>,
169        _option: Option<SynthesisOption>,
170    ) -> Result<()> {
171        self.synthesize_text(text.to_string(), cmd_seq).await
172    }
173
174    async fn stop(&mut self) -> Result<()> {
175        self.token.cancel();
176        self.tx = None;
177        Ok(())
178    }
179}
active_call/synthesis/supertonic.rs

active_call/synthesis/
supertonic.rs
