add examples/whisper.linux

This commit is contained in:
nnnet 2026-02-08 17:25:49 +03:00
parent 4b23ff249e
commit d92c33bc51
17 changed files with 5379 additions and 0 deletions

3
examples/whisper.linux/.gitignore vendored Normal file
View File

@ -0,0 +1,3 @@
__pycache__/
*.pyc
.pytest_cache/

View File

@ -0,0 +1,154 @@
# whisper.linux
Voice typing for Linux desktop using [whisper.cpp](https://github.com/ggerganov/whisper.cpp).
Transcribes speech from microphone and injects text at the cursor position.
Supports X11 and Wayland, hotkey and continuous listening modes, voice commands.
## Requirements
- Python 3.10+
- PyQt5: `pip install PyQt5`
- Built `whisper-cli` from whisper.cpp (see root README)
- A GGML model file (e.g. `ggml-base.bin`)
- `xdotool` (X11/XWayland) or `wtype` (Wayland) for text injection
- `xclip` (X11) or `wl-copy` (Wayland) for clipboard fallback
## Quick Start
```bash
cd examples/whisper.linux
# Start
./whisper-linux
# Start with debug logging
./whisper-linux --debug
# Toggle recording on a running instance
./whisper-linux --toggle
# Stop
pkill -f whisper-linux
```
## CLI Options
```
--toggle Send toggle signal to running instance
--language LANG Override language (ru, en, auto)
--model PATH Override model path
--input-mode MODE hotkey or listen
--output-mode MODE batch or stream
--stream Shortcut for --output-mode stream
--wake-word WORD Override wake word (for listen mode)
--wake-model PATH Lighter model for wake word detection
--debug Enable debug logging
```
## Input / Output Modes
Two independent axes control behavior:
| | batch | stream |
|---|---|---|
| **hotkey** | Record all, transcribe, inject at once | Each speech segment transcribed and injected live |
| **listen** | Wake word activates, text accumulated, injected on stop | Wake word activates, each segment injected live |
Default: `hotkey` + `batch` (press hotkey to record, press again to transcribe).
## Voice Commands
When `voice_commands = True` (default), spoken command words trigger key presses
instead of being typed literally. Editable via tray menu: Settings > Edit voice commands.
| Word (EN) | Word (RU) | Action |
|---|---|---|
| enter | энтер, ввод | Press Enter |
| backspace | бэкспейс, назад | Delete previous word |
| tab | таб, табуляция | Press Tab |
| escape, stop | эскейп, стоп | Press Escape |
Commands use fuzzy matching (threshold 0.75), so slight mispronunciations are tolerated.
**Backspace** has special behavior: if there are buffered words not yet injected,
it removes the last word from the buffer. If the buffer is empty, it sends
`Ctrl+BackSpace` to delete the previous word in the editor.
## Keyboard Shortcut
In hotkey mode, you toggle recording via `--toggle`. Set up a global keyboard shortcut
to trigger it from anywhere:
**GNOME** (Settings → Keyboard → Custom Shortcuts → Add):
| Field | Value |
|---|---|
| Name | whisper-linux |
| Command | `/path/to/whisper.linux/whisper-linux --toggle` |
| Shortcut | `Super+V` or any key you prefer |
**Or via command line (GNOME):**
```bash
# Replace /path/to/whisper.linux with the actual path
TOGGLE_CMD="/path/to/whisper.linux/whisper-linux --toggle"
gsettings set org.gnome.settings-daemon.plugins.media-keys custom-keybindings \
"['/org/gnome/settings-daemon/plugins/media-keys/custom-keybindings/whisper-linux/']"
dconf write /org/gnome/settings-daemon/plugins/media-keys/custom-keybindings/whisper-linux/name "'whisper-linux'"
dconf write /org/gnome/settings-daemon/plugins/media-keys/custom-keybindings/whisper-linux/command "'$TOGGLE_CMD'"
dconf write /org/gnome/settings-daemon/plugins/media-keys/custom-keybindings/whisper-linux/binding "'F8'"
```
**KDE:** System Settings → Shortcuts → Custom Shortcuts → Add.
## Configuration
Config file: `~/.config/whisper-linux/config.ini`
All settings are configurable via the system tray menu (right-click the tray icon).
## Autostart
Copy the desktop file to autostart:
```bash
cp whisper-linux.desktop ~/.config/autostart/
```
## Project Structure
```
whisper.linux/
whisper-linux # Launcher script (unique process name)
app/ # Python package
__init__.py # Re-exports public API
__main__.py # Entry point
config.py # Config, AppState, constants, helpers
audio.py # AudioRecorder, AudioStream, SimpleVAD
transcriber.py # Transcriber, WakeWordDetector
injector.py # TextInjector (xdotool/wtype/clipboard)
commands.py # VoiceCommands (Enter, Backspace, etc.)
tray.py # TrayIcon, system tray menu, settings
app.py # WhisperLinuxApp, state machine, CLI
tests/
conftest.py # Fixtures
test_whisper_linux.py # Tests (all mocked, no hardware needed)
run_tests.sh # Run all tests with one command
whisper-linux.desktop # Desktop entry for autostart
```
## Running Tests
```bash
# All tests
./run_tests.sh
# With options
./run_tests.sh --debug -x -k "test_toggle"
# Or directly
python3 -m pytest tests/ -v
```

View File

@ -0,0 +1,21 @@
"""whisper.linux — Voice typing for Linux desktop."""
# Re-export public API for backward compatibility (tests import from whisper_linux)
import os
import subprocess
import threading
import time
from .config import (
Config, AppState, CONFIG_DIR, CONFIG_FILE, PID_FILE,
AVAILABLE_MODELS, DEFAULT_VOICE_COMMANDS, _REPO_ROOT,
_find_executable, _find_model, _list_models, _list_audio_devices,
_WHISPER_CLI_NAMES, _WHISPER_SEARCH_DIRS, _MODEL_SEARCH_DIRS,
LOG_FORMAT, log,
)
from .audio import AudioRecorder, AudioStream, SimpleVAD, _write_wav
from .transcriber import Transcriber, WakeWordDetector, _is_hallucination
from .injector import TextInjector
from .commands import VoiceCommands
from .tray import TrayIcon, _create_icon
from .app import WhisperLinuxApp, send_toggle, main

View File

@ -0,0 +1,4 @@
"""Entry point for `python3 -m whisper_linux`."""
from .app import main
main()

View File

@ -0,0 +1,621 @@
"""Main application logic, state machine, and CLI for whisper.linux."""
import argparse
import logging
import os
import queue
import signal
import subprocess
import sys
import tempfile
import threading
import time
from pathlib import Path
from typing import Optional
from . import config as _cfg
from .config import Config, AppState, log
from .audio import AudioRecorder, AudioStream, SimpleVAD, _write_wav
from .transcriber import Transcriber, WakeWordDetector
from .injector import TextInjector
from .commands import VoiceCommands
from .tray import TrayIcon
class WhisperLinuxApp:
"""Main application: state machine, threading, PID file, signal handling."""
def __init__(self, config: Optional[Config] = None):
self.config = config or Config()
self.state = AppState.IDLE
self.recorder = AudioRecorder(self.config)
self.transcriber = Transcriber(self.config)
self.injector = TextInjector(self.config)
self._tray: Optional[TrayIcon] = None
self._qt_app = None
self._prev_window_id: Optional[str] = None
self._audio_stream: Optional[AudioStream] = None
self._vad: Optional[SimpleVAD] = None
self._wake_detector: Optional[WakeWordDetector] = None
self._silence_timer: Optional[threading.Timer] = None
self._segment_queue: Optional[queue.Queue] = None
self._segment_worker: Optional[threading.Thread] = None
self._accumulated_texts: list = []
self._voice_commands = VoiceCommands(self.config.voice_commands_map)
self._mute_until: float = 0
def _inject_text(self, text: str):
"""Inject text, processing voice commands if enabled."""
if self.config.voice_commands:
self._voice_commands.process(
text, self.injector.inject, self.injector.send_key,
)
else:
self.injector.inject(text)
# -- PID file management --
@staticmethod
def write_pid():
_cfg.PID_FILE.write_text(str(os.getpid()))
@staticmethod
def read_pid() -> Optional[int]:
if _cfg.PID_FILE.exists():
try:
pid = int(_cfg.PID_FILE.read_text().strip())
os.kill(pid, 0)
return pid
except (ValueError, ProcessLookupError, PermissionError):
return None
return None
@staticmethod
def remove_pid():
_cfg.PID_FILE.unlink(missing_ok=True)
# -- Signal handling --
def _setup_signals(self):
signal.signal(signal.SIGUSR1, self._on_sigusr1)
signal.signal(signal.SIGTERM, self._on_sigterm)
signal.signal(signal.SIGINT, self._on_sigterm)
def _on_sigusr1(self, signum, frame):
log.info("Received SIGUSR1 \u2014 toggling")
if self._qt_app:
from PyQt5.QtCore import QTimer
QTimer.singleShot(0, self.toggle)
def _on_sigterm(self, signum, frame):
log.info("Received %s \u2014 quitting", signal.Signals(signum).name)
self.quit()
# -- State machine --
def toggle(self):
log.info("Toggle: state=%s, input=%s, output=%s",
self.state.value, self.config.input_mode, self.config.output_mode)
if self.config.input_mode == "listen":
self._toggle_listen()
else:
self._toggle_hotkey()
def _toggle_hotkey(self):
if self.state == AppState.IDLE:
if self.config.output_mode == "batch":
self._start_recording()
else:
self._start_hotkey_stream()
elif self.state == AppState.RECORDING:
self._stop_recording()
elif self.state == AppState.DICTATING:
self._stop_hotkey_stream()
def _toggle_listen(self):
if self.state == AppState.IDLE:
self._start_listening()
elif self.state in (AppState.LISTENING, AppState.DICTATING):
self._stop_listening()
def _force_idle(self):
if self.state == AppState.RECORDING:
self.recorder.stop()
elif self.state in (AppState.LISTENING, AppState.DICTATING):
if self.config.input_mode == "hotkey" and self.state == AppState.DICTATING:
self._stop_hotkey_stream()
return
self._stop_listening()
self._set_state(AppState.IDLE)
# -- Streaming mode --
def _start_listening(self):
self._save_active_window()
self._wake_detector = WakeWordDetector(self.config.wake_word)
self._vad = SimpleVAD(self.config, on_speech_end=self._on_speech_end,
on_speech_start=self._on_speech_start)
self._audio_stream = AudioStream(self.config)
self._segment_queue = queue.Queue()
self._segment_worker = threading.Thread(
target=self._segment_worker_loop, daemon=True,
)
self._segment_worker.start()
try:
self._audio_stream.start(on_data=self._on_audio_data)
self._set_state(AppState.LISTENING)
log.info("Streaming: LISTENING (waiting for wake word '%s')", self.config.wake_word)
except Exception as e:
log.error("Failed to start audio stream: %s", e)
if self._tray:
self._tray.notify("Error", f"Stream failed: {e}")
if self._segment_queue:
self._segment_queue.put(None)
self._segment_worker = None
self._segment_queue = None
def _stop_listening(self):
self._cancel_silence_timer()
self._flush_accumulated_text()
if self._audio_stream:
self._audio_stream.stop()
self._audio_stream = None
self._vad = None
self._wake_detector = None
if self._segment_queue:
self._segment_queue.put(None)
if self._segment_worker:
self._segment_worker.join(timeout=5)
self._segment_worker = None
self._segment_queue = None
self._set_state(AppState.IDLE)
log.info("Streaming: stopped")
def _start_hotkey_stream(self):
self._save_active_window()
self._wake_detector = None
self._vad = SimpleVAD(self.config, on_speech_end=self._on_speech_end,
on_speech_start=self._on_speech_start)
self._audio_stream = AudioStream(self.config)
self._accumulated_texts.clear()
self._segment_queue = queue.Queue()
self._segment_worker = threading.Thread(
target=self._segment_worker_loop, daemon=True,
)
self._segment_worker.start()
try:
self._audio_stream.start(on_data=self._on_audio_data)
self._set_state(AppState.DICTATING)
self._play_start_signal()
log.info("Hotkey+Stream: DICTATING started")
except Exception as e:
log.error("Failed to start audio stream: %s", e)
if self._tray:
self._tray.notify("Error", f"Stream failed: {e}")
if self._segment_queue:
self._segment_queue.put(None)
self._segment_worker = None
self._segment_queue = None
def _stop_hotkey_stream(self):
self._cancel_silence_timer()
self._flush_accumulated_text()
if self._audio_stream:
self._audio_stream.stop()
self._audio_stream = None
self._vad = None
if self._segment_queue:
self._segment_queue.put(None)
if self._segment_worker:
self._segment_worker.join(timeout=5)
self._segment_worker = None
self._segment_queue = None
self._set_state(AppState.IDLE)
self._play_end_signal()
log.info("Hotkey+Stream: stopped")
def _flush_accumulated_text(self):
if self._accumulated_texts:
all_text = " ".join(self._accumulated_texts)
self._restore_active_window()
self._inject_text(all_text)
self._accumulated_texts.clear()
def _on_audio_data(self, chunk: bytes):
if self._mute_until and time.time() < self._mute_until:
return
if self._vad:
self._vad.feed(chunk)
def _on_speech_start(self):
if self.state == AppState.DICTATING:
self._cancel_silence_timer()
log.debug("Silence timer cancelled (speech started)")
def _on_speech_end(self, pcm_data: bytes):
if self.state == AppState.DICTATING:
self._cancel_silence_timer()
if self._segment_queue:
self._segment_queue.put(pcm_data)
def _segment_worker_loop(self):
while True:
try:
pcm_data = self._segment_queue.get(timeout=1.0)
except queue.Empty:
if not self._segment_queue:
break
continue
if pcm_data is None:
break
self._process_speech_segment(pcm_data)
def _process_speech_segment(self, pcm_data: bytes):
duration_s = len(pcm_data) / (16000 * 2)
fd, wav_path = tempfile.mkstemp(suffix=".wav", prefix="whisper-stream-")
os.close(fd)
try:
_write_wav(wav_path, pcm_data)
model = None
if self.state == AppState.LISTENING and self.config.wake_model:
model = self.config.wake_model
text = self.transcriber.transcribe(wav_path, model=model,
duration_s=duration_s)
if not text:
return
log.info("Stream segment: %r (state=%s)", text[:80], self.state.value)
if self._wake_detector and self._wake_detector.contains_wake_word(text):
remaining = self._wake_detector.strip_wake_word(text)
if self.state == AppState.LISTENING:
self._cancel_silence_timer()
self._accumulated_texts.clear()
self.state = AppState.DICTATING
self._marshal_set_state(AppState.DICTATING)
self._play_start_signal()
log.info("Streaming: DICTATING (wake word detected)")
self._marshal_notify("whisper.linux", "Dictation started")
elif self.state == AppState.DICTATING:
self._cancel_silence_timer()
if remaining:
if self.config.output_mode == "stream":
self._restore_active_window()
self._inject_text(remaining)
else:
self._accumulated_texts.append(remaining)
self._flush_accumulated_text()
self.state = AppState.LISTENING
self._marshal_set_state(AppState.LISTENING)
log.info("Streaming: LISTENING (wake word \u2192 stop dictation)")
self._marshal_notify("whisper.linux", "Dictation paused")
return
if self.state == AppState.DICTATING:
if self.config.output_mode == "stream":
self._restore_active_window()
self._inject_text(text)
else:
self._accumulated_texts.append(text)
if self.config.input_mode == "listen":
self._reset_silence_timer()
log.debug("Silence timer started after segment (%.1fs)",
self.config.silence_timeout)
if self.config.notification:
preview = text[:80] + ("..." if len(text) > 80 else "")
self._marshal_notify("whisper.linux", preview)
except Exception as e:
log.error("Stream segment processing failed: %s", e)
finally:
if os.path.exists(wav_path):
os.unlink(wav_path)
def _marshal_set_state(self, state: AppState):
if self._qt_app:
from PyQt5.QtCore import QTimer
QTimer.singleShot(0, lambda: self._set_state(state))
else:
self._set_state(state)
def _marshal_notify(self, title: str, message: str):
if not self._tray:
return
if self._qt_app:
from PyQt5.QtCore import QTimer
QTimer.singleShot(0, lambda: self._tray.notify(title, message))
else:
self._tray.notify(title, message)
def _reset_silence_timer(self):
self._cancel_silence_timer()
self._silence_timer = threading.Timer(
self.config.silence_timeout,
self._on_silence_timeout,
)
self._silence_timer.daemon = True
self._silence_timer.start()
def _cancel_silence_timer(self):
if self._silence_timer:
self._silence_timer.cancel()
self._silence_timer = None
def _on_silence_timeout(self):
if self.state == AppState.DICTATING:
if self._vad and self._vad._in_speech:
log.debug("Silence timeout skipped (speech active), restarting")
self._reset_silence_timer()
return
log.info("Streaming: silence timeout \u2192 LISTENING")
self._flush_accumulated_text()
self.state = AppState.LISTENING
self._marshal_set_state(AppState.LISTENING)
self._marshal_notify("whisper.linux", "Dictation paused (silence)")
self._play_end_signal()
def _play_sound(self, sounds: tuple):
"""Play the first available sound file."""
players = ("pw-play", "paplay", "canberra-gtk-play")
for sound in sounds:
if not os.path.isfile(sound):
continue
for player in players:
try:
subprocess.Popen(
[player, sound],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
log.debug("Sound played via %s: %s", player, sound)
return
except FileNotFoundError:
continue
log.debug("No sound player available")
def _play_start_signal(self):
if not self.config.end_signal:
return
self._mute_until = time.time() + 0.6
if self._vad:
self._vad.reset()
self._play_sound((
"/usr/share/sounds/freedesktop/stereo/message-new-instant.oga",
"/usr/share/sounds/freedesktop/stereo/message.oga",
"/usr/share/sounds/freedesktop/stereo/bell.oga",
))
def _play_end_signal(self):
if not self.config.end_signal:
return
self._mute_until = time.time() + 0.6
if self._vad:
self._vad.reset()
self._play_sound((
"/usr/share/sounds/freedesktop/stereo/complete.oga",
"/usr/share/sounds/freedesktop/stereo/bell.oga",
"/usr/share/sounds/freedesktop/stereo/message.oga",
))
def _set_state(self, state: AppState):
self.state = state
if self._tray:
self._tray.set_state(state)
def _save_active_window(self):
try:
result = subprocess.run(
["xdotool", "getactivewindow"],
capture_output=True, text=True, timeout=2,
)
if result.returncode == 0 and result.stdout.strip():
self._prev_window_id = result.stdout.strip()
log.debug("Saved active window: %s", self._prev_window_id)
except Exception:
self._prev_window_id = None
def _restore_active_window(self):
if not self._prev_window_id:
return
try:
subprocess.run(
["xdotool", "windowactivate", "--sync", self._prev_window_id],
check=True, timeout=3,
)
time.sleep(0.15)
log.debug("Restored focus to window: %s", self._prev_window_id)
except Exception as e:
log.debug("Could not restore window focus: %s", e)
def _start_recording(self):
self._save_active_window()
try:
self.recorder.start()
self._set_state(AppState.RECORDING)
except Exception as e:
log.error("Failed to start recording: %s", e)
if self._tray:
self._tray.notify("Error", f"Recording failed: {e}")
def _stop_recording(self):
wav_path = self.recorder.stop()
if not wav_path:
self._set_state(AppState.IDLE)
return
self._set_state(AppState.PROCESSING)
t = threading.Thread(target=self._transcribe_and_inject, args=(wav_path,), daemon=True)
t.start()
def _transcribe_and_inject(self, wav_path: str):
try:
text = self.transcriber.transcribe(wav_path)
if text:
self._restore_active_window()
self._inject_text(text)
if self._tray and self.config.notification:
preview = text[:80] + ("..." if len(text) > 80 else "")
self._tray.notify("whisper.linux", preview)
else:
log.info("Empty transcription")
if self._tray:
self._tray.notify("whisper.linux", "(no speech detected)")
except Exception as e:
log.error("Transcription/injection failed: %s", e)
if self._tray:
self._tray.notify("Error", str(e)[:100])
finally:
if os.path.exists(wav_path):
os.unlink(wav_path)
self._set_state(AppState.IDLE)
self._play_end_signal()
def quit(self):
log.info("Shutting down")
self._cancel_silence_timer()
if self._audio_stream:
self._audio_stream.stop()
if self._segment_queue:
self._segment_queue.put(None)
if self._segment_worker:
self._segment_worker.join(timeout=3)
self.recorder.cleanup()
self.remove_pid()
if self._qt_app:
self._qt_app.quit()
# -- Main entry --
def run(self):
from PyQt5.QtWidgets import QApplication
self._qt_app = QApplication(sys.argv)
self._qt_app.setQuitOnLastWindowClosed(False)
self.write_pid()
self._setup_signals()
from PyQt5.QtCore import QTimer
signal_timer = QTimer()
signal_timer.timeout.connect(lambda: None)
signal_timer.start(200)
self._tray = TrayIcon(self)
self._set_state(AppState.IDLE)
log.info("whisper.linux started (pid %d)", os.getpid())
log.info(" whisper-cli : %s", self.config.whisper_cli)
log.info(" model : %s", self.config.model)
log.info(" models_dir : %s", self.config.models_dir)
log.info(" language : %s", self.config.language)
log.info(" threads : %d", self.config.threads)
log.info(" gpu_device : %d", self.config.gpu_device)
log.info(" audio_device: %s", self.config.audio_device)
log.info(" display : %s", self.config.display_server)
log.info(" paste_keys : %s", self.config.paste_keys)
log.info(" clipboard : %s", self.config.use_clipboard_fallback)
log.info(" input_mode : %s", self.config.input_mode)
log.info(" output_mode : %s", self.config.output_mode)
log.info(" wake_word : %s", self.config.wake_word)
log.info(" wake_model : %s", self.config.wake_model or "(same as main)")
log.info(" voice_cmds : %s", self.config.voice_commands)
if self.config.input_mode == "listen":
from PyQt5.QtCore import QTimer as _QT
_QT.singleShot(0, self.toggle)
try:
sys.exit(self._qt_app.exec_())
finally:
self.remove_pid()
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def send_toggle():
pid = WhisperLinuxApp.read_pid()
if pid is None:
print("whisper-linux is not running.", file=sys.stderr)
sys.exit(1)
os.kill(pid, signal.SIGUSR1)
log.info("Sent SIGUSR1 to pid %d", pid)
def main():
parser = argparse.ArgumentParser(
description="whisper.linux \u2014 Voice typing for Linux desktop",
)
parser.add_argument("--toggle", action="store_true",
help="Toggle recording on a running instance (sends SIGUSR1)")
parser.add_argument("--language", "-l", default=None,
help="Override transcription language (e.g. ru, en, auto)")
parser.add_argument("--model", "-m", default=None,
help="Override model path")
parser.add_argument("--input-mode", choices=["hotkey", "listen"], default=None,
help="Override input mode (hotkey or listen)")
parser.add_argument("--output-mode", choices=["batch", "stream"], default=None,
help="Override output mode (batch or stream)")
parser.add_argument("--stream", action="store_true",
help="Shortcut for --output-mode stream")
parser.add_argument("--wake-word", default=None,
help="Override wake word for listen mode")
parser.add_argument("--wake-model", default=None,
help="Override wake model (lighter model for wake word detection)")
parser.add_argument("--debug", action="store_true",
help="Enable debug logging")
args = parser.parse_args()
if args.debug:
logging.getLogger().setLevel(logging.DEBUG)
if args.toggle:
send_toggle()
return
existing_pid = WhisperLinuxApp.read_pid()
if existing_pid is not None:
print(f"whisper-linux is already running (pid {existing_pid}).", file=sys.stderr)
print("Use --toggle to start/stop recording, or kill the existing instance.", file=sys.stderr)
sys.exit(1)
config = Config()
if args.language:
config.language = args.language
if args.model:
config.model = args.model
if args.stream:
config.output_mode = "stream"
if args.input_mode:
config.input_mode = args.input_mode
if args.output_mode:
config.output_mode = args.output_mode
if args.wake_word:
config.wake_word = args.wake_word
if args.wake_model:
config.wake_model = args.wake_model
if not config.model or not os.path.isfile(config.model):
if config.model:
log.warning("Model not found: %s — searching for alternatives", config.model)
from .config import _find_model
fallback = _find_model(config.model_search_dirs)
if fallback:
log.info("Using model: %s", fallback)
config.model = fallback
config.save()
else:
print("Error: No model found. Run install.sh or set model path in config.", file=sys.stderr)
sys.exit(1)
if config.wake_model and not os.path.isfile(config.wake_model):
log.warning("Wake model not found: %s — using main model", config.wake_model)
config.wake_model = ""
app = WhisperLinuxApp(config)
app.run()
if __name__ == "__main__":
main()

View File

@ -0,0 +1,288 @@
"""Audio recording, streaming, and VAD for whisper.linux."""
import array
import math
import os
import signal
import struct
import subprocess
import tempfile
import threading
import time
from typing import Optional
from .config import Config, log
# ---------------------------------------------------------------------------
# AudioRecorder
# ---------------------------------------------------------------------------
class AudioRecorder:
"""Records audio to a temporary WAV file."""
def __init__(self, config: Config):
self._config = config
self._proc: Optional[subprocess.Popen] = None
self._wav_path: Optional[str] = None
@property
def wav_path(self) -> Optional[str]:
return self._wav_path
@property
def is_recording(self) -> bool:
return self._proc is not None and self._proc.poll() is None
def _build_record_cmd(self, path: str) -> list:
audio_device = self._config.audio_device
if audio_device.startswith("pw") or audio_device == "auto":
pw = self._find_pw_record()
if pw:
return [pw, "--format", "s16", "--rate", "16000", "--channels", "1", path]
if audio_device and audio_device not in ("auto", "default"):
return ["arecord", "-D", audio_device, "-f", "S16_LE", "-r", "16000",
"-c", "1", "-t", "wav", path]
pw = self._find_pw_record()
if pw:
return [pw, "--format", "s16", "--rate", "16000", "--channels", "1", path]
return ["arecord", "-f", "S16_LE", "-r", "16000", "-c", "1", "-t", "wav", path]
@staticmethod
def _find_pw_record():
for p in ("/usr/bin/pw-record", "/usr/local/bin/pw-record"):
if os.path.isfile(p) and os.access(p, os.X_OK):
return p
return None
def start(self) -> str:
if self.is_recording:
raise RuntimeError("Already recording")
fd, path = tempfile.mkstemp(suffix=".wav", prefix="whisper-linux-")
os.close(fd)
self._wav_path = path
cmd = self._build_record_cmd(path)
self._proc = subprocess.Popen(
cmd,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
)
time.sleep(0.2)
if self._proc.poll() is not None:
stderr = self._proc.stderr.read().decode(errors="replace").strip()
self._proc = None
os.unlink(path)
self._wav_path = None
raise RuntimeError(f"Recording failed to start ({cmd[0]}): {stderr}")
log.info("Recording started → %s (pid %d, cmd=%s)", path, self._proc.pid, cmd[0])
return path
def stop(self) -> Optional[str]:
if self._proc is None:
return None
if self._proc.poll() is None:
self._proc.send_signal(signal.SIGINT)
try:
self._proc.wait(timeout=5)
except subprocess.TimeoutExpired:
self._proc.kill()
self._proc.wait()
log.info("Recording stopped → %s", self._wav_path)
self._proc = None
return self._wav_path
def cleanup(self):
self.stop()
if self._wav_path and os.path.exists(self._wav_path):
os.unlink(self._wav_path)
self._wav_path = None
# ---------------------------------------------------------------------------
# _write_wav
# ---------------------------------------------------------------------------
def _write_wav(path: str, pcm_data: bytes, sample_rate: int = 16000, channels: int = 1, bits: int = 16):
"""Write raw PCM s16le data to a WAV file with RIFF header."""
data_size = len(pcm_data)
byte_rate = sample_rate * channels * bits // 8
block_align = channels * bits // 8
with open(path, "wb") as f:
f.write(b"RIFF")
f.write(struct.pack("<I", 36 + data_size))
f.write(b"WAVE")
f.write(b"fmt ")
f.write(struct.pack("<I", 16))
f.write(struct.pack("<H", 1))
f.write(struct.pack("<H", channels))
f.write(struct.pack("<I", sample_rate))
f.write(struct.pack("<I", byte_rate))
f.write(struct.pack("<H", block_align))
f.write(struct.pack("<H", bits))
f.write(b"data")
f.write(struct.pack("<I", data_size))
f.write(pcm_data)
# ---------------------------------------------------------------------------
# SimpleVAD
# ---------------------------------------------------------------------------
class SimpleVAD:
"""Energy-based Voice Activity Detection on raw s16le PCM frames."""
FRAME_MS = 30
TRAILING_SILENCE_MS = 300
def __init__(self, config: Config, on_speech_end=None, on_speech_start=None):
self._threshold = config.vad_threshold
self._min_speech_ms = config.min_speech_ms
self._max_speech_s = config.max_speech_s
self._sample_rate = 16000
self._frame_bytes = self._sample_rate * 2 * self.FRAME_MS // 1000
self._silence_frames = self.TRAILING_SILENCE_MS // self.FRAME_MS
self.on_speech_end = on_speech_end
self.on_speech_start = on_speech_start
self._buffer = bytearray()
self._speech_buffer = bytearray()
self._in_speech = False
self._silent_count = 0
def reset(self):
self._buffer.clear()
self._speech_buffer.clear()
self._in_speech = False
self._silent_count = 0
@staticmethod
def _rms(frame_bytes: bytes) -> float:
if len(frame_bytes) < 2:
return 0.0
samples = array.array("h")
samples.frombytes(frame_bytes[:len(frame_bytes) - len(frame_bytes) % 2])
if not samples:
return 0.0
return math.sqrt(sum(s * s for s in samples) / len(samples))
def feed(self, chunk: bytes):
self._buffer.extend(chunk)
while len(self._buffer) >= self._frame_bytes:
frame = bytes(self._buffer[:self._frame_bytes])
del self._buffer[:self._frame_bytes]
rms = self._rms(frame)
if rms >= self._threshold:
if not self._in_speech:
self._in_speech = True
self._silent_count = 0
log.debug("VAD: speech start (RMS=%.0f)", rms)
if self.on_speech_start:
self.on_speech_start()
self._speech_buffer.extend(frame)
self._silent_count = 0
elif self._in_speech:
self._speech_buffer.extend(frame)
self._silent_count += 1
if self._silent_count >= self._silence_frames:
self._emit_speech()
max_bytes = int(self._max_speech_s * self._sample_rate * 2)
if self._in_speech and len(self._speech_buffer) >= max_bytes:
self._emit_speech()
def _emit_speech(self):
pcm = bytes(self._speech_buffer)
duration_ms = len(pcm) * 1000 // (self._sample_rate * 2)
self._speech_buffer.clear()
self._in_speech = False
self._silent_count = 0
if duration_ms >= self._min_speech_ms:
log.debug("VAD: speech end (%dms)", duration_ms)
if self.on_speech_end:
self.on_speech_end(pcm)
else:
log.debug("VAD: speech too short (%dms), discarded", duration_ms)
# ---------------------------------------------------------------------------
# AudioStream
# ---------------------------------------------------------------------------
class AudioStream:
"""Streams raw PCM from pw-record to a callback in a reader thread."""
SAMPLE_RATE = 16000
BYTES_PER_SEC = 32000
CHUNK_MS = 100
CHUNK_BYTES = BYTES_PER_SEC * CHUNK_MS // 1000
def __init__(self, config: Config):
self._config = config
self._proc: Optional[subprocess.Popen] = None
self._thread: Optional[threading.Thread] = None
self._on_data = None
self._running = False
@property
def is_running(self) -> bool:
return self._running and self._proc is not None and self._proc.poll() is None
def _build_stream_cmd(self) -> list:
audio_device = self._config.audio_device
pw = AudioRecorder._find_pw_record()
if pw:
return [pw, "--format", "s16", "--rate", "16000", "--channels", "1", "-"]
cmd = ["arecord", "-f", "S16_LE", "-r", "16000", "-c", "1", "-t", "raw"]
if audio_device and audio_device not in ("auto", "default"):
cmd.extend(["-D", audio_device])
return cmd
def start(self, on_data):
if self._running:
raise RuntimeError("AudioStream already running")
self._on_data = on_data
self._running = True
cmd = self._build_stream_cmd()
self._proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
time.sleep(0.2)
if self._proc.poll() is not None:
stderr = self._proc.stderr.read().decode(errors="replace").strip()
self._running = False
self._proc = None
raise RuntimeError(f"AudioStream failed to start ({cmd[0]}): {stderr}")
self._thread = threading.Thread(target=self._reader_loop, daemon=True)
self._thread.start()
log.info("AudioStream started (pid %d, cmd=%s)", self._proc.pid, cmd[0])
def stop(self):
self._running = False
if self._proc and self._proc.poll() is None:
self._proc.send_signal(signal.SIGINT)
try:
self._proc.wait(timeout=3)
except subprocess.TimeoutExpired:
self._proc.kill()
self._proc.wait()
if self._thread:
self._thread.join(timeout=3)
self._thread = None
self._proc = None
log.info("AudioStream stopped")
def _reader_loop(self):
try:
while self._running and self._proc and self._proc.poll() is None:
data = self._proc.stdout.read(self.CHUNK_BYTES)
if not data:
break
if self._on_data and self._running:
self._on_data(data)
except Exception as e:
log.error("AudioStream reader error: %s", e)
finally:
self._running = False

View File

@ -0,0 +1,81 @@
"""Voice command processing for whisper.linux.
Detects command words in transcribed text (e.g. "Enter", "Backspace")
and executes corresponding key presses instead of typing them literally.
"""
import difflib
from typing import Callable, Optional
from .config import DEFAULT_VOICE_COMMANDS, log
class VoiceCommands:
"""Processes transcribed text for voice commands (Enter, Backspace, etc.)."""
DEFAULT_COMMANDS = DEFAULT_VOICE_COMMANDS
FUZZY_THRESHOLD = 0.75
def __init__(self, commands: dict = None):
self._commands = commands if commands is not None else dict(self.DEFAULT_COMMANDS)
def process(self, text: str, inject_fn: Callable, send_key_fn: Callable) -> bool:
"""Process text, calling inject_fn for text and send_key_fn for key presses.
For "backspace", removes the previous word from the buffer. If the buffer
is empty, sends Ctrl+BackSpace to delete the word in the editor.
Returns True if any commands were found.
"""
words = text.split()
if not words:
return False
buffer = []
had_commands = False
for word in words:
clean = word.strip(".,!?;:-\"'()[]").lower()
action = self._match_command(clean)
if action:
had_commands = True
if action == "backspace":
if buffer:
removed = buffer.pop()
log.info("Voice command: backspace (removed '%s' from buffer)", removed)
else:
send_key_fn("ctrl+BackSpace")
log.info("Voice command: backspace (Ctrl+BackSpace sent)")
else:
# Flush text buffer first, then send key
if buffer:
inject_fn(" ".join(buffer))
buffer.clear()
key = action[4:] # strip "key:" prefix
send_key_fn(key)
log.info("Voice command: %s -> %s", clean, key)
else:
buffer.append(word)
# Flush remaining text
if buffer:
inject_fn(" ".join(buffer))
return had_commands
def _match_command(self, word: str) -> Optional[str]:
"""Match a word to a command (exact, then fuzzy)."""
if not word:
return None
# Exact match
if word in self._commands:
return self._commands[word]
# Fuzzy match — pick best above threshold
best_ratio = 0.0
best_action = None
for cmd_word, action in self._commands.items():
ratio = difflib.SequenceMatcher(None, cmd_word, word).ratio()
if ratio >= self.FUZZY_THRESHOLD and ratio > best_ratio:
best_ratio = ratio
best_action = action
return best_action

View File

@ -0,0 +1,355 @@
"""Configuration, constants, and helpers for whisper.linux."""
import configparser
import enum
import logging
import os
import re
import subprocess
from dataclasses import dataclass
from pathlib import Path
LOG_FORMAT = "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
log = logging.getLogger("whisper-linux")
# ---------------------------------------------------------------------------
# AppState
# ---------------------------------------------------------------------------
class AppState(enum.Enum):
IDLE = "idle"
RECORDING = "recording"
PROCESSING = "processing"
LISTENING = "listening" # stream: waiting for wake word
DICTATING = "dictating" # stream: actively typing text
# ---------------------------------------------------------------------------
# Paths and constants
# ---------------------------------------------------------------------------
_REPO_ROOT = Path(__file__).resolve().parent.parent.parent.parent
CONFIG_DIR = Path.home() / ".config" / "whisper-linux"
CONFIG_FILE = CONFIG_DIR / "config.ini"
PID_FILE = Path("/tmp") / "whisper-linux.pid"
_WHISPER_CLI_NAMES = ["whisper-cli", "main"]
_WHISPER_SEARCH_DIRS = [
_REPO_ROOT / "build" / "bin",
_REPO_ROOT / "build",
Path.home() / ".local" / "bin",
Path("/usr/local/bin"),
Path("/usr/bin"),
]
_MODEL_SEARCH_DIRS = [
_REPO_ROOT / "models",
Path.home() / ".local" / "share" / "whisper-linux" / "models",
]
DEFAULT_VOICE_COMMANDS = {
"enter": "key:Return",
"энтер": "key:Return",
"ввод": "key:Return",
"backspace": "backspace",
"бэкспейс": "backspace",
"бекспейс": "backspace",
"назад": "backspace",
"tab": "key:Tab",
"таб": "key:Tab",
"табуляция": "key:Tab",
"escape": "key:Escape",
"эскейп": "key:Escape",
"стоп": "key:Escape",
}
AVAILABLE_MODELS = [
("tiny", "~75 MB"),
("base", "~142 MB"),
("small", "~466 MB"),
("medium", "~1.5 GB"),
("large-v3", "~3.1 GB"),
]
# ---------------------------------------------------------------------------
# Helper functions
# ---------------------------------------------------------------------------
def _find_executable(names, search_dirs):
"""Find first matching executable in search dirs, then PATH."""
for d in search_dirs:
for name in names:
p = d / name
if p.is_file() and os.access(p, os.X_OK):
return str(p)
for name in names:
for pdir in os.environ.get("PATH", "").split(os.pathsep):
p = Path(pdir) / name
if p.is_file() and os.access(p, os.X_OK):
return str(p)
return None
def _find_model(search_dirs, preferred="ggml-base.bin"):
"""Find first .bin model file."""
for d in search_dirs:
p = d / preferred
if p.is_file():
return str(p)
for d in search_dirs:
if d.is_dir():
for f in sorted(d.glob("ggml-*.bin")):
return str(f)
return None
def _list_models(search_dirs):
"""Return list of (name, path) for all available ggml models, sorted by size."""
seen = set()
models = []
for d in search_dirs:
if d.is_dir():
for f in sorted(d.glob("ggml-*.bin")):
if f.name not in seen:
seen.add(f.name)
name = f.stem.replace("ggml-", "")
models.append((name, str(f)))
models.sort(key=lambda x: Path(x[1]).stat().st_size)
return models
def _list_audio_devices():
"""Return list of (label, device_id) for audio input devices."""
devices = []
if os.path.isfile("/usr/bin/pw-record"):
devices.append(("Auto (PipeWire)", "auto"))
else:
devices.append(("Auto", "auto"))
try:
result = subprocess.run(
["arecord", "-l"], capture_output=True, text=True, timeout=5,
)
for line in result.stdout.splitlines():
m = re.match(r"card\s+(\d+):\s*(\w+).*device\s+(\d+):\s*(.*?)(?:\[|$)", line)
if m:
card, card_name, dev, dev_name = m.groups()
device_id = f"plughw:{card},{dev}"
label = dev_name.strip() if dev_name.strip() else device_id
devices.append((label, device_id))
except Exception:
pass
return devices
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
@dataclass
class Config:
whisper_cli: str = ""
model: str = ""
models_dir: str = ""
language: str = "ru"
threads: int = 4
gpu_device: int = -1
display_server: str = ""
audio_device: str = ""
paste_keys: str = "shift+Insert"
use_clipboard_fallback: bool = False
notification: bool = True
input_mode: str = "hotkey"
output_mode: str = "batch"
wake_word: str = "марфуша"
wake_model: str = ""
silence_timeout: float = 3.0
vad_threshold: int = 300
min_speech_ms: int = 500
max_speech_s: float = 30.0
end_signal: bool = True
voice_commands: bool = True
def __post_init__(self):
self._gpu_devices = []
self.voice_commands_map: dict = dict(DEFAULT_VOICE_COMMANDS)
self._load()
def _load(self):
"""Load from config.ini, auto-detect missing values."""
cp = configparser.ConfigParser()
if CONFIG_FILE.exists():
cp.read(CONFIG_FILE)
sec = "whisper-linux"
if cp.has_section(sec):
self.whisper_cli = cp.get(sec, "whisper_cli", fallback=self.whisper_cli)
self.model = cp.get(sec, "model", fallback=self.model)
self.models_dir = cp.get(sec, "models_dir", fallback=self.models_dir)
self.language = cp.get(sec, "language", fallback=self.language)
self.threads = cp.getint(sec, "threads", fallback=self.threads)
self.gpu_device = cp.getint(sec, "gpu_device", fallback=self.gpu_device)
self.display_server = cp.get(sec, "display_server", fallback=self.display_server)
self.audio_device = cp.get(sec, "audio_device", fallback=self.audio_device)
self.paste_keys = cp.get(sec, "paste_keys", fallback=self.paste_keys)
self.use_clipboard_fallback = cp.getboolean(
sec, "use_clipboard_fallback", fallback=self.use_clipboard_fallback
)
self.notification = cp.getboolean(sec, "notification", fallback=self.notification)
old_mode = cp.get(sec, "mode", fallback=None)
if old_mode and not cp.has_option(sec, "input_mode"):
if old_mode == "stream":
self.input_mode = "listen"
self.output_mode = "stream"
else:
self.input_mode = "hotkey"
self.output_mode = "batch"
self.input_mode = cp.get(sec, "input_mode", fallback=self.input_mode)
self.output_mode = cp.get(sec, "output_mode", fallback=self.output_mode)
self.wake_word = cp.get(sec, "wake_word", fallback=self.wake_word)
self.wake_model = cp.get(sec, "wake_model", fallback=self.wake_model)
self.silence_timeout = cp.getfloat(sec, "silence_timeout", fallback=self.silence_timeout)
self.vad_threshold = cp.getint(sec, "vad_threshold", fallback=self.vad_threshold)
self.min_speech_ms = cp.getint(sec, "min_speech_ms", fallback=self.min_speech_ms)
self.max_speech_s = cp.getfloat(sec, "max_speech_s", fallback=self.max_speech_s)
self.end_signal = cp.getboolean(sec, "end_signal", fallback=self.end_signal)
self.voice_commands = cp.getboolean(sec, "voice_commands", fallback=self.voice_commands)
vc_sec = "voice-commands"
if cp.has_section(vc_sec):
saved = dict(cp.items(vc_sec))
# Merge: keep saved commands, add new defaults that aren't overridden
merged = dict(DEFAULT_VOICE_COMMANDS)
merged.update(saved)
self.voice_commands_map = merged
if not self.whisper_cli:
self.whisper_cli = _find_executable(_WHISPER_CLI_NAMES, _WHISPER_SEARCH_DIRS) or "whisper-cli"
if not self.models_dir:
self.models_dir = self._detect_models_dir()
if not self.model:
self.model = _find_model(self.model_search_dirs) or ""
if not self.display_server:
self.display_server = self._detect_display_server()
if not self.audio_device:
self.audio_device = self._detect_audio_device()
detected_gpu = self._detect_gpu_device()
if self.gpu_device < 0:
self.gpu_device = detected_gpu
@staticmethod
def _detect_display_server():
xdg = os.environ.get("XDG_SESSION_TYPE", "").lower()
if "wayland" in xdg:
return "wayland"
if "x11" in xdg:
return "x11"
if os.environ.get("WAYLAND_DISPLAY"):
return "wayland"
if os.environ.get("DISPLAY"):
return "x11"
return "x11"
@staticmethod
def _detect_audio_device():
if os.path.isfile("/usr/bin/pw-record"):
log.info("PipeWire detected (pw-record available), using auto audio device")
return "auto"
try:
result = subprocess.run(
["arecord", "-l"], capture_output=True, text=True, timeout=5,
)
for line in result.stdout.splitlines():
m = re.match(r"card\s+(\d+):.*device\s+(\d+):", line)
if m:
card, dev = m.group(1), m.group(2)
device = f"plughw:{card},{dev}"
log.info("Auto-detected ALSA audio device: %s (%s)",
device, line.strip())
return device
except Exception as e:
log.warning("Failed to detect audio device: %s", e)
return "default"
@staticmethod
def _detect_models_dir():
for d in _MODEL_SEARCH_DIRS:
if d.is_dir() and any(d.glob("ggml-*.bin")):
return str(d)
return str(_MODEL_SEARCH_DIRS[0])
@property
def model_search_dirs(self):
dirs = []
if self.models_dir:
dirs.append(Path(self.models_dir))
for d in _MODEL_SEARCH_DIRS:
if d not in dirs:
dirs.append(d)
return dirs
def _detect_gpu_device(self):
cli = self.whisper_cli or _find_executable(_WHISPER_CLI_NAMES, _WHISPER_SEARCH_DIRS)
model = self.model or _find_model(self.model_search_dirs)
if not cli or not model:
return 0
try:
result = subprocess.run(
[cli, "-m", model, "-f", "/dev/null"],
capture_output=True, text=True, timeout=10,
)
devices = {}
for line in result.stderr.splitlines():
m = re.match(r"ggml_vulkan:\s+(\d+)\s*=\s*(.+?)(?:\s*\|.*)?$", line)
if m:
idx, name = int(m.group(1)), m.group(2).strip()
devices[idx] = name
log.info("GPU device %d: %s", idx, name)
self._gpu_devices = [(name, idx) for idx, name in sorted(devices.items())]
for idx, name in devices.items():
if "NVIDIA" in name.upper():
log.info("Auto-selected NVIDIA GPU device: %d", idx)
return idx
for idx, name in devices.items():
if "AMD" in name.upper() or "RADEON" in name.upper():
log.info("Auto-selected AMD GPU device: %d", idx)
return idx
except Exception as e:
log.debug("GPU device detection failed: %s", e)
return 0
def save(self):
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
cp = configparser.ConfigParser()
sec = "whisper-linux"
cp.add_section(sec)
cp.set(sec, "whisper_cli", self.whisper_cli)
cp.set(sec, "model", self.model)
cp.set(sec, "models_dir", self.models_dir)
cp.set(sec, "language", self.language)
cp.set(sec, "threads", str(self.threads))
cp.set(sec, "gpu_device", str(self.gpu_device))
cp.set(sec, "display_server", self.display_server)
cp.set(sec, "audio_device", self.audio_device)
cp.set(sec, "paste_keys", self.paste_keys)
cp.set(sec, "use_clipboard_fallback", str(self.use_clipboard_fallback))
cp.set(sec, "notification", str(self.notification))
cp.set(sec, "input_mode", self.input_mode)
cp.set(sec, "output_mode", self.output_mode)
cp.set(sec, "wake_word", self.wake_word)
cp.set(sec, "wake_model", self.wake_model)
cp.set(sec, "silence_timeout", str(self.silence_timeout))
cp.set(sec, "vad_threshold", str(self.vad_threshold))
cp.set(sec, "min_speech_ms", str(self.min_speech_ms))
cp.set(sec, "max_speech_s", str(self.max_speech_s))
cp.set(sec, "end_signal", str(self.end_signal))
cp.set(sec, "voice_commands", str(self.voice_commands))
vc_sec = "voice-commands"
cp.add_section(vc_sec)
for word, action in self.voice_commands_map.items():
cp.set(vc_sec, word, action)
with open(CONFIG_FILE, "w") as f:
cp.write(f)
log.info("Config saved to %s", CONFIG_FILE)

View File

@ -0,0 +1,158 @@
"""Text injection (xdotool/ydotool/clipboard) for whisper.linux."""
import os
import subprocess
import time
from .config import Config, log
class TextInjector:
"""Injects text at cursor position using xdotool, ydotool, or clipboard."""
# X11 keysym → evdev name (for ydotool which uses evdev names)
_YDOTOOL_KEY_MAP = {
"Return": "Enter",
"Escape": "Esc",
}
def __init__(self, config: Config):
self.config = config
def inject(self, text: str):
if not text:
return
ds = self.config.display_server
if ds == "wayland":
self._inject_wayland(text)
else:
self._inject_x11(text)
def _inject_x11(self, text: str):
if self.config.use_clipboard_fallback or not text.isascii():
self._inject_clipboard_x11(text)
return
try:
subprocess.run(
["xdotool", "type", "--clearmodifiers", "--", text],
check=True, timeout=10,
)
except (subprocess.CalledProcessError, FileNotFoundError):
log.warning("xdotool type failed, falling back to clipboard")
self._inject_clipboard_x11(text)
def _inject_clipboard_x11(self, text: str):
subprocess.run(
["xclip", "-selection", "clipboard"],
input=text.encode("utf-8"), check=True, timeout=5,
)
time.sleep(0.1)
subprocess.run(
["xdotool", "key", "--clearmodifiers", "ctrl+v"],
check=True, timeout=5,
)
def _inject_wayland(self, text: str):
try:
subprocess.run(["wtype", "--", text], check=True, timeout=10)
return
except (subprocess.CalledProcessError, FileNotFoundError):
pass
try:
subprocess.run(
["wl-copy", "--", text], check=True, timeout=5,
)
subprocess.run(
["wl-copy", "--primary", "--", text], check=True, timeout=5,
)
except (subprocess.CalledProcessError, FileNotFoundError):
log.warning("wl-copy failed, falling back to xclip (XWayland only)")
self._inject_x11(text)
return
paste_keys = self.config.paste_keys
log.info("Clipboard set via wl-copy, sending %s", paste_keys)
time.sleep(0.3)
try:
subprocess.run(
["ydotool", "key", "--delay", "100", paste_keys],
check=True, timeout=5, capture_output=True,
)
log.info("Paste sent via ydotool (%s)", paste_keys)
return
except (subprocess.CalledProcessError, FileNotFoundError, RuntimeError) as e:
log.debug("ydotool failed: %s", e)
try:
subprocess.run(
["xdotool", "key", "--clearmodifiers", paste_keys],
check=True, timeout=5,
)
log.info("Paste sent via xdotool (%s)", paste_keys)
return
except (subprocess.CalledProcessError, FileNotFoundError):
pass
log.warning("Text copied to clipboard (Ctrl+V to paste). "
"For auto-paste: sudo chmod 0660 /dev/uinput && "
"sudo chown root:$USER /dev/uinput")
def send_key(self, key: str):
"""Send a key press (e.g. 'Return', 'Tab', 'ctrl+BackSpace')."""
ds = self.config.display_server
if ds == "wayland":
self._send_key_wayland(key)
else:
self._send_key_x11(key)
def _send_key_x11(self, key: str):
try:
subprocess.run(
["xdotool", "key", "--clearmodifiers", key],
check=True, timeout=5,
)
except (subprocess.CalledProcessError, FileNotFoundError) as e:
log.warning("xdotool key '%s' failed: %s", key, e)
def _send_key_wayland(self, key: str):
# wtype supports XKB key names natively on Wayland
try:
if "+" in key:
parts = key.split("+")
cmd = ["wtype"]
for mod in parts[:-1]:
cmd.extend(["-M", mod.lower()])
cmd.extend(["-k", parts[-1]])
else:
cmd = ["wtype", "-k", key]
subprocess.run(cmd, check=True, timeout=5)
return
except (subprocess.CalledProcessError, FileNotFoundError):
pass
# ydotool uses evdev key names (Enter, not Return; Esc, not Escape)
try:
if "+" in key:
parts = key.split("+")
evdev_parts = [self._YDOTOOL_KEY_MAP.get(p, p) for p in parts]
evdev_key = "+".join(evdev_parts)
else:
evdev_key = self._YDOTOOL_KEY_MAP.get(key, key)
subprocess.run(
["ydotool", "key", "--delay", "50", evdev_key],
check=True, timeout=5, capture_output=True,
)
return
except (subprocess.CalledProcessError, FileNotFoundError):
pass
# xdotool via XWayland
try:
subprocess.run(
["xdotool", "key", "--clearmodifiers", key],
check=True, timeout=5,
)
return
except (subprocess.CalledProcessError, FileNotFoundError):
pass
log.warning("Failed to send key '%s' — no working key sender", key)

View File

@ -0,0 +1,182 @@
"""Transcriber and WakeWordDetector for whisper.linux."""
import difflib
import os
import re
import subprocess
from .config import Config, log
# ---------------------------------------------------------------------------
# Hallucination filter — whisper generates these from training data on silence
# ---------------------------------------------------------------------------
# Known hallucination phrases (training data leaks)
_HALLUCINATION_PATTERNS = [
r"субтитр",
r"редактор\s+субтитр",
r"перевод\w*\s+субтитр",
r"подредактир",
r"продолжение\s+следует",
r"спасибо\s+за\s+(просмотр|подписк|внимание)",
r"подписывайтесь",
r"не\s+забудьте\s+подписаться",
r"ставьте\s+лайк",
r"с\s+вами\s+был[аио]?\s",
r"а[пп]+ортизатор",
r"добро\s+пожаловать",
r"до\s+новых\s+встреч",
r"до\s+свидания.*друзья",
r"subtitles?\s+(by|made|edited|created)",
r"thanks?\s+for\s+watching",
r"subscribe\s+(to|and)",
r"please\s+(like|subscribe)",
r"don'?t\s+forget\s+to\s+subscribe",
r"copyright\s+\d{4}",
]
_HALLUCINATION_RE = re.compile(
"|".join(_HALLUCINATION_PATTERNS), re.IGNORECASE,
)
# Speech rate limits — normal speech is ~2-3 words/sec, ~12-15 chars/sec.
# Hallucinations on short audio produce way more text than physically possible.
_MAX_WORDS_PER_SEC = 5 # generous upper bound
_MAX_CHARS_PER_SEC = 25 # generous upper bound
def _is_hallucination(text: str, duration_s: float = 0) -> bool:
"""Return True if text looks like a whisper hallucination.
Uses two layers:
1. Pattern matching known hallucination phrases.
2. Speech rate check if audio duration is known, rejects text that
is impossibly long for the given duration (e.g. 5 words from 0.5s).
"""
# Layer 1: known patterns
if _HALLUCINATION_RE.search(text):
return True
# Layer 2: speech rate sanity check (only when duration is known)
if duration_s > 0:
words = text.split()
word_count = len(words)
char_count = len(text.replace(" ", ""))
max_words = max(2, duration_s * _MAX_WORDS_PER_SEC)
max_chars = max(10, duration_s * _MAX_CHARS_PER_SEC)
if word_count > max_words:
log.debug("Hallucination (words): %d words in %.1fs (max %.0f)",
word_count, duration_s, max_words)
return True
if char_count > max_chars:
log.debug("Hallucination (chars): %d chars in %.1fs (max %.0f)",
char_count, duration_s, max_chars)
return True
return False
# ---------------------------------------------------------------------------
# WakeWordDetector
# ---------------------------------------------------------------------------
class WakeWordDetector:
"""Checks transcription text for the wake word using exact + fuzzy matching."""
FUZZY_THRESHOLD = 0.7
def __init__(self, wake_word: str):
self._wake_word = wake_word.lower().strip()
self._variants = self._build_variants(self._wake_word)
@staticmethod
def _build_variants(word):
variants = {word}
for suffix in ("а", "я", "ша", "жа"):
if word.endswith(suffix) and len(word) > len(suffix) + 2:
variants.add(word[:-len(suffix)])
return variants
def contains_wake_word(self, text: str) -> bool:
text_lower = text.lower().strip()
if self._wake_word in text_lower:
return True
for word in text_lower.split():
word = word.strip(".,!?;:-\"'()[]")
if word and self._is_fuzzy_match(word):
return True
return False
def strip_wake_word(self, text: str) -> str:
import re
pattern = re.compile(re.escape(self._wake_word), re.IGNORECASE)
if pattern.search(text):
result = pattern.sub("", text)
result = re.sub(r"\s+", " ", result).strip()
return result.strip(".,!?;:- ")
words = text.split()
kept = []
for w in words:
clean = w.strip(".,!?;:-\"'()[]").lower()
if clean and self._is_fuzzy_match(clean):
continue
kept.append(w)
result = " ".join(kept).strip()
return result.strip(".,!?;:- ")
def _is_fuzzy_match(self, word: str) -> bool:
ratio = difflib.SequenceMatcher(None, self._wake_word, word).ratio()
if ratio >= self.FUZZY_THRESHOLD:
return True
for variant in self._variants:
ratio = difflib.SequenceMatcher(None, variant, word).ratio()
if ratio >= self.FUZZY_THRESHOLD:
return True
return False
# ---------------------------------------------------------------------------
# Transcriber
# ---------------------------------------------------------------------------
class Transcriber:
"""Runs whisper-cli and returns transcribed text."""
def __init__(self, config: Config):
self.config = config
def transcribe(self, wav_path: str, model: str = None,
duration_s: float = 0) -> str:
if not os.path.isfile(wav_path):
raise FileNotFoundError(f"WAV file not found: {wav_path}")
cmd = [
self.config.whisper_cli,
"-m", model or self.config.model,
"-f", wav_path,
"-nt",
"-np",
"-t", str(self.config.threads),
"-l", self.config.language,
"-dev", str(self.config.gpu_device),
]
log.info("Transcribing (%.1fs): %s", duration_s, " ".join(cmd))
result = subprocess.run(
cmd, capture_output=True, text=True, timeout=300,
)
if result.returncode != 0:
log.error("whisper-cli stderr: %s", result.stderr)
raise RuntimeError(f"whisper-cli failed (rc={result.returncode}): {result.stderr[:200]}")
text = result.stdout.strip()
text = text.replace("[BLANK_AUDIO]", "").strip()
if text and _is_hallucination(text, duration_s):
log.info("Hallucination filtered (%.1fs): %r", duration_s, text[:100])
return ""
log.info("Transcription: %r", text[:100])
return text

View File

@ -0,0 +1,696 @@
"""System tray icon and menu for whisper.linux."""
import subprocess
import threading
from pathlib import Path
from typing import TYPE_CHECKING
from .config import (
Config, AppState, AVAILABLE_MODELS, DEFAULT_VOICE_COMMANDS, _REPO_ROOT,
_list_models, _list_audio_devices, log,
)
if TYPE_CHECKING:
from .app import WhisperLinuxApp
def _create_icon(color, size=64):
"""Create a microphone tray icon programmatically via QPainter."""
from PyQt5.QtCore import Qt, QRect, QPoint
from PyQt5.QtGui import QPixmap, QPainter, QColor, QBrush, QPen, QIcon
pixmap = QPixmap(size, size)
pixmap.fill(Qt.transparent)
p = QPainter(pixmap)
p.setRenderHint(QPainter.Antialiasing)
c = QColor(color)
p.setBrush(QBrush(c))
p.setPen(QPen(c.darker(130), 2))
# Mic body (rounded rect)
bw, bh = size // 3, size // 2
bx = (size - bw) // 2
by = size // 8
p.drawRoundedRect(bx, by, bw, bh, bw // 3, bw // 3)
# Arc below mic
p.setBrush(Qt.NoBrush)
p.setPen(QPen(c, max(2, size // 16)))
arc_w = int(bw * 1.6)
arc_h = size // 4
arc_x = (size - arc_w) // 2
arc_y = by + bh - arc_h // 2
p.drawArc(arc_x, arc_y, arc_w, arc_h, 0, -180 * 16)
# Stem
mid_x = size // 2
stem_top = arc_y + arc_h
stem_bot = stem_top + size // 6
p.drawLine(mid_x, stem_top, mid_x, stem_bot)
# Base
base_w = size // 3
p.drawLine(mid_x - base_w // 2, stem_bot, mid_x + base_w // 2, stem_bot)
p.end()
return QIcon(pixmap)
class _CallHelper:
"""Helper to marshal calls from background threads to Qt main thread."""
def __init__(self):
from PyQt5.QtCore import QObject, pyqtSignal
class _Obj(QObject):
sig = pyqtSignal(object)
self._obj = _Obj()
self._obj.sig.connect(self._run)
@staticmethod
def _run(func):
func()
def call(self, func):
self._obj.sig.emit(func)
class TrayIcon:
"""System tray icon with right-click menu."""
COLORS = {
AppState.IDLE: "#888888",
AppState.RECORDING: "#e53935",
AppState.PROCESSING: "#ffb300",
AppState.LISTENING: "#2196F3",
AppState.DICTATING: "#4CAF50",
}
# Minimum expected sizes (bytes) for downloaded models
_MODEL_MIN_SIZES = {
"tiny": 70_000_000,
"base": 130_000_000,
"small": 450_000_000,
"medium": 1_400_000_000,
"large-v1": 2_900_000_000,
"large-v2": 2_900_000_000,
"large-v3": 2_900_000_000,
"large-v3-turbo": 1_500_000_000,
}
def __init__(self, app_ref: "WhisperLinuxApp"):
from PyQt5.QtWidgets import QSystemTrayIcon
self._app_ref = app_ref
self._icons = {state: _create_icon(color) for state, color in self.COLORS.items()}
self._downloading = set()
self._kept_actions = []
self._call_helper = _CallHelper()
self.tray = QSystemTrayIcon()
self.tray.setIcon(self._icons[AppState.IDLE])
self.tray.setToolTip("whisper.linux \u2014 Voice Typing")
self.tray.activated.connect(self._on_activated)
self._build_menu()
self.tray.show()
# -- Menu construction --
def _build_menu(self):
from PyQt5.QtWidgets import QMenu, QAction, QActionGroup
self.menu = QMenu()
self.action_toggle = QAction("Start Recording", self.menu)
self.action_toggle.triggered.connect(lambda: self._app_ref.toggle())
self.menu.addAction(self.action_toggle)
self.menu.addSeparator()
# Language
lang_menu = self.menu.addMenu("Language")
self._lang_group = QActionGroup(lang_menu)
self._lang_group.setExclusive(True)
for code, label in [("ru", "Russian"), ("en", "English"), ("auto", "Auto-detect")]:
a = QAction(label, lang_menu, checkable=True)
a.setData(code)
if code == self._app_ref.config.language:
a.setChecked(True)
a.triggered.connect(self._on_language_changed)
self._lang_group.addAction(a)
lang_menu.addAction(a)
# Model
self._model_menu = self.menu.addMenu("Model")
self._model_group = QActionGroup(self._model_menu)
self._model_group.setExclusive(True)
self._rebuild_model_menu()
# Settings
self._build_settings_menu()
self.menu.addSeparator()
quit_action = QAction("Quit", self.menu)
quit_action.triggered.connect(self._app_ref.quit)
self.menu.addAction(quit_action)
self.tray.setContextMenu(self.menu)
def _rebuild_model_menu(self):
from PyQt5.QtWidgets import QAction
menu = self._model_menu
menu.clear()
self._kept_actions.clear()
for a in self._model_group.actions():
self._model_group.removeAction(a)
current_model = Path(self._app_ref.config.model).name if self._app_ref.config.model else ""
downloaded = _list_models(self._app_ref.config.model_search_dirs)
downloaded_names = {name for name, _ in downloaded}
if not downloaded and self._app_ref.config.model:
name = Path(self._app_ref.config.model).stem.replace("ggml-", "")
downloaded = [(name, self._app_ref.config.model)]
downloaded_names = {name}
for name, path in downloaded:
size_mb = Path(path).stat().st_size / (1024 * 1024)
label = f"{name} ({size_mb:.0f} MB)"
a = QAction(label, menu, checkable=True)
a.setData(path)
if Path(path).name == current_model:
a.setChecked(True)
a.triggered.connect(self._on_model_changed)
self._model_group.addAction(a)
menu.addAction(a)
available = [(n, s) for n, s in AVAILABLE_MODELS if n not in downloaded_names]
if available and downloaded:
menu.addSeparator()
for name, size in available:
if name in self._downloading:
a = QAction(f"{name} {size} \u23f3", menu)
a.setEnabled(False)
else:
a = QAction(f"{name} {size} \u2193 Download", menu)
a.setData(name)
a.triggered.connect(lambda checked, n=name: self._on_download_model(n))
menu.addAction(a)
self._kept_actions.append(a)
def _rebuild_wake_model_menu(self):
from PyQt5.QtWidgets import QAction
menu = self._wake_model_menu
menu.clear()
for a in self._wake_model_group.actions():
self._wake_model_group.removeAction(a)
config = self._app_ref.config
same_label = "Same as main model"
if config.model and Path(config.model).is_file():
main_mb = Path(config.model).stat().st_size / (1024 * 1024)
same_label = f"Same as main model ({main_mb:.0f} MB)"
a = QAction(same_label, menu, checkable=True)
a.setData("")
if not config.wake_model:
a.setChecked(True)
a.triggered.connect(self._on_wake_model_changed)
self._wake_model_group.addAction(a)
menu.addAction(a)
for name, path in _list_models(config.model_search_dirs):
size_mb = Path(path).stat().st_size / (1024 * 1024)
label = f"{name} ({size_mb:.0f} MB)"
a = QAction(label, menu, checkable=True)
a.setData(path)
if path == config.wake_model:
a.setChecked(True)
a.triggered.connect(self._on_wake_model_changed)
self._wake_model_group.addAction(a)
menu.addAction(a)
def _build_settings_menu(self):
from PyQt5.QtWidgets import QAction, QActionGroup
settings_menu = self.menu.addMenu("Settings")
config = self._app_ref.config
# Input mode
input_menu = settings_menu.addMenu("Input mode")
self._input_mode_group = QActionGroup(input_menu)
self._input_mode_group.setExclusive(True)
for val, label in [("hotkey", "Hotkey (push-to-talk)"), ("listen", "Listening (wake word)")]:
a = QAction(label, input_menu, checkable=True)
a.setData(val)
if val == config.input_mode:
a.setChecked(True)
a.triggered.connect(self._on_input_mode_changed)
self._input_mode_group.addAction(a)
input_menu.addAction(a)
# Output mode
output_menu = settings_menu.addMenu("Output mode")
self._output_mode_group = QActionGroup(output_menu)
self._output_mode_group.setExclusive(True)
for val, label in [("batch", "Batch (all at once)"), ("stream", "Streaming (per segment)")]:
a = QAction(label, output_menu, checkable=True)
a.setData(val)
if val == config.output_mode:
a.setChecked(True)
a.triggered.connect(self._on_output_mode_changed)
self._output_mode_group.addAction(a)
output_menu.addAction(a)
settings_menu.addSeparator()
# Threads
threads_menu = settings_menu.addMenu("Threads")
self._threads_group = QActionGroup(threads_menu)
self._threads_group.setExclusive(True)
for n in [1, 2, 4, 8, 16]:
a = QAction(str(n), threads_menu, checkable=True)
a.setData(n)
if n == config.threads:
a.setChecked(True)
a.triggered.connect(self._on_threads_changed)
self._threads_group.addAction(a)
threads_menu.addAction(a)
# GPU
gpu_menu = settings_menu.addMenu("GPU")
self._gpu_group = QActionGroup(gpu_menu)
self._gpu_group.setExclusive(True)
gpu_devices = getattr(config, '_gpu_devices', [])
if gpu_devices:
for name, idx in gpu_devices:
label = f"{idx}: {name}"
a = QAction(label, gpu_menu, checkable=True)
a.setData(idx)
if idx == config.gpu_device:
a.setChecked(True)
a.triggered.connect(self._on_gpu_changed)
self._gpu_group.addAction(a)
gpu_menu.addAction(a)
else:
a = QAction(f"Device {config.gpu_device}", gpu_menu, checkable=True)
a.setData(config.gpu_device)
a.setChecked(True)
self._gpu_group.addAction(a)
gpu_menu.addAction(a)
# Audio
audio_menu = settings_menu.addMenu("Audio")
self._audio_group = QActionGroup(audio_menu)
self._audio_group.setExclusive(True)
audio_devices = _list_audio_devices()
for label, device_id in audio_devices:
a = QAction(label, audio_menu, checkable=True)
a.setData(device_id)
if device_id == config.audio_device:
a.setChecked(True)
a.triggered.connect(self._on_audio_changed)
self._audio_group.addAction(a)
audio_menu.addAction(a)
# Paste mode
paste_menu = settings_menu.addMenu("Paste mode")
self._paste_group = QActionGroup(paste_menu)
self._paste_group.setExclusive(True)
for keys, label in [("shift+Insert", "Shift+Insert (universal)"),
("ctrl+v", "Ctrl+V (regular apps)"),
("ctrl+shift+v", "Ctrl+Shift+V (terminals)")]:
a = QAction(label, paste_menu, checkable=True)
a.setData(keys)
if keys == config.paste_keys:
a.setChecked(True)
a.triggered.connect(self._on_paste_keys_changed)
self._paste_group.addAction(a)
paste_menu.addAction(a)
settings_menu.addSeparator()
# Models directory
models_dir = config.models_dir or "auto"
self._models_dir_action = QAction(f"Models: {models_dir}", settings_menu)
self._models_dir_action.triggered.connect(self._on_models_dir_change)
settings_menu.addAction(self._models_dir_action)
settings_menu.addSeparator()
# Wake word
self._wake_word_action = QAction(f"Wake word: {config.wake_word}", settings_menu)
self._wake_word_action.triggered.connect(self._on_wake_word_change)
settings_menu.addAction(self._wake_word_action)
# Wake model
self._wake_model_menu = settings_menu.addMenu("Wake model")
self._wake_model_group = QActionGroup(self._wake_model_menu)
self._wake_model_group.setExclusive(True)
self._rebuild_wake_model_menu()
settings_menu.addSeparator()
# Silence timeout (inline spinbox)
from PyQt5.QtWidgets import QWidgetAction, QWidget, QHBoxLayout, QLabel, QDoubleSpinBox
silence_widget = QWidget()
silence_layout = QHBoxLayout(silence_widget)
silence_layout.setContentsMargins(8, 2, 8, 2)
silence_label = QLabel("Silence timeout:")
self._silence_spin = QDoubleSpinBox()
self._silence_spin.setRange(0.5, 300.0)
self._silence_spin.setSingleStep(0.5)
self._silence_spin.setDecimals(1)
self._silence_spin.setSuffix("s")
self._silence_spin.setValue(config.silence_timeout)
self._silence_spin.valueChanged.connect(self._on_silence_timeout_spin)
silence_layout.addWidget(silence_label)
silence_layout.addWidget(self._silence_spin)
self._silence_widget_action = QWidgetAction(settings_menu)
self._silence_widget_action.setDefaultWidget(silence_widget)
settings_menu.addAction(self._silence_widget_action)
# End signal toggle
self._end_signal_action = QAction("End signal (beep)", settings_menu, checkable=True)
self._end_signal_action.setChecked(config.end_signal)
self._end_signal_action.triggered.connect(self._on_end_signal_toggled)
settings_menu.addAction(self._end_signal_action)
# Voice commands toggle
self._voice_cmd_action = QAction("Voice commands", settings_menu, checkable=True)
self._voice_cmd_action.setChecked(config.voice_commands)
self._voice_cmd_action.triggered.connect(self._on_voice_commands_toggled)
settings_menu.addAction(self._voice_cmd_action)
# Voice commands editor
self._edit_cmds_action = QAction("Edit voice commands...", settings_menu)
self._edit_cmds_action.triggered.connect(self._on_edit_voice_commands)
settings_menu.addAction(self._edit_cmds_action)
# -- Event handlers --
def _on_activated(self, reason):
from PyQt5.QtWidgets import QSystemTrayIcon
if reason == QSystemTrayIcon.Trigger:
self._app_ref.toggle()
def _on_language_changed(self):
action = self._lang_group.checkedAction()
if action:
self._app_ref.config.language = action.data()
self._app_ref.config.save()
log.info("Language changed to: %s", action.data())
def _on_model_changed(self):
action = self._model_group.checkedAction()
if action:
self._app_ref.config.model = action.data()
self._app_ref.config.save()
self._rebuild_wake_model_menu()
log.info("Model changed to: %s", action.data())
def _on_download_model(self, name):
if name in self._downloading:
return
self._downloading.add(name)
self._rebuild_model_menu()
log.info("Download requested: model '%s'", name)
self.notify("whisper.linux", f"Downloading {name}...")
t = threading.Thread(target=self._do_download_model, args=(name,), daemon=True)
t.start()
def _marshal_call(self, func):
"""Schedule a callable on the Qt main thread (thread-safe)."""
self._call_helper.call(func)
def _do_download_model(self, name):
script = _REPO_ROOT / "models" / "download-ggml-model.sh"
if not script.is_file():
self._downloading.discard(name)
log.error("Download script not found: %s", script)
self._marshal_call(
lambda: self.notify("Error", "download-ggml-model.sh not found"))
return
models_dir = self._app_ref.config.models_dir
Path(models_dir).mkdir(parents=True, exist_ok=True)
model_path = Path(models_dir) / f"ggml-{name}.bin"
# Remove incomplete downloads so the script doesn't skip them
min_size = self._MODEL_MIN_SIZES.get(name, 0)
if model_path.is_file() and min_size and model_path.stat().st_size < min_size:
log.warning("Removing incomplete model %s (%d bytes, expected >= %d)",
model_path, model_path.stat().st_size, min_size)
model_path.unlink()
log.info("Download started: model '%s' \u2192 %s", name, models_dir)
try:
result = subprocess.run(
["bash", str(script), name, models_dir],
capture_output=True, text=True, timeout=3600,
)
if result.returncode != 0:
log.error("Download script failed (rc=%d): %s",
result.returncode, result.stderr[:200])
if model_path.is_file():
size = model_path.stat().st_size
if min_size and size < min_size:
log.error("Downloaded model too small (%d bytes, expected >= %d), removing",
size, min_size)
model_path.unlink()
self._downloading.discard(name)
msg = f"Download incomplete ({size // 1_000_000}MB), please retry"
self._marshal_call(lambda: self.notify("Error", msg))
return
self._downloading.discard(name)
size_mb = size / (1024 * 1024)
log.info("Download complete: model '%s' (%d MB)", name, size_mb)
msg = f"Model '{name}' ready ({size_mb:.0f} MB)"
self._marshal_call(lambda: (
self._rebuild_model_menu(),
self._rebuild_wake_model_menu(),
self.notify("whisper.linux", msg),
))
else:
self._downloading.discard(name)
err = result.stderr[:100]
log.error("Download failed: model file not found after script: %s", model_path)
self._marshal_call(
lambda: self.notify("Error", f"Download failed: {err}"))
except subprocess.TimeoutExpired:
self._downloading.discard(name)
log.error("Download timed out for model '%s' (1 hour limit)", name)
self._marshal_call(
lambda: self.notify("Error", f"Download timed out for '{name}' (1 hour limit)"))
except Exception as e:
self._downloading.discard(name)
log.error("Download failed for model '%s': %s", name, e)
err_str = str(e)
self._marshal_call(
lambda: self.notify("Error", f"Download failed: {err_str}"))
def _on_threads_changed(self):
action = self._threads_group.checkedAction()
if action:
self._app_ref.config.threads = action.data()
self._app_ref.config.save()
log.info("Threads changed to: %d", action.data())
def _on_gpu_changed(self):
action = self._gpu_group.checkedAction()
if action:
self._app_ref.config.gpu_device = action.data()
self._app_ref.config.save()
log.info("GPU device changed to: %d", action.data())
def _on_audio_changed(self):
action = self._audio_group.checkedAction()
if action:
self._app_ref.config.audio_device = action.data()
self._app_ref.config.save()
log.info("Audio device changed to: %s", action.data())
def _on_paste_keys_changed(self):
action = self._paste_group.checkedAction()
if action:
self._app_ref.config.paste_keys = action.data()
self._app_ref.config.save()
log.info("Paste mode changed to: %s", action.data())
def _on_models_dir_change(self):
from PyQt5.QtWidgets import QFileDialog
current = self._app_ref.config.models_dir or str(Path.home())
new_dir = QFileDialog.getExistingDirectory(None, "Models directory", current)
if not new_dir:
return
self._app_ref.config.models_dir = new_dir
self._app_ref.config.save()
self._models_dir_action.setText(f"Models: {new_dir}")
self._rebuild_model_menu()
self._rebuild_wake_model_menu()
log.info("Models dir changed to: %s", new_dir)
def _on_input_mode_changed(self):
action = self._input_mode_group.checkedAction()
if action:
old_val = self._app_ref.config.input_mode
new_val = action.data()
if old_val != new_val and self._app_ref.state != AppState.IDLE:
self._app_ref._force_idle()
self._app_ref.config.input_mode = new_val
self._app_ref.config.save()
log.info("Input mode changed to: %s", new_val)
if new_val == "listen" and self._app_ref.state == AppState.IDLE:
self._app_ref.toggle()
def _on_output_mode_changed(self):
action = self._output_mode_group.checkedAction()
if action:
old_val = self._app_ref.config.output_mode
new_val = action.data()
if old_val != new_val and self._app_ref.state != AppState.IDLE:
self._app_ref._force_idle()
self._app_ref.config.output_mode = new_val
self._app_ref.config.save()
log.info("Output mode changed to: %s", new_val)
def _on_wake_word_change(self):
from PyQt5.QtWidgets import QInputDialog
current = self._app_ref.config.wake_word
text, ok = QInputDialog.getText(None, "Wake Word", "Enter wake word:", text=current)
if ok and text.strip():
self._app_ref.config.wake_word = text.strip()
self._app_ref.config.save()
self._wake_word_action.setText(f"Wake word: {text.strip()}")
log.info("Wake word changed to: %s", text.strip())
def _on_wake_model_changed(self):
action = self._wake_model_group.checkedAction()
if action:
self._app_ref.config.wake_model = action.data()
self._app_ref.config.save()
log.info("Wake model changed to: %s", action.data() or "(same as main)")
def _on_silence_timeout_spin(self, value):
self._app_ref.config.silence_timeout = value
self._app_ref.config.save()
log.info("Silence timeout changed to: %.1fs", value)
def _on_end_signal_toggled(self):
self._app_ref.config.end_signal = self._end_signal_action.isChecked()
self._app_ref.config.save()
log.info("End signal: %s", self._app_ref.config.end_signal)
def _on_voice_commands_toggled(self):
self._app_ref.config.voice_commands = self._voice_cmd_action.isChecked()
self._app_ref.config.save()
log.info("Voice commands: %s", self._app_ref.config.voice_commands)
def _on_edit_voice_commands(self):
from PyQt5.QtWidgets import (
QDialog, QVBoxLayout, QHBoxLayout, QTableWidget,
QTableWidgetItem, QPushButton, QHeaderView, QAbstractItemView,
)
from PyQt5.QtCore import Qt
config = self._app_ref.config
cmds = dict(config.voice_commands_map)
dlg = QDialog()
dlg.setWindowTitle("Voice Commands")
dlg.setMinimumSize(450, 400)
layout = QVBoxLayout(dlg)
table = QTableWidget(len(cmds), 2)
table.setHorizontalHeaderLabels(["Word", "Action"])
table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
table.setSelectionBehavior(QAbstractItemView.SelectRows)
for row, (word, action) in enumerate(cmds.items()):
table.setItem(row, 0, QTableWidgetItem(word))
table.setItem(row, 1, QTableWidgetItem(action))
layout.addWidget(table)
btn_layout = QHBoxLayout()
add_btn = QPushButton("Add")
remove_btn = QPushButton("Remove")
reset_btn = QPushButton("Reset defaults")
btn_layout.addWidget(add_btn)
btn_layout.addWidget(remove_btn)
btn_layout.addStretch()
btn_layout.addWidget(reset_btn)
layout.addLayout(btn_layout)
ok_layout = QHBoxLayout()
ok_btn = QPushButton("OK")
cancel_btn = QPushButton("Cancel")
ok_layout.addStretch()
ok_layout.addWidget(ok_btn)
ok_layout.addWidget(cancel_btn)
layout.addLayout(ok_layout)
def add_row():
row = table.rowCount()
table.insertRow(row)
table.setItem(row, 0, QTableWidgetItem(""))
table.setItem(row, 1, QTableWidgetItem("key:Return"))
table.editItem(table.item(row, 0))
def remove_row():
rows = sorted({idx.row() for idx in table.selectedIndexes()}, reverse=True)
for row in rows:
table.removeRow(row)
def reset_defaults():
table.setRowCount(0)
for word, action in DEFAULT_VOICE_COMMANDS.items():
row = table.rowCount()
table.insertRow(row)
table.setItem(row, 0, QTableWidgetItem(word))
table.setItem(row, 1, QTableWidgetItem(action))
add_btn.clicked.connect(add_row)
remove_btn.clicked.connect(remove_row)
reset_btn.clicked.connect(reset_defaults)
ok_btn.clicked.connect(dlg.accept)
cancel_btn.clicked.connect(dlg.reject)
if dlg.exec_() == QDialog.Accepted:
new_cmds = {}
for row in range(table.rowCount()):
w = (table.item(row, 0).text() or "").strip().lower()
a = (table.item(row, 1).text() or "").strip()
if w and a:
new_cmds[w] = a
config.voice_commands_map = new_cmds
config.save()
# Update the live VoiceCommands instance
self._app_ref._voice_commands._commands = new_cmds
log.info("Voice commands updated: %d entries", len(new_cmds))
# -- State & notifications --
def set_state(self, state: AppState):
self.tray.setIcon(self._icons[state])
if state == AppState.IDLE:
self.action_toggle.setText("Start Recording")
self.tray.setToolTip("whisper.linux \u2014 Idle")
elif state == AppState.RECORDING:
self.action_toggle.setText("Stop Recording")
self.tray.setToolTip("whisper.linux \u2014 Recording...")
elif state == AppState.PROCESSING:
self.action_toggle.setText("Processing...")
self.tray.setToolTip("whisper.linux \u2014 Transcribing...")
elif state == AppState.LISTENING:
self.action_toggle.setText("Stop Listening")
self.tray.setToolTip("whisper.linux \u2014 Listening (say wake word)...")
elif state == AppState.DICTATING:
self.action_toggle.setText("Stop Dictating")
self.tray.setToolTip("whisper.linux \u2014 Dictating...")
def notify(self, title: str, message: str):
from PyQt5.QtWidgets import QSystemTrayIcon
self.tray.showMessage(title, message, QSystemTrayIcon.Information, 3000)

155
examples/whisper.linux/install.sh Executable file
View File

@ -0,0 +1,155 @@
#!/usr/bin/env bash
# install.sh — Install dependencies and set up whisper.linux
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
MODEL_NAME="${1:-base}"
MODEL_FILE="ggml-${MODEL_NAME}.bin"
CONFIG_DIR="$HOME/.config/whisper-linux"
echo "=== whisper.linux installer ==="
echo ""
# --- 1. Install system dependencies ---
echo "[1/4] Installing system dependencies..."
sudo apt-get update -qq
sudo apt-get install -y -qq \
python3-pyqt5 \
xdotool \
xclip \
wl-clipboard \
ydotool \
alsa-utils \
build-essential \
cmake
# Install Vulkan dev deps for GPU acceleration (optional, non-fatal)
sudo apt-get install -y -qq libvulkan-dev glslc 2>/dev/null || true
echo " ✓ System dependencies installed"
# --- 2. Build whisper-cli ---
echo "[2/4] Building whisper-cli..."
BUILD_DIR="$REPO_ROOT/build"
mkdir -p "$BUILD_DIR"
# Enable Vulkan GPU acceleration if available
CMAKE_EXTRA=""
if pkg-config --exists vulkan 2>/dev/null || [ -f /usr/include/vulkan/vulkan.h ]; then
CMAKE_EXTRA="-DGGML_VULKAN=ON"
echo " → Vulkan detected, building with GPU support"
else
echo " → Vulkan not found, building CPU-only"
fi
cmake -S "$REPO_ROOT" -B "$BUILD_DIR" -DCMAKE_BUILD_TYPE=Release $CMAKE_EXTRA
cmake --build "$BUILD_DIR" -j "$(nproc)"
WHISPER_CLI="$BUILD_DIR/bin/whisper-cli"
if [ ! -f "$WHISPER_CLI" ]; then
# Fallback: some builds put it directly in build/
WHISPER_CLI="$BUILD_DIR/whisper-cli"
fi
if [ ! -x "$WHISPER_CLI" ]; then
echo " ✗ Failed to build whisper-cli"
exit 1
fi
echo " ✓ whisper-cli built: $WHISPER_CLI"
# --- 3. Download model ---
echo "[3/4] Downloading model ($MODEL_NAME)..."
MODEL_DIR="$REPO_ROOT/models"
MODEL_PATH="$MODEL_DIR/$MODEL_FILE"
if [ -f "$MODEL_PATH" ]; then
echo " ✓ Model already exists: $MODEL_PATH"
else
bash "$REPO_ROOT/models/download-ggml-model.sh" "$MODEL_NAME"
if [ ! -f "$MODEL_PATH" ]; then
echo " ✗ Model download failed"
exit 1
fi
echo " ✓ Model downloaded: $MODEL_PATH"
fi
# --- 4. Create config ---
echo "[4/4] Creating config..."
mkdir -p "$CONFIG_DIR"
CONFIG_FILE="$CONFIG_DIR/config.ini"
if [ -f "$CONFIG_FILE" ]; then
echo " → Config exists: $CONFIG_FILE"
# Update model and whisper_cli paths to match current build
if grep -q "^model\s*=" "$CONFIG_FILE"; then
sed -i "s|^model\s*=.*|model = $MODEL_PATH|" "$CONFIG_FILE"
fi
if grep -q "^whisper_cli\s*=" "$CONFIG_FILE"; then
sed -i "s|^whisper_cli\s*=.*|whisper_cli = $WHISPER_CLI|" "$CONFIG_FILE"
fi
echo " ✓ Config updated (model=$MODEL_PATH)"
else
cat > "$CONFIG_FILE" <<EOF
[whisper-linux]
whisper_cli = $WHISPER_CLI
model = $MODEL_PATH
language = ru
threads = 4
display_server =
use_clipboard_fallback = False
notification = True
EOF
echo " ✓ Config created: $CONFIG_FILE"
fi
# --- Optional: uinput access for ydotool (Wayland text injection) ---
if [ -c /dev/uinput ]; then
echo "[+] Setting up /dev/uinput access for ydotool..."
# Persistent udev rule
UINPUT_RULE="/etc/udev/rules.d/99-uinput-whisper.rules"
if [ ! -f "$UINPUT_RULE" ]; then
echo "KERNEL==\"uinput\", MODE=\"0660\", GROUP=\"$(id -gn)\"" | sudo tee "$UINPUT_RULE" > /dev/null
sudo udevadm control --reload-rules
sudo udevadm trigger /dev/uinput
fi
# Immediate fix for current session
sudo chmod 0660 /dev/uinput
sudo chown "root:$(id -gn)" /dev/uinput
echo " ✓ /dev/uinput access configured"
fi
# --- Optional: install desktop file for autostart ---
AUTOSTART_DIR="$HOME/.config/autostart"
DESKTOP_FILE="$SCRIPT_DIR/whisper-linux.desktop"
if [ -f "$DESKTOP_FILE" ]; then
mkdir -p "$AUTOSTART_DIR"
# Update Exec path in desktop file
sed "s|Exec=.*|Exec=python3 -m app|" \
"$DESKTOP_FILE" > "$AUTOSTART_DIR/whisper-linux.desktop"
echo " ✓ Autostart desktop file installed"
fi
# --- Set up keyboard shortcut (GNOME) ---
LAUNCHER="$SCRIPT_DIR/whisper-linux"
TOGGLE_CMD="$LAUNCHER --toggle"
echo ""
echo "=== Installation complete ==="
echo ""
echo "Usage:"
echo " $LAUNCHER # Start (tray icon)"
echo " $LAUNCHER --toggle # Toggle recording"
echo " $LAUNCHER --debug # Start with debug logging"
echo " pkill -f whisper-linux # Stop"
echo ""
echo "=== Keyboard shortcut setup ==="
echo " GNOME: Settings → Keyboard → Custom Shortcuts → Add:"
echo " Name: whisper-linux"
echo " Command: $TOGGLE_CMD"
echo " Shortcut: Super+V (or any key you like)"
echo ""
echo " Or run this command to set it up automatically:"
echo " gsettings set org.gnome.settings-daemon.plugins.media-keys custom-keybindings \"['/org/gnome/settings-daemon/plugins/media-keys/custom-keybindings/whisper-linux/']\""
echo " gsettings set org.gnome.settings-daemon.plugins.media-keys.custom-keybinding:/org/gnome/settings-daemon/plugins/media-keys/custom-keybindings/whisper-linux/ name 'whisper-linux'"
echo " gsettings set org.gnome.settings-daemon.plugins.media-keys.custom-keybinding:/org/gnome/settings-daemon/plugins/media-keys/custom-keybindings/whisper-linux/ command '$TOGGLE_CMD'"
echo " gsettings set org.gnome.settings-daemon.plugins.media-keys.custom-keybinding:/org/gnome/settings-daemon/plugins/media-keys/custom-keybindings/whisper-linux/ binding 'F8'"

View File

@ -0,0 +1,6 @@
#!/bin/bash
# Run all whisper.linux tests.
# Usage: ./run_tests.sh [-v] [--debug] [pytest-options...]
cd "$(dirname "$0")"
python3 -m pytest tests/ -v "$@"

View File

@ -0,0 +1,133 @@
"""Pytest fixtures for whisper.linux tests."""
import os
import sys
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
# Add project root to path so we can import the 'app' package
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
@pytest.fixture
def tmp_config_dir(tmp_path):
"""Temporary config directory."""
config_dir = tmp_path / ".config" / "whisper-linux"
config_dir.mkdir(parents=True)
return config_dir
@pytest.fixture
def tmp_config_file(tmp_config_dir):
"""Temporary config.ini with valid paths."""
config_file = tmp_config_dir / "config.ini"
config_file.write_text(
"[whisper-linux]\n"
"whisper_cli = /usr/bin/whisper-cli\n"
"model = /tmp/ggml-base.bin\n"
"models_dir = /tmp\n"
"language = ru\n"
"threads = 4\n"
"display_server = x11\n"
"audio_device = plughw:1,0\n"
"paste_keys = shift+Insert\n"
"use_clipboard_fallback = False\n"
"notification = True\n"
)
return config_file
@pytest.fixture
def mock_config(tmp_config_file, monkeypatch):
"""Config that reads from temp config file."""
import app as wl
monkeypatch.setattr(wl.config, "CONFIG_DIR", tmp_config_file.parent)
monkeypatch.setattr(wl.config, "CONFIG_FILE", tmp_config_file)
return wl.Config()
@pytest.fixture
def mock_config_wayland(mock_config):
"""Config set to Wayland."""
mock_config.display_server = "wayland"
return mock_config
@pytest.fixture
def tmp_wav_file():
"""Create a temporary WAV file for testing."""
fd, path = tempfile.mkstemp(suffix=".wav", prefix="test-whisper-")
os.write(fd, b"RIFF" + b"\x00" * 40) # Minimal WAV header stub
os.close(fd)
yield path
if os.path.exists(path):
os.unlink(path)
@pytest.fixture
def tmp_pid_file(tmp_path, monkeypatch):
"""Temporary PID file."""
import app as wl
pid_file = tmp_path / "whisper-linux.pid"
monkeypatch.setattr(wl.config, "PID_FILE", pid_file)
return pid_file
@pytest.fixture
def mock_config_stream(mock_config):
"""Config set to listen+stream mode (old 'stream' mode)."""
mock_config.input_mode = "listen"
mock_config.output_mode = "stream"
mock_config.wake_word = "дуняша"
mock_config.silence_timeout = 3.0
mock_config.vad_threshold = 300
mock_config.min_speech_ms = 500
mock_config.max_speech_s = 30.0
return mock_config
@pytest.fixture
def mock_config_hotkey_stream(mock_config):
"""Config set to hotkey+stream mode."""
mock_config.input_mode = "hotkey"
mock_config.output_mode = "stream"
mock_config.wake_word = "дуняша"
mock_config.silence_timeout = 3.0
mock_config.vad_threshold = 300
mock_config.min_speech_ms = 500
mock_config.max_speech_s = 30.0
return mock_config
@pytest.fixture
def mock_config_listen_batch(mock_config):
"""Config set to listen+batch mode."""
mock_config.input_mode = "listen"
mock_config.output_mode = "batch"
mock_config.wake_word = "дуняша"
mock_config.silence_timeout = 3.0
mock_config.vad_threshold = 300
mock_config.min_speech_ms = 500
mock_config.max_speech_s = 30.0
return mock_config
@pytest.fixture
def silence_pcm():
"""Return 1 second of silence (s16le mono 16kHz)."""
return b"\x00\x00" * 16000
@pytest.fixture
def speech_pcm():
"""Return 1 second of loud 'speech' (s16le mono 16kHz, high amplitude sine)."""
import array, math
samples = array.array("h")
for i in range(16000):
samples.append(int(10000 * math.sin(2 * math.pi * 440 * i / 16000)))
return samples.tobytes()

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,5 @@
#!/bin/bash
# whisper-linux — Voice typing for Linux desktop
# Unique process name for easy pkill -f whisper-linux
cd "$(dirname "$0")"
exec python3 -m app "$@"

View File

@ -0,0 +1,10 @@
[Desktop Entry]
Type=Application
Name=whisper.linux
Comment=Voice typing for Linux desktop
Exec=~/whisper.linux/whisper-linux
Icon=audio-input-microphone
Terminal=false
Categories=Utility;Audio;
StartupNotify=false
X-GNOME-Autostart-enabled=true