active_call/synthesis/
deepgram.rs

1use crate::synthesis::{SynthesisClient, SynthesisEvent, SynthesisOption, SynthesisType};
2use anyhow::Result;
3use anyhow::anyhow;
4use async_trait::async_trait;
5use bytes::Bytes;
6use futures::SinkExt;
7use futures::StreamExt;
8use futures::TryStreamExt;
9use futures::future;
10use futures::future::FutureExt;
11use futures::stream;
12use futures::stream::SplitSink;
13use futures::{Stream, stream::BoxStream};
14use serde::Deserialize;
15use serde::Serialize;
16use tokio::net::TcpStream;
17use tokio::sync::mpsc;
18use tokio_stream::wrappers::UnboundedReceiverStream;
19use tokio_tungstenite::MaybeTlsStream;
20use tokio_tungstenite::WebSocketStream;
21use tokio_tungstenite::connect_async;
22use tokio_tungstenite::tungstenite::Message;
23use tokio_tungstenite::tungstenite::client::IntoClientRequest;
24use tracing::warn;
25use url::Url;
26
27type WsStream = WebSocketStream<MaybeTlsStream<TcpStream>>;
28type WsSink = SplitSink<WsStream, Message>;
29
30const DEEPGRAM_BASE_URL: &str = "https://api.deepgram.com/v1/speak";
31const TERMINATORS: [char; 3] = ['.', '?', '!'];
32
33// https://developers.deepgram.com/docs/tts-rest
34pub struct RestClient {
35    option: SynthesisOption,
36    tx: Option<mpsc::UnboundedSender<(String, Option<usize>, Option<SynthesisOption>)>>,
37}
38
39#[derive(Serialize)]
40struct Payload {
41    text: String,
42}
43
44impl RestClient {
45    pub fn new(option: SynthesisOption) -> Self {
46        Self { option, tx: None }
47    }
48}
49
50// deepgram model parameter format: details: https://developers.deepgram.com/docs/tts-models
51// model: voice and language: [modelname]-[voicename]-[language],
52// encoding: linear16,mulaw,alaw, etc.
53// sample_rate: 8000, 16000, etc.
54fn request_url(option: &SynthesisOption, protocol: &str) -> Url {
55    let mut url = Url::parse(DEEPGRAM_BASE_URL).expect("Deepgram base url is invalid");
56    url.set_scheme(protocol).expect("illegal url scheme");
57
58    let extra = option.extra.as_ref();
59    let encoding = option
60        .codec
61        .as_deref()
62        .map(|codec| match codec {
63            "pcm" | "linear16" => "linear16",
64            "pcmu" | "mulaw" => "mulaw",
65            "pcma" | "alaw" => "alaw",
66            other => other,
67        })
68        .unwrap_or("linear16");
69
70    let mut query = url.query_pairs_mut();
71
72    if extra.map(|e| !e.contains_key("model")).unwrap_or(true) {
73        if let Some(model) = option.model.as_ref().or(option.speaker.as_ref()) {
74            query.append_pair("model", model);
75        }
76    }
77
78    if extra.map(|e| !e.contains_key("encoding")).unwrap_or(true) {
79        query.append_pair("encoding", encoding);
80    }
81
82    if extra.map(|e| !e.contains_key("sample_rate")).unwrap_or(true)
83        && matches!(encoding, "linear16" | "mulaw" | "alaw")
84    {
85        let samplerate = option.samplerate.unwrap_or(16000);
86        query.append_pair("sample_rate", samplerate.to_string().as_str());
87    }
88
89    if extra.map(|e| !e.contains_key("container")).unwrap_or(true)
90        && matches!(encoding, "linear16" | "mulaw" | "alaw")
91    {
92        query.append_pair("container", "none");
93    }
94
95    if let Some(extra) = extra {
96        for (key, value) in extra {
97            query.append_pair(key, value);
98        }
99    }
100
101    drop(query);
102    url
103}
104
105async fn chunked_stream(
106    option: SynthesisOption,
107    text: String,
108) -> Result<impl Stream<Item = Result<Bytes>>> {
109    let url = request_url(&option, "https");
110    let token = option
111        .secret_key
112        .as_ref()
113        .ok_or_else(|| anyhow!("Deepegram tts: missing api key"))?;
114    let payload = Payload { text };
115    let client = reqwest::Client::new();
116    let resp = client
117        .post(url)
118        .header("Content-Type", "application/json")
119        .header("Authorization", format!("Token {}", token))
120        .json(&payload)
121        .send()
122        .await?;
123    if !resp.status().is_success() {
124        let status = resp.status();
125        let body = resp.text().await.unwrap_or_default();
126        return Err(anyhow!("Deepgram TTS request failed: {} {}", status, body));
127    }
128    Ok(resp.bytes_stream().map_err(anyhow::Error::from))
129}
130
131#[async_trait]
132impl SynthesisClient for RestClient {
133    fn provider(&self) -> SynthesisType {
134        SynthesisType::Deepgram
135    }
136
137    async fn start(
138        &mut self,
139    ) -> Result<BoxStream<'static, (Option<usize>, Result<SynthesisEvent>)>> {
140        let (tx, rx) = mpsc::unbounded_channel();
141        self.tx = Some(tx);
142        let max_concurrent_tasks = self.option.max_concurrent_tasks.unwrap_or(1);
143        let client_option = self.option.clone();
144        let stream = UnboundedReceiverStream::new(rx).flat_map_unordered(
145            max_concurrent_tasks,
146            move |(text, cmd_seq, cmd_option)| {
147                let option = client_option.merge_with(cmd_option);
148                chunked_stream(option, text)
149                    .map(move |res| match res {
150                        Ok(stream) => stream
151                            .map(move |res| res.map(|bytes| SynthesisEvent::AudioChunk(bytes)))
152                            .chain(stream::once(future::ready(Ok(SynthesisEvent::Finished))))
153                            .boxed(),
154                        Err(e) => stream::once(future::ready(Err(e))).boxed(),
155                    })
156                    .flatten_stream()
157                    .map(move |res| (cmd_seq, res))
158                    .boxed()
159            },
160        );
161
162        Ok(stream.boxed())
163    }
164
165    async fn synthesize(
166        &mut self,
167        text: &str,
168        cmd_seq: Option<usize>,
169        option: Option<SynthesisOption>,
170    ) -> Result<()> {
171        if let Some(tx) = &self.tx {
172            tx.send((text.to_string(), cmd_seq, option))?;
173        } else {
174            return Err(anyhow::anyhow!("Deepgram TTS: missing client sender"));
175        };
176        Ok(())
177    }
178
179    async fn stop(&mut self) -> Result<()> {
180        self.tx.take();
181        Ok(())
182    }
183}
184
185struct StreamingClient {
186    option: SynthesisOption,
187    sink: Option<WsSink>,
188}
189
190impl StreamingClient {
191    pub fn new(option: SynthesisOption) -> Self {
192        Self { option, sink: None }
193    }
194}
195
196#[derive(Serialize)]
197#[serde(tag = "type")]
198enum Command {
199    Speak { text: String },
200    Flush,
201    Close,
202}
203
204#[allow(dead_code)]
205#[derive(Deserialize, Debug)]
206#[serde(tag = "type")]
207enum Event {
208    Metadata {
209        request_id: String,
210        model_name: String,
211        model_version: String,
212        model_uuid: String,
213    },
214    Flushed {
215        sequence_id: usize,
216    },
217    Cleared {
218        sequence_id: usize,
219    },
220    Warning {
221        description: String,
222        code: String,
223    },
224}
225
226async fn connect(option: SynthesisOption) -> Result<WsStream> {
227    let url = request_url(&option, "wss");
228    let mut request = url.as_str().into_client_request()?;
229    let token = option
230        .secret_key
231        .as_ref()
232        .ok_or_else(|| anyhow!("Deepegram tts: missing api key"))?;
233    request
234        .headers_mut()
235        .insert("Authorization", format!("Token {}", token).parse()?);
236    let (ws_stream, _) = connect_async(request).await?;
237    Ok(ws_stream)
238}
239
240#[async_trait]
241impl SynthesisClient for StreamingClient {
242    fn provider(&self) -> SynthesisType {
243        SynthesisType::Deepgram
244    }
245
246    async fn start(
247        &mut self,
248    ) -> Result<BoxStream<'static, (Option<usize>, Result<SynthesisEvent>)>> {
249        let (sink, source) = connect(self.option.clone()).await?.split();
250        self.sink = Some(sink);
251        let stream = source
252            .filter_map(async move |message| match message {
253                Ok(Message::Binary(bytes)) => Some(Ok(SynthesisEvent::AudioChunk(bytes))),
254                Ok(Message::Text(text)) => {
255                    let event: Event =
256                        serde_json::from_str(&text).expect("Deepgram TTS API changed!");
257
258                    if let Event::Warning { description, code } = event {
259                        warn!("Deepgram TTS: warning: {}, {}", description, code);
260                    }
261
262                    None
263                }
264                Ok(Message::Close(_)) => Some(Ok(SynthesisEvent::Finished)),
265                Err(e) => Some(Err(anyhow!("Deepgram TTS: websocket error: {:?}", e))),
266                _ => None,
267            })
268            .map(|res| (None, res))
269            .boxed();
270        Ok(stream)
271    }
272
273    async fn synthesize(
274        &mut self,
275        text: &str,
276        _cmd_seq: Option<usize>,
277        _option: Option<SynthesisOption>,
278    ) -> Result<()> {
279        if let Some(sink) = &mut self.sink {
280            // deepgram should mannualy flush, use terminators to split text
281
282            for sentence in text.split_inclusive(&TERMINATORS[..]) {
283                if !sentence.is_empty() {
284                    let speak_cmd = Command::Speak {
285                        text: sentence.to_string(),
286                    };
287                    let speak_json = serde_json::to_string(&speak_cmd)?;
288                    sink.send(Message::text(speak_json)).await?;
289                }
290
291                if sentence.ends_with(&TERMINATORS[..]) {
292                    let flush_cmd = Command::Flush;
293                    let flush_json = serde_json::to_string(&flush_cmd)?;
294                    sink.send(Message::text(flush_json)).await?;
295                }
296            }
297        } else {
298            return Err(anyhow::anyhow!("Deepgram TTS: missing sink"));
299        };
300        Ok(())
301    }
302
303    async fn stop(&mut self) -> Result<()> {
304        if let Some(mut sink) = self.sink.take() {
305            let close_cmd = Command::Close;
306            let close_json = serde_json::to_string(&close_cmd)?;
307            sink.send(Message::text(close_json)).await?;
308        } else {
309            warn!("Deepgram TTS: missing sink");
310        }
311        Ok(())
312    }
313}
314
315pub struct DeepegramTtsClient;
316
317impl DeepegramTtsClient {
318    pub fn create(streaming: bool, option: &SynthesisOption) -> Result<Box<dyn SynthesisClient>> {
319        if streaming {
320            Ok(Box::new(StreamingClient::new(option.clone())))
321        } else {
322            Ok(Box::new(RestClient::new(option.clone())))
323        }
324    }
325}
active_call/synthesis/deepgram.rs

active_call/synthesis/
deepgram.rs
