exordium.text.whisper

Whisper speech-to-text model wrapper.

Attributes

logger

Module-level logger.

Classes

WhisperWrapper

Wrapper for HuggingFace Whisper speech-to-text models.

Module Contents

exordium.text.whisper.logger

Module-level logger.

class exordium.text.whisper.WhisperWrapper(model_name: str = 'distil-whisper/distil-large-v3', device_id: int = 0, dtype: torch.dtype | None = None, pretrained: bool = True)[source]

Bases: exordium.text.base.StreamingMixin, exordium.text.base.SpeechToText

Wrapper for HuggingFace Whisper speech-to-text models.

Uses transformers for native MPS/CUDA/CPU support with optional word-by-word streaming via TextIteratorStreamer.

The pipeline:

  • preprocess() — converts any audio input to a feature dict ready for the model.

  • inference() — runs model.generate and returns the decoded string. For streaming, use stream() instead.

  • __call__() — chains preprocess + inference; returns the full transcript as a string.

  • stream() — chains preprocess + streaming generation; yields text chunks word-by-word as they are decoded.

Parameters:

  • model_name – HuggingFace model identifier. Recommended: "distil-whisper/distil-large-v3" for quality/speed on GPU, "openai/whisper-large-v3-turbo" for highest quality.

  • device_id – Device index. -1 or None → CPU, 0+ → GPU/MPS.

  • dtype – Torch dtype for the model weights. Defaults to torch.float16 on GPU/MPS, torch.float32 on CPU.

Example — full transcript:

wrapper = WhisperWrapper()
text = wrapper(waveform)
text = wrapper("audio.wav")

Example — streaming word-by-word:

for chunk in wrapper.stream(waveform):
    print(chunk, end="", flush=True)
preprocess(audio: pathlib.Path | str | numpy.ndarray | torch.Tensor, sample_rate: int = 16000) → dict[str, torch.Tensor][source]

Convert any audio input to model-ready input features on self.device.

Parameters:

  • audio –

    One of:

    • str | Path — loaded via torchaudio, resampled to 16 kHz.

    • np.ndarray — float32, 1-D or (1, T), 16 kHz mono.

    • torch.Tensor — same shape/dtype as ndarray.

  • sample_rate – Sample rate of the input waveform when passing an array or tensor directly. Ignored for file paths (detected automatically). Defaults to 16000.

Returns:

Dict with input_features on self.device, plus an attention_mask when the audio exceeds Whisper’s 30 s window.

inference(inputs: object, **kwargs: object) → str[source]

Run greedy/beam-search decoding and return the full transcript.

Parameters:

  • inputs – Feature dict from preprocess() on self.device.

  • **kwargs – Optional language (str) and beam_size (int).

Returns:

Transcribed text as a plain string.

transcribe_segments(audio: pathlib.Path | str | numpy.ndarray | torch.Tensor, language: str | None = None, beam_size: int = 5, sample_rate: int = 16000) → list[exordium.text.base.Segment][source]

Transcribe audio into timestamped Segment chunks.

Works for audio of any length: short clips decode in one pass, longer recordings use Whisper’s sequential long-form decoding. The segments are what align_segments() consumes to keep forced alignment bounded on long files.

Parameters:

  • audio – Audio file path, numpy array, or torch tensor.

  • language – Language code or None for auto-detection.

  • beam_size – Decoder beam width.

  • sample_rate – Sample rate of the input waveform (ignored for paths).

Returns:

Segments in chronological order; empty if nothing was transcribed.

__call__(audio: pathlib.Path | str | numpy.ndarray | torch.Tensor, language: str | None = None, beam_size: int = 5, sample_rate: int = 16000) → str[source]

Transcribe audio and return the full text as a plain string.

Parameters:

  • audio – Audio file path, numpy array, or torch tensor.

  • language – Language code or None for auto-detection.

  • beam_size – Decoder beam width.

  • sample_rate – Sample rate of the input waveform (ignored for paths).

Returns:

Transcribed text.

stream(audio: pathlib.Path | str | numpy.ndarray | torch.Tensor, language: str | None = None, beam_size: int = 1, sample_rate: int = 16000) → collections.abc.Iterator[str][source]

Transcribe audio with word-by-word streaming output.

Runs model.generate in a background thread and yields decoded text chunks to the caller as they are produced — no need to wait for the full transcript. Beam size defaults to 1 (greedy) for lowest latency; increase for better quality at the cost of first-token delay.

Parameters:

  • audio – Audio file path, numpy array, or torch tensor.

  • language – Language code or None for auto-detection.

  • beam_size – Decoder beam width. Defaults to 1 for streaming.

  • sample_rate – Sample rate of the input waveform (ignored for paths).

Audio longer than Whisper’s 30 s window is streamed window by window: TextIteratorStreamer only observes the first chunk of Whisper’s sequential long-form decoding, so streaming the whole file in one generate call would silently stop after 30 s. Each window is decoded short-form instead, which keeps the full transcript flowing. A word spanning a window boundary may be split.

Yields:

Decoded text chunks (words or sub-word tokens) as strings.

Raises:

Exception – Whatever model.generate raised in the worker thread, re-raised on the caller’s thread once the stream drains.

Example:

for chunk in wrapper.stream("audio.wav"):
    print(chunk, end="", flush=True)
print()  # newline after full transcript