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

perf: faster Settings, typing and dictation; mypy in CI, engine interface, architecture docs (#968)

Browse Source 
* fix(perf): reduce startup and inference overhead

* fix(perf): report effective whisper.cpp fallback settings

* fix(injection): add opt-in X11 paste for long dictations

* perf(vad): lower default silence timeout to 1.2s

* fix(types): resolve mypy errors across application source

* ci: run mypy in the lint job

* ci: include mypy in lint dependency group

* fix(types): narrow evdev timestamp comparison

* refactor(recognition): add SpeechEngine protocol and engine dispatch table

* refactor(settings): move About section and engine tables out of settings_dialog

* docs: add ARCHITECTURE.md and refresh docs index and AGENTS layout

* fix(settings): preserve GTK dialog compatibility after extraction

* style: format touched Python modules

* fix(types): keep local typecheck from syncing dependencies

* test(vad): align main startup expectation with new default

* style: sort engine registry and protocol test imports

* test(settings): follow extracted About builder attribute alias

* test(settings): isolate GTK dialog base in About test

* test(settings): isolate About color scheme build test

* test(settings): restore GTK base after About scheme test

* test(settings): isolate About scheme check from GTK mocks

* test(settings): support extracted dialog mixin methods

* docs: add before/after performance benchmark charts

* fix(injection): type instead of pasting when the clipboard holds non-text data
This commit is contained in:
Jatin K Malik
2026-10-20 20:23:08 -07:00
committed by GitHub
parent 4799f06bd2
commit 65de3427fb
48 changed files with 7354 additions and 905 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/vocalinux/audio/capture.py
+8 -8
View File
@@ -15,7 +15,7 @@ import ctypes
import logging
import subprocess
import time
from typing import TYPE_CHECKING, Any, Optional
from typing import TYPE_CHECKING, Any, Optional, cast
from .pipewire import (
PIPEWIRE_INDEX_BASE,
@@ -280,7 +280,7 @@ def _resolve_valid_input_device(audio: Any, preferred_index: Optional[int] = Non
Returns:
A valid device index with input channels, or None if none found.
"""
input_device_indices = []
input_device_indices: list[int] = []
default_input_index = None
try:
@@ -345,7 +345,7 @@ def _resolve_valid_input_device(audio: Any, preferred_index: Optional[int] = Non
)
if default_input_index is not None and default_input_index in input_device_indices:
return default_input_index
return cast(int, default_input_index)
return input_device_indices[0]
@@ -644,8 +644,8 @@ def test_audio_input(device_index: Optional[int] = None, duration: float = 1.0)
"error": None,
}
if is_pipewire_device_index(device_index):
return _test_pipewire_input(device_index, duration)
if is_pipewire_device_index(cast(int, device_index)):
return _test_pipewire_input(cast(int, device_index), duration)
try:
import numpy as np
@@ -700,12 +700,12 @@ def test_audio_input(device_index: Optional[int] = None, duration: float = 1.0)
return result
# Record and analyze
all_amplitudes = []
all_amplitudes: list[Any] = []
frames_to_read = int(RATE * duration / CHUNK)
for _ in range(frames_to_read):
try:
data = stream.read(CHUNK, exception_on_overflow=False)
data = cast(Any, stream).read(CHUNK, exception_on_overflow=False)
audio_data = np.frombuffer(data, dtype=np.int16)
amplitudes = np.abs(audio_data)
all_amplitudes.extend(amplitudes)
@@ -724,7 +724,7 @@ def test_audio_input(device_index: Optional[int] = None, duration: float = 1.0)
result["mean_amplitude"] = float(np.mean(all_amplitudes))
# Signal present if max amplitude is above typical digital noise floor
# 16-bit audio has max value of 32768, noise floor is typically < 100
result["has_signal"] = result["max_amplitude"] > 200
result["has_signal"] = cast(float, result["max_amplitude"]) > 200
except ImportError as e:
result["error"] = f"Missing dependency: {e}"
src/vocalinux/auto_pause_monitor.py
+1 -1
View File
@@ -119,7 +119,7 @@ class AutoPauseMonitor:
self._paused = False
self._timeout_id: Optional[int] = None
self._running = False
self._last_poll_interval = DEFAULT_POLL_INTERVAL_SECONDS
self._last_poll_interval: float = DEFAULT_POLL_INTERVAL_SECONDS
@property
def paused(self) -> bool:
src/vocalinux/common_types.py
+45 -1
View File
@@ -4,7 +4,7 @@ This module provides type definitions to avoid circular imports.
"""
from enum import Enum, auto
from typing import Any, Callable, Iterable, Optional, Protocol # noqa: F401
from typing import Any, Callable, Iterable, Optional, Protocol, runtime_checkable # noqa: F401
class RecognitionState(Enum):
@@ -51,10 +51,40 @@ class Engine(Protocol):
...
@runtime_checkable
class SpeechEngine(Protocol):
"""Runtime-checkable contract for speech transcription engines."""
def init(self) -> None:
"""Initialize the engine and load any required models."""
...
def is_ready(self) -> bool:
"""Return True if the engine is initialized and ready to transcribe."""
...
def transcribe(
self,
audio_buffer: list[bytes],
language: Optional[str] = None,
initial_prompt: Optional[str] = None,
) -> str:
"""Transcribe audio chunks with optional per-call language and prompt."""
...
def cleanup(self) -> None:
"""Release any engine resources."""
...
class SpeechRecognitionManagerProtocol(Protocol):
"""Protocol defining the interface for SpeechRecognitionManager."""
state: RecognitionState
state_callbacks: list[Callable[[RecognitionState], None]]
buffer_during_reload: bool
test_capture_floor: Optional[float]
test_capture_ceiling: Optional[float]
def start_recognition(self, mode: str = "toggle") -> bool:
"""Start the speech recognition process. Returns True if listening began."""
@@ -76,6 +106,14 @@ class SpeechRecognitionManagerProtocol(Protocol):
"""Register a callback for recognized text."""
...
def get_text_callbacks(self) -> list[Callable[[str], None]]:
"""Return registered transcript callbacks."""
...
def set_text_callbacks(self, callbacks: list[Callable[[str], None]]) -> None:
"""Replace registered transcript callbacks."""
...
def register_segment_callback(self, callback: Callable[[str, float], None]) -> None:
"""Register a callback for recognized text with its audio-capture start time."""
...
@@ -123,6 +161,12 @@ class SpeechRecognitionManagerProtocol(Protocol):
"""Register a callback for download progress (fraction, mbps, status)."""
...
def set_audio_device(
self, device_index: Optional[int], device_name: Optional[str] = None
) -> None:
"""Set the preferred audio input device."""
...
def download_whispercpp_model(self, model_name: str) -> None:
"""Download a catalog whisper.cpp model without changing engine config."""
...
src/vocalinux/main.py
+1 -1
View File
@@ -527,7 +527,7 @@ def main():
logger.info(f"Using model={model_size} (saved for engine {engine})")
vad_sensitivity = saved_settings.get("vad_sensitivity", 3)
silence_timeout = saved_settings.get("silence_timeout", 2.0)
silence_timeout = saved_settings.get("silence_timeout", 1.2)
stop_sound_guard_ms = saved_settings.get("stop_sound_guard_ms", 200)
voice_commands_enabled = saved_settings.get("voice_commands_enabled") # None = auto
audio_device_index = audio_settings.get("device_index", None)
src/vocalinux/speech_recognition/command_phrases.py
+4 -3
View File
@@ -7,7 +7,7 @@ punctuation / line-break aliases keyed by Whisper ISO language codes
from __future__ import annotations
from typing import Optional
from typing import Optional, cast
from vocalinux.utils.vosk_model_info import SUPPORTED_LANGUAGES
@@ -192,8 +192,9 @@ def normalize_command_language(language: Optional[str]) -> str:
if not language or language == "auto":
return "en"
info = SUPPORTED_LANGUAGES.get(language)
if info is not None and info.get("whisper"):
return info["whisper"]
language_info = cast(Optional[dict[str, str]], info)
if language_info is not None and language_info.get("whisper"):
return language_info["whisper"]
if language.startswith("en"):
return "en"
return language
src/vocalinux/speech_recognition/command_processor.py
+1 -1
View File
@@ -111,7 +111,7 @@ class CommandProcessor:
}
# Active format modifiers
self.active_formats = set()
self.active_formats: set[str] = set()
# Compile regex patterns for faster matching
self._compile_patterns()
src/vocalinux/speech_recognition/engines/__init__.py
+49 -1
View File
@@ -4,10 +4,14 @@ New engine implementations are registered here so the recognition manager can
discover them without hard-coding every backend.
"""
import importlib.util
import logging
from typing import Type
from ...common_types import Engine, EngineType
logger = logging.getLogger(__name__)
ENGINES: dict[EngineType, Type[Engine]] = {}
# Register faster-whisper when available.
@@ -18,4 +22,48 @@ try:
except Exception:
pass
__all__ = ["ENGINES"]
def _module_available(name: str) -> bool:
try:
return importlib.util.find_spec(name) is not None
except (ImportError, ValueError):
return False
def available_engines() -> list[str]:
"""Return installed speech engines in the Settings display order."""
engines = {
"vosk": _module_available("vosk"),
"whisper": _module_available("whisper"),
"whisper_cpp": False,
"parakeet": False,
"faster_whisper": _module_available("faster_whisper"),
"remote_api": False,
}
try:
from pywhispercpp.model import Model
engines["whisper_cpp"] = True
except ImportError:
pass
try:
import sherpa_onnx # noqa: F401
engines["parakeet"] = True
except ImportError:
pass
try:
import requests # noqa: F401
engines["remote_api"] = True
except ImportError:
pass
logger.debug("Available engines: %s", engines)
return [name for name, available in engines.items() if available]
__all__ = ["ENGINES", "available_engines"]
src/vocalinux/speech_recognition/recognition_manager.py
+110 -83
View File
@@ -10,6 +10,7 @@ import importlib.util
import inspect
import json
import logging
import multiprocessing
import os
import queue
import re
@@ -18,7 +19,7 @@ import sys
import threading
import time
from pathlib import Path
from typing import TYPE_CHECKING, Callable, Optional, Union
from typing import TYPE_CHECKING, Any, Callable, Optional, Union, cast
if TYPE_CHECKING:
import numpy as np
@@ -74,6 +75,7 @@ from ..utils.whisper_model_info import (
from ..utils.whispercpp_model_info import (
NON_DICTATION_MODELS,
WHISPERCPP_MODEL_INFO,
ComputeBackend,
get_model_path,
is_english_only_model,
is_model_downloaded,
@@ -145,8 +147,9 @@ def resolve_whisper_language(language: str) -> Optional[str]:
if language == "auto":
return None
info = SUPPORTED_LANGUAGES.get(language)
if info is not None and "whisper" in info:
return info["whisper"]
language_info = cast(Optional[dict[str, str]], info)
if language_info is not None and "whisper" in language_info:
return language_info["whisper"]
# Legacy fallback for codes not present in the catalog.
if language == "en-us":
return "en"
@@ -409,9 +412,9 @@ class SpeechRecognitionManager:
)
self._vosk_dictionary_warned = False
self.state = RecognitionState.IDLE
self.audio_thread = None
self.recognition_thread = None
self.model = None
self.audio_thread: Optional[threading.Thread] = None
self.recognition_thread: Optional[threading.Thread] = None
self.model: Any = None
self.recognizer = None # Added for VOSK
self.command_processor = CommandProcessor(language=self.language)
@@ -461,7 +464,7 @@ class SpeechRecognitionManager:
# Speech detection parameters (load defaults, will be overridden by configure)
self.vad_sensitivity = kwargs.get("vad_sensitivity", 3)
self.silence_timeout = kwargs.get("silence_timeout", 2.0)
self.silence_timeout = kwargs.get("silence_timeout", 1.2)
# Silero VAD (neural-network-based, falls back to amplitude if unavailable)
self._silero_vad = load_silero_vad()
@@ -486,7 +489,8 @@ class SpeechRecognitionManager:
self.whispercpp_entropy_thold = kwargs.get("whispercpp_entropy_thold", 2.4)
self.whispercpp_logprob_thold = kwargs.get("whispercpp_logprob_thold", -1.0)
self.whispercpp_no_speech_thold = kwargs.get("whispercpp_no_speech_thold", 0.6)
self.whispercpp_n_threads = kwargs.get("whispercpp_n_threads", None)
self.whispercpp_n_threads = cast(Optional[int], kwargs.get("whispercpp_n_threads", None))
self._whispercpp_effective_threads: Optional[int] = None
self.whispercpp_gpu_device = kwargs.get("whispercpp_gpu_device", None)
# Remote API settings
@@ -496,7 +500,7 @@ class SpeechRecognitionManager:
self.remote_api_model = kwargs.get("remote_api_model", "whisper-1")
self._http_session = None
self._faster_whisper_engine = None
self._faster_whisper_engine: Any = None
# Audio diagnostics tracking
self._last_audio_level = 0.0
@@ -505,14 +509,14 @@ class SpeechRecognitionManager:
# Recording control flags
self.should_record = False
self._recognition_mode = "toggle" # "toggle" or "push_to_talk"
self.audio_buffer = []
self.audio_buffer: list[bytes] = []
# Monotonic time the current buffer's first chunk was captured,
# propagated onto each enqueued _AudioSegment.
self._segment_started_at: Optional[float] = None
self._recording_segment_has_speech = False
self._buffer_lock = threading.Lock() # Thread safety for audio_buffer
self._model_lock = threading.Lock() # Thread safety for model/recognizer access
self._segment_queue = queue.Queue(maxsize=32)
self._segment_queue: queue.Queue[tuple[_AudioSegment, str]] = queue.Queue(maxsize=32)
# Reliability improvements - Issue #92
self._max_buffer_size = 5000 # Maximum number of audio chunks in buffer
@@ -1124,7 +1128,7 @@ class SpeechRecognitionManager:
else:
logger.debug("Whisper returned empty transcription")
return text
return cast(str, text)
except (RuntimeError, OSError, ValueError) as e:
logger.error(f"Error in Whisper transcription: {e}", exc_info=True)
@@ -1335,16 +1339,19 @@ class SpeechRecognitionManager:
logger.warning("faster-whisper engine not ready during transcription")
return ""
return self._faster_whisper_engine.transcribe(
audio_buffer,
language=self._dictation_language(language),
initial_prompt=self._get_dictionary_prompt(),
return cast(
str,
self._faster_whisper_engine.transcribe(
audio_buffer,
language=self._dictation_language(language),
initial_prompt=self._get_dictionary_prompt(),
),
)
except (RuntimeError, OSError, ValueError) as e:
logger.error(f"Error in faster-whisper transcription: {e}", exc_info=True)
return ""
def _init_whispercpp(self):
def _init_whispercpp(self) -> None:
"""Initialize the whisper.cpp speech recognition engine."""
try:
_preload_pywhispercpp_shared_libraries()
@@ -1434,9 +1441,15 @@ class SpeechRecognitionManager:
logger.info("Removed the unverified model; it will be downloaded again")
return False
def _build_whispercpp_model_kwargs(self, n_threads: int) -> dict:
def _whispercpp_cpu_thread_count(self) -> int:
if self.whispercpp_n_threads is not None and self.whispercpp_n_threads > 0:
return self.whispercpp_n_threads
return max(1, min(multiprocessing.cpu_count() - 2, 8))
def _build_whispercpp_model_kwargs(self, n_threads: int) -> dict[str, object]:
model_kwargs = {
"n_threads": n_threads,
"print_progress": False,
"suppress_blank": True,
"no_speech_thold": self.whispercpp_no_speech_thold,
"entropy_thold": self.whispercpp_entropy_thold,
@@ -1565,11 +1578,7 @@ class SpeechRecognitionManager:
Args:
model_path: Filesystem path to the GGML model file.
"""
import multiprocessing
import time
from ..utils.whispercpp_model_info import (
ComputeBackend,
_prefer_discrete_vulkan_device,
detect_compute_backend,
detect_cpu_info,
@@ -1671,13 +1680,14 @@ class SpeechRecognitionManager:
has_gpu_libs = actual_gpu_backend in ("vulkan", "cuda")
if self.whispercpp_n_threads is not None and self.whispercpp_n_threads > 0:
n_threads = self.whispercpp_n_threads
elif has_gpu_libs:
if has_gpu_libs and not (
self.whispercpp_n_threads is not None and self.whispercpp_n_threads > 0
):
n_threads = max(1, multiprocessing.cpu_count() // 4)
else:
n_threads = max(1, min(multiprocessing.cpu_count() - 2, 8))
n_threads = self._whispercpp_cpu_thread_count()
self._whispercpp_effective_threads = n_threads
load_start_time = time.time()
loaded_backend = runtime_backend
@@ -1692,23 +1702,30 @@ class SpeechRecognitionManager:
model_error, model_path, model_kwargs, ComputeBackend.CPU
)
effective_threads = self._whispercpp_effective_threads
if effective_threads is None:
effective_threads = n_threads
load_duration = time.time() - load_start_time
if has_gpu_libs:
logger.info(
f"whisper.cpp configured with n_threads={n_threads} "
f"(GPU backend: {actual_gpu_backend})"
)
else:
logger.info(
f"whisper.cpp configured with n_threads={n_threads} "
f"(CPU-only; pywhispercpp lacks GPU libraries)"
)
logger.info(f"whisper.cpp model loaded in {load_duration:.2f}s ({loaded_backend} backend)")
logger.info(
f"whisper.cpp configured with n_threads={effective_threads} "
f"({loaded_backend} backend)"
)
logger.info(
f"whisper.cpp model loaded in {load_duration:.2f}s "
f"({loaded_backend} backend, n_threads={effective_threads})"
)
self._model_initialized = True
logger.info("whisper.cpp engine initialized successfully.")
def _handle_gpu_fallback(self, error, model_path: str, model_kwargs: dict, cpu_backend):
def _handle_gpu_fallback(
self,
error: RuntimeError,
model_path: str,
model_kwargs: dict[str, object],
cpu_backend: str,
) -> str:
"""Handle GPU backend failure by falling back to CPU.
Args:
@@ -1741,7 +1758,12 @@ class SpeechRecognitionManager:
# Force CPU backend by disabling GPU backends
os.environ["GGML_VULKAN"] = "0"
os.environ["GGML_CUDA"] = "0"
self.model = self._load_model_with_compatible_params(model_path, model_kwargs)
cpu_model_kwargs = model_kwargs.copy()
cpu_thread_count = self._whispercpp_cpu_thread_count()
cpu_model_kwargs["n_threads"] = cpu_thread_count
self._whispercpp_effective_threads = cpu_thread_count
self.model = self._load_model_with_compatible_params(model_path, cpu_model_kwargs)
logger.info("Reloaded on CPU with n_threads=%d", cpu_thread_count)
logger.info("Successfully loaded model with CPU backend")
return cpu_backend
@@ -1815,7 +1837,10 @@ class SpeechRecognitionManager:
",".join(unsupported),
)
if candidate_probs:
lang = max(candidate_probs, key=candidate_probs.get)
lang = max(
candidate_probs,
key=cast(Callable[[str], float], candidate_probs.get),
)
logger.info(
"whisper.cpp restricted language detection: detected=%s %.3f, candidates=%s, using=%s %.3f",
detected[0],
@@ -2107,7 +2132,7 @@ class SpeechRecognitionManager:
stream = self.model.create_stream()
stream.accept_waveform(16000, audio_float)
self.model.decode_stream(stream)
return stream.result.text.strip()
return cast(str, stream.result.text.strip())
except (RuntimeError, OSError, ValueError) as e:
audio_info = (
@@ -2370,7 +2395,7 @@ class SpeechRecognitionManager:
result = response.json()
# whisper.cpp server format returns {"text": "..."}
return result.get("text", "")
return cast(str, result.get("text", ""))
except requests.exceptions.ConnectionError as e:
logger.error(f"Cannot connect to remote server {url}: {e}")
@@ -2644,7 +2669,7 @@ class SpeechRecognitionManager:
system_model_path = os.path.join(system_dir, model_name)
if os.path.exists(system_model_path):
logger.info(f"Found pre-installed model at: {system_model_path}")
return system_model_path
return cast(str, system_model_path)
# If not found anywhere, return the user path (will be created if needed)
logger.debug(f"No existing model found, will use: {user_model_path}")
@@ -3663,6 +3688,29 @@ class SpeechRecognitionManager:
self._process_audio_buffer(audio_buffer)
def _transcribe_with_vosk(self, audio_buffer: list[bytes]) -> Optional[str]:
"""Transcribe one audio segment with the initialized Vosk recognizer."""
with self._model_lock:
if self.recognizer is None:
logger.warning("Recognizer is None during processing, returning empty result")
return None
for data in audio_buffer:
self.recognizer.AcceptWaveform(data)
result = json.loads(self.recognizer.FinalResult())
return result.get("text", "")
def _transcribe_remote_api_segment(
self, audio_buffer: list[bytes], language: Optional[str]
) -> Optional[str]:
"""Snapshot the HTTP session and transcribe one remote API segment."""
with self._model_lock:
session = self._http_session
if session is None:
logger.error("Remote API HTTP session not initialized")
return None
return self._transcribe_with_remote_api(audio_buffer, session, language)
def _process_audio_buffer(
self, audio_buffer: list[bytes], language: Optional[str] = None
) -> None:
@@ -3678,48 +3726,27 @@ class SpeechRecognitionManager:
language = getattr(audio_buffer, "language", None)
dictation_language = self._dictation_language(language)
if self.engine == "vosk":
# Lock recognizer access to prevent race condition with reconfigure
with self._model_lock:
# Check if recognizer is still valid
if self.recognizer is None:
logger.warning("Recognizer is None during processing, returning empty result")
return
for data in audio_buffer:
self.recognizer.AcceptWaveform(data)
result = json.loads(self.recognizer.FinalResult())
text = result.get("text", "")
elif self.engine == "whisper":
text = self._transcribe_with_whisper(audio_buffer, dictation_language)
elif self.engine == "whisper_cpp":
text = self._transcribe_with_whispercpp(audio_buffer, dictation_language)
elif self.engine == "parakeet":
text = self._transcribe_with_parakeet(audio_buffer)
elif self.engine == "faster_whisper":
text = self._transcribe_with_faster_whisper(audio_buffer, dictation_language)
elif self.engine == "remote_api":
# Snapshot the HTTP session under lock to prevent race with
# reconfigure() / reinitialize_after_resume() which close/recreate
# the session under _model_lock. The snapshot (a local reference)
# remains valid even if another thread closes the old session —
# urllib3's PoolManager keeps existing connections alive until the
# in-flight request completes.
with self._model_lock:
session = self._http_session
if session is None:
logger.error("Remote API HTTP session not initialized")
return
text = self._transcribe_with_remote_api(audio_buffer, session, dictation_language)
else:
transcribers: dict[str, Callable[[], Optional[str]]] = {
"vosk": lambda: self._transcribe_with_vosk(audio_buffer),
"whisper": lambda: self._transcribe_with_whisper(audio_buffer, dictation_language),
"whisper_cpp": lambda: self._transcribe_with_whispercpp(
audio_buffer, dictation_language
),
"parakeet": lambda: self._transcribe_with_parakeet(audio_buffer),
"faster_whisper": lambda: self._transcribe_with_faster_whisper(
audio_buffer, dictation_language
),
"remote_api": lambda: self._transcribe_remote_api_segment(
audio_buffer, dictation_language
),
}
transcriber = transcribers.get(self.engine)
if transcriber is None:
logger.error(f"Unknown engine: {self.engine}")
return
text = transcriber()
if text is None:
return
# Process text - either with voice commands or pass through directly
logger.debug(f"_process_audio_buffer got text='{text[:50] if text else '(empty)'}...'")
src/vocalinux/text_injection/ibus_engine.py
+8 -4
View File
@@ -25,7 +25,7 @@ import sys
import threading
import time
from pathlib import Path
from typing import Callable, Optional
from typing import Any, Callable, Mapping, Optional
# Package import when loaded normally; absolute / inline fallbacks when this
# file is executed by path (Vocalinux start_engine_process / IBus component exec).
@@ -39,7 +39,7 @@ except ImportError: # pragma: no cover - exercised via script-path subprocess
except ImportError:
# Last resort: same semantics as the vocalinux.utils originals, except
# that host_env cannot strip a bundle it can no longer import.
def host_env(base=None):
def host_env(base: Optional[Mapping[str, str]] = None) -> dict[str, str]:
return dict(os.environ if base is None else base)
def xdg_data_home() -> str:
@@ -1135,7 +1135,10 @@ def stop_engine_process(wait_timeout: float = 2.0) -> None:
PID_FILE.unlink()
class VocalinuxEngine(IBus.Engine if IBUS_AVAILABLE else object):
IBusEngineBase: Any = IBus.Engine if IBUS_AVAILABLE else object
class VocalinuxEngine(IBusEngineBase):
"""
IBus proxy engine for Vocalinux text injection.
@@ -1464,7 +1467,7 @@ class IBusTextInjector:
# Empty until a real X11 layout is captured. A non-empty default (e.g.
# "us") would make stop()/restore re-apply a layout that was never
# captured, flipping non-US users to us. See issue #474.
self._previous_xkb_layout: tuple = ("", "", "")
self._previous_xkb_layout: Optional[tuple[str, str, str]] = ("", "", "")
if auto_activate:
self._setup_engine()
@@ -1598,6 +1601,7 @@ class IBusTextInjector:
when the live engine is Vocalinux or unavailable (failed inject restore).
"""
current = get_current_engine()
restore_engine: Optional[str]
if current and current != ENGINE_NAME:
restore_engine = current
else:
src/vocalinux/text_injection/text_injector.py
+119 -32
View File
@@ -14,7 +14,7 @@ import subprocess
import threading
import time
from enum import Enum
from typing import List, Optional, Tuple
from typing import List, Optional, Tuple, cast
from ..utils.host_process import host_env
from ..utils.paths import config_dir
@@ -32,6 +32,18 @@ from .remote_desktop_portal import (
)
_PORTAL_SUBMIT_TIMEOUT_S = 195.0 # portal's _START_TIMEOUT_S + _REQUEST_TIMEOUT_S
_X11_PASTE_MIN_CHARS = 30
_X11_TEXT_CLIPBOARD_TARGETS = frozenset(
{
"utf8_string",
"string",
"text",
"text/plain",
"text/plain;charset=utf-8",
"compound_text",
}
)
_X11_CLIPBOARD_META_TARGETS = frozenset({"targets", "timestamp", "multiple", "save_targets"})
logger = logging.getLogger(__name__)
@@ -149,7 +161,7 @@ class TextInjector:
self._state_lock = threading.Lock()
self._abort_injections = threading.Event()
self.last_typed_count = 0
self._clipboard_tool_health = {}
self._clipboard_tool_health: dict[str, bool] = {}
self._clipboard_timeout = 0.35
# Overlapping ydotool pastes: bump generation to cancel stale restores;
# target keeps the original pre-injection clipboard across the window.
@@ -1564,7 +1576,9 @@ class TextInjector:
if os.path.exists(config_path):
with open(config_path, "r") as f:
config = json.load(f)
return config.get("text_injection", {}).get("copy_to_clipboard", False)
return cast(
bool, config.get("text_injection", {}).get("copy_to_clipboard", False)
)
except Exception as e:
logger.debug(f"Could not read copy_to_clipboard setting: {e}")
return False
@@ -1785,7 +1799,7 @@ class TextInjector:
logger.warning("Could not import audio feedback module")
return False
def _inject_with_xdotool(self, text: str):
def _inject_with_xdotool(self, text: str) -> None:
"""
Inject text using xdotool for X11 environments.
@@ -1845,6 +1859,21 @@ class TextInjector:
except Exception as e:
logger.debug(f"Window focus command failed: {e}")
if (
self.environment == DesktopEnvironment.X11
and len(text) >= _X11_PASTE_MIN_CHARS
and self._paste_long_text_enabled()
):
self._wait_for_modifiers_released()
if self._clipboard_holds_non_text():
logger.info(
"Clipboard holds non-text content; typing instead of pasting to preserve it"
)
elif self._inject_via_clipboard_paste(text, env=env):
return
else:
logger.warning("Long-text clipboard paste failed; falling back to xdotool typing")
# Inject text using xdotool
try:
max_retries = 2
@@ -1921,10 +1950,6 @@ class TextInjector:
)
typed += 1 # the '\n' the Return produced
# Add a larger delay between chunks
if i + chunk_size < len(text):
time.sleep(0.1)
logger.info(
f"Text injected using xdotool: '{text[:20]}...' ({len(text)} chars)"
)
@@ -1996,10 +2021,14 @@ class TextInjector:
Returns the text, "" if a tool reports a verifiably empty clipboard,
or None if unreadable as text (non-text data, no tool, or error).
On WAYLAND_XDOTOOL this reads the X11 CLIPBOARD (xclip/xsel), matching
the selection the xdotool paste wrote. wl-paste is a different clipboard.
On X11 and WAYLAND_XDOTOOL this reads the X11 CLIPBOARD (xclip/xsel),
matching the selection the xdotool paste wrote. wl-paste is a different
clipboard.
"""
prefer_x11 = self.environment == DesktopEnvironment.WAYLAND_XDOTOOL
prefer_x11 = self.environment in (
DesktopEnvironment.X11,
DesktopEnvironment.WAYLAND_XDOTOOL,
)
host_is_wayland = (
self._session_environment == DesktopEnvironment.WAYLAND
or os.environ.get("XDG_SESSION_TYPE", "").lower() == "wayland"
@@ -2041,6 +2070,40 @@ class TextInjector:
return "" if saw_empty else None
def _clipboard_holds_non_text(self) -> bool:
"""Return whether the X11 clipboard has no readable text target."""
if self.environment != DesktopEnvironment.X11:
return False
if not shutil.which("xclip"):
return self._read_clipboard() is None
try:
result = subprocess.run(
["xclip", "-selection", "clipboard", "-t", "TARGETS", "-o"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
timeout=1.0,
text=True,
encoding="utf-8",
errors="replace",
env=host_env(),
)
except (subprocess.TimeoutExpired, OSError, UnicodeDecodeError):
logger.debug("Could not inspect X11 clipboard targets")
return self._read_clipboard() is None
if result.returncode != 0 or not result.stdout.strip():
return False
targets = {
target.strip().casefold() for target in result.stdout.splitlines() if target.strip()
}
if not targets:
return False
content_targets = targets.difference(_X11_CLIPBOARD_META_TARGETS)
return not any(target in _X11_TEXT_CLIPBOARD_TARGETS for target in content_targets)
def _paste_shortcut_preference(self) -> str:
"""Return the configured clipboard-paste shortcut id."""
from ..ui.config_manager import DEFAULT_PASTE_SHORTCUT, normalize_paste_shortcut
@@ -2059,6 +2122,24 @@ class TextInjector:
logger.debug(f"Could not read paste_shortcut setting: {exc}")
return DEFAULT_PASTE_SHORTCUT
def _paste_long_text_enabled(self) -> bool:
"""Return whether long X11 dictations should use clipboard paste."""
try:
import json
config_path = os.path.join(config_dir(), "config.json")
if not os.path.exists(config_path):
return False
with open(config_path, "r") as handle:
config = json.load(handle)
text_injection = config.get("text_injection", {})
if not isinstance(text_injection, dict):
return False
return bool(text_injection.get("paste_long_text", False))
except (OSError, ValueError, TypeError, AttributeError) as exc:
logger.debug("Could not read paste_long_text setting: %s", exc)
return False
def _should_use_terminal_paste(self) -> bool:
"""Return True when clipboard injection should send Ctrl+Shift+V.
@@ -2085,12 +2166,12 @@ class TextInjector:
that chord instead. Workaround for ydotool's US-ASCII-only key events
(see issue #362). The chord itself is sent with wtype keysyms when
that tool is usable, otherwise with a layout-resolved ydotool keycode
for Latin 'v' (issue #787). On WAYLAND_XDOTOOL the copy uses xclip/xsel
and the chord is xdotool's, because the focused XWayland window pastes
the X11 CLIPBOARD (#657). Saves the previous clipboard and restores it
after a short delay. Overlapping pastes share one restore target
(pre-first-injection content) and a generation counter so stale restore
threads exit.
for Latin 'v' (issue #787). On X11 and WAYLAND_XDOTOOL the copy uses
xclip/xsel and the chord is xdotool's, because the focused X11/XWayland
window pastes the X11 CLIPBOARD (#657). Saves the previous clipboard and
restores it after a short delay. Overlapping pastes share one restore
target (pre-first-injection content) and a generation counter so stale
restore threads exit.
Args:
text: The text to inject.
@@ -2100,12 +2181,12 @@ class TextInjector:
Returns:
True if successful, False otherwise
"""
logger.debug(
"Using clipboard-paste injection for non-ASCII text "
"(saving clipboard to restore after paste)"
)
logger.debug("Using clipboard-paste injection (saving clipboard to restore after paste)")
prefer_x11 = self.environment == DesktopEnvironment.WAYLAND_XDOTOOL
prefer_x11 = self.environment in (
DesktopEnvironment.X11,
DesktopEnvironment.WAYLAND_XDOTOOL,
)
tools = self._x11_clipboard_tools() if prefer_x11 else None
# Inherit a pending restore target so a second paste in the delay window
@@ -2133,7 +2214,7 @@ class TextInjector:
):
self._clipboard_restore_target = previous_clipboard
# Simulate paste. XWayland uses xdotool against the X11 CLIPBOARD;
# Simulate paste. X11/XWayland uses xdotool against the X11 CLIPBOARD;
# native Wayland prefers wtype keysyms, then layout-resolved ydotool.
# ydotool syntax differs by version:
# - 0.1.x (distro packages): named sequences, e.g. ctrl+v
@@ -2408,11 +2489,14 @@ class TextInjector:
def _clipboard_paste_command(self, *, terminal: bool = False) -> list:
"""Return argv for the clipboard paste chord.
On WAYLAND_XDOTOOL, xdotool against the X11 CLIPBOARD (what an XWayland
window pastes from). Otherwise prefers wtype keysyms, then ydotool
with a layout-resolved keycode for 'v'.
On X11 and WAYLAND_XDOTOOL, xdotool pastes from the X11 CLIPBOARD.
Otherwise prefer wtype keysyms, then ydotool with a layout-resolved
keycode for 'v'.
"""
if self.environment == DesktopEnvironment.WAYLAND_XDOTOOL:
if self.environment in (
DesktopEnvironment.X11,
DesktopEnvironment.WAYLAND_XDOTOOL,
):
chord = "ctrl+shift+v" if terminal else "ctrl+v"
return ["xdotool", "key", "--clearmodifiers", chord]
if self._wtype_usable_for_paste():
@@ -3190,8 +3274,11 @@ class TextInjector:
except (TypeError, ValueError):
return 0.012
def _log_current_window_info(self):
def _log_current_window_info(self) -> None:
"""Log information about the current window/application for debugging."""
if not logger.isEnabledFor(logging.DEBUG):
return
try:
if (
self.environment == DesktopEnvironment.X11
@@ -3203,7 +3290,7 @@ class TextInjector:
except Exception as e:
logger.debug(f"Could not get window info: {e}")
def _log_x11_window_info(self):
def _log_x11_window_info(self) -> None:
"""Log X11 window information."""
env = os.environ.copy()
@@ -3238,7 +3325,7 @@ class TextInjector:
timeout=2,
)
window_name = result.stdout.strip()
logger.info(f"Target window: '{window_name}' (ID: {window_id})")
logger.debug(f"Target window: '{window_name}' (ID: {window_id})")
# Prefer xprop; xdotool classname is last-resort when xprop is missing
# (see focused_window.read_wm_class).
@@ -3262,12 +3349,12 @@ class TextInjector:
try:
with open(f"/proc/{window_pid}/comm", "r") as f:
process_name = f.read().strip()
logger.info(f"Target process: {process_name} (PID: {window_pid})")
logger.debug(f"Target process: {process_name} (PID: {window_pid})")
except Exception:
pass
except subprocess.TimeoutExpired:
logger.warning("Timeout getting window information")
logger.debug("Timeout getting window information")
except subprocess.CalledProcessError as e:
logger.debug(f"xdotool command failed: {e.stderr}")
except Exception as e:
src/vocalinux/ui/config_manager.py
+20 -8
View File
@@ -9,7 +9,7 @@ import json
import logging
import os
import threading
from typing import Any, Optional
from typing import Any, Optional, cast
from ..utils.paths import config_dir
from ..utils.vosk_model_info import SUPPORTED_LANGUAGES
@@ -126,7 +126,7 @@ DEFAULT_CONFIG = {
"parakeet_model_size": "v3-european", # Parakeet TDT 0.6B v3 (25 European languages)
"faster_whisper_model_size": "tiny", # Default model for faster-whisper engine
"vad_sensitivity": 3, # Voice Activity Detection sensitivity (1-5)
"silence_timeout": 2.0, # Seconds of silence before stopping
"silence_timeout": 1.2, # Seconds of silence before stopping
"stop_sound_guard_ms": 200, # Small tail trim to avoid the stop sound without clipping speech
"voice_commands_enabled": None, # None = auto (enabled for VOSK, disabled for Whisper)
"remote_api_url": "", # Remote speech recognition server URL (e.g. http://192.168.1.100:8080)
@@ -196,6 +196,7 @@ DEFAULT_CONFIG = {
# Clipboard-paste chord: auto-detect terminals, or force Ctrl+V /
# Ctrl+Shift+V when a nested terminal panel is not detected.
"paste_shortcut": "auto",
"paste_long_text": False, # Faster clipboard paste for long X11 dictations
# Text-injection backend: "auto" autodetects, or pin "ibus"/"wtype"/
# "ydotool"/"xdotool" when autodetection is wrong (#476).
# VOCALINUX_FORCE_BACKEND overrides this for a single run.
@@ -346,7 +347,8 @@ def resolve_whispercpp_variant(
# The "layout" sentinel is deliberately not in the catalog, so it reads as
# non-English here and derives the multilingual variant -- which is what a
# mode that can land on any language needs (#821).
language_is_english = SUPPORTED_LANGUAGES.get(language_id, {}).get("whisper") == "en"
language_info = cast(dict[str, str], SUPPORTED_LANGUAGES.get(language_id, {}))
language_is_english = language_info.get("whisper") == "en"
def stand_in(variant: str) -> str:
# Only language-derived results reach this helper; an explicit pick is
@@ -429,7 +431,7 @@ class ConfigManager:
return
try:
detected = detect_system_language(SUPPORTED_LANGUAGES)
detected = detect_system_language(cast(dict[str, dict[str, str]], SUPPORTED_LANGUAGES))
except Exception as exc: # pragma: no cover - detection must never block startup
logger.debug(f"Language detection failed: {exc}")
return
@@ -743,7 +745,7 @@ class ConfigManager:
def get_settings(self) -> dict[str, Any]:
"""Get the entire configuration dictionary."""
return self.config
return cast(dict[str, Any], self.config)
def get_model_size_for_engine(self, engine: str) -> str:
"""Return the model id the engine should load for ``engine``.
@@ -763,7 +765,7 @@ class ConfigManager:
saved = sr_config.get("model_size", "small" if engine == "vosk" else "tiny")
if engine.lower() != "whisper_cpp":
return saved
return cast(str, saved)
# Unpinned configs store the bare size; the engine still needs the
# language-derived loadable id (and leftover ``{size}.en`` must not
@@ -825,9 +827,9 @@ class ConfigManager:
if enabled is None:
# Auto mode: enabled for VOSK, disabled for Whisper engines
engine = sr_config.get("engine", "whisper_cpp")
return engine == "vosk"
return bool(engine == "vosk")
return enabled
return cast(bool, enabled)
def update_speech_recognition_settings(self, settings: dict[str, Any]) -> None:
"""Update multiple speech recognition settings at once."""
@@ -913,6 +915,16 @@ class ConfigManager:
self.config["text_injection"] = {}
self.config["text_injection"]["paste_shortcut"] = normalize_paste_shortcut(shortcut)
def get_paste_long_text(self) -> bool:
"""Return whether long X11 dictations should use clipboard paste."""
return self.get_bool("text_injection", "paste_long_text", False)
def set_paste_long_text(self, enabled: bool) -> None:
"""Save whether long X11 dictations should use clipboard paste."""
if "text_injection" not in self.config:
self.config["text_injection"] = {}
self.config["text_injection"]["paste_long_text"] = bool(enabled)
def get_language_shortcuts(self) -> list[dict[str, str]]:
"""Return validated ``[{"shortcut", "language"}]`` bindings (#805).
src/vocalinux/ui/keyboard_backends/base.py
+2 -2
View File
@@ -8,7 +8,7 @@ the required methods.
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Callable, Optional, Tuple
from typing import Callable, Optional, Tuple, cast
Callback = Callable[[], None]
@@ -392,7 +392,7 @@ def format_shortcut_label(spec: ShortcutSpec) -> str:
return SHORTCUT_DISPLAY_NAMES[legacy_id]
return _MODIFIER_LABELS.get(spec.primary_modifier, spec.primary_modifier)
parts = [_MODIFIER_LABELS.get(m, m) for m in spec.modifiers]
parts.append(_main_key_label(spec.key))
parts.append(_main_key_label(cast(str, spec.key)))
return "+".join(parts)
src/vocalinux/ui/keyboard_backends/evdev_backend.py
+4 -4
View File
@@ -26,7 +26,7 @@ import select
import threading
import time
import weakref
from typing import Any, Mapping, Optional, Sequence, TextIO
from typing import Any, Mapping, Optional, Sequence, TextIO, cast
# Try to import evdev
try:
@@ -165,7 +165,7 @@ def _is_post_grab(event: InputEvent, grab_ts: float) -> bool:
unreadable is treated as pre-grab — refusing the grab is the safe side.
"""
try:
return event.timestamp() >= grab_ts
return cast(bool, event.timestamp() >= grab_ts)
except (AttributeError, TypeError):
return False
@@ -1610,8 +1610,8 @@ class EvdevKeyboardBackend(KeyboardBackend):
self._hub._known_engines.add(self)
self._hub.reset_if_idle()
self.last_trigger_time = 0
self.last_key_press_time = 0
self.last_trigger_time: float = 0.0
self.last_key_press_time: float = 0.0
self.double_tap_threshold = 0.3 # seconds
self.key_pressed_devices: set[int] = set()
src/vocalinux/ui/keyboard_backends/pynput_backend.py
+14 -14
View File
@@ -8,7 +8,7 @@ Works on X11 and XWayland, but NOT on pure Wayland.
import logging
import threading
import time
from typing import Optional
from typing import Any, Optional, cast
# Try to import pynput
try:
@@ -24,14 +24,14 @@ from .base import DEFAULT_SHORTCUT, DEFAULT_SHORTCUT_MODE, KeyboardBackend
logger = logging.getLogger(__name__)
def _optional_key(key_name: str):
def _optional_key(key_name: str) -> Any:
if not PYNPUT_AVAILABLE:
return None
return getattr(keyboard.Key, key_name, None)
def _key_set(*key_names: str) -> set:
keys = set()
def _key_set(*key_names: str) -> set[Any]:
keys: set[Any] = set()
for key_name in key_names:
key = _optional_key(key_name)
if key is not None:
@@ -149,10 +149,10 @@ def pynput_key_token(key) -> Optional[str]:
return None
token = _PYNPUT_NAMED_TOKENS.get(key)
if token is not None:
return token
return cast(str, token)
char = getattr(key, "char", None)
if char:
lowered = char.lower()
lowered = cast(str, char.lower())
if len(lowered) == 1 and lowered.isalnum():
return lowered
code = ord(char)
@@ -179,16 +179,16 @@ class PynputKeyboardBackend(KeyboardBackend):
mode: The shortcut mode ("toggle" or "push_to_talk")
"""
super().__init__(shortcut, mode)
self.listener = None
self.last_trigger_time = 0
self.last_key_press_time = 0
self.listener: Optional[Any] = None
self.last_trigger_time: float = 0.0
self.last_key_press_time: float = 0.0
self.double_tap_threshold = 0.3 # seconds
self.current_keys = set()
self.current_keys: set[Any] = set()
# Combo (modifier+key) state.
self._combo_required_modifier_keys: list = []
self._combo_required_modifier_keys: list[Any] = []
self._combo_key_token: Optional[str] = None
self._combo_held_keys: set = set()
self._combo_held_keys: set[Any] = set()
self._combo_active = False
self._resolve_combo_targets()
@@ -242,11 +242,11 @@ class PynputKeyboardBackend(KeyboardBackend):
return False
normalized_key = self._normalize_modifier_key(key)
return normalized_key == target_key
return cast(bool, normalized_key == target_key)
def is_available(self) -> bool:
"""Check if pynput is available."""
return PYNPUT_AVAILABLE
return cast(bool, PYNPUT_AVAILABLE)
def get_permission_hint(self) -> Optional[str]:
"""
src/vocalinux/ui/logging_dialog.py
+1 -1
View File
@@ -696,7 +696,7 @@ class LoggingDialog(Gtk.Dialog):
dialog.format_secondary_text("Do you want to replace it?")
dialog.add_button("_Cancel", Gtk.ResponseType.CANCEL)
dialog.add_button("_Replace", Gtk.ResponseType.OK)
confirmed = dialog.run() == Gtk.ResponseType.OK
confirmed = bool(dialog.run() == Gtk.ResponseType.OK)
dialog.destroy()
return confirmed
src/vocalinux/ui/logging_manager.py
+2 -2
View File
@@ -217,12 +217,12 @@ class LoggingManager:
return {"total": 0, "by_level": {}, "by_module": {}, "oldest": None, "newest": None}
# Count by level
by_level = {}
by_level: dict[str, int] = {}
for record in records:
by_level[record.level] = by_level.get(record.level, 0) + 1
# Count by module
by_module = {}
by_module: dict[str, int] = {}
for record in records:
module = (
record.logger_name.split(".")[0]
src/vocalinux/ui/settings_about.py
+522
View File
@@ -0,0 +1,522 @@
"""About page content and trusted family links for Settings."""
import logging
import os
from typing import Any, Optional
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk # noqa: E402
from ..utils.update_checker import is_trusted_release_url, normalize_channel
from ..version import __copyright__, __url__, __version__
logger = logging.getLogger(__name__)
VOCALINUX_SITE_URL = "https://vocalinux.com"
VOCAHQ_SITE_URL = "https://vocahq.com"
VOCAMAC_SITE_URL = "https://vocamac.com"
VOCAWIN_SITE_URL = "https://vocawin.com"
VOCAPHONE_SITE_URL = "https://vocaphone.vocahq.com"
VOCAGATEWAY_SITE_URL = "https://vocagateway.vocahq.com"
GITHUB_REPO_URL = __url__
GITHUB_ISSUES_URL = "https://github.com/VocaHQ/vocalinux/issues"
VOCAHQ_DISCORD_URL = "https://discord.gg/t6muquAJbm"
VOCAHQ_X_URL = "https://x.com/vocahq"
VOCAHQ_MAILTO_URL = "mailto:hello@vocahq.com"
# About page links that are not GitHub release URLs.
_ABOUT_OPEN_URLS = frozenset(
{
VOCALINUX_SITE_URL,
VOCAHQ_SITE_URL,
VOCAMAC_SITE_URL,
VOCAWIN_SITE_URL,
VOCAPHONE_SITE_URL,
VOCAGATEWAY_SITE_URL,
GITHUB_REPO_URL,
GITHUB_ISSUES_URL,
VOCAHQ_DISCORD_URL,
VOCAHQ_X_URL,
VOCAHQ_MAILTO_URL,
}
)
# Official Talk-to-us marks from VocaHQ/.github brand/vocahq/social (commit 61c8eee).
# Do not redraw. fill is currentColor; paint to match the dialog foreground so
# the marks stay visible on both light and dark surfaces.
_ABOUT_INK = "#14231C"
_ABOUT_INK_ON_DARK = "#E6E1D8"
_ABOUT_ICON_TEXT_PX = 18
_FAMILY_ICON_PX = 22
def _color_luminance(color: Any) -> float:
"""Relative luminance of a Gdk.RGBA (0 = black, 1 = white)."""
return float(0.2126 * color.red + 0.7152 * color.green + 0.0722 * color.blue)
def _rgba_to_hex(color: Any) -> str:
"""Format a Gdk.RGBA as #rrggbb."""
r = max(0, min(255, int(round(color.red * 255))))
g = max(0, min(255, int(round(color.green * 255))))
b = max(0, min(255, int(round(color.blue * 255))))
return f"#{r:02x}{g:02x}{b:02x}"
def _about_surface_is_dark() -> bool:
"""Return True when GTK is using a dark theme."""
settings = Gtk.Settings.get_default()
if settings is not None:
try:
if bool(settings.get_property("gtk-application-prefer-dark-theme")):
return True
except Exception:
pass
try:
theme = str(settings.get_property("gtk-theme-name") or "").lower()
if "dark" in theme:
return True
except Exception:
pass
try:
from ..utils.gtk_color_scheme import read_os_color_scheme
if read_os_color_scheme() == "prefer-dark":
return True
except Exception:
pass
return False
def _about_ink_hex(
widget: Optional[Gtk.Widget] = None,
dark: Optional[bool] = None,
) -> str:
"""Ink for currentColor marks: theme fg when known, else light/dark defaults."""
if dark is None:
dark = _about_surface_is_dark()
if widget is not None:
ctx = widget.get_style_context()
for name in ("theme_fg_color", "theme_text_color", "fg_color"):
try:
found, color = ctx.lookup_color(name)
except Exception:
continue
if not found:
continue
lum = _color_luminance(color)
if dark and lum >= 0.45:
return _rgba_to_hex(color)
if not dark and lum <= 0.40:
return _ABOUT_INK
return _ABOUT_INK_ON_DARK if dark else _ABOUT_INK
# Official platform marks from VocaHQ/.github brand/promo/cards/platform
# (currentColor copies live in web/public/brand/platforms). Home is a simple
# house so HQ does not reuse a product lockup.
_VOCAHQ_FAMILY_LINKS = (
(
VOCAHQ_SITE_URL,
"VocaHQ",
"Family site",
("platform-home",),
"Open vocahq.com",
),
(
VOCALINUX_SITE_URL,
"Vocalinux",
"Linux, available now",
("platform-linux",),
"Open vocalinux.com",
),
(
VOCAMAC_SITE_URL,
"VocaMac",
"macOS, available now",
("platform-apple",),
"Open vocamac.com",
),
(
VOCAWIN_SITE_URL,
"VocaWin",
"Windows, unsigned beta",
("platform-windows",),
"Open vocawin.com",
),
(
VOCAPHONE_SITE_URL,
"VocaPhone",
"Android beta / iOS TestFlight",
("platform-android", "platform-apple"),
"Open vocaphone.vocahq.com",
),
(
VOCAGATEWAY_SITE_URL,
"VocaGateway",
"Self-hosted, headless",
("platform-server",),
"Open vocagateway.vocahq.com",
),
)
def _can_open_url(url: str) -> bool:
"""Return True for GitHub project URLs or the About page allowlist."""
return bool(url) and (url in _ABOUT_OPEN_URLS or is_trusted_release_url(url))
class AboutSectionMixin(Gtk.Dialog):
"""Build About content on the Settings dialog."""
def _about_mark_image(
self: Any,
icon_name: str,
pixel_size: int = _ABOUT_ICON_TEXT_PX,
dark: Optional[bool] = None,
) -> Optional[Gtk.Image]:
"""Load a currentColor SVG and paint it to match the dialog foreground."""
from gi.repository import GdkPixbuf
from ..utils.resource_manager import ResourceManager
path = ResourceManager().get_icon_path(icon_name)
if not os.path.exists(path):
return None
try:
ink = _about_ink_hex(self, dark)
with open(path, encoding="utf-8") as handle:
svg = handle.read().replace("currentColor", ink)
loader = GdkPixbuf.PixbufLoader.new_with_type("svg")
loader.set_size(pixel_size, pixel_size)
loader.write(svg.encode("utf-8"))
loader.close()
pixbuf = loader.get_pixbuf()
if pixbuf is None:
return None
image = Gtk.Image.new_from_pixbuf(pixbuf)
image.set_pixel_size(pixel_size)
return image
except Exception as exc:
logger.warning("Failed to load About mark %s: %s", icon_name, exc)
return None
def _about_open_button(
self: Any,
url: str,
tooltip: str,
label: str = "Open",
icon_name: Optional[str] = None,
width: int = 100,
dark: Optional[bool] = None,
) -> Gtk.Button:
"""Button that opens a trusted About URL.
Unique labels (Website, Source code, Report a bug or idea) are the
accessible name. Shared "Open" labels use the tooltip instead.
"""
from .settings_dialog import _set_accessible_name
button = Gtk.Button(label=label)
button.set_size_request(width, -1)
button.set_tooltip_text(tooltip)
if label == "Open":
_set_accessible_name(button, tooltip)
if icon_name:
image = self._about_mark_image(icon_name, dark=dark)
if image is not None:
image.set_margin_end(8)
button.set_image(image)
button.set_always_show_image(True)
button.get_style_context().add_class("about-mark-button")
button.connect("clicked", lambda *_args, dest=url: self._open_web_url(dest))
return button
def _family_tile(
self: Any,
url: str,
title: str,
subtitle: str,
icon_names: tuple[str, ...],
tooltip: str,
dark: Optional[bool] = None,
) -> Gtk.Button:
"""Compact icon + name tile that opens a family site."""
from .settings_dialog import _set_accessible_name
inner = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=16)
marks = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
marks.set_valign(Gtk.Align.CENTER)
for icon_name in icon_names:
icon = self._about_mark_image(
icon_name,
pixel_size=_FAMILY_ICON_PX,
dark=dark,
)
if icon is not None:
marks.pack_start(icon, False, False, 0)
inner.pack_start(marks, False, False, 0)
text = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0)
name_label = Gtk.Label(label=title, xalign=0)
name_label.get_style_context().add_class("family-tile-title")
status_label = Gtk.Label(label=subtitle, xalign=0)
status_label.get_style_context().add_class("family-tile-subtitle")
status_label.set_line_wrap(True)
text.pack_start(name_label, False, False, 0)
text.pack_start(status_label, False, False, 0)
inner.pack_start(text, True, True, 0)
button = Gtk.Button()
button.add(inner)
button.get_style_context().add_class("family-tile")
button.set_tooltip_text(tooltip)
button.set_hexpand(True)
button.set_halign(Gtk.Align.FILL)
_set_accessible_name(button, tooltip)
button.connect("clicked", lambda *_args, dest=url: self._open_web_url(dest))
return button
def _build_about_section(self: Any) -> None:
"""Build the About page using the same PreferenceRow cards as other pages."""
about_tab = getattr(self, "about_tab")
config_manager = getattr(self, "config_manager")
on_check_updates_clicked = getattr(self, "_on_check_updates_clicked")
on_update_channel_changed = getattr(self, "_on_update_channel_changed")
on_open_release_clicked = getattr(self, "_on_open_release_clicked")
from gi.repository import GdkPixbuf
from ..utils.resource_manager import ResourceManager
from .settings_dialog import (
_CONTROL_WIDTH,
DEFAULT_UPDATE_CHANNEL,
PreferenceRow,
PreferencesGroup,
_combo_chrome,
_combo_with_suffix,
_set_accessible_name,
_style_action_button,
)
dark = _about_surface_is_dark()
about_icon = None
logo_path = ResourceManager().get_icon_path("vocalinux")
if os.path.exists(logo_path):
try:
pixbuf = GdkPixbuf.Pixbuf.new_from_file(logo_path)
scaled = pixbuf.scale_simple(48, 48, GdkPixbuf.InterpType.BILINEAR)
about_icon = Gtk.Image.new_from_pixbuf(scaled)
about_icon.set_pixel_size(48)
except Exception as exc:
logger.warning("Failed to load About icon: %s", exc)
app_group = PreferencesGroup(
title="Vocalinux",
description=(
"Available now on Linux (X11 and Wayland). After a model is downloaded, "
"speech is processed on this PC."
),
keywords=("about", "version", "app"),
header_icon=about_icon,
)
self.about_version_label = Gtk.Label(label=__version__)
self.about_version_label.set_selectable(True)
self.about_version_label.get_style_context().add_class("preference-row-subtitle")
version_row = PreferenceRow(
title="Version",
subtitle="Currently installed Vocalinux build",
widget=self.about_version_label,
keywords=("version", "build"),
)
app_group.add_row(version_row)
link_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
link_box.pack_start(
self._about_open_button(
VOCALINUX_SITE_URL,
"Open vocalinux.com website",
label="Website",
width=110,
dark=dark,
),
False,
False,
0,
)
link_box.pack_start(
self._about_open_button(
GITHUB_REPO_URL,
"Open the Vocalinux GitHub repository",
label="Source code",
width=120,
dark=dark,
),
False,
False,
0,
)
links_row = PreferenceRow(
title="Open source",
subtitle="Product site at vocalinux.com, source on GitHub.",
widget=link_box,
keywords=("website", "vocalinux.com", "github", "source", "repo", "code", "agpl"),
)
app_group.add_row(links_row)
license_row = PreferenceRow(
title="License",
subtitle=f"GNU AGPL v3 · {__copyright__}",
keywords=("license", "agpl", "gpl", "copyright"),
)
app_group.add_row(license_row)
about_tab.pack_start(app_group, False, False, 0)
talk_group = PreferencesGroup(
title="Talk to us",
description=(
"Bugs, feedback, and feature ideas open a GitHub issue. You pick the "
"template on the next screen."
),
keywords=("github", "discord", "contact", "bug", "feedback", "email"),
)
report_btn = self._about_open_button(
GITHUB_ISSUES_URL,
"Open GitHub issues for a bug or idea",
label="Report a bug or idea",
icon_name="github",
width=200,
dark=dark,
)
talk_group.add_row(
PreferenceRow(
title="GitHub issues",
subtitle="https://github.com/VocaHQ/vocalinux/issues",
widget=report_btn,
keywords=("github", "issue", "bug", "idea"),
)
)
contact_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
for url, label, tooltip, icon_name, width in (
(VOCAHQ_DISCORD_URL, "Join Discord", "Open the VocaHQ Discord", "discord", 140),
(VOCAHQ_X_URL, "Follow on X", "Open X @vocahq", "x", 130),
(VOCAHQ_MAILTO_URL, "Email us", "Email hello@vocahq.com", "mail", 120),
):
contact_box.pack_start(
self._about_open_button(
url,
tooltip,
label=label,
icon_name=icon_name,
width=width,
dark=dark,
),
False,
False,
0,
)
talk_group.add_row(
PreferenceRow(
title="Community",
subtitle="Discord, X, and email",
widget=contact_box,
keywords=("discord", "x", "twitter", "email", "mail"),
)
)
about_tab.pack_start(talk_group, False, False, 0)
updates_group = PreferencesGroup(
title="Updates",
description="Stable for numbered releases. Nightly for dated development builds.",
keywords=("update", "version", "release", "upgrade", "appimage", "nightly", "channel"),
)
self.update_channel_combo = Gtk.ComboBoxText()
self.update_channel_combo.append("stable", "Stable")
self.update_channel_combo.append("nightly", "Nightly")
_combo_chrome(
self.update_channel_combo,
tooltip=(
"Stable uses the latest numbered release. Nightly uses the newest "
"nightly-YYYY-MM-DD build."
),
)
saved_channel = normalize_channel(
config_manager.get_str("updates", "channel", DEFAULT_UPDATE_CHANNEL)
)
self.update_channel_combo.set_active_id(saved_channel)
self.check_updates_btn = Gtk.Button(label="Check")
self.check_updates_btn.get_style_context().add_class("suffix-button")
_style_action_button(self.check_updates_btn, width=80)
self.check_updates_btn.set_tooltip_text(
"Look up the latest release for this channel on GitHub"
)
self.check_updates_btn.connect("clicked", on_check_updates_clicked)
self.update_status_row = PreferenceRow(
title="Channel",
subtitle="Not checked yet",
widget=_combo_with_suffix(
self.update_channel_combo,
self.check_updates_btn,
combo_width=_CONTROL_WIDTH - 88,
),
keywords=("channel", "stable", "nightly", "beta", "status", "check"),
)
updates_group.add_row(self.update_status_row)
about_tab.pack_start(updates_group, False, False, 0)
self.update_channel_combo.connect("changed", on_update_channel_changed)
self.release_notes_group = PreferencesGroup(
title="What's New",
keywords=("changelog", "what's new", "release notes", "latest", "download"),
)
self.open_release_btn = Gtk.Button(label="Open")
_style_action_button(self.open_release_btn)
self.open_release_btn.set_sensitive(False)
self.open_release_btn.set_tooltip_text(
"Open the release page in your browser for download links and install steps"
)
_set_accessible_name(
self.open_release_btn,
"Open the latest release page",
)
self.open_release_btn.connect("clicked", on_open_release_clicked)
self.release_notes_row = PreferenceRow(
title="Release notes",
subtitle="Notes appear here after a successful update check.",
widget=self.open_release_btn,
keywords=("notes", "changelog", "latest"),
)
self.latest_release_row = self.release_notes_row
if self.release_notes_row.subtitle_label is not None:
self.release_notes_row.subtitle_label.set_max_width_chars(72)
self.release_notes_group.add_row(self.release_notes_row)
about_tab.pack_start(self.release_notes_group, False, False, 0)
self.release_notes_group.hide()
family_group = PreferencesGroup(
title="Part of VocaHQ",
description="Private dictation for Linux, Mac, Windows, and phone.",
keywords=("vocahq", "vocamac", "vocawin", "vocaphone", "vocagateway", "family"),
)
family_grid = Gtk.Grid()
family_grid.set_column_spacing(8)
family_grid.set_row_spacing(4)
family_grid.set_column_homogeneous(True)
family_grid.set_margin_start(8)
family_grid.set_margin_end(8)
family_grid.set_margin_top(4)
family_grid.set_margin_bottom(8)
for index, (url, title, subtitle, icon_names, open_name) in enumerate(_VOCAHQ_FAMILY_LINKS):
tile = self._family_tile(url, title, subtitle, icon_names, open_name, dark=dark)
family_grid.attach(tile, index % 2, index // 2, 1, 1)
family_row = Gtk.ListBoxRow()
family_row.set_activatable(False)
family_row.add(family_grid)
family_group.add_row(family_row)
about_tab.pack_start(family_group, False, False, 0)
src/vocalinux/ui/settings_dialog.py
+120 -596
View File
File diff suppressed because it is too large Load Diff
src/vocalinux/ui/settings_engines.py
+50
View File
@@ -0,0 +1,50 @@
"""Speech-engine metadata used by the Settings dialog."""
from ..speech_recognition.engines import available_engines
from ..utils import parakeet_model_info as parakeet
from ..utils.faster_whisper_model_info import FASTER_WHISPER_MODEL_INFO
from ..utils.vosk_model_info import SUPPORTED_LANGUAGES, VOSK_MODEL_INFO
from ..utils.whispercpp_model_info import MODEL_SIZES as WHISPERCPP_MODEL_SIZES
from ..utils.whispercpp_model_info import (
NON_DICTATION_MODELS,
WHISPERCPP_MODEL_INFO,
default_variant_for_size,
)
WHISPER_MODEL_INFO = {
"tiny": {"size_mb": 75, "desc": "Fastest, lowest accuracy", "params": "39M"},
"base": {"size_mb": 142, "desc": "Fast, good for basic use", "params": "74M"},
"small": {"size_mb": 466, "desc": "Balanced speed/accuracy", "params": "244M"},
"medium": {"size_mb": 1500, "desc": "High accuracy, slower", "params": "769M"},
"large": {"size_mb": 2900, "desc": "Highest accuracy, slowest", "params": "1550M"},
}
def get_available_engines() -> dict[str, bool]:
"""Preserve the legacy engine-to-availability mapping."""
available = set(available_engines())
return {
name: name in available
for name in (
"vosk",
"whisper",
"whisper_cpp",
"parakeet",
"faster_whisper",
"remote_api",
)
}
__all__ = [
"FASTER_WHISPER_MODEL_INFO",
"NON_DICTATION_MODELS",
"SUPPORTED_LANGUAGES",
"VOSK_MODEL_INFO",
"WHISPERCPP_MODEL_INFO",
"WHISPERCPP_MODEL_SIZES",
"WHISPER_MODEL_INFO",
"default_variant_for_size",
"get_available_engines",
"parakeet",
]
src/vocalinux/ui/transcript_dialog.py
+1 -1
View File
@@ -272,7 +272,7 @@ class TranscriptDialog(Gtk.Dialog):
dialog.format_secondary_text("Do you want to replace it?")
dialog.add_button("_Cancel", Gtk.ResponseType.CANCEL)
dialog.add_button("_Replace", Gtk.ResponseType.OK)
confirmed = dialog.run() == Gtk.ResponseType.OK
confirmed = bool(dialog.run() == Gtk.ResponseType.OK)
dialog.destroy()
return confirmed
src/vocalinux/utils/resource_manager.py
+3 -3
View File
@@ -192,7 +192,7 @@ class ResourceManager:
Dictionary with validation results
"""
resources_dir = self.resources_dir
results = {
results: dict[str, object] = {
"resources_dir_exists": os.path.exists(resources_dir),
"icons_dir_exists": os.path.exists(self.icons_dir),
"sounds_dir_exists": os.path.exists(self.sounds_dir),
@@ -211,7 +211,7 @@ class ResourceManager:
for icon in expected_icons:
icon_path = self.get_icon_path(icon)
if not os.path.exists(icon_path):
results["missing_icons"].append(icon)
cast(list[str], results["missing_icons"]).append(icon)
# Expected sounds
expected_sounds = ["start_recording", "stop_recording", "error"]
@@ -219,6 +219,6 @@ class ResourceManager:
for sound in expected_sounds:
sound_path = self.get_sound_path(sound)
if not os.path.exists(sound_path):
results["missing_sounds"].append(sound)
cast(list[str], results["missing_sounds"]).append(sound)
return results
src/vocalinux/utils/system_language.py
+16 -12
View File
@@ -16,7 +16,7 @@ import logging
import os
import re
import subprocess
from typing import Optional
from typing import Any, Optional, cast
from .host_process import host_env
@@ -161,22 +161,24 @@ def detect_keyboard_layout() -> Optional[str]:
return chosen.split(",")[0].strip() or None
def detect_locale_language(environ: Optional[dict] = None) -> Optional[str]:
def detect_locale_language(environ: Optional[dict[str, str]] = None) -> Optional[str]:
"""Return the first locale value the environment offers, or None.
LANGUAGE is returned intact (a colon-separated preference list). Mapping
that list onto a supported catalogue entry happens in
``_language_from_locale_env``.
"""
environ = os.environ if environ is None else environ
environment = cast(dict[str, str], os.environ if environ is None else environ)
for name in _LOCALE_ENV_VARS:
value = environ.get(name)
value = environment.get(name)
if value:
return value
return None
def _language_from_locale_env(environ: dict, supported: set[str] | dict) -> Optional[str]:
def _language_from_locale_env(
environ: dict[str, str], supported: set[str] | dict[str, dict[str, str]]
) -> Optional[str]:
"""Map locale env vars onto a catalogue entry, first supported win.
LANGUAGE is a colon-separated preference list: each entry is tried through
@@ -199,7 +201,7 @@ def _language_from_locale_env(environ: dict, supported: set[str] | dict) -> Opti
def detect_system_language(
supported: set[str] | dict, environ: Optional[dict] = None
supported: set[str] | dict[str, dict[str, str]], environ: Optional[dict[str, str]] = None
) -> Optional[str]:
"""Return the language id to start from, or None when nothing is decisive.
@@ -215,8 +217,8 @@ def detect_system_language(
logger.info(f"Language {layout_language} taken from keyboard layout {layout!r}")
return layout_language
environ = os.environ if environ is None else environ
locale_language = _language_from_locale_env(environ, supported)
environment = cast(dict[str, str], os.environ if environ is None else environ)
locale_language = _language_from_locale_env(environment, supported)
if locale_language:
logger.info(f"Language {locale_language} taken from locale environment")
return locale_language
@@ -255,7 +257,7 @@ def _gsettings_get(key: str) -> Optional[str]:
return result.stdout.strip() or None
def _parse_sources(value: Optional[str]) -> list:
def _parse_sources(value: Optional[str]) -> list[tuple[str, str]]:
"""Parse a gsettings input-source list, tolerating the ``@a(ss)`` prefix."""
text = (value or "").strip()
if text.startswith("@a(ss) "):
@@ -265,7 +267,9 @@ def _parse_sources(value: Optional[str]) -> list:
except (ValueError, SyntaxError):
logger.debug(f"Could not parse an input-source list: {value!r}")
return []
return list(parsed) if isinstance(parsed, (list, tuple)) else []
if not isinstance(parsed, (list, tuple)):
return []
return cast(list[tuple[str, str]], list(parsed))
def _xkb_layout_from_source(source: object) -> Optional[str]:
@@ -283,7 +287,7 @@ def _xkb_layout_from_source(source: object) -> Optional[str]:
return source_id.split("+", 1)[0].strip() or None
def _source_at_index(parsed: list, current: str) -> Optional[object]:
def _source_at_index(parsed: list[tuple[str, str]], current: str) -> Optional[tuple[str, str]]:
"""Return the ``(type, id)`` pair the ``current`` index points at."""
if not parsed:
return None
@@ -307,7 +311,7 @@ def _parse_active_source(sources: str, current: str) -> Optional[str]:
return _xkb_layout_from_source(_source_at_index(_parse_sources(sources), current))
def _active_input_source() -> Optional[object]:
def _active_input_source() -> Optional[tuple[str, str]]:
"""Return GNOME's active ``(type, id)`` input source, or None if unreadable.
None means "no answer available" -- not GNOME, or the keys are missing. An
src/vocalinux/utils/whispercpp_model_info.py
+7 -7
View File
@@ -11,7 +11,7 @@ import os
import re
import subprocess
from functools import lru_cache
from typing import Optional
from typing import Optional, cast
from .host_process import host_env
from .model_checksums import whispercpp_revision
@@ -352,8 +352,8 @@ def _prefer_discrete_vulkan_device() -> Optional[int]:
devices = _hardware_vulkan_devices()
for device in devices:
if device["device_type"] == "discrete":
return device["index"]
return devices[0]["index"] if devices else None
return cast(int, device["index"])
return cast(int, devices[0]["index"]) if devices else None
def _vulkan_device_name_by_index(devices: list[dict], device_index: Optional[int]) -> Optional[str]:
@@ -361,11 +361,11 @@ def _vulkan_device_name_by_index(devices: list[dict], device_index: Optional[int
if not devices:
return None
if device_index is None:
return devices[0]["name"]
return cast(str, devices[0]["name"])
for device in devices:
if device["index"] == device_index:
return device["name"]
return devices[0]["name"]
return cast(str, device["name"])
return cast(str, devices[0]["name"])
@lru_cache(maxsize=1)
@@ -546,7 +546,7 @@ def _model_file_path(model_name: str) -> str:
model_info = WHISPERCPP_MODEL_INFO.get(model_name)
if model_info and model_info.get("url"):
return os.path.join(whispercpp_dir, os.path.basename(model_info["url"]))
return os.path.join(whispercpp_dir, os.path.basename(cast(str, model_info["url"])))
return os.path.join(whispercpp_dir, f"ggml-{model_name}.bin")