spych.core

  1from faster_whisper import WhisperModel
  2from spych.utils import Notify, Recorder, get_clean_audio_buffer
  3from typing import Union
  4
  5
  6class Spych(Notify):
  7    def __init__(
  8        self,
  9        whisper_model: str = "base.en",
 10        whisper_device: str = "cpu",
 11        whisper_compute_type: str = "int8",
 12        no_speech_threshold: float = 0.3,
 13        vad_speech_threshold: float = 0.5,
 14        vad_silence_threshold: float = 0.35,
 15        vad_silence_frames_threshold: int = 20,
 16        vad_speech_pad_frames: int = 5,
 17        vad_max_speech_duration_s: float = 30.0,
 18    ) -> None:
 19        """
 20        Usage:
 21
 22        - Initializes a transcription object using faster-whisper for fully offline
 23          speech-to-text
 24
 25        Optional:
 26
 27        - `whisper_model`:
 28            - Type: str
 29            - What: The faster-whisper model name to use for transcription
 30            - Default: "base.en"
 31            - Note: Larger models (small, medium, large) provide better accuracy at
 32              the cost of speed; smaller models (tiny, base) are faster but less accurate
 33
 34        - `whisper_device`:
 35            - Type: str
 36            - What: The device to run the whisper model on
 37            - Default: "cpu"
 38            - Options: "cpu", "cuda"
 39            - Note: Use "cuda" for GPU acceleration if available
 40
 41        - `whisper_compute_type`:
 42            - Type: str
 43            - What: The compute type to use for the whisper model
 44            - Default: "int8"
 45            - Options: "int8", "float16", "float32"
 46            - Note: "int8" offers a good balance of speed and accuracy on both CPU and GPU
 47
 48        - `no_speech_threshold`:
 49            - Type: float
 50            - What: The threshold for the `no_speech_prob` returned by faster-whisper
 51            - Default: 0.3
 52            - Note: Segments with a `no_speech_prob` above this threshold will be
 53              ignored to reduce false positives from silence or background noise
 54
 55        - `vad_speech_threshold`:
 56            - Type: float (0.0–1.0)
 57            - What: Silero probability above which a frame is considered speech onset
 58              when `listen` is called with `duration="auto"` or `duration=0`
 59            - Default: 0.5
 60            - Note: Raise to reduce false positives in noisy environments
 61
 62        - `vad_silence_threshold`:
 63            - Type: float (0.0–1.0)
 64            - What: Silero probability below which a frame is considered silence
 65              during an active utterance; must be less than `vad_speech_threshold`
 66              to create a hysteresis band that prevents rapid toggling
 67            - Default: 0.35
 68
 69        - `vad_silence_frames_threshold`:
 70            - Type: int
 71            - What: Consecutive silent frames (~32ms each) required to confirm the
 72              utterance has ended and return the buffer
 73            - Default: 20  (~640ms)
 74
 75        - `vad_speech_pad_frames`:
 76            - Type: int
 77            - What: Pre-roll frames captured before onset confirmation; also the
 78              number of consecutive voiced frames required to confirm speech onset
 79            - Default: 5  (~160ms)
 80
 81        - `vad_max_speech_duration_s`:
 82            - Type: float
 83            - What: Hard cap on a single VAD-captured utterance in seconds; forces
 84              a return even if the speaker never pauses
 85            - Default: 30.0
 86        """
 87        self.wake_model = WhisperModel(
 88            whisper_model,
 89            device=whisper_device,
 90            compute_type=whisper_compute_type,
 91        )
 92        self.no_speech_threshold = no_speech_threshold
 93        self.vad_speech_threshold = vad_speech_threshold
 94        self.vad_silence_threshold = vad_silence_threshold
 95        self.vad_silence_frames_threshold = vad_silence_frames_threshold
 96        self.vad_speech_pad_frames = vad_speech_pad_frames
 97        self.vad_max_speech_duration_s = vad_max_speech_duration_s
 98        self.recorder = Recorder()
 99
100    def listen(
101        self,
102        duration: Union[int, float, str] = 0,
103        device_index: int = -1,
104    ) -> str:
105        """
106        Usage:
107
108        - Records audio from the microphone and returns the transcription as a string.
109          Supports both fixed-duration recording and VAD-gated recording that
110          automatically stops at the end of a natural utterance.
111
112        Optional:
113
114        - `duration`:
115            - Type: int | float | str
116            - What: Controls how long to record
117            - Default: 0
118            - Options:
119                - int | float : Record for exactly this many seconds
120                - "auto" or 0 : Use Silero VAD to detect a complete utterance and
121                                stop automatically when the speaker finishes;
122                                honours the `vad_*` parameters set at init time
123            - Note: "auto" and 0 are equivalent; either signals VAD-gated recording
124
125        - `device_index`:
126            - Type: int
127            - What: The microphone device index to record from
128            - Default: -1
129            - Note: Use `-1` to select the system default input device
130
131        Returns:
132
133        - `transcription`:
134            - Type: str
135            - What: The transcribed text from the recorded audio
136            - Note: Multiple segments are joined with a single space
137        """
138        if duration == "auto" or duration == 0:
139            buffer = self.recorder.record_vad(
140                device_index=device_index,
141                speech_threshold=self.vad_speech_threshold,
142                silence_threshold=self.vad_silence_threshold,
143                silence_frames_threshold=self.vad_silence_frames_threshold,
144                speech_pad_frames=self.vad_speech_pad_frames,
145                max_speech_duration_s=self.vad_max_speech_duration_s,
146            )
147        else:
148            buffer = self.recorder.record(
149                device_index=device_index, duration=duration
150            )
151
152        audio_buffer = get_clean_audio_buffer(buffer)
153        segments, _ = self.wake_model.transcribe(audio_buffer, beam_size=2)
154        return " ".join(
155            [
156                segment.text
157                for segment in segments
158                if segment.no_speech_prob <= self.no_speech_threshold
159            ]
160        )