active_call/media/vad/
mod.rs

1use crate::event::{EventSender, SessionEvent};
2use crate::media::processor::Processor;
3use crate::media::{AudioFrame, PcmBuf, Samples};
4use anyhow::Result;
5use serde::{Deserialize, Serialize};
6use serde_with::skip_serializing_none;
7use std::any::Any;
8use tokio_util::sync::CancellationToken;
9
10pub(crate) mod simd;
11pub mod tiny_silero;
12pub(crate) mod utils;
13pub use tiny_silero::TinySilero;
14
15#[cfg(test)]
16mod benchmark_all;
17#[cfg(test)]
18mod tests;
19
20#[skip_serializing_none]
21#[derive(Clone, Debug, Deserialize, Serialize)]
22#[serde(rename_all = "camelCase")]
23#[serde(default)]
24pub struct VADOption {
25    pub r#type: VadType,
26    pub samplerate: u32,
27    /// Padding before speech detection (in ms)
28    pub speech_padding: u64,
29    /// Padding after silence detection (in ms)
30    pub silence_padding: u64,
31    pub ratio: f32,
32    pub voice_threshold: f32,
33    pub max_buffer_duration_secs: u64,
34    /// Timeout duration for silence (in ms), None means disable this feature
35    pub silence_timeout: Option<u64>,
36    pub endpoint: Option<String>,
37    pub secret_key: Option<String>,
38    pub secret_id: Option<String>,
39    #[serde(skip)]
40    pub refer: Option<bool>,
41}
42
43impl Default for VADOption {
44    fn default() -> Self {
45        Self {
46            r#type: VadType::Silero,
47            samplerate: 16000,
48            speech_padding: 250, // min_speech_duration_ms (match silero_vad default)
49            silence_padding: 100, // min_silence_duration_ms
50            ratio: 0.5,
51            voice_threshold: 0.5,
52            max_buffer_duration_secs: 50,
53            silence_timeout: None,
54            endpoint: None,
55            secret_key: None,
56            secret_id: None,
57            refer: None,
58        }
59    }
60}
61
62#[derive(Clone, Debug, Serialize, Eq, Hash, PartialEq)]
63#[serde(rename_all = "lowercase")]
64pub enum VadType {
65    Silero,
66    Other(String),
67}
68
69impl<'de> Deserialize<'de> for VadType {
70    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
71    where
72        D: serde::Deserializer<'de>,
73    {
74        let value = String::deserialize(deserializer)?;
75        match value.as_str() {
76            "silero" => Ok(VadType::Silero),
77            _ => Ok(VadType::Other(value)),
78        }
79    }
80}
81
82impl std::fmt::Display for VadType {
83    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
84        match self {
85            VadType::Silero => write!(f, "silero"),
86            VadType::Other(provider) => write!(f, "{}", provider),
87        }
88    }
89}
90
91impl TryFrom<&String> for VadType {
92    type Error = String;
93
94    fn try_from(value: &String) -> std::result::Result<Self, Self::Error> {
95        match value.as_str() {
96            "silero" => Ok(VadType::Silero),
97            other => Ok(VadType::Other(other.to_string())),
98        }
99    }
100}
101struct SpeechBuf {
102    samples: PcmBuf,
103    timestamp: u64,
104}
105
106struct VadProcessorInner {
107    vad: Box<dyn VadEngine>,
108    event_sender: EventSender,
109    option: VADOption,
110    window_bufs: Vec<SpeechBuf>,
111    triggered: bool,
112    triggered_event_sent: bool,
113    current_speech_start: Option<u64>,
114    temp_end: Option<u64>,
115    refer: Option<bool>,
116}
117pub struct VadProcessor {
118    inner: VadProcessorInner,
119}
120
121pub trait VadEngine: Send + Sync + Any {
122    fn process(&mut self, frame: &mut AudioFrame) -> Vec<(bool, u64)>;
123}
124
125impl VadProcessorInner {
126    pub fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()> {
127        let samples = match &frame.samples {
128            Samples::PCM { samples } => samples,
129            _ => return Ok(()),
130        };
131
132        let samples_cloned = samples.to_owned();
133        let results = self.vad.process(frame);
134        for (is_speaking, timestamp) in results {
135            if is_speaking || self.triggered {
136                let current_buf = SpeechBuf {
137                    samples: samples_cloned.clone(),
138                    timestamp,
139                };
140                self.window_bufs.push(current_buf);
141            }
142            self.process_vad_logic(is_speaking, timestamp, &frame.track_id)?;
143
144            // Clean up old buffers periodically
145            if self.window_bufs.len() > 1000 || !self.triggered {
146                let cutoff = if self.triggered {
147                    timestamp.saturating_sub(5000)
148                } else {
149                    timestamp.saturating_sub(self.option.silence_padding)
150                };
151                self.window_bufs.retain(|buf| buf.timestamp > cutoff);
152            }
153        }
154
155        Ok(())
156    }
157
158    fn process_vad_logic(
159        &mut self,
160        is_speaking: bool,
161        timestamp: u64,
162        track_id: &str,
163    ) -> Result<()> {
164        if is_speaking && !self.triggered {
165            self.triggered = true;
166            self.current_speech_start = Some(timestamp);
167            self.triggered_event_sent = false;
168        } else if is_speaking && self.triggered {
169            // Already triggered, check if we need to emit Speaking event for the first time
170            if let Some(start_time) = self.current_speech_start {
171                let duration = timestamp.saturating_sub(start_time);
172                // L1 filter: only emit Speaking if duration is enough (e.g. 200ms)
173                // Use speech_padding as min_speech_duration
174                if duration >= self.option.speech_padding && !self.triggered_event_sent {
175                    let event = SessionEvent::Speaking {
176                        track_id: track_id.to_string(),
177                        timestamp: crate::media::get_timestamp(),
178                        start_time,
179                        is_filler: None, // Will be enriched by MFCC or ASR
180                        confidence: Some(1.0),
181                        refer: self.refer,
182                    };
183                    self.event_sender.send(event).ok();
184                    self.triggered_event_sent = true;
185                }
186            }
187        } else if !is_speaking {
188            if self.temp_end.is_none() {
189                self.temp_end = Some(timestamp);
190            }
191
192            if let Some(temp_end) = self.temp_end {
193                // Use saturating_sub to handle timestamp wrapping or out-of-order frames
194                let silence_duration = timestamp.saturating_sub(temp_end);
195
196                // Process regular silence detection for speech segments
197                if self.triggered && silence_duration >= self.option.silence_padding {
198                    if let Some(start_time) = self.current_speech_start {
199                        // Use safe duration calculation
200                        let duration = temp_end.saturating_sub(start_time);
201                        if duration >= self.option.speech_padding {
202                            let samples_vec = self
203                                .window_bufs
204                                .iter()
205                                .filter(|buf| {
206                                    buf.timestamp >= start_time && buf.timestamp <= temp_end
207                                })
208                                .flat_map(|buf| buf.samples.iter())
209                                .cloned()
210                                .collect();
211                            self.window_bufs.clear();
212
213                            let event = SessionEvent::Silence {
214                                track_id: track_id.to_string(),
215                                timestamp: crate::media::get_timestamp(),
216                                start_time,
217                                duration,
218                                samples: Some(samples_vec),
219                                refer: self.refer,
220                            };
221                            self.event_sender.send(event).ok();
222                        }
223                    }
224                    self.triggered = false;
225                    self.triggered_event_sent = false;
226                    self.current_speech_start = None;
227                    self.temp_end = Some(timestamp); // Update temp_end for silence timeout tracking
228                }
229
230                // Process silence timeout if configured
231                if let Some(timeout) = self.option.silence_timeout {
232                    // Use same safe calculation for silence timeout
233                    let timeout_duration = timestamp.saturating_sub(temp_end);
234
235                    if timeout_duration >= timeout {
236                        let event = SessionEvent::Silence {
237                            track_id: track_id.to_string(),
238                            timestamp: crate::media::get_timestamp(),
239                            start_time: temp_end,
240                            duration: timeout_duration,
241                            samples: None,
242                            refer: self.refer,
243                        };
244                        self.event_sender.send(event).ok();
245                        self.temp_end = Some(timestamp);
246                    }
247                }
248            }
249        }
250
251        if is_speaking && self.temp_end.is_some() {
252            self.temp_end = None;
253        }
254
255        Ok(())
256    }
257}
258
259impl VadProcessor {
260    pub fn create(
261        _token: CancellationToken,
262        event_sender: EventSender,
263        option: VADOption,
264    ) -> Result<Box<dyn Processor>> {
265        let vad: Box<dyn VadEngine> = match option.r#type {
266            VadType::Silero => Box::new(tiny_silero::TinySilero::new(option.clone())?),
267            _ => Box::new(NopVad::new()?),
268        };
269        Ok(Box::new(VadProcessor::new(vad, event_sender, option)?))
270    }
271
272    pub fn create_nop(
273        _token: CancellationToken,
274        event_sender: EventSender,
275        option: VADOption,
276    ) -> Result<Box<dyn Processor>> {
277        let vad: Box<dyn VadEngine> = match option.r#type {
278            _ => Box::new(NopVad::new()?),
279        };
280        Ok(Box::new(VadProcessor::new(vad, event_sender, option)?))
281    }
282
283    pub fn new(
284        engine: Box<dyn VadEngine>,
285        event_sender: EventSender,
286        option: VADOption,
287    ) -> Result<Self> {
288        let refer = option.refer;
289        let inner = VadProcessorInner {
290            vad: engine,
291            event_sender,
292            option,
293            window_bufs: Vec::new(),
294            triggered_event_sent: false,
295            triggered: false,
296            current_speech_start: None,
297            temp_end: None,
298            refer,
299        };
300        Ok(Self { inner })
301    }
302}
303
304impl Processor for VadProcessor {
305    fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()> {
306        self.inner.process_frame(frame)
307    }
308}
309
310struct NopVad {}
311
312impl NopVad {
313    pub fn new() -> Result<Self> {
314        Ok(Self {})
315    }
316}
317
318impl VadEngine for NopVad {
319    fn process(&mut self, frame: &mut AudioFrame) -> Vec<(bool, u64)> {
320        let samples = match &frame.samples {
321            Samples::PCM { samples } => samples,
322            _ => return vec![(false, frame.timestamp)],
323        };
324        // Check if there are any non-zero samples
325        let has_speech = samples.iter().any(|&x| x != 0);
326        vec![(has_speech, frame.timestamp)]
327    }
328}
active_call/media/vad/mod.rs

active_call/media/vad/
mod.rs
