Skip to content

Commit ee2309b

Browse files
shenjintishenjinti
committed
refactor(tts): synthesis clients with stream mode
1 parent 1c82f74 commit ee2309b

12 files changed

Lines changed: 714 additions & 552 deletions

File tree

‎examples/voice_demo.rs‎

Lines changed: 36 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ use futures::StreamExt;
77
use rustpbx::llm::LlmContent;
88
use rustpbx::media::codecs::bytes_to_samples;
99
use rustpbx::media::track::file::read_wav_file;
10-
use rustpbx::synthesis::{SynthesisClient, TTSEvent};
10+
use rustpbx::synthesis::{SynthesisClient, SynthesisEvent};
1111
use rustpbx::transcription::TencentCloudAsrClientBuilder;
1212
use rustpbx::{PcmBuf, Sample};
1313
use std::collections::VecDeque;
@@ -436,43 +436,44 @@ async fn main() -> Result<()> {
436436
if let Ok(LlmContent::Final(text)) = content {
437437
info!("LLM response: {}ms {}", st.elapsed().as_millis(), text);
438438
let st = Instant::now();
439-
if let Ok(mut audio_stream) =
440-
tts_client.synthesize(&text, None, None).await
441-
{
442-
let mut total_bytes = 0;
443-
while let Some(Ok(event)) = audio_stream.next().await {
444-
match event {
445-
TTSEvent::AudioChunk(chunk) => {
446-
total_bytes += chunk.len();
447-
let audio_data: PcmBuf =
448-
bytes_to_samples(&chunk);
449-
let final_audio =
450-
if sample_rate != output_sample_rate {
451-
resample::resample_mono(
452-
&audio_data,
453-
sample_rate,
454-
output_sample_rate,
455-
)
456-
} else {
457-
audio_data
458-
};
459-
output_buffer.push(&final_audio);
460-
}
461-
TTSEvent::Finished => {
462-
break;
463-
}
464-
_ => {}
439+
let mut audio_stream = tts_client
440+
.start()
441+
.await
442+
.expect("Failed to start TTS stream");
443+
tts_client
444+
.synthesize(&text, None, None, None)
445+
.await
446+
.expect("Failed to synthesize text");
447+
448+
let mut total_bytes = 0;
449+
while let Some(Ok(event)) = audio_stream.next().await {
450+
match event {
451+
SynthesisEvent::AudioChunk(chunk) => {
452+
total_bytes += chunk.len();
453+
let audio_data: PcmBuf = bytes_to_samples(&chunk);
454+
let final_audio =
455+
if sample_rate != output_sample_rate {
456+
resample::resample_mono(
457+
&audio_data,
458+
sample_rate,
459+
output_sample_rate,
460+
)
461+
} else {
462+
audio_data
463+
};
464+
output_buffer.push(&final_audio);
465465
}
466+
SynthesisEvent::Finished => {
467+
break;
468+
}
469+
_ => {}
466470
}
467-
info!(
468-
"TTS synthesis: {}ms ({}) bytes",
469-
st.elapsed().as_millis(),
470-
total_bytes
471-
);
472-
} else {
473-
error!("Error synthesizing TTS");
474-
break;
475471
}
472+
info!(
473+
"TTS synthesis: {}ms ({}) bytes",
474+
st.elapsed().as_millis(),
475+
total_bytes
476+
);
476477
} else if let Err(e) = content {
477478
error!("Error generating LLM response: {}", e);
478479
break;

‎src/bin/text2wav.rs‎

Lines changed: 24 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ use tracing::{debug, error, info};
1010

1111
use rustpbx::media::codecs::bytes_to_samples;
1212
use rustpbx::synthesis::{
13-
SynthesisClient, SynthesisOption, SynthesisType, TTSEvent, TencentCloudTtsClient,
13+
SynthesisClient, SynthesisEvent, SynthesisOption, SynthesisType, TencentCloudTtsClient,
1414
};
1515

1616
const SAMPLE_RATE: u32 = 16000;
@@ -228,42 +228,34 @@ async fn main() -> Result<()> {
228228
// Generate speech for the segment text
229229
if !segment.text.is_empty() {
230230
info!("Synthesizing text: {}", segment.text);
231-
232-
match tts_client.synthesize(&segment.text, None, None).await {
233-
Ok(mut audio_stream) => {
234-
let mut total_bytes = 0;
235-
while let Some(chunk_result) = audio_stream.next().await {
236-
match chunk_result {
237-
Ok(event) => match event {
238-
TTSEvent::AudioChunk(chunk) => {
239-
debug!("Received chunk of {} bytes", chunk.len());
240-
total_bytes += chunk.len();
241-
let samples: PcmBuf = bytes_to_samples(&chunk);
242-
for &sample in &samples {
243-
writer.write_sample(sample)?;
244-
}
245-
}
246-
TTSEvent::Finished => {
247-
break;
248-
}
249-
_ => {}
250-
},
251-
Err(e) => {
252-
error!("Error in audio stream chunk: {:?}", e);
253-
return Err(anyhow::anyhow!(
254-
"Failed to process audio chunk: {}",
255-
e
256-
));
231+
let mut audio_stream = tts_client.start().await?;
232+
tts_client
233+
.synthesize(&segment.text, None, None, None)
234+
.await?;
235+
let mut total_bytes = 0;
236+
while let Some(chunk_result) = audio_stream.next().await {
237+
match chunk_result {
238+
Ok(event) => match event {
239+
SynthesisEvent::AudioChunk(chunk) => {
240+
debug!("Received chunk of {} bytes", chunk.len());
241+
total_bytes += chunk.len();
242+
let samples: PcmBuf = bytes_to_samples(&chunk);
243+
for &sample in &samples {
244+
writer.write_sample(sample)?;
257245
}
258246
}
247+
SynthesisEvent::Finished => {
248+
break;
249+
}
250+
_ => {}
251+
},
252+
Err(e) => {
253+
error!("Error in audio stream chunk: {:?}", e);
254+
return Err(anyhow::anyhow!("Failed to process audio chunk: {}", e));
259255
}
260-
debug!("Received total of {} bytes of audio data", total_bytes);
261-
}
262-
Err(e) => {
263-
error!("Failed to synthesize text: {}", e);
264-
return Err(anyhow::anyhow!("Failed to synthesize text: {}", e));
265256
}
266257
}
258+
debug!("Received total of {} bytes of audio data", total_bytes);
267259
}
268260
}
269261

‎src/call/active_call.rs‎

Lines changed: 17 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,12 @@ use crate::{
1717
Track, TrackConfig,
1818
file::FileTrack,
1919
rtp::{RtpTrack, RtpTrackBuilder},
20-
tts::{TtsCommand, TtsHandle},
20+
tts::SynthesisHandle,
2121
webrtc::WebrtcTrack,
2222
websocket::{WebsocketBytesReceiver, WebsocketTrack},
2323
},
2424
},
25-
synthesis::SynthesisOption,
25+
synthesis::{SynthesisCommand, SynthesisOption},
2626
useragent::invitation::PendingDialog,
2727
};
2828
use anyhow::Result;
@@ -78,7 +78,7 @@ pub struct ActiveCall {
7878
pub session_id: String,
7979
pub media_stream: Arc<MediaStream>,
8080
pub track_config: TrackConfig,
81-
pub tts_handle: Mutex<Option<TtsHandle>>,
81+
pub tts_handle: Mutex<Option<SynthesisHandle>>,
8282
pub auto_hangup: Arc<Mutex<Option<(u32, CallRecordHangupReason)>>>,
8383
pub wait_input_timeout: Arc<Mutex<Option<u32>>>,
8484
pub event_sender: EventSender,
@@ -519,7 +519,7 @@ impl ActiveCall {
519519
Some(s) => Some(s),
520520
None => tts_option.speaker.clone(),
521521
};
522-
let mut play_command = TtsCommand {
522+
let mut play_command = SynthesisCommand {
523523
text,
524524
speaker,
525525
play_id: play_id.clone(),
@@ -533,7 +533,7 @@ impl ActiveCall {
533533
text = %play_command.text,
534534
speaker = ?play_command.speaker,
535535
auto_hangup = ?auto_hangup,
536-
"new tts command"
536+
"new synthesis command"
537537
);
538538

539539
let ssrc = rand::random::<u32>();
@@ -1181,14 +1181,21 @@ impl ActiveCall {
11811181
)
11821182
.await?;
11831183

1184+
let offer = rtp_track.local_description().ok();
11841185
let call_option = call_state_ref
1185-
.read()
1186-
.as_ref()
1187-
.and_then(|cs| Ok(cs.option.clone()))
1188-
.unwrap_or_default()
1186+
.write()
1187+
.as_mut()
1188+
.ok()
1189+
.map(|cs| {
1190+
cs.option.as_mut().map(|o| {
1191+
o.offer = offer.clone();
1192+
});
1193+
cs.start_time = Utc::now();
1194+
cs.option.clone()
1195+
})
1196+
.flatten()
11891197
.unwrap_or_default();
11901198

1191-
let offer = rtp_track.local_description().ok();
11921199
invite_option.offer = offer.clone().map(|s| s.into());
11931200

11941201
Self::setup_track_with_stream(
@@ -1255,24 +1262,6 @@ impl ActiveCall {
12551262
.update_remote_description(&track_id, &answer)
12561263
.await
12571264
.ok();
1258-
1259-
call_state_ref
1260-
.write()
1261-
.as_mut()
1262-
.and_then(|cs| {
1263-
if cs.dialog.is_none() {
1264-
Some(DialogGuard::new(
1265-
self.invitation.dialog_layer.clone(),
1266-
dialog_id,
1267-
));
1268-
}
1269-
cs.option.as_mut().map(|o| o.offer = offer);
1270-
cs.answer = Some(answer);
1271-
cs.answer_time = Some(Utc::now());
1272-
cs.last_status_code = 200;
1273-
Ok(())
1274-
})
1275-
.ok();
12761265
Ok(())
12771266
}
12781267

‎src/call/sip.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -257,7 +257,7 @@ pub async fn client_dialog_event_loop(
257257
info!(session_id, track_id, %dialog_id, "client dialog calling");
258258
}
259259
DialogState::Confirmed(dialog_id) => {
260-
info!(session_id, track_id, %dialog_id, "dialog confirmed");
260+
info!(session_id, track_id, %dialog_id, "client dialog confirmed");
261261
call_state
262262
.write()
263263
.as_mut()

‎src/media/engine.rs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,13 @@ use super::{
33
denoiser::NoiseReducer,
44
processor::Processor,
55
track::{
6-
tts::{TtsHandle, TtsTrack},
76
Track,
7+
tts::{SynthesisHandle, TtsTrack},
88
},
99
vad::{VADOption, VadProcessor, VadType},
1010
};
1111
use crate::{
12+
TrackId,
1213
call::{CallOption, EouOption},
1314
event::EventSender,
1415
synthesis::{
@@ -19,7 +20,6 @@ use crate::{
1920
AliyunAsrClientBuilder, TencentCloudAsrClientBuilder, TranscriptionClient,
2021
TranscriptionOption, TranscriptionType, VoiceApiAsrClientBuilder,
2122
},
22-
TrackId,
2323
};
2424
use anyhow::Result;
2525
use std::{collections::HashMap, future::Future, pin::Pin, sync::Arc};
@@ -235,9 +235,9 @@ impl StreamEngine {
235235
ssrc: u32,
236236
play_id: Option<String>,
237237
tts_option: &SynthesisOption,
238-
) -> Result<(TtsHandle, Box<dyn Track>)> {
238+
) -> Result<(SynthesisHandle, Box<dyn Track>)> {
239239
let (tx, rx) = mpsc::unbounded_channel();
240-
let new_handle = TtsHandle::new(tx, play_id);
240+
let new_handle = SynthesisHandle::new(tx, play_id);
241241
let tts_client = engine.create_tts_client(tts_option).await?;
242242
let tts_track = TtsTrack::new(track_id, session_id, rx, tts_client)
243243
.with_ssrc(ssrc)

0 commit comments

Comments
 (0)