diff --git a/AGENTS.md b/AGENTS.md index f86452b..c3fb3c6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -104,6 +104,7 @@ pytest tests/test_command_processor.py::TestCommandProcessor::test_initializatio pytest -m "not slow" pytest -m "not integration" python -m vocalinux.main --debug +python -m vocalinux.main --transcribe-file path/to/audio.wav # diarized file transcription ``` Website: `web/AGENTS.md`, `web/PRODUCT.md`, `web/DESIGN.md`. Do not duplicate site commands here. @@ -146,6 +147,8 @@ src/vocalinux/ ├── speech_recognition/ │ ├── recognition_manager.py # whisper.cpp / Whisper / Vosk / remote │ ├── command_processor.py # voice commands +│ ├── diarization.py # TinyDiarize per-speaker file transcription +│ ├── diarization.py # --transcribe-file / tray "Transcribe audio file" │ ├── silero_vad.py │ └── data/ # bundled silero_vad.onnx ├── text_injection/ @@ -153,6 +156,8 @@ src/vocalinux/ │ └── ibus_engine.py # Wayland IBus injection ├── ui/ │ ├── tray_indicator.py, settings_dialog.py, first_run_dialog.py +│ ├── transcript_dialog.py # speaker-tagged transcript viewer +│ ├── transcript_dialog.py # speaker-attributed transcript viewer/export │ ├── config_manager.py, action_handler.py, audio_feedback.py │ ├── autostart_manager.py, keyboard_shortcuts.py │ ├── logging_dialog.py, logging_manager.py @@ -160,6 +165,7 @@ src/vocalinux/ ├── utils/ │ ├── paths.py, resource_manager.py │ ├── update_checker.py, update_monitor.py +│ ├── pywhispercpp_loader.py # bundled-lib preloader for source-built installs │ └── whispercpp_model_info.py, vosk_model_info.py └── resources/ # SVG icons + WAV cues (also repo resources/) ``` diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index 3feb689..fed1f1f 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -209,8 +209,11 @@ vocalinux --engine parakeet vocalinux --model medium.en-q5_0 vocalinux --wayland vocalinux --start-minimized +vocalinux --transcribe-file meeting.wav # diarized transcript on stdout, then exits ``` +`--transcribe-file` uses the TinyDiarize model (`small.en-tdrz`) and needs it downloaded first — the tray "Transcribe Audio File…" entry fetches it on demand. + ## Troubleshooting ```bash diff --git a/scripts/generate-model-checksums.py b/scripts/generate-model-checksums.py index ea365ac..bc581fe 100644 --- a/scripts/generate-model-checksums.py +++ b/scripts/generate-model-checksums.py @@ -114,13 +114,12 @@ def _model_names() -> tuple[list[str], list[str]]: """Model names the application can actually download, straight from the package.""" sys.path.insert(0, str(SRC)) from vocalinux.utils.vosk_model_info import VOSK_MODEL_INFO - from vocalinux.utils.whispercpp_model_info import WHISPERCPP_MODEL_INFO, whispercpp_model_file + from vocalinux.utils.whispercpp_model_info import WHISPERCPP_MODEL_INFO vosk = { name for info in VOSK_MODEL_INFO.values() for name in info["languages"].values() if name } - whispercpp = {whispercpp_model_file(name) for name in WHISPERCPP_MODEL_INFO} - return sorted(whispercpp), sorted(vosk) + return sorted(WHISPERCPP_MODEL_INFO), sorted(vosk) def _parakeet_entries() -> list[Entry]: @@ -205,7 +204,27 @@ def _faster_whisper_entries() -> list[Entry]: return entries +def _hf_lfs_digests(repo: str, revision: str) -> dict[str, tuple[str, int]]: + """sha256/size per LFS file in ``repo`` at ``revision``, from the HF API.""" + payload = _fetch_json( + f"https://huggingface.co/api/models/{repo}/revision/{revision}?blobs=true" + ) + published = {} + for sibling in payload.get("siblings", []): + lfs = sibling.get("lfs") or {} + digest = lfs.get("sha256") or lfs.get("oid") + if digest: + published[sibling["rfilename"]] = (digest, int(lfs.get("size", 0))) + return published + + def _whispercpp_entries(wanted: Iterable[str]) -> tuple[list[Entry], str]: + sys.path.insert(0, str(SRC)) + from vocalinux.utils.whispercpp_model_info import ( + whispercpp_model_file, + whispercpp_model_source, + ) + payload = _fetch_json(HF_API) revision = payload.get("sha") if not revision: @@ -218,20 +237,36 @@ def _whispercpp_entries(wanted: Iterable[str]) -> tuple[list[Entry], str]: if digest: by_name[sibling["rfilename"]] = (digest, int(lfs.get("size", 0))) + # Catalog entries hosted outside the main repository pin their own commit + # in whispercpp_model_info.py (same contract parakeet bundles follow). + side_repo_digests: dict[str, dict[str, tuple[str, int]]] = {} + entries, missing = [], [] - for filename in wanted: - if filename not in by_name: + for model_name in wanted: + filename = whispercpp_model_file(model_name) + repo, pinned_revision = whispercpp_model_source(model_name) + if repo != HF_REPO: + if repo not in side_repo_digests: + side_repo_digests[repo] = _hf_lfs_digests(repo, pinned_revision) + found = side_repo_digests[repo].get(filename) + if found is None: + missing.append(f"{filename} ({repo})") + continue + digest, size = found + elif filename not in by_name: missing.append(filename) continue - digest, size = by_name[filename] + else: + digest, size = by_name[filename] entries.append(Entry(filename, "sha256", digest, size)) if missing: raise SystemExit( - f"Hugging Face {HF_REPO}@{revision} publishes no sha256 for: {', '.join(missing)}.\n" + f"Hugging Face publishes no sha256 for: {', '.join(missing)}.\n" "Either the model was renamed upstream or whispercpp_model_info.py lists " "a file that does not exist." ) + entries.sort(key=lambda entry: entry.filename) return entries, revision diff --git a/src/vocalinux/audio/capture.py b/src/vocalinux/audio/capture.py index 22a99e3..ed41f22 100644 --- a/src/vocalinux/audio/capture.py +++ b/src/vocalinux/audio/capture.py @@ -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. diff --git a/src/vocalinux/common_types.py b/src/vocalinux/common_types.py index 2fab15d..f438662 100644 --- a/src/vocalinux/common_types.py +++ b/src/vocalinux/common_types.py @@ -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.""" diff --git a/src/vocalinux/main.py b/src/vocalinux/main.py index 20487a9..4a23d03 100644 --- a/src/vocalinux/main.py +++ b/src/vocalinux/main.py @@ -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 diff --git a/src/vocalinux/speech_recognition/diarization.py b/src/vocalinux/speech_recognition/diarization.py new file mode 100644 index 0000000..92fe372 --- /dev/null +++ b/src/vocalinux/speech_recognition/diarization.py @@ -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) diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index 71e30a2..516d271 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -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"]) diff --git a/src/vocalinux/ui/config_manager.py b/src/vocalinux/ui/config_manager.py index 9e1d721..0e7ad17 100644 --- a/src/vocalinux/ui/config_manager.py +++ b/src/vocalinux/ui/config_manager.py @@ -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: diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index e665eb0..ed5a231 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -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 diff --git a/src/vocalinux/ui/transcript_dialog.py b/src/vocalinux/ui/transcript_dialog.py new file mode 100644 index 0000000..600e13f --- /dev/null +++ b/src/vocalinux/ui/transcript_dialog.py @@ -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() diff --git a/src/vocalinux/ui/tray_indicator.py b/src/vocalinux/ui/tray_indicator.py index ec893b9..b5ca42e 100644 --- a/src/vocalinux/ui/tray_indicator.py +++ b/src/vocalinux/ui/tray_indicator.py @@ -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. diff --git a/src/vocalinux/utils/model_checksums.txt b/src/vocalinux/utils/model_checksums.txt index 1049d44..9e2ac99 100644 --- a/src/vocalinux/utils/model_checksums.txt +++ b/src/vocalinux/utils/model_checksums.txt @@ -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 diff --git a/src/vocalinux/utils/pywhispercpp_loader.py b/src/vocalinux/utils/pywhispercpp_loader.py new file mode 100644 index 0000000..843c4f6 --- /dev/null +++ b/src/vocalinux/utils/pywhispercpp_loader.py @@ -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}, + ) diff --git a/src/vocalinux/utils/whispercpp_model_info.py b/src/vocalinux/utils/whispercpp_model_info.py index c238ca9..e265f68 100644 --- a/src/vocalinux/utils/whispercpp_model_info.py +++ b/src/vocalinux/utils/whispercpp_model_info.py @@ -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.""" diff --git a/tests/test_diarization.py b/tests/test_diarization.py new file mode 100644 index 0000000..3f4c297 --- /dev/null +++ b/tests/test_diarization.py @@ -0,0 +1,352 @@ +""" +Tests for the TinyDiarize file-transcription module. + +Covers speaker-turn parsing, transcript formatting, and audio decoding — +the parts of the "Transcribe audio file" flow that do not need GTK. +""" + +import importlib +import subprocess +import sys +import unittest +from dataclasses import dataclass +from unittest.mock import MagicMock, patch + +import pytest + + +@pytest.fixture(autouse=True) +def _real_wave_and_numpy(): + """Pin real wave/numpy in sys.modules for each test. + + Sibling test modules replace them with MagicMocks at import time (the same + hazard test_audio_feedback's fixture documents), and diarization binds + `import wave`/`import numpy`/`import tempfile` when first imported inside a + test. Popping before reimporting works no matter which module was imported + first. + """ + poisoned = ("wave", "numpy", "tempfile") + previous = {name: sys.modules.pop(name, None) for name in poisoned} + for name in poisoned: + sys.modules[name] = importlib.import_module(name) + # If another test imported diarization while wave/numpy were mocked, its + # module-level bindings are stale — force a reimport under the real ones. + sys.modules.pop("vocalinux.speech_recognition.diarization", None) + try: + yield + finally: + for name, module in previous.items(): + if module is None: + sys.modules.pop(name, None) + else: + sys.modules[name] = module + + +@dataclass +class _FakeSegment: + """Stand-in for pywhispercpp's Segment (t0/t1 in 10 ms units).""" + + t0: int + t1: int + text: str + probability: float = 0.9 + + +class TestSpeakerTurnBlocks(unittest.TestCase): + """TinyDiarize emits [SPEAKER_TURN] markers; each one starts a block.""" + + def test_single_speaker_run_stays_one_block(self): + from vocalinux.speech_recognition.diarization import speaker_turn_blocks + + segments = [ + _FakeSegment(0, 150, "Hello there."), + _FakeSegment(150, 320, "How are you?"), + ] + blocks = speaker_turn_blocks(segments) + + self.assertEqual(len(blocks), 1) + self.assertEqual(blocks[0].speaker, 1) + self.assertEqual(blocks[0].start_seconds, 0.0) + self.assertEqual(blocks[0].text, "Hello there. How are you?") + + def test_marker_starts_a_new_speaker(self): + from vocalinux.speech_recognition.diarization import speaker_turn_blocks + + segments = [ + _FakeSegment(0, 150, "First speaker."), + _FakeSegment(150, 320, "[SPEAKER_TURN] And a reply."), + ] + blocks = speaker_turn_blocks(segments) + + self.assertEqual(len(blocks), 2) + self.assertEqual(blocks[0].speaker, 1) + self.assertEqual(blocks[0].text, "First speaker.") + self.assertEqual(blocks[1].speaker, 2) + self.assertEqual(blocks[1].start_seconds, 1.5) + self.assertEqual(blocks[1].text, "And a reply.") + + def test_marker_mid_segment_splits_the_text(self): + from vocalinux.speech_recognition.diarization import speaker_turn_blocks + + segments = [_FakeSegment(50, 320, "Hello. [SPEAKER_TURN] Hi back.")] + blocks = speaker_turn_blocks(segments) + + self.assertEqual(len(blocks), 2) + self.assertEqual(blocks[0].start_seconds, 0.5) + self.assertEqual(blocks[0].text, "Hello.") + self.assertEqual(blocks[1].speaker, 2) + self.assertEqual(blocks[1].text, "Hi back.") + + def test_adjacent_turns_number_speakers_sequentially(self): + from vocalinux.speech_recognition.diarization import speaker_turn_blocks + + segments = [ + _FakeSegment(0, 100, "One."), + _FakeSegment(100, 200, "[SPEAKER_TURN] Two. [SPEAKER_TURN] Three."), + ] + blocks = speaker_turn_blocks(segments) + + self.assertEqual([block.speaker for block in blocks], [1, 2, 3]) + self.assertEqual([block.text for block in blocks], ["One.", "Two.", "Three."]) + + def test_silence_only_output_gives_no_blocks(self): + from vocalinux.speech_recognition.diarization import speaker_turn_blocks + + segments = [_FakeSegment(0, 100, "[SPEAKER_TURN]")] + self.assertEqual(speaker_turn_blocks(segments), []) + self.assertEqual(speaker_turn_blocks([]), []) + + +class TestFormatTimestamp(unittest.TestCase): + def test_formats_hours_minutes_seconds(self): + from vocalinux.speech_recognition.diarization import format_timestamp + + self.assertEqual(format_timestamp(0), "00:00:00") + self.assertEqual(format_timestamp(61.9), "00:01:01") + self.assertEqual(format_timestamp(3600 + 62 + 3), "01:01:05") + + def test_negative_times_clamp_to_zero(self): + from vocalinux.speech_recognition.diarization import format_timestamp + + self.assertEqual(format_timestamp(-5), "00:00:00") + + +class TestFormatTranscript(unittest.TestCase): + def test_renders_timestamped_speaker_lines(self): + from vocalinux.speech_recognition.diarization import TranscriptBlock, format_transcript + + blocks = [ + TranscriptBlock(0.0, 1, "Hello there."), + TranscriptBlock(12.5, 2, "Hi back."), + ] + self.assertEqual( + format_transcript(blocks), + "[00:00:00] Speaker 1: Hello there.\n\n[00:00:12] Speaker 2: Hi back.", + ) + + def test_empty_transcript_is_empty_string(self): + from vocalinux.speech_recognition.diarization import format_transcript + + self.assertEqual(format_transcript([]), "") + + +def _write_wav(path, sample_rate=16000, channels=1, sample_width=2, frames=160): + """Write a minimal PCM WAV: silence for ``frames`` frames.""" + # Resolve wave at call time: sibling test modules poison sys.modules. + import wave + + with wave.open(str(path), "wb") as wav_file: + wav_file.setnchannels(channels) + wav_file.setsampwidth(sample_width) + wav_file.setframerate(sample_rate) + wav_file.writeframes(b"\x00\x00" * frames * channels) + + +class TestLoadAudio(unittest.TestCase): + def test_strict_wav_loads_directly(self): + import tempfile + from pathlib import Path + + import numpy as np + + from vocalinux.speech_recognition.diarization import load_audio + + with tempfile.TemporaryDirectory() as tmpdir: + wav_path = Path(tmpdir) / "sample.wav" + _write_wav(wav_path) + audio = load_audio(str(wav_path)) + + self.assertIsInstance(audio, np.ndarray) + self.assertEqual(audio.dtype, np.float32) + self.assertEqual(audio.size, 160) + + def test_stereo_wav_is_downmixed(self): + import tempfile + from pathlib import Path + + import numpy as np + + from vocalinux.speech_recognition.diarization import load_audio + + with tempfile.TemporaryDirectory() as tmpdir: + wav_path = Path(tmpdir) / "stereo.wav" + _write_wav(wav_path, channels=2) + audio = load_audio(str(wav_path)) + + self.assertEqual(audio.size, 160) + + def test_wrong_rate_wav_falls_back_to_ffmpeg(self): + import tempfile + from pathlib import Path + + from vocalinux.speech_recognition.diarization import load_audio + + with tempfile.TemporaryDirectory() as tmpdir: + wav_path = Path(tmpdir) / "speech.wav" + _write_wav(wav_path, sample_rate=44100) + + with patch("vocalinux.speech_recognition.diarization._load_with_ffmpeg") as mock_ffmpeg: + mock_ffmpeg.return_value = MagicMock() + load_audio(str(wav_path)) + mock_ffmpeg.assert_called_once_with(str(wav_path)) + + def test_non_wav_goes_to_ffmpeg(self): + import tempfile + from pathlib import Path + + import numpy as np + + from vocalinux.speech_recognition.diarization import load_audio + + with tempfile.TemporaryDirectory() as tmpdir: + media_path = Path(tmpdir) / "clip.mp3" + media_path.write_bytes(b"ID3\x04\x00") + + with patch("vocalinux.speech_recognition.diarization._load_with_ffmpeg") as mock_ffmpeg: + mock_ffmpeg.return_value = np.zeros(16, dtype=np.float32) + audio = load_audio(str(media_path)) + + self.assertIsInstance(audio, np.ndarray) + + def test_missing_ffmpeg_reports_a_clean_error(self): + import tempfile + from pathlib import Path + + from vocalinux.speech_recognition.diarization import _load_with_ffmpeg + + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "clip.ogg" + path.write_bytes(b"OggS") + with patch("shutil.which", return_value=None): + with self.assertRaises(RuntimeError) as ctx: + _load_with_ffmpeg(str(path)) + self.assertIn("ffmpeg", str(ctx.exception)) + + def test_ffmpeg_decode_runs_as_host_process(self): + """ffmpeg is a host binary: it must run under host_env(), not the bundle.""" + import tempfile + from pathlib import Path + + from vocalinux.speech_recognition.diarization import _load_with_ffmpeg + + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "clip.mp3" + path.write_bytes(b"ID3") + + def fake_run(cmd, **kwargs): + # Produce a valid WAV so the load completes. + _write_wav(Path(cmd[-1])) + return MagicMock(returncode=0) + + with ( + patch("shutil.which", return_value="/usr/bin/ffmpeg"), + patch("subprocess.run", side_effect=fake_run) as mock_run, + patch( + "vocalinux.speech_recognition.diarization.host_env", + return_value={"HOST": "1"}, + ) as mock_env, + ): + audio = _load_with_ffmpeg(str(path)) + + self.assertIsNotNone(audio) + self.assertEqual(mock_run.call_args.kwargs["env"], {"HOST": "1"}) + mock_env.assert_called_once() + + def test_ffmpeg_failure_surfaces_as_runtime_error(self): + import tempfile + from pathlib import Path + + from vocalinux.speech_recognition.diarization import _load_with_ffmpeg + + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "broken.mp3" + path.write_bytes(b"not real mp3") + with ( + patch("shutil.which", return_value="/usr/bin/ffmpeg"), + patch( + "subprocess.run", + side_effect=subprocess.CalledProcessError(1, "ffmpeg"), + ), + ): + with self.assertRaises(RuntimeError) as ctx: + _load_with_ffmpeg(str(path)) + self.assertIn("could not decode", str(ctx.exception)) + + +class TestTranscribeAudioFile(unittest.TestCase): + def test_missing_model_fails_before_touching_pywhispercpp(self): + from vocalinux.speech_recognition.diarization import transcribe_audio_file + + with patch( + "vocalinux.speech_recognition.diarization.is_model_downloaded", + return_value=False, + ): + with self.assertRaises(FileNotFoundError) as ctx: + transcribe_audio_file("/tmp/whatever.wav") + self.assertIn("not downloaded", str(ctx.exception)) + + def test_tdrz_enabled_model_transcribes_to_blocks(self): + import numpy as np + + from vocalinux.speech_recognition import diarization + + fake_model = MagicMock() + fake_model.transcribe.return_value = [ + _FakeSegment(0, 100, "Hello."), + _FakeSegment(100, 200, "[SPEAKER_TURN] Hi."), + ] + model_cls = MagicMock(return_value=fake_model) + + with ( + patch.object(diarization, "is_model_downloaded", return_value=True), + patch.object( + diarization, "get_model_path", return_value="/models/ggml-small.en-tdrz.bin" + ), + patch.object(diarization, "verify_model_file"), + patch("vocalinux.utils.pywhispercpp_loader.preload_shared_libraries"), + patch.object(diarization, "load_audio", return_value=np.zeros(1600, dtype=np.float32)), + ): + # pywhispercpp is imported lazily inside the function. + fake_module = MagicMock() + fake_module.Model = model_cls + with patch.dict("sys.modules", {"pywhispercpp.model": fake_module}): + blocks = diarization.transcribe_audio_file("/tmp/in.wav") + + model_cls.assert_called_once_with("/models/ggml-small.en-tdrz.bin", tdrz_enable=True) + self.assertEqual(len(blocks), 2) + self.assertEqual(blocks[1].speaker, 2) + + def test_empty_audio_is_a_clear_error(self): + import numpy as np + + from vocalinux.speech_recognition import diarization + + with ( + patch.object(diarization, "is_model_downloaded", return_value=True), + patch.object(diarization, "verify_model_file"), + patch("vocalinux.utils.pywhispercpp_loader.preload_shared_libraries"), + patch.object(diarization, "load_audio", return_value=np.zeros(0, dtype=np.float32)), + ): + with self.assertRaises(ValueError) as ctx: + diarization.transcribe_audio_file("/tmp/empty.wav") + self.assertIn("no audio", str(ctx.exception)) diff --git a/tests/test_main_args_deps.py b/tests/test_main_args_deps.py index 95012b8..8d3e3ba 100644 --- a/tests/test_main_args_deps.py +++ b/tests/test_main_args_deps.py @@ -51,6 +51,7 @@ class TestParseArguments(unittest.TestCase): assert args.engine is None assert args.wayland is False assert args.start_minimized is False + assert args.transcribe_file is None def test_parse_args_debug_flag(self): """Test parsing with debug flag.""" @@ -114,6 +115,42 @@ class TestParseArguments(unittest.TestCase): args = parse_arguments() assert args.start_minimized is True + def test_parse_args_transcribe_file(self): + """Test parsing with the headless --transcribe-file flag.""" + with patch.object(sys, "argv", ["vocalinux", "--transcribe-file", "/tmp/a.wav"]): + from vocalinux.main import parse_arguments + + args = parse_arguments() + assert args.transcribe_file == "/tmp/a.wav" + + def test_run_file_transcription_prints_transcript(self): + """The headless path prints formatted blocks and skips GTK entirely.""" + from vocalinux.main import _run_file_transcription + from vocalinux.speech_recognition.diarization import TranscriptBlock + + blocks = [TranscriptBlock(0.0, 1, "Hello there.")] + with patch( + "vocalinux.speech_recognition.diarization.transcribe_audio_file", + return_value=blocks, + ): + buffer = StringIO() + with redirect_stdout(buffer): + assert _run_file_transcription("/tmp/a.wav") == 0 + assert "Speaker 1: Hello there." in buffer.getvalue() + + def test_run_file_transcription_reports_errors_on_stderr(self): + """Decode/load failures exit non-zero with the error on stderr.""" + from vocalinux.main import _run_file_transcription + + stderr_buffer = StringIO() + with patch( + "vocalinux.speech_recognition.diarization.transcribe_audio_file", + side_effect=RuntimeError("ffmpeg missing"), + ): + with patch.object(sys, "stderr", stderr_buffer): + assert _run_file_transcription("/tmp/a.wav") == 1 + assert "ffmpeg missing" in stderr_buffer.getvalue() + def test_parse_args_multiple_arguments(self): """Test parsing with multiple arguments.""" with patch.object( diff --git a/tests/test_model_checksums.py b/tests/test_model_checksums.py index 1050094..e9e5fd1 100644 --- a/tests/test_model_checksums.py +++ b/tests/test_model_checksums.py @@ -17,9 +17,7 @@ from tempfile import TemporaryDirectory from unittest.mock import patch from vocalinux.utils import model_checksums -from vocalinux.utils.faster_whisper_model_info import ( - FASTER_WHISPER_MODEL_INFO, -) +from vocalinux.utils.faster_whisper_model_info import FASTER_WHISPER_MODEL_INFO from vocalinux.utils.faster_whisper_model_info import manifest_key as faster_whisper_manifest_key from vocalinux.utils.faster_whisper_model_info import model_files as faster_whisper_model_files from vocalinux.utils.model_checksums import ( @@ -39,6 +37,7 @@ from vocalinux.utils.vosk_model_info import VOSK_MODEL_INFO from vocalinux.utils.whispercpp_model_info import ( WHISPERCPP_MODEL_INFO, whispercpp_model_file, + whispercpp_model_source, ) REPO_ROOT = Path(__file__).resolve().parents[1] @@ -130,9 +129,12 @@ class TestManifestCoverage(unittest.TestCase): self.assertRegex(revision, r"^[0-9a-f]{40}$") def test_download_urls_use_the_pinned_revision(self): + # Models hosted outside ggerganov/whisper.cpp (e.g. TinyDiarize) pin + # their own repo's commit instead of the main-repo revision. revision = whispercpp_revision() for name, info in WHISPERCPP_MODEL_INFO.items(): - self.assertIn(f"/resolve/{revision}/", info["url"], name) + _, model_revision = whispercpp_model_source(name) + self.assertIn(f"/resolve/{model_revision or revision}/", info["url"], name) class TestVerifyFile(unittest.TestCase): diff --git a/tests/test_recognition_manager_helpers.py b/tests/test_recognition_manager_helpers.py index 7d5bbf4..7cf3b48 100644 --- a/tests/test_recognition_manager_helpers.py +++ b/tests/test_recognition_manager_helpers.py @@ -47,6 +47,7 @@ from vocalinux.speech_recognition.recognition_manager import ( from vocalinux.speech_recognition.recognition_manager import ( # noqa: E402 test_audio_input as _test_audio_input, ) +from vocalinux.utils import pywhispercpp_loader # Restore immediately for _k, _v in _ORIG.items(): @@ -224,9 +225,11 @@ class TestPywhispercppLibraryHelpers: loaded_handles.append(handle) return handle - monkeypatch.setattr(rm, "_PYWHISPERCPP_PRELOADED_LIBS", []) - monkeypatch.setattr(rm, "_find_pywhispercpp_shared_library_dirs", lambda: [str(libs_dir)]) - monkeypatch.setattr(rm.ctypes, "CDLL", fake_cdll) + monkeypatch.setattr(pywhispercpp_loader, "_PRELOADED_LIBS", []) + monkeypatch.setattr( + pywhispercpp_loader, "find_shared_library_dirs", lambda: [str(libs_dir)] + ) + monkeypatch.setattr(pywhispercpp_loader.ctypes, "CDLL", fake_cdll) _preload_pywhispercpp_shared_libraries() @@ -234,7 +237,7 @@ class TestPywhispercppLibraryHelpers: str(ggml_lib), str(whisper_lib), ] - assert rm._PYWHISPERCPP_PRELOADED_LIBS == loaded_handles + assert pywhispercpp_loader._PRELOADED_LIBS == loaded_handles class TestTestAudioInput(unittest.TestCase): diff --git a/tests/test_settings_dialog.py b/tests/test_settings_dialog.py index ccf027e..9f5c88a 100644 --- a/tests/test_settings_dialog.py +++ b/tests/test_settings_dialog.py @@ -635,7 +635,12 @@ class TestSettingsDialogHelperFunctions(unittest.TestCase): ) self.assertIn("large-v3-turbo", WHISPERCPP_MODEL_INFO) self.assertIn("large-v3-turbo-q5_0", WHISPERCPP_MODEL_INFO) - self.assertNotIn("small.en-tdrz", WHISPERCPP_MODEL_INFO) + # TinyDiarize is catalogued for the file-transcription flow but must + # never appear as a dictation size-bucket choice. + self.assertIn("small.en-tdrz", WHISPERCPP_MODEL_INFO) + from vocalinux.utils.whispercpp_model_info import NON_DICTATION_MODELS + + self.assertIn("small.en-tdrz", NON_DICTATION_MODELS) def test_faster_whisper_picker_lists_english_only_catalog_ids(self): """Settings must list the same .en ids the tray can persist.""" diff --git a/tests/test_whispercpp_model_info_ext.py b/tests/test_whispercpp_model_info_ext.py index d5c0745..b779e93 100644 --- a/tests/test_whispercpp_model_info_ext.py +++ b/tests/test_whispercpp_model_info_ext.py @@ -123,10 +123,7 @@ class TestModelInfo(unittest.TestCase): def test_available_models_contains_all_models(self): """Test that AVAILABLE_MODELS contains all defined models.""" - from vocalinux.utils.whispercpp_model_info import ( - AVAILABLE_MODELS, - WHISPERCPP_MODEL_INFO, - ) + from vocalinux.utils.whispercpp_model_info import AVAILABLE_MODELS, WHISPERCPP_MODEL_INFO self.assertEqual(set(AVAILABLE_MODELS), set(WHISPERCPP_MODEL_INFO.keys())) @@ -164,6 +161,7 @@ class TestModelInfo(unittest.TestCase): "large-v3-turbo", "large-v3-turbo-q5_0", "large-v3-turbo-q8_0", + "small.en-tdrz", ] self.assertEqual(AVAILABLE_MODELS, expected_models) @@ -173,6 +171,7 @@ class TestModelInfo(unittest.TestCase): from vocalinux.utils.whispercpp_model_info import ( MODEL_SIZES, MODEL_VARIANTS_BY_SIZE, + NON_DICTATION_MODELS, WHISPERCPP_MODEL_INFO, ) @@ -185,12 +184,42 @@ class TestModelInfo(unittest.TestCase): grouped_models = { model_name for variants in MODEL_VARIANTS_BY_SIZE.values() for model_name in variants } - self.assertEqual(grouped_models, set(WHISPERCPP_MODEL_INFO.keys())) + self.assertEqual(grouped_models, set(WHISPERCPP_MODEL_INFO.keys()) - NON_DICTATION_MODELS) self.assertFalse( any(model_name.endswith("-tdrz") for model_name in grouped_models), "TinyDiarize needs diarization-specific runtime support, not the base dictation picker", ) + def test_tdrz_model_is_catalogued_but_not_a_dictation_model(self): + """TinyDiarize is fetchable through the catalog, never dictation-selectable.""" + from vocalinux.utils.whispercpp_model_info import ( + NON_DICTATION_MODELS, + TDRZ_MODEL, + WHISPERCPP_MODEL_INFO, + is_dictation_model, + whispercpp_model_file, + whispercpp_model_source, + ) + + self.assertEqual(TDRZ_MODEL, "small.en-tdrz") + self.assertIn(TDRZ_MODEL, WHISPERCPP_MODEL_INFO) + self.assertIn(TDRZ_MODEL, NON_DICTATION_MODELS) + self.assertFalse(is_dictation_model(TDRZ_MODEL)) + self.assertTrue(is_dictation_model("small.en")) + self.assertFalse(is_dictation_model("not-a-model")) + + self.assertEqual(whispercpp_model_file(TDRZ_MODEL), "ggml-small.en-tdrz.bin") + url = WHISPERCPP_MODEL_INFO[TDRZ_MODEL]["url"] + self.assertIn("akashmjn/tinydiarize-whisper.cpp", url) + repo, revision = whispercpp_model_source(TDRZ_MODEL) + self.assertEqual(repo, "akashmjn/tinydiarize-whisper.cpp") + self.assertTrue(revision) + self.assertIn(f"/resolve/{revision}/", url) + + main_repo, main_revision = whispercpp_model_source("small.en") + self.assertEqual(main_repo, "ggerganov/whisper.cpp") + self.assertEqual(main_revision, "") + def test_model_variant_helpers_identify_size_and_english_only_variants(self): """Test helpers that drive the split whisper.cpp model selectors.""" from vocalinux.utils.whispercpp_model_info import ( @@ -575,10 +604,7 @@ class TestDetectVulkanSupport(unittest.TestCase): return MagicMock(returncode=1, stdout="") with patch("subprocess.run", side_effect=run_side_effect): - from vocalinux.utils.whispercpp_model_info import ( - ComputeBackend, - detect_compute_backend, - ) + from vocalinux.utils.whispercpp_model_info import ComputeBackend, detect_compute_backend backend, backend_info = detect_compute_backend() self.assertEqual(backend, ComputeBackend.VULKAN) @@ -745,10 +771,7 @@ class TestDetectComputeBackend(unittest.TestCase): mock_vulkan.return_value = (True, "Intel Arc GPU") mock_cuda.return_value = (True, "NVIDIA RTX 3080 (10GB)") - from vocalinux.utils.whispercpp_model_info import ( - ComputeBackend, - detect_compute_backend, - ) + from vocalinux.utils.whispercpp_model_info import ComputeBackend, detect_compute_backend backend, info = detect_compute_backend() @@ -764,10 +787,7 @@ class TestDetectComputeBackend(unittest.TestCase): mock_vulkan.return_value = (False, None) mock_cuda.return_value = (True, "NVIDIA RTX 3080 (10GB)") - from vocalinux.utils.whispercpp_model_info import ( - ComputeBackend, - detect_compute_backend, - ) + from vocalinux.utils.whispercpp_model_info import ComputeBackend, detect_compute_backend backend, info = detect_compute_backend() @@ -785,10 +805,7 @@ class TestDetectComputeBackend(unittest.TestCase): mock_cuda.return_value = (False, None) mock_cpu.return_value = "Intel Core i7" - from vocalinux.utils.whispercpp_model_info import ( - ComputeBackend, - detect_compute_backend, - ) + from vocalinux.utils.whispercpp_model_info import ComputeBackend, detect_compute_backend backend, info = detect_compute_backend() @@ -1148,10 +1165,7 @@ class TestGetBackendDisplayName(unittest.TestCase): def test_get_backend_display_name_returns_string(self): """Test that get_backend_display_name returns a string.""" - from vocalinux.utils.whispercpp_model_info import ( - ComputeBackend, - get_backend_display_name, - ) + from vocalinux.utils.whispercpp_model_info import ComputeBackend, get_backend_display_name result = get_backend_display_name(ComputeBackend.CPU) self.assertIsInstance(result, str) @@ -1159,30 +1173,21 @@ class TestGetBackendDisplayName(unittest.TestCase): def test_get_backend_display_name_vulkan(self): """Test display name for Vulkan backend.""" - from vocalinux.utils.whispercpp_model_info import ( - ComputeBackend, - get_backend_display_name, - ) + from vocalinux.utils.whispercpp_model_info import ComputeBackend, get_backend_display_name result = get_backend_display_name(ComputeBackend.VULKAN) self.assertEqual(result, "Vulkan GPU") def test_get_backend_display_name_cuda(self): """Test display name for CUDA backend.""" - from vocalinux.utils.whispercpp_model_info import ( - ComputeBackend, - get_backend_display_name, - ) + from vocalinux.utils.whispercpp_model_info import ComputeBackend, get_backend_display_name result = get_backend_display_name(ComputeBackend.CUDA) self.assertEqual(result, "NVIDIA CUDA") def test_get_backend_display_name_cpu(self): """Test display name for CPU backend.""" - from vocalinux.utils.whispercpp_model_info import ( - ComputeBackend, - get_backend_display_name, - ) + from vocalinux.utils.whispercpp_model_info import ComputeBackend, get_backend_display_name result = get_backend_display_name(ComputeBackend.CPU) self.assertEqual(result, "CPU")