xavierk/voculinux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

feat(diarization): per-speaker transcripts via TinyDiarize + --transcribe-file (#884)

Browse Source 
* feat(transcribe): add TinyDiarize file transcription surface

Catalog small.en-tdrz (hosted at akashmjn/tinydiarize-whisper.cpp,
pinned commit) while keeping it out of every dictation-model selection
path: its [SPEAKER_TURN] markup would type noise into the focused
window. The checksum generator resolves side-repo models at their own
pinned revision, parakeet-style.

Add speech_recognition.diarization (strict-WAV fast path with ffmpeg
fallback, tdrz transcription, speaker-turn parsing, timestamped
formatting) and surface it through a tray Transcribe Audio File flow
with on-demand model download plus a headless --transcribe-file CLI.
TranscriptDialog offers copy and save of the diarized transcript.

Closes #756

* docs(user-guide): document --transcribe-file

* fix(transcribe): pass tdrz_enable as a pywhispercpp kwarg

pywhispercpp's Model takes decode params as keyword arguments; a
params= dict crashes on the C-level whisper_full_params object. Also
let --transcribe-file report unexpected failures cleanly instead of
dumping a traceback.

* fix(tray): dismiss the tdrz download dialog on response

* test(diarization): pin real wave/numpy/tempfile around sibling mocks

Several test modules replace sys.modules entries with MagicMocks at
import time and never restore them, so diarization's wave/numpy/
tempfile bindings resolved to mocks when this file ran after them in
CI. An autouse fixture reimports the real stdlib modules per test, the
same remedy test_audio_feedback already applies for wave.

* fix(transcribe): verify tdrz model, preload libs, gate formats on ffmpeg

- Hash the on-disk TinyDiarize model against the pinned digest before
  native code maps it, matching the dictation load path
- Extract the bundled-lib preloader to utils/pywhispercpp_loader.py and
  preload before the lazy pywhispercpp import so the headless
  --transcribe-file path works on source-built installs without RPATH
- Offer only WAV in the file picker when ffmpeg is absent; name the
  dependency in the filter label
- Type the new tray/diarization callbacks; document --transcribe-file
  and the new modules in AGENTS.md

* style(audio): finish type annotations on capture helpers and download path

_setup_alsa_error_handler now declares Optional[Any] and its ctypes
callback gets concrete parameter types; the remaining capture helpers
take audio: Any (PyAudio has no importable type) and test_audio_input's
device_index becomes Optional[int], which also clears the preexisting
implicit-Optional error. _download_whispercpp_model gains -> None and
its HF url is narrowed to str so the concatenation and download call
stop adding new mypy findings.

* style(tests): wrap long monkeypatch line in preload test

* refactor(main): narrow transcribe-file errors and add file context

* style(ui): annotate TranscriptDialog constructor

* fix(diarize): log worker tracebacks, isort test helpers, document --transcribe-file

---------

Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
This commit is contained in:
devin-ai-integration[bot]
2026-09-30 00:44:55 -07:00
committed by GitHub
co-authored by Jatin K Malik
parent b4babcaeca
commit 8ebc4ce2ba
21 changed files with 1421 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/vocalinux/audio/capture.py
+12 -10
View File
@@ -26,7 +26,7 @@ CAPTURE_CHUNK = 1024
# ALSA error handler to suppress warnings during PyAudio initialization
def _setup_alsa_error_handler():
def _setup_alsa_error_handler() -> Optional[Any]:
"""Set up an error handler to suppress ALSA warnings."""
try:
# Try multiple library name variations for cross-distro compatibility
@@ -45,7 +45,9 @@ def _setup_alsa_error_handler():
)
# Create a no-op error handler
def _error_handler(filename, line, function, err, fmt):
def _error_handler(
filename: bytes, line: int, function: bytes, err: int, fmt: bytes
) -> None:
pass
_alsa_error_handler = ERROR_HANDLER_FUNC(_error_handler)
@@ -131,7 +133,7 @@ def _is_bluetooth_device(device_name: Optional[str]) -> bool:
return any(pattern in name_lower for pattern in bluetooth_patterns)
def _safe_close_stream(stream) -> None:
def _safe_close_stream(stream: Any) -> None:
"""Stop and close a PortAudio stream without raising.
Closing an active stream (especially Bluetooth SCO/HFP capture) without
@@ -154,7 +156,7 @@ def _safe_close_stream(stream) -> None:
pass
def _get_device_info_safe(audio, device_index: Optional[int] = None) -> dict:
def _get_device_info_safe(audio: Any, device_index: Optional[int] = None) -> dict:
"""Fetch PortAudio device info, returning {} on failure."""
try:
if device_index is not None:
@@ -211,7 +213,7 @@ def get_audio_input_devices() -> list:
def _resolve_device_by_name(
audio, device_name: Optional[str], fallback_index: Optional[int] = None
audio: Any, device_name: Optional[str], fallback_index: Optional[int] = None
) -> Optional[int]:
"""Resolve a device index by name, falling back to index, then system default.
@@ -241,7 +243,7 @@ def _resolve_device_by_name(
return _resolve_valid_input_device(audio, fallback_index)
def _resolve_valid_input_device(audio, preferred_index: Optional[int] = None) -> Optional[int]:
def _resolve_valid_input_device(audio: Any, preferred_index: Optional[int] = None) -> Optional[int]:
"""Resolve a valid audio input device, skipping unsafe or output-only devices.
Checks that the device has maxInputChannels > 0. Falls back from
@@ -324,7 +326,7 @@ def _resolve_valid_input_device(audio, preferred_index: Optional[int] = None) ->
return input_device_indices[0]
def _open_capture_stream(audio, device_index: Optional[int] = None) -> tuple[int, int, object]:
def _open_capture_stream(audio: Any, device_index: Optional[int] = None) -> tuple[int, int, object]:
"""
Negotiate a working (channels, sample_rate) and return the opened stream.
@@ -498,7 +500,7 @@ def _downmix_to_mono(
return mono, None
def _get_supported_channels(audio, device_index: Optional[int] = None) -> int:
def _get_supported_channels(audio: Any, device_index: Optional[int] = None) -> int:
"""
Detect the supported number of channels for the audio device.
@@ -519,7 +521,7 @@ def _get_supported_channels(audio, device_index: Optional[int] = None) -> int:
return channels
def _get_supported_sample_rate(audio, device_index: Optional[int], channels: int = 1) -> int:
def _get_supported_sample_rate(audio: Any, device_index: Optional[int], channels: int = 1) -> int:
"""
Get a supported sample rate for the audio device.
@@ -589,7 +591,7 @@ def _get_supported_sample_rate(audio, device_index: Optional[int], channels: int
return 16000
def test_audio_input(device_index: int = None, duration: float = 1.0) -> dict:
def test_audio_input(device_index: Optional[int] = None, duration: float = 1.0) -> dict:
"""
Test audio input from a device and return diagnostic information.
src/vocalinux/common_types.py
+39
View File
@@ -76,6 +76,45 @@ class SpeechRecognitionManagerProtocol(Protocol):
"""Register a callback for recognized text with its audio-capture start time."""
...
@property
def model_ready(self) -> bool:
"""Whether the engine has a model loaded and can recognize speech."""
...
def reconfigure(
self,
engine: Optional[str] = None,
model_size: Optional[str] = None,
language: Optional[str] = None,
force_download: bool = True,
force_reinit: bool = False,
**kwargs: Any,
) -> None:
"""Reconfigure the engine on the fly."""
...
def try_begin_download(self) -> bool:
"""Claim the engine for a model download. False if one is running."""
...
def end_download(self) -> None:
"""Release the claim taken by try_begin_download()."""
...
def cancel_download(self) -> None:
"""Request cancellation of the current download."""
...
def set_download_progress_callback(
self, callback: Optional[Callable[[float, float, str], None]]
) -> None:
"""Register a callback for download progress (fraction, mbps, status)."""
...
def download_whispercpp_model(self, model_name: str) -> None:
"""Download a catalog whisper.cpp model without changing engine config."""
...
class TextInjectorProtocol(Protocol):
"""Protocol defining the interface for TextInjector."""
src/vocalinux/main.py
+32
View File
@@ -110,6 +110,15 @@ def parse_arguments():
action="store_true",
help="Stop voice typing on a running instance (via D-Bus) and exit",
)
parser.add_argument(
"--transcribe-file",
type=str,
metavar="PATH",
help=(
"Transcribe an audio file with speaker attribution using the "
"TinyDiarize model, print the transcript, and exit"
),
)
return parser.parse_args()
@@ -145,6 +154,24 @@ def _dispatch_trigger(command: str) -> int:
return 1
def _run_file_transcription(path: str) -> int:
"""Headless ``--transcribe-file`` path: transcribe and print, no GTK."""
from .speech_recognition.diarization import format_transcript, transcribe_audio_file
try:
blocks = transcribe_audio_file(path)
except (OSError, ValueError, RuntimeError) as error:
print(
f"vocalinux: could not transcribe {path}: {type(error).__name__}: {error}",
file=sys.stderr,
)
return 1
transcript = format_transcript(blocks)
print(transcript if transcript else "vocalinux: no speech detected")
return 0
def check_dependencies():
"""Check for required dependencies and provide helpful error messages."""
missing_system_deps = []
@@ -314,6 +341,11 @@ def main():
if trigger is not None:
sys.exit(_dispatch_trigger(trigger))
# Headless file transcription exits before the instance lock and GTK, so
# it works alongside a running tray app and on a displayless shell.
if isinstance(args.transcribe_file, str):
sys.exit(_run_file_transcription(args.transcribe_file))
# Check for single instance BEFORE any initialization
from . import single_instance
src/vocalinux/speech_recognition/diarization.py
+210
View File
@@ -0,0 +1,210 @@
"""
Diarized file transcription with the TinyDiarize (tdrz) whisper.cpp model.
The TinyDiarize weights emit ``[SPEAKER_TURN]`` tokens at speaker boundaries
(whisper.cpp's tinydiarize mode, which pywhispercpp exposes through its
``tdrz_enable`` flag). This module turns that output into a readable
speaker-attributed transcript for the "Transcribe audio file" flow; it never
touches the dictation pipeline.
"""
import logging
import os
import re
import shutil
import subprocess
import tempfile
import wave
from dataclasses import dataclass
from typing import TYPE_CHECKING, Sequence
import numpy as np
from ..utils.host_process import host_env
from ..utils.model_checksums import verify_model_file
from ..utils.whispercpp_model_info import TDRZ_MODEL, get_model_path, is_model_downloaded
if TYPE_CHECKING:
from pywhispercpp.model import Segment
logger = logging.getLogger(__name__)
#: Sample rate whisper.cpp requires on its float32 input.
WHISPER_SAMPLE_RATE = 16000
#: Marker TinyDiarize embeds in segment text at each speaker boundary.
SPEAKER_TURN_MARKER = "[SPEAKER_TURN]"
#: pywhispercpp reports segment times in 10 ms units.
_SEGMENT_TIME_SCALE = 0.01
_SPEAKER_TURN_PATTERN = re.compile(r"\[SPEAKER_TURN\]")
#: Label prefix for each detected speaker block.
SPEAKER_LABEL = "Speaker"
@dataclass
class TranscriptBlock:
"""One speaker's run of consecutive text in a diarized transcript."""
start_seconds: float
speaker: int
text: str
def format_timestamp(total_seconds: float) -> str:
"""Render a position in the audio as ``HH:MM:SS``."""
total_seconds = max(0, int(total_seconds))
hours, remainder = divmod(total_seconds, 3600)
minutes, seconds = divmod(remainder, 60)
return f"{hours:02d}:{minutes:02d}:{seconds:02d}"
def speaker_turn_blocks(segments: Sequence["Segment"]) -> list[TranscriptBlock]:
"""Group raw tdrz segments into per-speaker transcript blocks.
TinyDiarize emits ``[SPEAKER_TURN]`` where it detects a speaker change;
each marker starts a new block, so runs between markers are one speaker's
consecutive text. Consecutive blocks get sequential speaker indexes —
attribution is positional ("Speaker 1", "Speaker 2", ...), not identity.
"""
blocks: list[TranscriptBlock] = []
speaker = 1
current_text_parts: list[str] = []
current_start = 0.0
def flush() -> None:
text = " ".join(current_text_parts).strip()
if text:
blocks.append(TranscriptBlock(current_start, speaker, text))
current_text_parts.clear()
for segment in segments:
pieces = _SPEAKER_TURN_PATTERN.split(segment.text)
for index, piece in enumerate(pieces):
piece = piece.strip()
if index > 0:
flush()
speaker += 1
current_start = segment.t0 * _SEGMENT_TIME_SCALE
elif not current_text_parts:
# A block starts at the segment its first text came from.
current_start = segment.t0 * _SEGMENT_TIME_SCALE
if piece:
current_text_parts.append(piece)
flush()
return blocks
def format_transcript(blocks: Sequence[TranscriptBlock]) -> str:
"""Render speaker blocks as the transcript the dialog shows and exports."""
lines = []
for block in blocks:
timestamp = format_timestamp(block.start_seconds)
lines.append(f"[{timestamp}] {SPEAKER_LABEL} {block.speaker}: {block.text}")
return "\n\n".join(lines)
def _load_wav(path: str) -> np.ndarray:
"""Load a 16-bit mono/stereo WAV at 16 kHz as float32 samples."""
with wave.open(path, "rb") as wav_file:
channels = wav_file.getnchannels()
if channels not in (1, 2):
raise ValueError(f"WAV file must be mono or stereo: {path}")
if wav_file.getsampwidth() != 2:
raise ValueError(f"WAV file must be 16-bit: {path}")
if wav_file.getframerate() != WHISPER_SAMPLE_RATE:
raise ValueError(f"WAV file must be {WHISPER_SAMPLE_RATE} Hz: {path}")
raw = wav_file.readframes(wav_file.getnframes())
audio = np.frombuffer(raw, dtype=np.int16).astype(np.float32)
if channels == 1:
return audio / 32768.0
stereo = audio.reshape(-1, 2)
return (stereo[:, 0] + stereo[:, 1]) / 65536.0
def ffmpeg_available() -> bool:
"""Whether ffmpeg can decode non-WAV formats for file transcription."""
return shutil.which("ffmpeg") is not None
def _load_with_ffmpeg(path: str) -> np.ndarray:
"""Decode any ffmpeg-supported media to mono 16 kHz float32 samples."""
if shutil.which("ffmpeg") is None:
raise RuntimeError(
"ffmpeg is not installed, so only 16-bit 16 kHz WAV files can be "
"transcribed. Install ffmpeg to transcribe other audio formats."
)
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file:
temp_path = temp_file.name
try:
subprocess.run(
[
"ffmpeg",
"-i",
path,
"-ac",
"1",
"-ar",
str(WHISPER_SAMPLE_RATE),
"-y",
temp_path,
],
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
env=host_env(),
)
return _load_wav(temp_path)
except subprocess.CalledProcessError as error:
raise RuntimeError(f"ffmpeg could not decode {path}") from error
finally:
os.remove(temp_path)
def load_audio(path: str) -> np.ndarray:
"""Decode ``path`` to mono 16 kHz float32 samples for whisper.cpp."""
if path.endswith(".wav"):
try:
return _load_wav(path)
except (wave.Error, EOFError, ValueError):
logger.info("%s is not a strict whisper WAV; decoding with ffmpeg", path)
return _load_with_ffmpeg(path)
return _load_with_ffmpeg(path)
def transcribe_audio_file(audio_path: str, model_name: str = TDRZ_MODEL) -> list[TranscriptBlock]:
"""Transcribe ``audio_path`` with TinyDiarize into per-speaker blocks.
Requires the model to be downloaded already — callers go through the
catalog download flow (or ``--transcribe-file`` errors out asking for it)
so this function never surprises the network in a worker thread.
"""
if not is_model_downloaded(model_name):
raise FileNotFoundError(
f"The {model_name} model is not downloaded yet; fetch it through "
"the catalog before transcribing"
)
model_path = get_model_path(model_name)
# A file merely present on disk (copied in, truncated, tampered) is hashed
# against the pinned digest before native code maps it, same as dictation.
verify_model_file(model_path)
from ..utils.pywhispercpp_loader import preload_shared_libraries
# Source builds without RPATH need the bundled libwhisper/libggml loaded
# first; the headless path never reaches engine initialization.
preload_shared_libraries()
from pywhispercpp.model import Model # deferred: heavy native import
audio = load_audio(audio_path)
if audio.size == 0:
raise ValueError(f"{audio_path} contains no audio")
model = Model(model_path, tdrz_enable=True)
segments = model.transcribe(audio)
return speaker_turn_blocks(segments)
src/vocalinux/speech_recognition/recognition_manager.py
+30 -124
View File
@@ -55,6 +55,12 @@ from ..utils.model_checksums import (
write_verification_stamp,
)
from ..utils.paths import models_dir
from ..utils.pywhispercpp_loader import (
find_shared_library_dirs as _find_pywhispercpp_shared_library_dirs,
)
from ..utils.pywhispercpp_loader import (
preload_shared_libraries as _preload_pywhispercpp_shared_libraries,
)
from ..utils.system_language import LANGUAGE_FOLLOWS_LAYOUT, language_for_active_layout
from ..utils.vosk_model_info import SUPPORTED_LANGUAGES, VOSK_MODEL_INFO
from ..utils.whisper_model_info import (
@@ -63,6 +69,7 @@ from ..utils.whisper_model_info import (
whisper_model_url,
)
from ..utils.whispercpp_model_info import (
NON_DICTATION_MODELS,
WHISPERCPP_MODEL_INFO,
get_model_path,
is_english_only_model,
@@ -172,122 +179,6 @@ def resolve_language_preference(language: str) -> str:
return resolved or "auto"
_PYWHISPERCPP_PRELOADED_LIBS: list[ctypes.CDLL] = []
def _find_pywhispercpp_shared_library_dirs() -> list[str]:
"""Find bundled pywhispercpp native library directories without importing it."""
candidate_dirs: list[Path] = []
for module_name in ("_pywhispercpp", "pywhispercpp"):
try:
spec = importlib.util.find_spec(module_name)
except (ImportError, AttributeError, ValueError):
spec = None
if spec is None:
continue
if spec.origin:
module_dir = Path(spec.origin).resolve().parent
candidate_dirs.extend(
[
module_dir,
module_dir / ".libs",
module_dir / "lib",
module_dir / "pywhispercpp.libs",
module_dir.parent / "pywhispercpp.libs",
]
)
if spec.submodule_search_locations:
for location in spec.submodule_search_locations:
package_dir = Path(location).resolve()
candidate_dirs.extend(
[
package_dir,
package_dir / ".libs",
package_dir / "lib",
package_dir.parent / "pywhispercpp.libs",
]
)
for path_entry in sys.path:
if not path_entry:
continue
path_root = Path(path_entry).resolve()
candidate_dirs.append(path_root / "pywhispercpp.libs")
library_dirs: list[str] = []
seen: set[str] = set()
for candidate_dir in candidate_dirs:
try:
resolved_dir = str(candidate_dir.resolve())
if resolved_dir in seen or not candidate_dir.is_dir():
continue
has_native_lib = any(candidate_dir.glob("libwhisper*.so*")) or any(
candidate_dir.glob("libggml*.so*")
)
except (OSError, TypeError, ValueError):
# TypeError/ValueError can surface when tests monkey-patch os.stat or
# when pathlib internals receive unexpected types from mocks.
continue
if has_native_lib:
seen.add(resolved_dir)
library_dirs.append(resolved_dir)
return library_dirs
def _preload_pywhispercpp_shared_libraries() -> None:
"""Preload bundled pywhispercpp shared libraries for source-built installs.
Some source builds place libwhisper/libggml next to the Python extension
without an RPATH. Preloading by absolute path lets the dynamic loader satisfy
the extension's libwhisper.so.1 dependency before importing pywhispercpp.
"""
if _PYWHISPERCPP_PRELOADED_LIBS:
return
libraries: list[Path] = []
for library_dir in _find_pywhispercpp_shared_library_dirs():
root = Path(library_dir)
libraries.extend(sorted(root.glob("libggml*.so*")))
libraries.extend(sorted(root.glob("libwhisper*.so*")))
if not libraries:
return
pending = list(dict.fromkeys(libraries))
loaded: list[ctypes.CDLL] = []
last_errors: dict[str, OSError] = {}
mode = getattr(ctypes, "RTLD_GLOBAL", 0)
# Native libs can depend on each other. Retry while progress is made so a
# dependency loaded earlier in the same directory can unlock later libraries.
while pending:
loaded_this_pass = False
for library_path in pending[:]:
try:
loaded.append(ctypes.CDLL(str(library_path), mode=mode))
pending.remove(library_path)
loaded_this_pass = True
except OSError as e:
last_errors[str(library_path)] = e
if not loaded_this_pass:
break
_PYWHISPERCPP_PRELOADED_LIBS.extend(loaded)
if pending:
logger.debug(
"Could not preload all pywhispercpp native libraries: %s",
{str(path): str(last_errors.get(str(path))) for path in pending},
)
logger = logging.getLogger(__name__)
@@ -1266,8 +1157,12 @@ class SpeechRecognitionManager:
_preload_pywhispercpp_shared_libraries()
from pywhispercpp.model import Model # noqa: F401 — fail fast if missing
# Validate model size for whisper.cpp
valid_models = list(WHISPERCPP_MODEL_INFO.keys())
# Validate model size for whisper.cpp. Non-dictation entries in the
# catalog (e.g. TinyDiarize) stay selectable for their own surfaces
# but emit markup that would inject noise into the focused window.
valid_models = [
name for name in WHISPERCPP_MODEL_INFO if name not in NON_DICTATION_MODELS
]
if self.model_size not in valid_models:
logger.warning(
f"Model size '{self.model_size}' not valid for whisper.cpp. "
@@ -2451,26 +2346,27 @@ class SpeechRecognitionManager:
os.remove(temp_file)
raise
def _download_whispercpp_model(self):
def _download_whispercpp_model(self, model_name: Optional[str] = None) -> None:
"""Download a whisper.cpp model with progress tracking."""
import requests
self._download_cancelled = False
model_info = WHISPERCPP_MODEL_INFO.get(self.model_size)
model_name = model_name or self.model_size
model_info = WHISPERCPP_MODEL_INFO.get(model_name)
if not model_info:
raise ValueError(f"Unknown whisper.cpp model size: {self.model_size}")
raise ValueError(f"Unknown whisper.cpp model size: {model_name}")
url = model_info["url"]
url = str(model_info["url"])
# Prefer explicit download=true (some HF edges serve HTML without it).
if "huggingface.co" in url and "download=" not in url:
url = url + ("&" if "?" in url else "?") + "download=true"
model_path = get_model_path(self.model_size)
model_path = get_model_path(model_name)
temp_file = model_path + ".tmp"
os.makedirs(os.path.dirname(model_path), exist_ok=True)
logger.info(f"Downloading whisper.cpp {self.model_size} model to {model_path}")
logger.info(f"Downloading whisper.cpp {model_name} model to {model_path}")
try:
self._stream_model_download(url, temp_file)
@@ -2509,6 +2405,16 @@ class SpeechRecognitionManager:
os.remove(temp_file)
raise
def download_whispercpp_model(self, model_name: str) -> None:
"""Download a catalog whisper.cpp model without changing engine config.
Used by surfaces that need a model the dictation engine does not load
(e.g. the TinyDiarize file-transcription flow). The engine's cancel
flag and progress callback apply, so the existing
try_begin_download/end_download coordination keeps working.
"""
self._download_whispercpp_model(model_name)
def _get_vosk_model_path(self) -> str:
"""Get the path to the VOSK model based on the selected size and language."""
model_name = self.vosk_model_map.get(self.model_size, self.vosk_model_map["small"])
src/vocalinux/ui/config_manager.py
+4 -2
View File
@@ -16,6 +16,7 @@ from ..utils.vosk_model_info import SUPPORTED_LANGUAGES
from ..utils.whispercpp_model_info import MODEL_SIZES as WHISPERCPP_MODEL_SIZES
from ..utils.whispercpp_model_info import WHISPERCPP_MODEL_INFO, default_variant_for_size
from ..utils.whispercpp_model_info import get_model_size as get_whispercpp_model_size
from ..utils.whispercpp_model_info import is_dictation_model
from ..utils.whispercpp_model_info import is_english_only_model as is_english_only_whispercpp_model
logger = logging.getLogger(__name__)
@@ -257,7 +258,7 @@ def resolve_whispercpp_variant(saved_model: str, pinned_variant: str, language_i
language_is_english = SUPPORTED_LANGUAGES.get(language_id, {}).get("whisper") == "en"
pinned = pinned_variant.lower() if isinstance(pinned_variant, str) else ""
if pinned in WHISPERCPP_MODEL_INFO:
if pinned in WHISPERCPP_MODEL_INFO and is_dictation_model(pinned):
if not language_is_english and is_english_only_whispercpp_model(pinned):
return _multilingual_sibling(pinned)
return pinned
@@ -270,6 +271,7 @@ def resolve_whispercpp_variant(saved_model: str, pinned_variant: str, language_i
# Honour true leftover specializations, but not a plain English-only id.
if (
saved in WHISPERCPP_MODEL_INFO
and is_dictation_model(saved)
and saved not in WHISPERCPP_MODEL_SIZES
and saved != f"{size}.en"
):
@@ -280,7 +282,7 @@ def resolve_whispercpp_variant(saved_model: str, pinned_variant: str, language_i
derived = default_variant_for_size(size, language_is_english)
if derived in WHISPERCPP_MODEL_INFO:
return derived
return saved if saved in WHISPERCPP_MODEL_INFO else "tiny"
return saved if saved in WHISPERCPP_MODEL_INFO and is_dictation_model(saved) else "tiny"
class ConfigManager:
src/vocalinux/ui/settings_dialog.py
+5
View File
@@ -121,6 +121,7 @@ from ..utils.faster_whisper_model_info import ( # isort:skip
FASTER_WHISPER_MODEL_INFO,
)
from ..utils.whispercpp_model_info import ( # isort:skip
NON_DICTATION_MODELS,
WHISPERCPP_MODEL_INFO,
default_variant_for_size,
)
@@ -250,6 +251,8 @@ def _model_display_name(model_name: str) -> str:
display_parts.append(part.upper())
elif part == "turbo":
display_parts.append("Turbo")
elif part == "tdrz":
display_parts.append("TinyDiarize")
elif part.startswith("v") and part[1:].isdigit():
display_parts.append(part)
else:
@@ -6502,6 +6505,8 @@ class SettingsDialog(Gtk.Dialog):
best = None
best_mb = None
for model_name in list_downloaded_whispercpp_models():
if model_name in NON_DICTATION_MODELS:
continue
info = WHISPERCPP_MODEL_INFO.get(model_name)
if not info:
continue
src/vocalinux/ui/transcript_dialog.py
+255
View File
@@ -0,0 +1,255 @@
"""
Diarized transcript dialog for VocaLinux.
Shows the speaker-attributed transcript produced by the "Transcribe audio
file" flow (TinyDiarize tdrz output via diarization.py) with copy and
save-to-file actions. GTK3, following the logging dialog's card-and-toolbar
conventions.
"""
import logging
import os
from typing import Optional, Sequence
import gi
gi.require_version("Gtk", "3.0")
gi.require_version("Gdk", "3.0")
from gi.repository import Gdk, Gtk # noqa: E402
from ..speech_recognition.diarization import ( # noqa: E402
TranscriptBlock,
format_timestamp,
format_transcript,
)
logger = logging.getLogger(__name__)
TRANSCRIPT_CSS = """
.transcript-dialog {
background-color: @theme_bg_color;
}
.toolbar-box {
margin-bottom: 8px;
}
.toolbar-box button {
border-radius: 6px;
min-height: 32px;
}
.transcript-view-container {
background-color: @theme_base_color;
border-radius: 12px;
border: 1px solid alpha(@borders, 0.5);
}
.transcript-view {
padding: 12px;
background-color: transparent;
}
.status-bar {
background-color: @theme_base_color;
border-radius: 8px;
padding: 8px 16px;
margin-top: 8px;
border: 1px solid alpha(@borders, 0.3);
}
"""
def _setup_css() -> None:
"""Set up CSS styling for the transcript dialog."""
css_provider = Gtk.CssProvider()
css_provider.load_from_data(TRANSCRIPT_CSS.encode())
Gtk.StyleContext.add_provider_for_screen(
Gdk.Screen.get_default(),
css_provider,
Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION,
)
class TranscriptDialog(Gtk.Dialog):
"""Dialog that shows a diarized transcript with copy and export actions."""
def __init__(self, parent: Optional[Gtk.Window], source_name: str) -> None:
super().__init__(
title=f"Transcript — {source_name}",
transient_for=parent,
flags=Gtk.DialogFlags.DESTROY_WITH_PARENT,
modal=False,
)
_setup_css()
self.set_default_size(720, 560)
self.get_style_context().add_class("transcript-dialog")
self.add_button("_Close", Gtk.ResponseType.CLOSE)
self.blocks: list[TranscriptBlock] = []
self.transcript_text = ""
content_area = self.get_content_area()
content_area.set_margin_top(16)
content_area.set_margin_bottom(8)
content_area.set_margin_start(16)
content_area.set_margin_end(16)
main_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
content_area.add(main_box)
main_box.pack_start(self._create_toolbar(), False, False, 0)
main_box.pack_start(self._create_transcript_view(), True, True, 0)
main_box.pack_start(self._create_status_bar(), False, False, 0)
main_box.show_all()
self.connect("response", self._on_response)
def _create_toolbar(self) -> Gtk.Box:
"""Toolbar with copy and save actions."""
toolbar_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
toolbar_box.get_style_context().add_class("toolbar-box")
toolbar_box.pack_start(Gtk.Box(), True, True, 0)
self.copy_button = Gtk.Button()
copy_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
copy_box.pack_start(
Gtk.Image.new_from_icon_name("edit-copy-symbolic", Gtk.IconSize.BUTTON),
False,
False,
0,
)
copy_box.pack_start(Gtk.Label(label="Copy"), False, False, 0)
self.copy_button.add(copy_box)
self.copy_button.set_tooltip_text("Copy the full transcript to the clipboard")
self.copy_button.connect("clicked", lambda widget: self._copy_to_clipboard())
toolbar_box.pack_start(self.copy_button, False, False, 0)
save_button = Gtk.Button()
save_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
save_box.pack_start(
Gtk.Image.new_from_icon_name("document-save-symbolic", Gtk.IconSize.BUTTON),
False,
False,
0,
)
save_box.pack_start(Gtk.Label(label="Save As"), False, False, 0)
save_button.add(save_box)
save_button.set_tooltip_text("Save the transcript to a text file")
save_button.connect("clicked", lambda widget: self._save_to_file())
toolbar_box.pack_start(save_button, False, False, 0)
return toolbar_box
def _create_transcript_view(self) -> Gtk.Box:
"""Scrollable read-only transcript view with speaker styling."""
container = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
container.get_style_context().add_class("transcript-view-container")
scrolled = Gtk.ScrolledWindow()
scrolled.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
scrolled.set_min_content_height(400)
scrolled.set_vexpand(True)
scrolled.set_hexpand(True)
self.text_view = Gtk.TextView()
self.text_view.set_editable(False)
self.text_view.set_cursor_visible(False)
self.text_view.set_wrap_mode(Gtk.WrapMode.WORD_CHAR)
self.text_view.set_vexpand(True)
self.text_view.set_hexpand(True)
self.text_view.set_left_margin(12)
self.text_view.set_right_margin(12)
self.text_view.set_top_margin(12)
self.text_view.set_bottom_margin(12)
self.text_view.get_style_context().add_class("transcript-view")
self.text_buffer = self.text_view.get_buffer()
header_tag = self.text_buffer.create_tag("speaker_header")
header_tag.set_property("weight", 600) # Pango.Weight.SEMIBOLD
timestamp_tag = self.text_buffer.create_tag("timestamp")
timestamp_tag.set_property("foreground", "#888888")
timestamp_tag.set_property("scale", 0.95)
scrolled.add(self.text_view)
container.pack_start(scrolled, True, True, 0)
return container
def _create_status_bar(self) -> Gtk.Box:
"""Bottom status line."""
status_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=16)
status_box.get_style_context().add_class("status-bar")
self.status_label = Gtk.Label(xalign=0)
status_box.pack_start(self.status_label, False, False, 0)
return status_box
def set_transcript(self, blocks: Sequence[TranscriptBlock]) -> None:
"""Populate the dialog with parsed speaker blocks."""
self.blocks = list(blocks)
self.text_buffer.set_text("")
speaker_count = 0
for block in self.blocks:
timestamp = format_timestamp(block.start_seconds)
end_iter = self.text_buffer.get_end_iter()
self.text_buffer.insert_with_tags_by_name(end_iter, f"[{timestamp}] ", "timestamp")
end_iter = self.text_buffer.get_end_iter()
self.text_buffer.insert_with_tags_by_name(
end_iter, f"Speaker {block.speaker}", "speaker_header"
)
self.text_buffer.insert(end_iter, f": {block.text}\n\n")
speaker_count = max(speaker_count, block.speaker)
self.status_label.set_text(
f"{len(self.blocks)} block(s), up to {speaker_count} speaker(s) detected"
if self.blocks
else "No speech detected"
)
self.transcript_text = format_transcript(self.blocks)
def _copy_to_clipboard(self) -> None:
"""Copy the full transcript to the clipboard."""
clipboard = Gtk.Clipboard.get(Gdk.SELECTION_CLIPBOARD)
clipboard.set_text(self.transcript_text, -1)
clipboard.store()
self.status_label.set_text("Transcript copied to clipboard")
def _save_to_file(self) -> None:
"""Write the transcript to a user-chosen text file."""
chooser = Gtk.FileChooserDialog(
title="Save Transcript",
transient_for=self,
action=Gtk.FileChooserAction.SAVE,
)
chooser.add_button("_Cancel", Gtk.ResponseType.CANCEL)
chooser.add_button("_Save", Gtk.ResponseType.OK)
chooser.set_do_overwrite_confirmation(True)
chooser.set_current_name("transcript.txt")
text_filter = Gtk.FileFilter()
text_filter.set_name("Text files")
text_filter.add_pattern("*.txt")
chooser.add_filter(text_filter)
try:
if chooser.run() == Gtk.ResponseType.OK:
path = chooser.get_filename()
try:
with open(path, "w", encoding="utf-8") as transcript_file:
transcript_file.write(self.transcript_text)
self.status_label.set_text(f"Saved to {os.path.basename(path)}")
except OSError as error:
logger.error("Could not save transcript to %s: %s", path, error)
self.status_label.set_text(f"Could not save: {error}")
finally:
chooser.destroy()
def _on_response(self, dialog: Gtk.Dialog, response_id: int) -> None:
"""Close destroys the dialog."""
if response_id in (Gtk.ResponseType.CLOSE, Gtk.ResponseType.DELETE_EVENT):
dialog.destroy()
src/vocalinux/ui/tray_indicator.py
+156 -3
View File
@@ -10,7 +10,7 @@ import os
import signal
import threading
import time
from typing import Any, Callable, Optional
from typing import Any, Callable, Optional, Sequence, cast
import gi
@@ -40,16 +40,22 @@ from ..common_types import RecognitionState, SpeechRecognitionManagerProtocol, T
from ..dbus_service import VocalinuxDBusService
from ..gateway_embed import GatewayStatus, get_gateway_embed_manager
from ..model_keepalive import DEFAULT_IDLE_TIMEOUT_SECONDS, ModelKeepAlive
from ..speech_recognition.diarization import (
TranscriptBlock,
ffmpeg_available,
transcribe_audio_file,
)
from ..suspend_handler import SuspendHandler
from ..utils.host_process import host_env
from ..utils.resource_manager import ResourceManager
from ..utils.update_checker import ReleaseInfo
from ..utils.update_monitor import UpdateMonitor
from ..utils.whispercpp_model_info import TDRZ_MODEL, WHISPERCPP_MODEL_INFO, is_model_downloaded
from . import notifications
from .config_manager import get_shared_config_manager
from .keyboard_backends import DEFAULT_SHORTCUT, DEFAULT_SHORTCUT_MODE
from .keyboard_shortcuts import KeyboardShortcutManager
from .settings_dialog import SettingsDialog, recommended_model_for_engine
from .settings_dialog import ModelDownloadDialog, SettingsDialog, recommended_model_for_engine
from .transcription_history import TranscriptionHistory
logger = logging.getLogger(__name__)
@@ -419,6 +425,7 @@ class TrayIndicator:
# Add menu items
self._add_menu_item("Start Voice Typing", self._on_start_clicked)
self._add_menu_item("Stop Voice Typing", self._on_stop_clicked)
self._add_menu_item("Transcribe Audio File\u2026", self._on_transcribe_file_clicked)
self._add_menu_separator()
# Recent snippets submenu (only when history is enabled)
@@ -761,7 +768,7 @@ class TrayIndicator:
"""Download the offered model, then report how it went."""
last_notified = 0.0
def on_progress(fraction, speed_mbps, status):
def on_progress(fraction: float, speed_mbps: float, status: str) -> None:
nonlocal last_notified
now = time.monotonic()
if now - last_notified < _DOWNLOAD_NOTIFY_INTERVAL_SECONDS:
@@ -821,6 +828,152 @@ class TrayIndicator:
self.speech_engine.end_download()
self._model_download_active = False
def _on_transcribe_file_clicked(self, widget: Gtk.Widget) -> None:
"""Pick an audio file and transcribe it with speaker attribution."""
chooser = Gtk.FileChooserDialog(
title="Transcribe Audio File",
transient_for=None,
action=Gtk.FileChooserAction.OPEN,
)
chooser.add_button("_Cancel", Gtk.ResponseType.CANCEL)
chooser.add_button("_Open", Gtk.ResponseType.OK)
# Non-WAV formats are decoded through ffmpeg; without it only the
# patterns the built-in WAV loader accepts are offered, so the picker
# never advertises a format that cannot be transcribed.
audio_filter = Gtk.FileFilter()
if ffmpeg_available():
audio_filter.set_name("Audio files")
for pattern in ("*.wav", "*.mp3", "*.ogg", "*.flac", "*.m4a", "*.opus"):
audio_filter.add_pattern(pattern)
else:
audio_filter.set_name("WAV audio (install ffmpeg for MP3/OGG/FLAC/M4A/Opus)")
audio_filter.add_pattern("*.wav")
chooser.add_filter(audio_filter)
try:
if chooser.run() == Gtk.ResponseType.OK:
path = chooser.get_filename()
else:
path = None
finally:
chooser.destroy()
if path:
self._start_file_transcription(path)
def _start_file_transcription(self, path: str) -> None:
"""Transcribe ``path`` on a worker, downloading TinyDiarize if needed."""
if not is_model_downloaded(TDRZ_MODEL):
self._offer_tdrz_download(path)
return
basename = os.path.basename(path)
progress = notifications.notify(
"Transcribing audio file",
f"{basename} — this can take a while on long files...",
"audio-x-generic",
)
def run() -> None:
try:
blocks = transcribe_audio_file(path)
except Exception as error:
logger.error("File transcription failed for %s: %s", path, error, exc_info=True)
_idle_once(notifications.close, progress)
_idle_once(
notifications.notify,
"Transcription failed",
f"{basename}: {error}",
"dialog-error",
)
return
_idle_once(notifications.close, progress)
_idle_once(self._show_transcript, basename, blocks)
threading.Thread(target=run, daemon=True, name="file-transcription").start()
def _show_transcript(self, basename: str, blocks: Sequence[TranscriptBlock]) -> None:
"""Open the transcript dialog for the finished blocks."""
from .transcript_dialog import TranscriptDialog
dialog = TranscriptDialog(None, basename)
dialog.set_transcript(blocks)
dialog.show_all()
def _offer_tdrz_download(self, path: str) -> None:
"""Ask to fetch TinyDiarize, then continue into transcription."""
info = WHISPERCPP_MODEL_INFO[TDRZ_MODEL]
prompt = Gtk.MessageDialog(
transient_for=None,
flags=Gtk.DialogFlags.MODAL,
message_type=Gtk.MessageType.QUESTION,
buttons=Gtk.ButtonsType.NONE,
text="Transcribing audio files needs the TinyDiarize model",
)
prompt.format_secondary_text(
f"Download it now? (~{info['size_mb']} MB, one-time). "
"It is a whisper.cpp model that marks speaker turns; it never "
"becomes your dictation model."
)
prompt.add_button("_Cancel", Gtk.ResponseType.CANCEL)
prompt.add_button("_Download", Gtk.ResponseType.OK)
try:
accepted = prompt.run() == Gtk.ResponseType.OK
finally:
prompt.destroy()
if not accepted:
return
if not self.speech_engine.try_begin_download():
notifications.notify(
"Download already in progress",
"Another speech model is being downloaded. Wait for it to finish, "
"then try again.",
"dialog-information",
)
return
dialog = ModelDownloadDialog(
None, TDRZ_MODEL, cast(int, info["size_mb"]), engine="whisper_cpp"
)
# The dialog's post-complete OK button emits a response but destroys
# nothing on its own.
dialog.connect("response", lambda *_args: dialog.destroy())
def run() -> None:
def on_progress(fraction: float, speed_mbps: float, status: str) -> None:
GLib.idle_add(dialog.update_progress, fraction, speed_mbps, status)
def check_cancelled() -> bool:
if dialog.cancelled:
self.speech_engine.cancel_download()
return not dialog.cancelled
cancel_check_id = GLib.timeout_add(100, check_cancelled)
try:
self.speech_engine.set_download_progress_callback(on_progress)
self.speech_engine.download_whispercpp_model(TDRZ_MODEL)
except Exception as error:
logger.error("TinyDiarize download failed: %s", error, exc_info=True)
message = (
"Download cancelled" if "cancelled" in str(error).lower() else str(error)[:100]
)
GLib.idle_add(dialog.set_complete, False, message)
return
finally:
GLib.source_remove(cancel_check_id)
self.speech_engine.set_download_progress_callback(None)
self.speech_engine.end_download()
GLib.idle_add(dialog.set_complete, True, "")
# The dialog shows its own completion; starting the transcription
# as soon as the model lands keeps the flow to a single click.
_idle_once(self._start_file_transcription, path)
threading.Thread(target=run, daemon=True, name="tdrz-download").start()
def _set_indicator_icon(self, icon_key: str, description: str) -> None:
"""Apply a tray icon, forcing a host redraw when cycling reused paths.
src/vocalinux/utils/model_checksums.txt
+1
View File
@@ -45,6 +45,7 @@ ggml-small-q5_1.bin sha256 ae85e4a935d7a567bd102f
ggml-small-q8_0.bin sha256 49c8fb02b65e6049d5fa6c04f81f53b867b5ec9540406812c643f177317f779f 264464607
ggml-small.bin sha256 1be3a9b2063867b937e64e2ec7483364a79917e157fa98c5d94b5c1fffea987b 487601967
ggml-small.en-q5_1.bin sha256 bfdff4894dcb76bbf647d56263ea2a96645423f1669176f4844a1bf8e478ad30 190098681
ggml-small.en-tdrz.bin sha256 ceac3ec06d1d98ef71aec665283564631055fd6129b79d8e1be4f9cc33cc54b4 487614184
ggml-small.en.bin sha256 c6138d6d58ecc8322097e0f987c32f1be8bb0a18532a3f88f734d1bbf9c41e5d 487614201
ggml-tiny-q5_1.bin sha256 818710568da3ca15689e31a743197b520007872ff9576237bda97bd1b469c3d7 32152673
ggml-tiny-q8_0.bin sha256 c2085835d3f50733e2ff6e4b41ae8a2b8d8110461e18821b09a15c40c42d1cca 43537433
src/vocalinux/utils/pywhispercpp_loader.py
+126
View File
@@ -0,0 +1,126 @@
"""Locate and preload pywhispercpp's bundled shared libraries.
Some source builds place libwhisper/libggml next to the Python extension
without an RPATH. Preloading by absolute path lets the dynamic loader satisfy
the extension's libwhisper.so.1 dependency before importing pywhispercpp, and
keeps non-dictation entry points (e.g. ``--transcribe-file``) working on such
installs.
"""
import ctypes
import importlib.util
import logging
import sys
from pathlib import Path
logger = logging.getLogger(__name__)
_PRELOADED_LIBS: list[ctypes.CDLL] = []
def find_shared_library_dirs() -> list[str]:
"""Find bundled pywhispercpp native library directories without importing it."""
candidate_dirs: list[Path] = []
for module_name in ("_pywhispercpp", "pywhispercpp"):
try:
spec = importlib.util.find_spec(module_name)
except (ImportError, AttributeError, ValueError):
spec = None
if spec is None:
continue
if spec.origin:
module_dir = Path(spec.origin).resolve().parent
candidate_dirs.extend(
[
module_dir,
module_dir / ".libs",
module_dir / "lib",
module_dir / "pywhispercpp.libs",
module_dir.parent / "pywhispercpp.libs",
]
)
if spec.submodule_search_locations:
for location in spec.submodule_search_locations:
package_dir = Path(location).resolve()
candidate_dirs.extend(
[
package_dir,
package_dir / ".libs",
package_dir / "lib",
package_dir.parent / "pywhispercpp.libs",
]
)
for path_entry in sys.path:
if not path_entry:
continue
path_root = Path(path_entry).resolve()
candidate_dirs.append(path_root / "pywhispercpp.libs")
library_dirs: list[str] = []
seen: set[str] = set()
for candidate_dir in candidate_dirs:
try:
resolved_dir = str(candidate_dir.resolve())
if resolved_dir in seen or not candidate_dir.is_dir():
continue
has_native_lib = any(candidate_dir.glob("libwhisper*.so*")) or any(
candidate_dir.glob("libggml*.so*")
)
except (OSError, TypeError, ValueError):
# TypeError/ValueError can surface when tests monkey-patch os.stat or
# when pathlib internals receive unexpected types from mocks.
continue
if has_native_lib:
seen.add(resolved_dir)
library_dirs.append(resolved_dir)
return library_dirs
def preload_shared_libraries() -> None:
"""Preload bundled pywhispercpp shared libraries for source-built installs."""
if _PRELOADED_LIBS:
return
libraries: list[Path] = []
for library_dir in find_shared_library_dirs():
root = Path(library_dir)
libraries.extend(sorted(root.glob("libggml*.so*")))
libraries.extend(sorted(root.glob("libwhisper*.so*")))
if not libraries:
return
pending = list(dict.fromkeys(libraries))
loaded: list[ctypes.CDLL] = []
last_errors: dict[str, OSError] = {}
mode = getattr(ctypes, "RTLD_GLOBAL", 0)
# Native libs can depend on each other. Retry while progress is made so a
# dependency loaded earlier in the same directory can unlock later libraries.
while pending:
loaded_this_pass = False
for library_path in pending[:]:
try:
loaded.append(ctypes.CDLL(str(library_path), mode=mode))
pending.remove(library_path)
loaded_this_pass = True
except OSError as e:
last_errors[str(library_path)] = e
if not loaded_this_pass:
break
_PRELOADED_LIBS.extend(loaded)
if pending:
logger.debug(
"Could not preload all pywhispercpp native libraries: %s",
{str(path): str(last_errors.get(str(path))) for path in pending},
)
src/vocalinux/utils/whispercpp_model_info.py
+50 -1
View File
@@ -26,7 +26,23 @@ logger = logging.getLogger(__name__)
# sha256 per file, and a file replaced upstream would turn every download into a
# checksum failure. Both the pin and the digests are refreshed together by
# scripts/generate-model-checksums.py.
_WHISPERCPP_REPO = "https://huggingface.co/ggerganov/whisper.cpp/resolve"
_WHISPERCPP_REPO_SLUG = "ggerganov/whisper.cpp"
_WHISPERCPP_REPO = f"https://huggingface.co/{_WHISPERCPP_REPO_SLUG}/resolve"
# TinyDiarize (tdrz) ships from its own upstream repository — the main one
# carries only dictation weights. The repo is static since 2023, so the
# commit is pinned here rather than in model_checksums.txt like the others.
_TDRZ_REPO = "akashmjn/tinydiarize-whisper.cpp"
_TDRZ_REVISION = "d44ba793fc67e509623a88a409723311fa677744"
#: Catalog model for speaker-turn diarization. TinyDiarize emits
#: ``[SPEAKER_TURN]`` markup rather than plain text, so it is fetchable through
#: the catalog but must never be offered or accepted as a dictation model:
#: bracketed tokens would type noise into the focused window.
TDRZ_MODEL = "small.en-tdrz"
#: whisper.cpp catalog models that are not usable for keystroke dictation.
NON_DICTATION_MODELS = frozenset({TDRZ_MODEL})
def whispercpp_model_file(model_name: str) -> str:
@@ -40,6 +56,11 @@ def whispercpp_model_file(model_name: str) -> str:
def _model_url(model_name: str) -> str:
"""Build the Hugging Face URL for a ggml whisper.cpp model."""
if model_name in NON_DICTATION_MODELS:
repo, revision = _WHISPERCPP_SIDE_SOURCES[model_name]
return (
f"https://huggingface.co/{repo}/resolve/{revision}/{whispercpp_model_file(model_name)}"
)
return f"{_WHISPERCPP_REPO}/{whispercpp_revision()}/{whispercpp_model_file(model_name)}"
@@ -73,8 +94,16 @@ _WHISPERCPP_MODEL_SPECS = [
("large-v3-turbo", 1620, "809M", "High accuracy, lower memory than large"),
("large-v3-turbo-q5_0", 574, "809M", "Quantized large v3 Turbo model"),
("large-v3-turbo-q8_0", 874, "809M", "Q8 quantized large v3 Turbo model"),
("small.en-tdrz", 465, "244M", "TinyDiarize speaker-turn model, English-only"),
]
#: Repository ``(slug, pinned_revision)`` for catalog models hosted outside the
#: main whisper.cpp repository. Keyed by model name; also drives
#: :func:`whispercpp_model_source`.
_WHISPERCPP_SIDE_SOURCES = {
TDRZ_MODEL: (_TDRZ_REPO, _TDRZ_REVISION),
}
WHISPERCPP_MODEL_INFO = {
spec[0]: {
"size_mb": spec[1],
@@ -150,6 +179,26 @@ def is_english_only_model(model_name: str) -> bool:
return ".en" in model_name.lower()
def is_dictation_model(model_name: str) -> bool:
"""Return whether a catalog model is a valid keystroke-dictation model."""
return model_name in WHISPERCPP_MODEL_INFO and model_name not in NON_DICTATION_MODELS
def whispercpp_model_source(model_name: str) -> tuple[str, str]:
"""Return the ``(repo_slug, pinned_revision)`` hosting a catalog model.
Main-repository models return an empty revision: their digests are pinned
to the commit recorded in ``model_checksums.txt`` and re-resolved by
``scripts/generate-model-checksums.py`` each run. Models hosted in a
separate repository pin their own commit in this module, the same way
parakeet bundles pin theirs.
"""
source = _WHISPERCPP_SIDE_SOURCES.get(model_name)
if source is not None:
return source
return _WHISPERCPP_REPO_SLUG, ""
# Compute backend types
class ComputeBackend:
"""Compute backend options for whisper.cpp."""