Mercurial

comparison dictation/audio.py @ 275:78699f810817

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Add Qwen3-VL and WebRTC dictation services Add Bazel targets for the CUDA-backed Qwen3-VL server and a local WebRTC faster-whisper dictation service. Co-authored-by: Copilot <[email protected]> Copilot-Session: e3d8cb06-6c95-4ae0-9757-651d3796ab00
author MrJuneJune <me@mrjunejune.com>
date Mon, 17 Aug 2026 10:58:47 -0700
parents
children 49e9e591c9bb
comparison
equal deleted inserted replaced
274:c9be578316a6 275:78699f810817
1 from __future__ import annotations
2
3 from dataclasses import dataclass
4 from enum import Enum
5
6 import numpy as np
7
8
9 class AudioEventKind(str, Enum):
10 SPEECH_STARTED = "speech_started"
11 PARTIAL_READY = "partial_ready"
12 FINAL_READY = "final_ready"
13
14
15 @dataclass(frozen=True)
16 class AudioEvent:
17 kind: AudioEventKind
18 samples: np.ndarray | None = None
19
20
21 class AudioSegmenter:
22 def __init__(
23 self,
24 *,
25 sample_rate: int = 16000,
26 speech_threshold: float = 0.012,
27 silence_ms: int = 700,
28 partial_interval_ms: int = 1200,
29 max_utterance_seconds: int = 30,
30 pre_roll_ms: int = 200,
31 ) -> None:
32 self.sample_rate = sample_rate
33 self.speech_threshold = speech_threshold
34 self.silence_samples = sample_rate * silence_ms // 1000
35 self.partial_samples = sample_rate * partial_interval_ms // 1000
36 self.max_samples = sample_rate * max_utterance_seconds
37 self.pre_roll_samples = sample_rate * pre_roll_ms // 1000
38 self._pre_roll = np.empty(0, dtype=np.float32)
39 self._utterance = np.empty(0, dtype=np.float32)
40 self._silence_count = 0
41 self._last_partial_size = 0
42 self._speaking = False
43
44 @property
45 def speaking(self) -> bool:
46 return self._speaking
47
48 def feed(self, samples: np.ndarray) -> list[AudioEvent]:
49 chunk = np.asarray(samples, dtype=np.float32).reshape(-1)
50 if chunk.size == 0:
51 return []
52 rms = float(np.sqrt(np.mean(np.square(chunk), dtype=np.float64)))
53 is_speech = rms >= self.speech_threshold
54 events: list[AudioEvent] = []
55
56 if not self._speaking:
57 if not is_speech:
58 self._append_pre_roll(chunk)
59 return events
60 self._speaking = True
61 self._utterance = np.concatenate((self._pre_roll, chunk))
62 self._pre_roll = np.empty(0, dtype=np.float32)
63 self._silence_count = 0
64 self._last_partial_size = 0
65 events.append(AudioEvent(AudioEventKind.SPEECH_STARTED))
66 else:
67 self._utterance = np.concatenate((self._utterance, chunk))
68
69 self._silence_count = 0 if is_speech else self._silence_count + chunk.size
70 utterance_size = self._utterance.size
71
72 if (
73 utterance_size - self._last_partial_size >= self.partial_samples
74 and self._silence_count < self.silence_samples
75 ):
76 self._last_partial_size = utterance_size
77 events.append(
78 AudioEvent(
79 AudioEventKind.PARTIAL_READY,
80 self._utterance.copy(),
81 )
82 )
83
84 if (
85 self._silence_count >= self.silence_samples
86 or utterance_size >= self.max_samples
87 ):
88 final = self._utterance.copy()
89 self._reset()
90 events.append(AudioEvent(AudioEventKind.FINAL_READY, final))
91
92 return events
93
94 def flush(self) -> AudioEvent | None:
95 if not self._speaking or self._utterance.size == 0:
96 self._reset()
97 return None
98 final = self._utterance.copy()
99 self._reset()
100 return AudioEvent(AudioEventKind.FINAL_READY, final)
101
102 def _append_pre_roll(self, chunk: np.ndarray) -> None:
103 self._pre_roll = np.concatenate((self._pre_roll, chunk))
104 if self._pre_roll.size > self.pre_roll_samples:
105 self._pre_roll = self._pre_roll[-self.pre_roll_samples :]
106
107 def _reset(self) -> None:
108 self._pre_roll = np.empty(0, dtype=np.float32)
109 self._utterance = np.empty(0, dtype=np.float32)
110 self._silence_count = 0
111 self._last_partial_size = 0
112 self._speaking = False