diff --git a/.github/workflows/snap-promote.yml b/.github/workflows/snap-promote.yml index 9e0c52c..45b2eff 100644 --- a/.github/workflows/snap-promote.yml +++ b/.github/workflows/snap-promote.yml @@ -65,8 +65,10 @@ jobs: - name: Install snapcraft run: sudo snap install snapcraft --classic - # `snapcraft status` prints the channel map grouped by track and arch; - # the candidate row is a continuation line, so column 1 is the channel. + # `snapcraft status` prints the channel map grouped by track and arch. + # The candidate row is normally a continuation line (column 1 is the + # channel), but if candidate ever leads the latest track block it rides + # the track line itself — column 3 is the channel then — so match both. - name: Resolve the candidate revision id: candidate env: @@ -78,7 +80,8 @@ jobs: /snap/bin/snapcraft status vocalinux | tee /tmp/snap-status.txt # POSIX: mawk has no \S — use a literal non-space class. row=$(awk '/^[^ \t]/ { track = $1 } - track == "latest" && $1 == "candidate" { print $2, $3; exit }' \ + track == "latest" && $1 == "candidate" { print $2, $3; exit } + track == "latest" && $3 == "candidate" { print $4, $5; exit }' \ /tmp/snap-status.txt) read -r seen_version revision <<<"$row" echo "latest/candidate holds version=${seen_version:-none} revision=${revision:-none}" @@ -105,7 +108,8 @@ jobs: set -euo pipefail VERSION="${TAG#v}" row=$(awk '/^[^ \t]/ { track = $1 } - track == "latest" && $1 == "candidate" { print $2, $3; exit }' \ + track == "latest" && $1 == "candidate" { print $2, $3; exit } + track == "latest" && $3 == "candidate" { print $4, $5; exit }' \ <(/snap/bin/snapcraft status vocalinux)) read -r seen_version revision <<<"$row" if [ "${revision:-}" != "$REVISION" ] || [ "$seen_version" != "$VERSION" ]; then diff --git a/AGENTS.md b/AGENTS.md index c3fb3c6..b74b92a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -144,6 +144,7 @@ src/vocalinux/ ├── auto_pause_monitor.py # unload model while configured apps run ├── model_keepalive.py # idle unload ├── suspend_handler.py # logind PrepareForSleep +├── custom_dictionary.py # terms bias + transcript correction file contracts ├── speech_recognition/ │ ├── recognition_manager.py # whisper.cpp / Whisper / Vosk / remote │ ├── command_processor.py # voice commands diff --git a/docs/CUSTOM_DICTIONARY_DESIGN.md b/docs/CUSTOM_DICTIONARY_DESIGN.md new file mode 100644 index 0000000..1a421b3 --- /dev/null +++ b/docs/CUSTOM_DICTIONARY_DESIGN.md @@ -0,0 +1,125 @@ +# Custom Dictionary Support Design + +## Scope + +Custom dictionary support has two separate jobs: + +1. **Custom terms** bias local Whisper, whisper.cpp, and Faster Whisper recognition + toward names and jargon. +2. **Transcript corrections** deterministically replace a known misheard phrase + after transcription. + +They appear together on the **Custom Dictionary** Settings page, but are not +interchangeable. A correction replacement does not implicitly become a prompt +term. That coupling would make recognition bias depend on correction order and +would make the external terms contract surprising. + +## File contracts + +Both files live in VocaLinux's XDG-aware configuration directory: +`$XDG_CONFIG_HOME/vocalinux` (or `~/.config/vocalinux` when unset). + +| File | Contract | Consumer | +| --- | --- | --- | +| `dictionary.txt` | UTF-8 (BOM accepted), one term per line. Blank lines and lines beginning with `#` are ignored. Terms are de-duplicated case-insensitively while retaining the first spelling. | VocaLinux and an external accessibility scanner. | +| `custom-dictionary-corrections.json` | UTF-8 JSON object: `{"version": 1, "corrections": [{"heard": "super base", "replacement": "Supabase"}]}`. | VocaLinux only. | + +The terms file remains deliberately simple and scanner-friendly. Corrections +need two unambiguous fields, a version, and exact replacement casing, so JSON is +used rather than overloading the line file with an ambiguous delimiter grammar. +The UI writes either file by atomic replacement; a failed UI save preserves the +old file and reports an error. + +The application reads both files for each completed transcript. An externally +edited file therefore takes effect on the next dictation segment without an app +restart. Missing, unreadable, malformed, or unsupported correction files are +ignored safely; dictation continues without their entries. + +The terms file may contain any number of lines, but VocaLinux limits a single +prompt to the first 200 valid terms and 2,000 characters. This bounds decoding +work while leaving the scanner's complete line-file contract intact. The Custom +Dictionary page keeps both an add/remove editor and a validated file chooser +for the terms file. + +## Processing and engine behavior + +For every completed audio segment, VocaLinux applies this order: + +1. The selected engine transcribes audio. +2. Live phrase corrections run against the raw transcript. +3. Voice-command interpretation runs on that corrected text. +4. Remaining text and actions are dispatched. + +Corrections run before commands so a correction can prevent a command-like +misrecognition from being acted on. This also means users must not create a +replacement that deliberately creates a voice command unless that is intended. + +Corrections match literal phrases case-insensitively, only where the adjacent +characters are not Unicode word characters (or common combining marks). Longer +phrases win over shorter overlaps; equal-length entries retain JSON order. +Replacement text is inserted exactly as written. Inputs are normalized to NFC +for matching, so composed/decomposed accents are handled consistently. + +| Engine | Custom terms | Corrections | +| --- | --- | --- | +| OpenAI Whisper | `initial_prompt` | Yes | +| whisper.cpp | Advanced initial prompt followed by custom terms; explicit empty prompt clears reused native state | Yes | +| Faster Whisper | `initial_prompt` | Yes | +| VOSK | No prompt-bias API | Yes | +| Parakeet | No prompt-bias API | Yes | +| Remote API | No prompt is added to current request formats | Yes | + +`--dictionary-file PATH` preserves the #767 session-only behavior: `PATH` is +used as the custom terms file, it enables terms for that session, and Settings +does not modify the saved terms file or enablement. Corrections remain loaded +from the standard JSON file. The previous experimental #768 +`text_injection.custom_dictionary` config list is read only as a fallback until +the JSON corrections file exists; it is never silently deleted or migrated. + +The #767 persisted configuration keys are retained unchanged: +`dictionary.enabled`, `dictionary.file_path` (default +`~/.config/vocalinux/dictionary.txt`), and `dictionary.max_words`. Existing +Peony settings therefore continue to select the scanner's `dictionary.txt` +file. New configured paths are accepted only when they can be expanded and are +either absent (so the UI can create them) or readable regular files; a failed +save leaves the prior setting in place. Invalid configured and `--dictionary-file` +paths are safely ignored and shown as unavailable rather than raising. + +## Alternatives rejected + +- **One mixed line file:** cannot express arbitrary phrases and replacements + without reserving/escaping delimiters, which weakens the scanner contract. +- **Corrections in `config.json`:** makes a scanner compete with application + settings writes and offers no durable external-file schema. +- **Post-command correction:** allows command-like transcription errors to be + acted on before they can be fixed. +- **Automatically adding correction replacements to prompts:** rejected because + replacements are not automatically included in the recognition prompt; + corrections work post-transcription on every engine. + +## Human test checklist + +1. Add `VocaLinux`, `PyGObject`, and a non-ASCII term in **Custom terms**; + confirm `dictionary.txt` is UTF-8, one line per term, and an external + edit changes the next Whisper, whisper.cpp, and Faster Whisper dictation segment. +2. Add `super base` → `Supabase`; test lowercase, uppercase, punctuation, an + overlapping short correction, `C++`, and non-ASCII text. Confirm replacement + casing is exact and embedded text (for example `supersize`) is not replaced. +3. With voice commands enabled, add a correction that changes a command-like + misrecognition into ordinary text. Confirm it is injected rather than acted + on. Also verify an intentionally command-like replacement is avoided. +4. Edit each file externally while VocaLinux is running. Test an invalid JSON + correction file and invalid UTF-8 terms file; confirm dictation continues and + the Settings status/save feedback is clear. +5. Test the UI with keyboard only: Add buttons via Enter, empty validation, + update/remove feedback, narrow window, long entries, and high-contrast theme. +6. Test all engines: Whisper, whisper.cpp, and Faster Whisper should use terms and + corrections; VOSK, Parakeet, and remote API should use corrections only, with + the VOSK terms warning. +7. Start with `--dictionary-file /path/to/terms.txt`; confirm the session uses + that file, Settings disables terms editing, saved settings remain unchanged, + and corrections still apply. +8. Start with an unresolved path such as `--dictionary-file ~missing-user/terms.txt` + and configure an unreadable path through the file chooser. Confirm Settings + reports the path safely, dictation continues without terms, and the prior + saved `dictionary.file_path` remains intact. diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index fed1f1f..5654121 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -212,6 +212,39 @@ vocalinux --start-minimized vocalinux --transcribe-file meeting.wav # diarized transcript on stdout, then exits ``` +## Custom Dictionary Support + +Open **Settings → Custom Dictionary** to configure two separate capabilities: + +- **Custom terms** bias recognition toward product names, people, and jargon. + They are read from `~/.config/vocalinux/dictionary.txt` by default, as UTF-8 + with one term per line. Blank lines and `#` comments are allowed, so the same + file remains friendly to an accessibility scanner. Use the add/remove editor + or choose a different readable terms file with the file picker. +- **Transcript corrections** replace a known misheard word or phrase after + transcription, for example `super base` with `Supabase`. Corrections are + stored separately in `~/.config/vocalinux/custom-dictionary-corrections.json`. + +VocaLinux re-reads both files before each completed dictation segment, so an +external edit applies to the next segment without restarting the app. Vocabulary +bias works with Whisper, whisper.cpp, and Faster Whisper. Corrections work with +every engine, including VOSK, Parakeet, and the configured remote API. + +Corrections run before voice-command interpretation. This can prevent a +command-like misrecognition from acting, but avoid replacements that create a +voice command unless that is intentional. Correction replacements are not +automatically added to recognition bias. + +For a one-session terms override, start Vocalinux with: + +```bash +vocalinux --dictionary-file /path/to/dictionary.txt +``` + +This temporarily enables terms from that file without changing saved settings; +the Custom terms controls are disabled for the session. It does not disable +transcript corrections. + `--transcribe-file` uses the TinyDiarize model (`small.en-tdrz`) and needs it downloaded first — the tray "Transcribe Audio File…" entry fetches it on demand. ## Troubleshooting diff --git a/src/vocalinux/custom_dictionary.py b/src/vocalinux/custom_dictionary.py new file mode 100644 index 0000000..199ac41 --- /dev/null +++ b/src/vocalinux/custom_dictionary.py @@ -0,0 +1,734 @@ +"""File-backed custom dictionary support for recognition bias and transcript fixes.""" + +import itertools +import json +import logging +import os +import re +import shutil +import unicodedata +import uuid +from pathlib import Path +from typing import TYPE_CHECKING, Any, Iterator, Optional + +from .utils.paths import config_dir + +if TYPE_CHECKING: + from .ui.config_manager import ConfigManager + +logger = logging.getLogger(__name__) + +TERMS_FILENAME = "dictionary.txt" +DEFAULT_TERMS_PATH = str(Path(config_dir()) / TERMS_FILENAME) +# Pre-XDG default persisted by older builds; keep as a configured path. +LEGACY_DEFAULT_TERMS_PATH = "~/.config/vocalinux/dictionary.txt" +CORRECTIONS_FILENAME = "custom-dictionary-corrections.json" +CORRECTIONS_VERSION = 1 +CORRECTIONS_TOP_LEVEL_KEYS = frozenset({"version", "corrections"}) +CORRECTIONS_ENTRY_KEYS = frozenset({"heard", "replacement"}) +DEFAULT_MAX_TERMS = 200 +MAX_TERM_CHARACTERS = 200 +MAX_PROMPT_CHARACTERS = 2_000 +MAX_CORRECTIONS = 500 +MAX_CORRECTION_CHARACTERS = 500 +# A terms file beyond this bound is invalid everywhere rather than read in +# part: the read itself stays bounded and no partial dictionary can reach +# recognition while Settings reports the file unusable. +MAX_TERMS_FILE_BYTES = 1_048_576 +# Even within the byte bound a scanner-managed file can hold ~130k terms; the +# dictionary is a prompt vocabulary, so yields stop here — Settings would +# otherwise build a row per term. +MAX_TERMS_YIELDED = 4_096 + + +class _DuplicateJsonKeyError(ValueError): + """Raised when a JSON object contains the same key more than once.""" + + +def _object_pairs_without_duplicates(pairs: list[tuple[Any, Any]]) -> dict[Any, Any]: + """Return a dict, raising if any key repeats at this object level.""" + payload: dict[Any, Any] = {} + for key, value in pairs: + if key in payload: + raise _DuplicateJsonKeyError(key) + payload[key] = value + return payload + + +def normalize_corrections(raw_entries: Any) -> list[dict[str, str]]: + """Validate correction entries, retaining the first case-insensitive source. + + Input order is meaningful for equal-length overlapping entries. Keeping the + first duplicate makes hand-edited JSON deterministic and agrees with the + order used when building the matching expression. + """ + if not isinstance(raw_entries, list): + return [] + + entries: list[dict[str, str]] = [] + seen: set[str] = set() + for index, entry in enumerate(raw_entries): + if len(entries) >= MAX_CORRECTIONS: + logger.warning("Ignoring corrections after the supported limit of %d", MAX_CORRECTIONS) + break + if not isinstance(entry, dict): + logger.warning("Ignoring correction %d: expected an object", index) + continue + heard = entry.get("heard") + replacement = entry.get("replacement") + if not isinstance(heard, str) or not isinstance(replacement, str): + logger.warning("Ignoring correction %d: both fields must be strings", index) + continue + heard = unicodedata.normalize("NFC", heard.strip()) + replacement = replacement.strip() + if not heard or not replacement: + logger.warning("Ignoring correction %d: both fields must be non-empty", index) + continue + if len(heard) > MAX_CORRECTION_CHARACTERS or len(replacement) > MAX_CORRECTION_CHARACTERS: + logger.warning("Ignoring correction %d: phrase is too long", index) + continue + normalized_heard = heard.casefold() + if normalized_heard in seen: + logger.warning("Ignoring duplicate correction source %r", heard) + continue + seen.add(normalized_heard) + entries.append({"heard": heard, "replacement": replacement}) + return entries + + +def apply_corrections(text: str, entries: list[dict[str, str]]) -> str: + """Replace configured phrases in *text* using the custom-dictionary policy. + + Sources are matched case-insensitively only when neither adjacent character + is a Unicode word character. Replacements retain the exact user-provided + spelling. Longer source phrases take precedence; ties retain JSON order. + """ + if not text or not entries: + return text + + normalized_text = unicodedata.normalize("NFC", text) + ordered = sorted( + entries, + key=lambda entry: (len(entry["heard"].split()), len(entry["heard"])), + reverse=True, + ) + alternatives = "|".join( + f"(?P{re.escape(entry['heard'])})" + for index, entry in enumerate(ordered) + ) + pattern = re.compile( + r"(? str: + """Return the replacement associated with the matched alternative.""" + for index, entry in enumerate(ordered): + if match.group(f"correction_{index}") is not None: + return entry["replacement"] + return match.group(0) + + corrected = pattern.sub(replace, normalized_text) + if corrected != text: + logger.debug("Applied custom dictionary corrections to final transcript") + return corrected + + +class CustomDictionaryManager: + """Read, write, and live-reload the two custom dictionary file contracts.""" + + def __init__( + self, config_manager: "ConfigManager", transient_terms_path: Optional[str] = None + ) -> None: + self.config = config_manager + self._transient_terms_path = transient_terms_path + + @property + def is_transient_terms(self) -> bool: + """Return whether a CLI session-only terms file supersedes the saved file.""" + return self._transient_terms_path is not None + + def terms_enabled(self) -> bool: + """Return whether custom terms should be offered to Whisper-family engines.""" + return self.is_transient_terms or bool(self.config.get("dictionary", "enabled", False)) + + def set_terms_enabled(self, enabled: bool) -> bool: + """Persist terms enablement, rolling back the in-memory setting on failure.""" + if self.is_transient_terms: + logger.info("Ignoring saved terms enablement while a CLI override is active") + return False + old_value = self.config.get("dictionary", "enabled", False) + if not self.config.set("dictionary", "enabled", bool(enabled)): + return False + if self.config.save_config(): + return True + self.config.set("dictionary", "enabled", old_value) + logger.warning("Could not save custom terms enablement; keeping previous setting") + return False + + def terms_path_text(self) -> str: + """Return the configured or session-only terms path without expansion. + + A missing or blank saved path uses the live XDG default. Historical + leftover pre-XDG default strings are kept, not rewritten to XDG. + """ + if self._transient_terms_path is not None: + return self._transient_terms_path + default_path = str(Path(config_dir()) / TERMS_FILENAME) + configured = self.config.get("dictionary", "file_path", default_path) + if not isinstance(configured, str) or not configured.strip(): + return default_path + configured = configured.strip() + if ( + self._is_legacy_default_terms_path(configured, default_path) + and not self._file_path_is_explicit() + ): + self._stamp_legacy_default_terms_path_explicit() + return configured + + def _file_path_is_explicit(self) -> bool: + """Return whether Settings persisted the current terms path on purpose.""" + return bool(self.config.get("dictionary", "file_path_explicit", False)) + + @staticmethod + def _is_legacy_default_terms_path(configured: str, default_path: str) -> bool: + """Return whether a persisted path matches the pre-XDG default string.""" + if configured == LEGACY_DEFAULT_TERMS_PATH: + return True + try: + expanded_legacy = str(Path.home() / ".config" / "vocalinux" / TERMS_FILENAME) + except RuntimeError: + return False + return configured == expanded_legacy and expanded_legacy != default_path + + def _stamp_legacy_default_terms_path_explicit(self) -> None: + """Best-effort persist that a leftover pre-XDG path is an explicit choice. + + Historical configs cannot distinguish leftover defaults from Settings + picks. Keep the stored path and mark it explicit so later reads stay + stable. Failures leave both the path and the marker unchanged. + """ + if self.is_transient_terms: + return + old_explicit = self.config.get("dictionary", "file_path_explicit", False) + if not self.config.set("dictionary", "file_path_explicit", True): + return + if self.config.save_config(): + return + self.config.set("dictionary", "file_path_explicit", old_explicit) + logger.warning( + "Could not persist custom terms path explicit marker; keeping the configured path" + ) + + def terms_path(self) -> Optional[Path]: + """Return the active expanded terms path, or None when it is invalid.""" + configured = self.terms_path_text() + try: + return Path(configured).expanduser() + except RuntimeError as error: + logger.warning("Could not expand custom terms path %r: %s", configured, error) + return None + + def _saved_terms_path(self) -> Optional[Path]: + """Return the persisted terms path, ignoring any CLI override.""" + default_path = str(Path(config_dir()) / TERMS_FILENAME) + configured = self.config.get("dictionary", "file_path", default_path) + if not isinstance(configured, str) or not configured.strip(): + configured = default_path + try: + return Path(configured.strip()).expanduser() + except RuntimeError as error: + logger.warning("Could not expand saved terms path %r: %s", configured, error) + return None + + def set_terms_path(self, path: str) -> bool: + """Persist a usable terms path, retaining the prior setting on save failure.""" + if self.is_transient_terms: + logger.info("Ignoring terms path change while a CLI override is active") + return False + if not isinstance(path, str) or not path.strip(): + logger.warning("Ignoring empty custom terms path") + return False + configured = path.strip() + try: + candidate = Path(configured).expanduser() + if candidate.exists() and (not candidate.is_file() or not self._is_readable(candidate)): + logger.warning("Ignoring unusable custom terms path %s", candidate) + return False + except (OSError, RuntimeError) as error: + logger.warning("Ignoring invalid custom terms path %r: %s", configured, error) + return False + + old_value = self.config.get( + "dictionary", "file_path", str(Path(config_dir()) / TERMS_FILENAME) + ) + old_explicit = self.config.get("dictionary", "file_path_explicit", False) + if not self.config.set("dictionary", "file_path", configured): + return False + if not self.config.set("dictionary", "file_path_explicit", True): + self.config.set("dictionary", "file_path", old_value) + return False + if self.config.save_config(): + return True + self.config.set("dictionary", "file_path", old_value) + self.config.set("dictionary", "file_path_explicit", old_explicit) + logger.warning("Could not save custom terms path; keeping previous setting") + return False + + def corrections_path(self) -> Path: + """Return the fixed structured corrections path. + + Corrections always live under config_dir(), the location promised by + the file contract, so a custom or session-only terms path cannot + strand or duplicate them. + """ + return Path(config_dir()) / CORRECTIONS_FILENAME + + def get_terms(self) -> list[str]: + """Read the current UTF-8 line file; external edits apply next segment.""" + path = self.terms_path() + if path is None: + return [] + return list(self._iter_terms(path)) + + @staticmethod + def _iter_terms(path: Path) -> Iterator[str]: + """Yield normalized, de-duplicated terms from a bounded read. + + The bounded read validates the whole file when it fits the limit, so + a partly invalid file yields nothing and recognition never consumes + terms a status check would call invalid. Files past the limit still + supply their complete leading lines — a scanner-managed file can grow + past the bound without losing its usable prefix. Iterating over the + decoded lines still lets callers stop processing early. + """ + try: + with path.open("rb") as terms_file: + raw = terms_file.read(MAX_TERMS_FILE_BYTES + 1) + except FileNotFoundError: + return + except OSError as error: + logger.warning("Could not read custom terms file: %s", error) + return + if len(raw) > MAX_TERMS_FILE_BYTES: + logger.warning( + "Custom terms file exceeds %d bytes; reading complete leading lines only", + MAX_TERMS_FILE_BYTES, + ) + # Keep the valid leading lines: a scanner-managed file can grow + # past the bound, but its usable prefix still supplies terms. A + # prefix with no complete line supplies nothing. + last_newline = raw.rfind(b"\n") + if last_newline == -1: + return + raw = raw[:last_newline] + try: + text = raw.decode("utf-8-sig") + except UnicodeError as error: + logger.warning("Could not read custom terms file: %s", error) + return + seen: set[str] = set() + for line in text.splitlines(): + term = unicodedata.normalize("NFC", line.strip()) + normalized_term = term.casefold() + if not term or term.startswith("#") or normalized_term in seen: + continue + seen.add(normalized_term) + yield term + if len(seen) >= MAX_TERMS_YIELDED: + logger.warning( + "Custom terms file supplies more than %d terms; trailing entries ignored", + MAX_TERMS_YIELDED, + ) + return + + def _iter_terms_for_prompt(self, path: Path, limit: int) -> Iterator[str]: + """Yield up to *limit* terms, streaming oversized files line by line. + + Files inside the read bound keep the same whole-file-validated + iteration as ``_iter_terms``. Past it — where the usable-prefix + policy already accepts complete leading lines — the prompt only + collects the first *limit* terms, but every complete line in the + bounded window is still decoded so an invalid file supplies no + terms at all. Each read is capped by the remaining window, so an + overlong line can never pull more than the byte bound into memory. + """ + if limit <= 0: + return + try: + oversized = path.stat().st_size > MAX_TERMS_FILE_BYTES + except OSError: + oversized = False + if not oversized: + yield from itertools.islice(self._iter_terms(path), limit) + return + terms: list[str] = [] + seen: set[str] = set() + consumed = 0 + try: + with path.open("rb") as terms_file: + index = 0 + while True: + raw_line = terms_file.readline(MAX_TERMS_FILE_BYTES + 2 - consumed) + consumed += len(raw_line) + if ( + not raw_line + or consumed > MAX_TERMS_FILE_BYTES + 1 + or not raw_line.endswith(b"\n") + ): + # Lines complete inside the byte bound are usable; a + # line straddling it is not a complete leading line. + break + try: + line = raw_line.decode("utf-8-sig" if index == 0 else "utf-8") + except UnicodeError as error: + logger.warning("Could not read custom terms file: %s", error) + return + index += 1 + # Collection stops at the prompt limit, but every + # complete leading line is still decoded: the prompt + # fails closed on the same invalid files the terms + # reader rejects. + if len(terms) >= limit: + continue + term = unicodedata.normalize("NFC", line.strip()) + normalized_term = term.casefold() + if not term or term.startswith("#") or normalized_term in seen: + continue + seen.add(normalized_term) + terms.append(term) + except FileNotFoundError: + return + except OSError as error: + logger.warning("Could not read custom terms file: %s", error) + return + yield from terms + + def save_terms(self, terms: list[str]) -> bool: + """Safely replace the standard terms file with normalized line entries.""" + if self.is_transient_terms: + logger.info("Ignoring terms edit while a CLI override is active") + return False + path = self.terms_path() + if path is None: + logger.warning("Cannot save custom terms because the configured path is invalid") + return False + normalized_terms = self._normalize_terms(terms) + contents = "\n".join(normalized_terms) + if contents: + contents += "\n" + return self._atomic_write(path, contents) + + def add_term(self, term: str) -> bool: + """Add one valid term without dropping or rewriting unrelated lines.""" + if self.is_transient_terms: + logger.info("Ignoring terms edit while a CLI override is active") + return False + cleaned = self._single_term(term) + if cleaned is None: + return False + path = self.terms_path() + if path is None: + logger.warning("Cannot add custom term because the configured path is invalid") + return False + contents = self._read_terms_contents(path, missing_value="") + if contents is None: + return False + existing_terms = self.get_terms() + if any(existing.casefold() == cleaned.casefold() for existing in existing_terms): + logger.warning("Ignoring duplicate custom term %r", cleaned) + return False + if ( + len(contents.encode("utf-8")) > MAX_TERMS_FILE_BYTES + or len(existing_terms) >= MAX_TERMS_YIELDED + ): + # A tail append would land past the read window where the new term + # stays invisible, while rewriting from a capped read would drop + # scanner-owned lines. Prepending keeps the addition readable and + # leaves every existing line, comment, and blank line intact. + return self._atomic_write(path, f"{cleaned}\n{contents}") + separator = "" if not contents or contents.endswith(("\n", "\r")) else "\n" + return self._atomic_write(path, f"{contents}{separator}{cleaned}\n") + + def remove_term(self, term: str) -> bool: + """Remove matching term lines without changing unrelated file content.""" + if self.is_transient_terms: + logger.info("Ignoring terms edit while a CLI override is active") + return False + cleaned = self._single_term(term) + if cleaned is None: + return False + path = self.terms_path() + if path is None: + logger.warning("Cannot remove custom term because the configured path is invalid") + return False + contents = self._read_terms_contents(path) + if contents is None: + return False + + remaining_lines: list[str] = [] + removed = False + for line in contents.splitlines(keepends=True): + line_term = unicodedata.normalize("NFC", line.rstrip("\r\n").strip()) + if not line_term.startswith("#") and line_term.casefold() == cleaned.casefold(): + removed = True + continue + remaining_lines.append(line) + if not removed: + logger.warning("Custom term %r was not present in the terms file", cleaned) + return False + return self._atomic_write(path, "".join(remaining_lines)) + + def build_initial_prompt(self) -> Optional[str]: + """Build the current Whisper prompt from enabled terms, if any.""" + if not self.terms_enabled(): + return None + max_terms = self.config.get("dictionary", "max_words", DEFAULT_MAX_TERMS) + try: + max_terms = max(0, int(max_terms)) + except (TypeError, ValueError): + logger.warning("Invalid custom terms limit %r; using %d", max_terms, DEFAULT_MAX_TERMS) + max_terms = DEFAULT_MAX_TERMS + path = self.terms_path() + if path is None: + return None + prompt_terms: list[str] = [] + prompt_characters = 0 + for term in self._iter_terms_for_prompt(path, max_terms): + additional_characters = len(term) + (1 if prompt_terms else 0) + if prompt_characters + additional_characters > MAX_PROMPT_CHARACTERS: + logger.warning("Custom terms prompt reached %d characters", MAX_PROMPT_CHARACTERS) + break + prompt_terms.append(term) + prompt_characters += additional_characters + return " ".join(prompt_terms) or None + + def get_corrections(self) -> list[dict[str, str]]: + """Read corrections from JSON for every segment, failing closed on errors.""" + entries = self._read_corrections(for_edit=False) + return entries if entries is not None else [] + + def get_corrections_for_edit(self) -> Optional[list[dict[str, str]]]: + """Return losslessly editable entries, or None for an unsafe source file. + + Runtime correction reads may safely ignore malformed individual entries, + but a UI edit must never rewrite the file from that filtered view. + """ + return self._read_corrections(for_edit=True) + + def _read_corrections_text(self, path: Path) -> Optional[str]: + """Return corrections JSON text, or None when no file exists. + + If *path* is missing, copy once from beside the terms file when a + leftover file from the earlier co-located contract exists there so + those entries survive the move to the fixed location. Copy failure + still returns the old text so entries are not lost. + """ + try: + return path.read_text(encoding="utf-8") + except FileNotFoundError: + pass + # Legacy corrections lived beside the persisted terms file. Resolve + # that path from the saved setting — never the CLI override — so an + # override's sibling is not adopted while a saved file's corrections + # still apply. + terms = self._saved_terms_path() + old_path = terms.parent / CORRECTIONS_FILENAME if terms is not None else None + if old_path is None or old_path == path or not old_path.is_file(): + return None + try: + contents = old_path.read_text(encoding="utf-8") + except FileNotFoundError: + return None + self._atomic_write(path, contents) + return contents + + def _read_corrections(self, *, for_edit: bool) -> Optional[list[dict[str, str]]]: + """Read corrections with stricter validation for write-back workflows.""" + path = self.corrections_path() + try: + contents = self._read_corrections_text(path) + if contents is None: + return self._legacy_corrections() + payload = json.loads( + contents, + object_pairs_hook=_object_pairs_without_duplicates if for_edit else None, + ) + except _DuplicateJsonKeyError: + logger.warning( + "Refusing to edit custom corrections because the source has duplicate JSON keys" + ) + return None + except (OSError, UnicodeError, json.JSONDecodeError) as error: + logger.warning("Could not read custom corrections file %s: %s", path, error) + return None if for_edit else [] + + version = payload.get("version") if isinstance(payload, dict) else None + if ( + not isinstance(payload, dict) + or not isinstance(version, int) + or isinstance(version, bool) + or version != CORRECTIONS_VERSION + ): + logger.warning("Ignoring custom corrections file with an unsupported schema") + return None if for_edit else [] + + raw_entries = payload.get("corrections") + entries = normalize_corrections(raw_entries) + if for_edit: + if set(payload) - CORRECTIONS_TOP_LEVEL_KEYS: + logger.warning( + "Refusing to edit custom corrections because the source has extra top-level fields" + ) + return None + if not isinstance(raw_entries, list) or len(entries) != len(raw_entries): + logger.warning( + "Refusing to edit custom corrections because some source entries are invalid" + ) + return None + if any( + not isinstance(entry, dict) or set(entry) - CORRECTIONS_ENTRY_KEYS + for entry in raw_entries + ): + logger.warning( + "Refusing to edit custom corrections because some source entries have extra fields" + ) + return None + if any( + entry.get("heard") != normalized["heard"] + or entry.get("replacement") != normalized["replacement"] + for entry, normalized in zip(raw_entries, entries) + ): + logger.warning( + "Refusing to edit custom corrections because normalization would change values" + ) + return None + return entries + + def save_corrections(self, entries: list[dict[str, str]]) -> bool: + """Safely write validated corrections in the versioned JSON contract.""" + payload = { + "version": CORRECTIONS_VERSION, + "corrections": normalize_corrections(entries), + } + return self._atomic_write( + self.corrections_path(), json.dumps(payload, ensure_ascii=False, indent=2) + "\n" + ) + + def apply_corrections(self, text: str) -> str: + """Live-reload and apply corrections to a completed transcript.""" + return apply_corrections(text, self.get_corrections()) + + def terms_status(self) -> str: + """Return a concise, user-facing description of the custom terms file.""" + path = self.terms_path() + if path is None: + return "Configured terms path is invalid or cannot be expanded." + try: + if not path.exists(): + return "Terms file does not exist yet; add a term to create it." + if not path.is_file(): + return "Terms path is not a regular file." + if path.stat().st_size > MAX_TERMS_FILE_BYTES: + count = len(self.get_terms()) + return f"{count} term(s) available from the file's leading lines only." + if len(list(self._iter_terms(path))) >= MAX_TERMS_YIELDED: + return f"Only the first {MAX_TERMS_YIELDED} terms are used." + path.read_text(encoding="utf-8-sig") + except UnicodeError: + return "Terms file is not valid UTF-8." + except OSError: + return "Terms file cannot be inspected." + return f"{len(self.get_terms())} term(s) available from the live file." + + @staticmethod + def _is_readable(path: Path) -> bool: + """Return whether a path can be opened for reading.""" + try: + with path.open("r", encoding="utf-8"): + return True + except OSError: + return False + + @staticmethod + def _normalize_terms(terms: list[str]) -> list[str]: + """Return trimmed, de-duplicated terms suitable for the line-file contract.""" + normalized: list[str] = [] + seen: set[str] = set() + for term in terms: + if not isinstance(term, str): + continue + cleaned = unicodedata.normalize("NFC", term.strip()) + if len(cleaned) > MAX_TERM_CHARACTERS: + logger.warning( + "Ignoring custom term longer than %d characters", MAX_TERM_CHARACTERS + ) + continue + key = cleaned.casefold() + if not cleaned or cleaned.startswith("#") or key in seen: + continue + seen.add(key) + normalized.append(cleaned) + return normalized + + @staticmethod + def _single_term(term: str) -> Optional[str]: + """Validate a single line-file term for an incremental edit.""" + if not isinstance(term, str) or "\n" in term or "\r" in term: + logger.warning("Ignoring invalid custom term") + return None + normalized = CustomDictionaryManager._normalize_terms([term]) + return normalized[0] if normalized else None + + @staticmethod + def _read_terms_contents(path: Path, missing_value: Optional[str] = None) -> Optional[str]: + """Read terms text for an edit while reporting invalid source files safely.""" + try: + return path.read_text(encoding="utf-8-sig") + except FileNotFoundError: + return missing_value + except (OSError, UnicodeError) as error: + logger.warning("Could not read custom terms file %s: %s", path, error) + return None + + def _legacy_corrections(self) -> list[dict[str, str]]: + """Read the never-released #768 config shape without migrating it silently.""" + raw_entries = self.config.get("text_injection", "custom_dictionary", []) + entries = normalize_corrections( + [ + {"heard": entry.get("spoken"), "replacement": entry.get("replacement")} + for entry in raw_entries + if isinstance(entry, dict) + ] + ) + if entries: + logger.warning("Using legacy config entries until %s is saved", CORRECTIONS_FILENAME) + return entries + + @staticmethod + def _atomic_write(path: Path, contents: str) -> bool: + """Atomically write UTF-8 user data and leave the old file intact on failure.""" + temporary_path: Optional[Path] = None + try: + path.parent.mkdir(parents=True, exist_ok=True) + temporary_path = path.with_name(f".{path.name}.{uuid.uuid4().hex}") + with temporary_path.open("x", encoding="utf-8") as temporary_file: + temporary_file.write(contents) + temporary_file.flush() + os.fsync(temporary_file.fileno()) + try: + # A replacement keeps the existing file's permissions; a new + # file keeps the umask-derived mode of the temporary file. + shutil.copymode(path, temporary_path) + except OSError: + pass + os.replace(temporary_path, path) + return True + except OSError as error: + logger.error("Could not save custom dictionary file %s: %s", path, error) + if temporary_path is not None: + try: + temporary_path.unlink(missing_ok=True) + except OSError as cleanup_error: + logger.warning("Could not remove temporary dictionary file: %s", cleanup_error) + return False diff --git a/src/vocalinux/main.py b/src/vocalinux/main.py index 632e813..a01ba6f 100644 --- a/src/vocalinux/main.py +++ b/src/vocalinux/main.py @@ -111,6 +111,14 @@ def parse_arguments(): action="store_true", help="Stop voice typing on a running instance (via D-Bus) and exit", ) + parser.add_argument( + "--dictionary-file", + type=str, + help=( + "Use this custom terms file for this session only (UTF-8, one term per line); " + "it does not change saved settings" + ), + ) parser.add_argument( "--transcribe-file", type=str, @@ -414,6 +422,7 @@ def main(): # Now it's safe to import GTK-dependent modules from .common_types import RecognitionState + from .custom_dictionary import CustomDictionaryManager from .speech_recognition import recognition_manager from .text_injection import text_injector from .ui import tray_indicator @@ -526,6 +535,16 @@ def main(): # sanitize on load so a bad preference cannot abort startup. history_max_items = sanitize_max_items(history_settings.get("max_items", DEFAULT_MAX_ITEMS)) + dictionary_file = getattr(args, "dictionary_file", None) + transient_terms_path = ( + dictionary_file.strip() + if isinstance(dictionary_file, str) and dictionary_file.strip() + else None + ) + if transient_terms_path is not None: + logger.info("Using session-only custom terms file: %s", transient_terms_path) + dictionary_manager = CustomDictionaryManager(config_manager, transient_terms_path) + logger.info(f"Final settings: engine={engine}, language={language}, model={model_size}") if audio_device_index is not None: logger.info( @@ -564,6 +583,7 @@ def main(): whispercpp_no_speech_thold=advanced_settings.get("whispercpp_no_speech_thold", 0.6), whispercpp_n_threads=advanced_settings.get("whispercpp_n_threads", 0), whispercpp_gpu_device=advanced_settings.get("whispercpp_gpu_device", None), + dictionary_manager=dictionary_manager, remote_api_url=saved_settings.get("remote_api_url", ""), remote_api_key=saved_settings.get("remote_api_key", ""), remote_api_endpoint=saved_settings.get("remote_api_endpoint", "/inference"), @@ -610,10 +630,19 @@ def main(): # TranscriptionHistory.add), so late segments extend that entry # rather than whichever snippet happens to be newest. latest_snippet_id: Optional[int] = None + # Worker thread that produced the most-recently-ended session's + # segments; deliveries on that same thread may extend its committed + # snippet once the next session has closed, so an older worker + # finishing two sessions later cannot leak into a newer entry. + ended_session_worker: Optional[threading.Thread] = None + # Every worker that has delivered in-session segments; a delivery on + # a thread never associated with a session is treated as the + # just-ended session's trailing decode, while a worker seen producing + # an earlier session can never merge into a newer entry. + session_workers_seen: set[threading.Thread] = set() # Clear epoch the most-recently-ended session was committed under; # late segments merging into its snippet are judged against it. ended_session_epoch = transcription_history.epoch - # In-app dictation pad: the Wayland-proof fallback target that # receives dictated text when "dictate to pad" capture is enabled # (#726). Constructed before the tray so the tray menu can open it. @@ -695,7 +724,7 @@ def main(): captured after the clear are kept, so dictation that continues across a clear is still recoverable. """ - nonlocal session_worker, latest_snippet_id + nonlocal session_worker, latest_snippet_id, ended_session_worker if not transcription_history.enabled: return segment = _normalize_segment_text(segment) @@ -736,23 +765,40 @@ def main(): ) ): session_worker = worker + session_workers_seen.add(worker) session_segments.append((segment, started_at)) return # Late segment from a session that already ended: merge into # its own snippet by id — a newer session may already have # committed on top, so the newest entry is not the target. - # Both writes are guarded by the ended session's epoch, so a + # Two kinds are admitted: trickles on the ended session's own + # worker thread (the engine's decode finishing after IDLE), + # and any straggler while a newer session is still open — + # the open session has not committed yet, so the tracked id + # still points to the ended session's own snippet. An older + # worker delivering after that newer session committed forms + # its own entry rather than growing the wrong snippet. Both + # writes are guarded by the ended session's epoch, so a # clear() landing between that commit and this delivery # still refuses the text. - if latest_snippet_id is not None and transcription_history.extend_entry( - latest_snippet_id, segment, expected_epoch=ended_session_epoch + if ( + latest_snippet_id is not None + and ( + worker is ended_session_worker + or (session_open and worker not in session_workers_seen) + ) + and transcription_history.extend_entry( + latest_snippet_id, segment, expected_epoch=ended_session_epoch + ) ): return # Otherwise the late segments are the session's only output - # and form their own snippet. + # and form their own snippet, which its worker keeps owning. snippet_id = transcription_history.add(segment, expected_epoch=ended_session_epoch) if snippet_id is not None: latest_snippet_id = snippet_id + ended_session_worker = worker + session_workers_seen.add(worker) def text_callback_wrapper(text: str) -> None: """Bridge between speech engine text events and the text injector. @@ -807,13 +853,14 @@ def main(): transcription history as a single snippet. """ nonlocal session_open, session_worker, latest_snippet_id - nonlocal ended_session_epoch, session_started_floor + nonlocal ended_session_epoch, ended_session_worker, session_started_floor if state in (RecognitionState.IDLE, RecognitionState.ERROR): if state == RecognitionState.IDLE: action_handler.set_last_injected_text("") last_injected["to_pad"] = False with session_lock: session_open = False + closing_worker = session_worker session_worker = None ended_session_epoch = transcription_history.epoch # Only segments captured after the last clear() join the @@ -832,10 +879,12 @@ def main(): latest_snippet_id = transcription_history.add( joined, expected_epoch=ended_session_epoch ) + ended_session_worker = closing_worker else: with session_lock: if not session_open: session_open = True + leftover_worker = session_worker session_worker = None session_started_floor = time.monotonic() # Segments left over by a session that ended without a @@ -850,8 +899,9 @@ def main(): for text, started_at in session_segments if started_at > cleared_at ), - expected_epoch=transcription_history.epoch, + expected_epoch=ended_session_epoch, ) + ended_session_worker = leftover_worker session_segments.clear() # Connect speech recognition to text injection and action handling. @@ -874,10 +924,18 @@ def main(): if not action_handler.last_injected_text: return True if last_injected["to_pad"]: - # The pad's own segment bookkeeping wins over the recorded - # text: a pad "undo" may have popped that segment, leaving - # its length stale. - target = dictation_pad.last_segment or action_handler.last_injected_text + # Only the pad's own segment bookkeeping may size the + # deletion: a manual edit or a pad "undo" blurs the + # boundaries (last_segment is None), and falling back to + # the recorded text's length could erase characters the + # user typed after dictating. Refuse rather than + # misdelete — the tracking is still cleared so a repeated + # command cannot retry the stale length. + target = dictation_pad.last_segment + if target is None: + action_handler.set_last_injected_text("") + last_injected["to_pad"] = False + return True deleted = dictation_pad.delete_last_chars(len(target)) if deleted: action_handler.set_last_injected_text("") diff --git a/src/vocalinux/speech_recognition/engines/faster_whisper_engine.py b/src/vocalinux/speech_recognition/engines/faster_whisper_engine.py index 478a5c7..9941d4a 100644 --- a/src/vocalinux/speech_recognition/engines/faster_whisper_engine.py +++ b/src/vocalinux/speech_recognition/engines/faster_whisper_engine.py @@ -39,6 +39,7 @@ class _WhisperModel(Protocol): beam_size: int, best_of: int, condition_on_previous_text: bool, + initial_prompt: Optional[str], ) -> tuple[Iterable[_Segment], object]: """Transcribe audio and return segments with metadata.""" ... @@ -130,7 +131,12 @@ class FasterWhisperEngine: return "en" return language - def transcribe(self, audio_buffer: list[bytes], language: Optional[str] = None) -> str: + def transcribe( + self, + audio_buffer: list[bytes], + language: Optional[str] = None, + initial_prompt: Optional[str] = None, + ) -> str: """Transcribe the provided audio buffer. Args: @@ -139,6 +145,7 @@ class FasterWhisperEngine: (#805): a one-shot dictation binds its session language here so a later restore cannot rewrite it mid-flight. None uses ``self.language``. + initial_prompt: Optional vocabulary prompt used to bias recognition. Returns: Recognized text. @@ -162,6 +169,7 @@ class FasterWhisperEngine: beam_size=5, best_of=5, condition_on_previous_text=False, + initial_prompt=initial_prompt, ) text_parts = [segment.text.strip() for segment in segments if segment.text] diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index f31c329..9fe5fbb 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -81,6 +81,9 @@ from ..version import __version__ from .command_processor import CommandProcessor from .silero_vad import SILERO_CHUNK_SIZE, load_silero_vad +if TYPE_CHECKING: + from ..custom_dictionary import CustomDictionaryManager + class _AudioSegment(list): """A queued audio segment that knows when its capture began. @@ -400,6 +403,10 @@ class SpeechRecognitionManager: # change a queued segment's language (#805). self._session_language: Optional[str] = None self.stop_sound_guard_ms = kwargs.get("stop_sound_guard_ms", 200) + self.dictionary_manager: Optional["CustomDictionaryManager"] = kwargs.get( + "dictionary_manager" + ) + self._vosk_dictionary_warned = False self.state = RecognitionState.IDLE self.audio_thread = None self.recognition_thread = None @@ -837,6 +844,36 @@ class SpeechRecognitionManager: self._faster_whisper_engine.language = previous logger.debug(f"Restored dictation language {previous} after one-shot") + def _get_dictionary_prompt(self) -> Optional[str]: + """Return a live custom-terms prompt without interrupting dictation on errors.""" + if self.dictionary_manager is None: + return None + try: + return self.dictionary_manager.build_initial_prompt() + except (OSError, TypeError, ValueError) as error: + logger.warning("Could not build custom terms prompt: %s", error) + return None + + def _get_whispercpp_prompt(self) -> Optional[str]: + """Combine the Advanced prompt with the live custom terms prompt. + + The explicit Advanced prompt remains first and custom terms are appended, + so enabling custom dictionary support never discards user configuration. + """ + parts = [self.whispercpp_initial_prompt.strip(), self._get_dictionary_prompt() or ""] + prompt = " ".join(part for part in parts if part) + return prompt or None + + def _apply_dictionary_corrections(self, text: str) -> str: + """Apply live corrections before voice-command interpretation.""" + if self.dictionary_manager is None: + return text + try: + return self.dictionary_manager.apply_corrections(text) + except (OSError, TypeError, ValueError) as error: + logger.warning("Could not apply custom dictionary corrections: %s", error) + return text + def _init_vosk(self): """Initialize the VOSK speech recognition engine.""" # VOSK doesn't support auto-detect, so fall back to en-us for "auto" @@ -884,6 +921,15 @@ class SpeechRecognitionManager: self.recognizer = KaldiRecognizer(self.model, 16000) self._model_initialized = True logger.info("VOSK engine initialized successfully.") + if ( + self.dictionary_manager is not None + and self.dictionary_manager.terms_enabled() + and not self._vosk_dictionary_warned + ): + self._vosk_dictionary_warned = True + logger.warning( + "Custom terms are ignored by VOSK; transcript corrections still apply." + ) except ImportError: logger.error("Failed to import VOSK. Please install it with 'pip install vosk'") @@ -1067,6 +1113,7 @@ class SpeechRecognitionManager: temperature=0.0, # Greedy decoding for consistency no_speech_threshold=0.6, fp16=use_fp16, # Explicitly set to avoid warning on CPU + initial_prompt=self._get_dictionary_prompt(), ) text = result.get("text", "").strip() @@ -1288,7 +1335,9 @@ class SpeechRecognitionManager: return "" return self._faster_whisper_engine.transcribe( - audio_buffer, language=self._dictation_language(language) + 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) @@ -1787,7 +1836,11 @@ class SpeechRecognitionManager: # Transcribe with whisper.cpp # pywhispercpp expects audio as numpy array transcribe_start = time.time() - segments = self.model.transcribe(audio_float, language=lang) + transcribe_kwargs = {"language": lang} + # pywhispercpp reuses native parameter state. Passing an empty + # value explicitly clears a prompt that was active last segment. + transcribe_kwargs["initial_prompt"] = self._get_whispercpp_prompt() or "" + segments = self.model.transcribe(audio_float, **transcribe_kwargs) transcribe_duration = time.time() - transcribe_start # Extract text from segments, filtering non-speech tokens @@ -3221,6 +3274,18 @@ class SpeechRecognitionManager: # stop_recognition owns the final buffer. Do not finish or allow a # new session until key release has completed that handoff. self._capture_finished.wait() + # The worker can exit during the bounded audio-thread join in + # stop_recognition, leaving the released key's final buffer — and + # any stragglers it never consumed — queued behind the stop + # sentinel. Transcribe what is still here before replacing the + # queue so the tail of the recording is not silently dropped. + # A failed capture or an explicit cancel still discards it. + if ( + self.model_ready + and not self._buffered_capture_failed + and not self._cancel_buffered_session.is_set() + ): + self._transcribe_queued_segments() with self._buffer_lock: self.audio_buffer = [] self._segment_queue = queue.Queue(maxsize=32) @@ -3656,6 +3721,9 @@ class SpeechRecognitionManager: # Process text - either with voice commands or pass through directly logger.debug(f"_process_audio_buffer got text='{text[:50] if text else '(empty)'}...'") if text: + # Corrections must precede command parsing: a safe replacement can + # stop a model mishearing from triggering a destructive action. + text = self._apply_dictionary_corrections(text) if self._voice_commands_enabled: # Process with voice commands (original behavior) processed_text, actions = self.command_processor.process_text(text) @@ -3782,6 +3850,30 @@ class SpeechRecognitionManager: except queue.Empty: logger.debug("Recognition queue emptied before stop signal") + def _transcribe_queued_segments(self) -> None: + """Transcribe audio segments still queued after the worker exited. + + Runs on the reload thread once key release has finished its handoff, + so segments that missed the worker's final drain — the released + key's tail buffer queued during its bounded join — are still + transcribed instead of being thrown away with the queue. + """ + while True: + try: + queued = self._segment_queue.get_nowait() + except queue.Empty: + return + if queued is None: + continue + # Queue items are (segment, language) tuples stamped at enqueue + # time — the language must reach the decoder or the buffer is + # transcribed under the wrong language. + segment, segment_language = queued + try: + self._process_audio_buffer(segment, segment_language) + except (ChecksumError, ImportError, OSError, RuntimeError, ValueError): + logger.exception("Failed to transcribe a leftover buffered segment") + def reconfigure( self, engine: Optional[str] = None, diff --git a/src/vocalinux/ui/config_manager.py b/src/vocalinux/ui/config_manager.py index 976a767..be6fe38 100644 --- a/src/vocalinux/ui/config_manager.py +++ b/src/vocalinux/ui/config_manager.py @@ -207,6 +207,12 @@ DEFAULT_CONFIG = { "enabled": True, # Keep recent dictation snippets in the tray menu "max_items": 10, # How many snippets to retain (in-memory only, cleared on quit) }, + "dictionary": { + # Preserve the custom-dictionary contract used by the accessibility scanner. + "enabled": False, + "file_path": os.path.join(CONFIG_DIR, "dictionary.txt"), + "max_words": 200, + }, "advanced": { "power_user_mode": False, "debug_logging": False, diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index e4ddddd..c264b5a 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -17,6 +17,7 @@ UX Design Notes: - Modal dialog for model downloads (explicit confirmation for large downloads) """ +import itertools import logging import math import os @@ -35,6 +36,7 @@ gi.require_version("Gdk", "3.0") from gi.repository import Gdk, GLib, GObject, Gtk, Pango # noqa: E402 from ..common_types import RecognitionState, _EvdevCaptureDevice # noqa: E402 +from ..custom_dictionary import normalize_corrections # noqa: E402 from ..gateway_embed import GatewayStatus, get_gateway_embed_manager # noqa: E402 from ..speech_recognition.silero_vad import is_silero_available # noqa: E402 from ..utils import parakeet_model_info as parakeet # noqa: E402 @@ -480,6 +482,12 @@ _ACTION_WIDTH = 96 _SPIN_WIDTH = 88 _ICON_BUTTON_WIDTH = 36 _PAIRED_COMBO_WIDTH = _CONTROL_WIDTH - _ICON_BUTTON_WIDTH - 8 +# Term rows are built on the UI thread, so a scanner-managed file with +# thousands of entries is capped and the remainder summarized in one row. +_MAX_TERMS_DISPLAYED = 500 +# Rows appended per GLib idle slice: building thousands at once stalls the +# UI the same way an uncapped list would. +_TERMS_ROWS_PER_IDLE = 150 def _style_combo(combo: Gtk.ComboBox, width: int = _CONTROL_WIDTH) -> Gtk.ComboBox: @@ -947,6 +955,21 @@ SETTINGS_CSS = """ border: 1px solid alpha(@borders, 0.5); } +/* Keep child list backgrounds inside the card's rounded lower corners. */ +.preferences-group-list { + background-color: transparent; + border-radius: 0 0 11px 11px; +} + +.preferences-group-list > row:last-child, +.preferences-group-list > row:last-child:hover { + border-radius: 0 0 11px 11px; +} + +.dictionary-entry-list { + background-color: transparent; +} + .preferences-group-title { font-weight: bold; font-size: 0.9em; @@ -2121,6 +2144,7 @@ class PreferencesGroup(Gtk.Box): # Content area with listbox for rows self.listbox = Gtk.ListBox() + self.listbox.get_style_context().add_class("preferences-group-list") self.listbox.set_selection_mode(Gtk.SelectionMode.NONE) self.listbox.set_activate_on_single_click(False) self.pack_start(self.listbox, False, False, 0) @@ -2587,6 +2611,7 @@ class SettingsDialog(Gtk.Dialog): self.config_manager = config_manager self.speech_engine = speech_engine + self.dictionary_manager = getattr(speech_engine, "dictionary_manager", None) self.shortcut_update_callback = shortcut_update_callback self.update_status_callback = update_status_callback self.overlay_enabled_callback = overlay_enabled_callback @@ -2675,6 +2700,7 @@ class SettingsDialog(Gtk.Dialog): # every distro (some system-monitor icons only ship with Yaru). self._pages = [ SettingsPage("dictation", "Dictation", "input-keyboard-symbolic"), + SettingsPage("dictionary", "Custom Dictionary", "accessories-dictionary-symbolic"), SettingsPage("model", "Speech Model", "audio-input-microphone-symbolic"), SettingsPage("audio", "Audio", "audio-speakers-symbolic"), SettingsPage("performance", "Performance", "power-profile-performance-symbolic"), @@ -2688,6 +2714,7 @@ class SettingsDialog(Gtk.Dialog): self.dictation_page = pages_by_name["dictation"] self.shortcuts_tab = self.dictation_page.box self.recognition_settings_tab = self.dictation_page.box + self.dictionary_tab = pages_by_name["dictionary"].box self.speech_engine_tab = pages_by_name["model"].box self.audio_tab = pages_by_name["audio"].box self.power_tab = pages_by_name["performance"].box @@ -2711,7 +2738,10 @@ class SettingsDialog(Gtk.Dialog): self.settings_stack.set_transition_duration(120) self.settings_stack.set_hexpand(True) for page in self._pages: - self.settings_stack.add_titled(_scrollable(page.box), page.name, page.title) + # The dictionary page owns two independently scrollable management + # panes so its pane chooser stays visible above long term lists. + content = page.box if page.name == "dictionary" else _scrollable(page.box) + self.settings_stack.add_titled(content, page.name, page.title) self.settings_stack.add_named(self._build_search_empty_page(), "search-empty") self.settings_stack.connect("notify::visible-child", self._on_settings_page_changed) @@ -2748,6 +2778,23 @@ class SettingsDialog(Gtk.Dialog): # Search state: baseline visibility snapshot while a query is active. self._search_baseline = None self._search_previous_page = None + self._search_previous_dictionary_pane = None + # Marks dictionary pane switches made by the search filter itself so a + # user-initiated switch can drop the saved pre-search pane. + self._search_pane_programmatic = False + # Custom-terms list stays capped for GTK row count until the user + # asks to render the whole file. + self._show_all_terms = False + # Bumps on every refresh so idle-sliced row builders left over from + # an older pass stop instead of re-adding cleared rows. + self._terms_build_token = 0 + # Terms added in this dialog that land past the display cap; they are + # pinned to the top so a successful save is never invisible. + self._pinned_terms: list[str] = [] + # A term saved under "show all" scrolls into view once the idle-sliced + # rebuild finishes: (row index, displayed term) so a stale target left + # by a superseded refresh can never jump to the wrong row. + self._scroll_terms_to_row: Optional[tuple[int, str]] = None # Set content_box to speech_engine_tab for backward compatibility self.content_box = self.speech_engine_tab @@ -2757,6 +2804,7 @@ class SettingsDialog(Gtk.Dialog): self._build_language_shortcuts_section() self._build_recognition_section() self._build_simple_model_section() + self._build_dictionary_section() self._build_engine_section() self._build_remote_server_section() self._build_gateway_embed_section() @@ -2964,6 +3012,9 @@ class SettingsDialog(Gtk.Dialog): visible_page = self.settings_stack.get_visible_child_name() page_names = {page.name for page in self._pages} self._search_previous_page = visible_page if visible_page in page_names else "dictation" + self._search_previous_dictionary_pane = ( + self.dictionary_management_stack.get_visible_child_name() + ) baseline = {"rows": {}, "groups": {}, "extras": {}, "islands": {}} for page in self._pages: @@ -3018,6 +3069,27 @@ class SettingsDialog(Gtk.Dialog): self.settings_stack.set_visible_child_name(page.name) self._search_previous_page = None + if self._search_previous_dictionary_pane is not None: + # A search may have auto-switched the dictionary pane to reveal a + # match; clearing the query returns to the pane the user selected. + self.dictionary_management_stack.set_visible_child_name( + self._search_previous_dictionary_pane + ) + self._search_previous_dictionary_pane = None + + def _on_dictionary_pane_changed(self, *_args: Any) -> None: + """Keep a manual dictionary pane switch made during an active search. + + The filter auto-switches panes to reveal a match and restores the + pre-search pane when the query clears. If the user clicks the pane + switcher themselves mid-search, that newer choice wins and clearing + the search must not roll it back. + """ + if self._search_pane_programmatic: + return + if self._search_baseline is not None: + self._search_previous_dictionary_pane = None + def _on_search_changed(self, entry): """Live-filter settings rows across all pages.""" query = entry.get_text().strip() @@ -3057,6 +3129,21 @@ class SettingsDialog(Gtk.Dialog): for extra in page.extras: extra.hide() + if page.name == "dictionary": + # A search can hide the selected management pane while the + # other pane contains the match. Keep matching controls visible. + current = self.dictionary_management_stack.get_visible_child_name() + terms_visible = self.dictionary_terms_group.get_visible() + corrections_visible = self.dictionary_corrections_group.get_visible() + self._search_pane_programmatic = True + try: + if current == "terms" and not terms_visible and corrections_visible: + self.dictionary_management_stack.set_visible_child_name("corrections") + elif current == "corrections" and not corrections_visible and terms_visible: + self.dictionary_management_stack.set_visible_child_name("terms") + finally: + self._search_pane_programmatic = False + if page_matches > 0: if page.update_badge_label is not None: page.update_badge_label.hide() @@ -3292,6 +3379,546 @@ class SettingsDialog(Gtk.Dialog): self.history_enabled_switch.connect("state-set", self._on_history_enabled_toggled) self.history_max_items_spin.connect("value-changed", self._on_history_max_items_changed) + def _build_dictionary_section(self) -> None: + """Build the Custom Dictionary page for terms and transcript corrections.""" + self.dictionary_management_stack = Gtk.Stack() + self.dictionary_management_stack.set_transition_type(Gtk.StackTransitionType.CROSSFADE) + self.dictionary_management_stack.set_transition_duration(120) + self.dictionary_management_stack.set_homogeneous(False) + self.dictionary_management_stack.set_hexpand(True) + self.dictionary_management_stack.set_vexpand(True) + self.dictionary_management_stack.connect( + "notify::visible-child-name", self._on_dictionary_pane_changed + ) + + self.dictionary_management_switcher = Gtk.StackSwitcher() + self.dictionary_management_switcher.set_stack(self.dictionary_management_stack) + self.dictionary_management_switcher.set_halign(Gtk.Align.CENTER) + self.dictionary_management_switcher.get_accessible().set_name("Custom dictionary section") + self.dictionary_tab.pack_start(self.dictionary_management_switcher, False, False, 0) + + self.dictionary_feedback_label = Gtk.Label(xalign=0) + self.dictionary_feedback_label.set_line_wrap(True) + self.dictionary_feedback_label.get_accessible().set_name("Custom dictionary status") + self.dictionary_feedback_label.get_style_context().add_class("tip-label") + self.dictionary_tab.pack_start(self.dictionary_feedback_label, False, False, 0) + + self.dictionary_tab.pack_start(self.dictionary_management_stack, True, True, 0) + + terms_group = PreferencesGroup( + title="Custom terms", + description=( + "Terms bias Whisper, whisper.cpp, and Faster Whisper recognition. The UTF-8 " + "terms file is live-reloaded and has one term per line, so accessibility tools " + "can edit it." + ), + keywords=("dictionary", "terms", "vocabulary", "whisper", "scanner"), + ) + self.dictionary_terms_group = terms_group + self.dictionary_terms_enabled_switch = Gtk.Switch() + terms_group.add_row( + PreferenceRow( + title="Use custom terms", + subtitle=( + "Vocabulary bias works with Whisper, whisper.cpp, and Faster Whisper; " + "corrections work with every engine." + ), + widget=self.dictionary_terms_enabled_switch, + ) + ) + terms_path_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6) + self.dictionary_terms_path_entry = Gtk.Entry() + self.dictionary_terms_path_entry.set_hexpand(True) + self.dictionary_terms_path_entry.set_placeholder_text("~/.config/vocalinux/dictionary.txt") + self.dictionary_terms_path_entry.set_tooltip_text( + "UTF-8 line file used for custom terms and the accessibility scanner" + ) + self.dictionary_terms_path_entry.get_accessible().set_name("Custom terms file") + terms_path_box.pack_start(self.dictionary_terms_path_entry, True, True, 0) + self.dictionary_terms_file_button = Gtk.FileChooserButton(title="Choose Terms File") + terms_path_box.pack_start(self.dictionary_terms_file_button, False, False, 0) + terms_group.add_row( + PreferenceRow( + title="Terms file", + subtitle="UTF-8, one term per line; # starts a comment", + widget=terms_path_box, + ) + ) + self.dictionary_terms_status_label = Gtk.Label(xalign=0) + self.dictionary_terms_status_label.set_line_wrap(True) + self.dictionary_terms_status_label.get_style_context().add_class("tip-label") + terms_group.add_row( + PreferenceRow( + title="Live status", + subtitle="The terms file is re-read before every transcription.", + widget=self.dictionary_terms_status_label, + ) + ) + + terms_add_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + terms_add_box.set_margin_top(8) + terms_add_box.set_margin_bottom(4) + terms_add_box.set_margin_start(16) + terms_add_box.set_margin_end(16) + self.dictionary_term_entry = Gtk.Entry() + self.dictionary_term_entry.set_placeholder_text("Add a term") + self.dictionary_term_entry.set_tooltip_text( + "Term to add to the scanner-friendly terms file" + ) + self.dictionary_term_entry.get_accessible().set_name("Custom term") + self.dictionary_term_entry.set_hexpand(True) + terms_add_box.pack_start(self.dictionary_term_entry, True, True, 0) + self.dictionary_add_term_button = Gtk.Button(label="Add term") + self.dictionary_add_term_button.connect("clicked", self._on_dictionary_add_term) + self.dictionary_term_entry.connect("activate", self._on_dictionary_add_term) + terms_add_box.pack_start(self.dictionary_add_term_button, False, False, 0) + terms_add_row = Gtk.ListBoxRow() + terms_add_row.set_activatable(False) + terms_add_row.add(terms_add_box) + terms_group.add_row(terms_add_row) + + self.dictionary_terms_listbox = Gtk.ListBox() + self.dictionary_terms_listbox.get_style_context().add_class("dictionary-entry-list") + self.dictionary_terms_listbox.set_selection_mode(Gtk.SelectionMode.NONE) + self.dictionary_terms_listbox.set_placeholder( + Gtk.Label(label="No custom terms yet.", xalign=0.5) + ) + terms_list_row = Gtk.ListBoxRow() + terms_list_row.set_activatable(False) + terms_list_row.add(self.dictionary_terms_listbox) + terms_group.add_row(terms_list_row) + + self.dictionary_terms_scroller = Gtk.ScrolledWindow() + self.dictionary_terms_scroller.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC) + self.dictionary_terms_scroller.set_shadow_type(Gtk.ShadowType.NONE) + self.dictionary_terms_scroller.add(terms_group) + self.dictionary_management_stack.add_titled( + self.dictionary_terms_scroller, "terms", "Custom terms" + ) + + corrections_group = PreferencesGroup( + title="Transcript corrections", + description=( + "Corrections replace a misheard whole word or phrase in every completed " + "transcript. They run before voice commands, so avoid replacement text that " + "is itself a voice command." + ), + keywords=("dictionary", "correction", "replacement", "misheard", "transcript"), + ) + self.dictionary_corrections_group = corrections_group + corrections_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + corrections_box.set_margin_top(8) + corrections_box.set_margin_bottom(4) + corrections_box.set_margin_start(16) + corrections_box.set_margin_end(16) + self.dictionary_heard_entry = Gtk.Entry() + self.dictionary_heard_entry.set_placeholder_text("Heard as, e.g. super base") + self.dictionary_heard_entry.set_tooltip_text("Misheard phrase, matched case-insensitively") + self.dictionary_heard_entry.get_accessible().set_name("Heard phrase") + self.dictionary_heard_entry.set_hexpand(True) + corrections_box.pack_start(self.dictionary_heard_entry, True, True, 0) + self.dictionary_replacement_entry = Gtk.Entry() + self.dictionary_replacement_entry.set_placeholder_text("Replace with, e.g. Supabase") + self.dictionary_replacement_entry.set_tooltip_text( + "Replacement inserted with exactly this spelling" + ) + self.dictionary_replacement_entry.get_accessible().set_name("Replacement text") + self.dictionary_replacement_entry.set_hexpand(True) + corrections_box.pack_start(self.dictionary_replacement_entry, True, True, 0) + self.dictionary_add_correction_button = Gtk.Button(label="Add correction") + self.dictionary_add_correction_button.connect("clicked", self._on_dictionary_add_correction) + self.dictionary_heard_entry.connect("activate", self._on_dictionary_add_correction) + self.dictionary_replacement_entry.connect("activate", self._on_dictionary_add_correction) + corrections_box.pack_start(self.dictionary_add_correction_button, False, False, 0) + corrections_add_row = Gtk.ListBoxRow() + corrections_add_row.set_activatable(False) + corrections_add_row.add(corrections_box) + corrections_group.add_row(corrections_add_row) + + self.dictionary_corrections_listbox = Gtk.ListBox() + self.dictionary_corrections_listbox.get_style_context().add_class("dictionary-entry-list") + self.dictionary_corrections_listbox.set_selection_mode(Gtk.SelectionMode.NONE) + self.dictionary_corrections_listbox.set_placeholder( + Gtk.Label(label="No transcript corrections yet.", xalign=0.5) + ) + corrections_list_row = Gtk.ListBoxRow() + corrections_list_row.set_activatable(False) + corrections_list_row.add(self.dictionary_corrections_listbox) + corrections_group.add_row(corrections_list_row) + + corrections_scroller = Gtk.ScrolledWindow() + corrections_scroller.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC) + corrections_scroller.set_shadow_type(Gtk.ShadowType.NONE) + corrections_scroller.add(corrections_group) + self.dictionary_management_stack.add_titled( + corrections_scroller, "corrections", "Corrections" + ) + + self.dictionary_terms_enabled_switch.connect("state-set", self._on_dictionary_terms_enabled) + self.dictionary_terms_path_entry.connect("activate", self._on_dictionary_terms_path_changed) + self.dictionary_terms_path_entry.connect( + "focus-out-event", self._on_dictionary_terms_path_changed + ) + self.dictionary_terms_file_button.connect("file-set", self._on_dictionary_terms_file_chosen) + + def _on_dictionary_terms_enabled(self, widget: Any, state: bool) -> bool: + """Persist terms enablement and restore the control when persistence fails.""" + if self._initializing or self._applying_settings or self.dictionary_manager is None: + return False + if not self.dictionary_manager.set_terms_enabled(bool(state)): + self.dictionary_feedback_label.set_text("Could not save custom terms setting.") + self._refresh_dictionary_ui() + return False + + def _on_dictionary_terms_path_changed(self, widget: Any, *args: Any) -> bool: + """Persist a validated custom terms path and restore it on failure.""" + if self._initializing or self._applying_settings or self.dictionary_manager is None: + return False + if self.dictionary_manager.set_terms_path(self.dictionary_terms_path_entry.get_text()): + self.dictionary_feedback_label.set_text("Custom terms path saved.") + else: + self.dictionary_feedback_label.set_text( + "Could not save that custom terms path; keeping the previous path." + ) + self._refresh_dictionary_ui() + return False + + def _on_dictionary_terms_file_chosen(self, widget: Any) -> None: + """Apply a terms path selected through the GTK file chooser.""" + path = widget.get_filename() + if path: + self.dictionary_terms_path_entry.set_text(path) + self._on_dictionary_terms_path_changed(self.dictionary_terms_path_entry) + + def _on_dictionary_add_term(self, widget: Any) -> None: + """Add a term to the fixed line file, reporting an observable result.""" + if self._initializing or self._applying_settings or self.dictionary_manager is None: + return + term = self.dictionary_term_entry.get_text().strip() + if not term: + self.dictionary_feedback_label.set_text("Enter a term before adding it.") + return + terms = self.dictionary_manager.get_terms() + if any(existing.casefold() == term.casefold() for existing in terms): + self.dictionary_feedback_label.set_text("That term is already in the terms file.") + return + if not self.dictionary_manager.add_term(term): + self.dictionary_feedback_label.set_text( + "Could not save the terms file; no term was added." + ) + return + self.dictionary_term_entry.set_text("") + terms_now = self.dictionary_manager.get_terms() + if any(existing.casefold() == term.casefold() for existing in terms_now): + self.dictionary_feedback_label.set_text("Term saved to the live terms file.") + if self._show_all_terms: + # Oversized or yield-capped files prepend the term, other + # files append it — scroll to wherever it landed once the + # idle-sliced rebuild finishes. Under "show all" the rebuilt + # rows match terms_now one to one. + new_index = next( + index + for index, existing in enumerate(terms_now) + if existing.casefold() == term.casefold() + ) + self._scroll_terms_to_row = (new_index, terms_now[new_index]) + else: + # A term that lands past the display cap is pinned to the top + # of the capped list — a saved term is never invisible. The + # next refresh drops the pin if the term is already visible. + self._pinned_terms.append(term) + else: + self.dictionary_feedback_label.set_text("That term is not valid for the terms file.") + self._refresh_dictionary_ui() + + def _on_terms_show_all(self, widget: Any) -> None: + """Render every term, including those past the display cap.""" + self._show_all_terms = True + self._refresh_dictionary_ui() + + def _on_terms_show_fewer(self, widget: Any) -> None: + """Collapse the terms list back to the display cap.""" + self._show_all_terms = False + self._refresh_dictionary_ui() + + def _on_dictionary_remove_term(self, widget: Any, term: str) -> None: + """Remove one term from the fixed line file.""" + if self._initializing or self._applying_settings or self.dictionary_manager is None: + return + if self.dictionary_manager.remove_term(term): + self.dictionary_feedback_label.set_text("Term removed from the live terms file.") + else: + self.dictionary_feedback_label.set_text( + "Could not save the terms file; the term was not removed." + ) + self._refresh_dictionary_ui() + + def _on_dictionary_add_correction(self, widget: Any) -> None: + """Add or update a phrase correction in the structured corrections file.""" + if self._initializing or self._applying_settings or self.dictionary_manager is None: + return + heard = self.dictionary_heard_entry.get_text().strip() + replacement = self.dictionary_replacement_entry.get_text().strip() + if not heard or not replacement: + self.dictionary_feedback_label.set_text( + "Enter both the heard phrase and its replacement." + ) + return + entries = self.dictionary_manager.get_corrections_for_edit() + if entries is None: + self.dictionary_feedback_label.set_text( + "Could not edit corrections: fix or replace the malformed corrections file first." + ) + return + candidate = {"heard": heard, "replacement": replacement} + normalized_entries = normalize_corrections([candidate]) + if not normalized_entries: + self.dictionary_feedback_label.set_text("That correction is not valid for the file.") + return + candidate = normalized_entries[0] + heard_key = candidate["heard"].casefold() + existing = [entry for entry in entries if entry["heard"].casefold() == heard_key] + entries = [entry for entry in entries if entry["heard"].casefold() != heard_key] + entries.append(candidate) + if not self.dictionary_manager.save_corrections(entries): + self.dictionary_feedback_label.set_text( + "Could not save corrections; no correction was changed." + ) + return + self.dictionary_heard_entry.set_text("") + self.dictionary_replacement_entry.set_text("") + if any( + entry["heard"].casefold() == heard_key + for entry in self.dictionary_manager.get_corrections() + ): + self.dictionary_feedback_label.set_text( + "Correction updated." + if existing + else "Correction saved to the live corrections file." + ) + else: + self.dictionary_feedback_label.set_text("That correction is not valid for the file.") + self._refresh_dictionary_ui() + + def _on_dictionary_remove_correction(self, widget: Any, heard: str) -> None: + """Remove one correction from the structured corrections file.""" + if self._initializing or self._applying_settings or self.dictionary_manager is None: + return + editable_entries = self.dictionary_manager.get_corrections_for_edit() + if editable_entries is None: + self.dictionary_feedback_label.set_text( + "Could not edit corrections: fix or replace the malformed corrections file first." + ) + return + entries = [ + entry for entry in editable_entries if entry["heard"].casefold() != heard.casefold() + ] + if self.dictionary_manager.save_corrections(entries): + self.dictionary_feedback_label.set_text("Correction removed.") + else: + self.dictionary_feedback_label.set_text( + "Could not save corrections; nothing was removed." + ) + self._refresh_dictionary_ui() + + def _make_term_row(self, term: str, transient: bool) -> Gtk.ListBoxRow: + """Build one removable term row for the custom terms list.""" + row = Gtk.ListBoxRow() + row.set_activatable(False) + row_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + row_box.set_margin_top(6) + row_box.set_margin_bottom(6) + row_box.set_margin_start(16) + row_box.set_margin_end(16) + label = Gtk.Label(label=term, xalign=0) + label.set_hexpand(True) + row_box.pack_start(label, True, True, 0) + remove_button = Gtk.Button(label="Remove") + remove_button.set_tooltip_text(f"Remove term {term}") + remove_button.get_accessible().set_name(f"Remove term {term}") + remove_button.set_sensitive(not transient) + remove_button.connect("clicked", self._on_dictionary_remove_term, term) + row_box.pack_start(remove_button, False, False, 0) + row.add(row_box) + return row + + def _build_term_rows( + self, terms: list[str], total_terms: int, transient: bool, token: int + ) -> None: + """Populate the terms listbox in idle slices of bounded size. + + Building thousands of GTK rows synchronously stalls the whole + dialog, so each slice appends at most ``_TERMS_ROWS_PER_IDLE`` rows + and reschedules itself. Every refresh bumps the build token, which + retires pending slices from an older pass instead of letting them + re-add rows the refresh already cleared. + """ + pending = iter(terms) + + def build_chunk() -> bool: + if token != self._terms_build_token: + return False + batch = list(itertools.islice(pending, _TERMS_ROWS_PER_IDLE)) + for term in batch: + self.dictionary_terms_listbox.add(self._make_term_row(term, transient)) + self.dictionary_terms_listbox.show_all() + if len(batch) == _TERMS_ROWS_PER_IDLE: + return True + self._finish_term_rows(total_terms) + return False + + # The first slice runs inline so small lists appear immediately; + # a slice never exceeds _TERMS_ROWS_PER_IDLE rows of work. + if build_chunk(): + GLib.idle_add(build_chunk) + + def _finish_term_rows(self, total_terms: int) -> None: + """Append the overflow summary row after the last term slice.""" + if total_terms > _MAX_TERMS_DISPLAYED: + overflow_row = Gtk.ListBoxRow() + overflow_row.set_activatable(False) + overflow_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + overflow_box.set_margin_top(6) + overflow_box.set_margin_bottom(6) + overflow_box.set_margin_start(16) + overflow_box.set_margin_end(16) + if self._show_all_terms: + overflow_text = f"Showing all {total_terms} terms." + toggle_button = Gtk.Button(label=f"Show first {_MAX_TERMS_DISPLAYED}") + toggle_button.connect("clicked", self._on_terms_show_fewer) + else: + overflow_text = f"Showing the first {_MAX_TERMS_DISPLAYED} of {total_terms} terms." + toggle_button = Gtk.Button(label=f"Show all {total_terms}") + toggle_button.connect("clicked", self._on_terms_show_all) + overflow_label = Gtk.Label(label=overflow_text, xalign=0) + overflow_label.set_line_wrap(True) + overflow_label.set_hexpand(True) + overflow_label.get_style_context().add_class("tip-label") + overflow_box.pack_start(overflow_label, True, True, 0) + overflow_box.pack_start(toggle_button, False, False, 0) + overflow_row.add(overflow_box) + self.dictionary_terms_listbox.add(overflow_row) + self.dictionary_terms_listbox.show_all() + if self._scroll_terms_to_row is not None: + index, expected_term = self._scroll_terms_to_row + self._scroll_terms_to_row = None + row = self.dictionary_terms_listbox.get_row_at_index(index) + row_text = self._term_row_text(row) if row is not None else None + if row_text is not None and row_text.casefold() == expected_term.casefold(): + self._scroll_terms_row_into_view(row) + + @staticmethod + def _term_row_text(row: Gtk.ListBoxRow) -> Optional[str]: + """Return the term displayed by a row built by ``_make_term_row``.""" + row_box = row.get_child() + if not isinstance(row_box, Gtk.Box) or not row_box.get_children(): + return None + label = row_box.get_children()[0] + return label.get_label() if isinstance(label, Gtk.Label) else None + + def _scroll_terms_row_into_view(self, row: Gtk.ListBoxRow) -> None: + """Bring a term row inside the terms scroller's viewport once laid out.""" + retries_left = 20 + + def scroll_once() -> bool: + nonlocal retries_left + retries_left -= 1 + if row.get_parent() is not self.dictionary_terms_listbox: + return False + allocation = row.get_allocation() + if allocation.height <= 0: + # Not laid out yet — retry briefly instead of never scrolling. + return retries_left > 0 + # The row's y is relative to the terms listbox, but the scroller's + # adjustment addresses the whole card (header and controls above + # the list) — translate into that space or the scroll stops short. + viewport = self.dictionary_terms_scroller.get_child() + content = viewport.get_child() if isinstance(viewport, Gtk.Viewport) else None + position = row.translate_coordinates(content, 0, 0) if content else None + if position is None: + return False + _, row_y = position + adjustment = self.dictionary_terms_scroller.get_vadjustment() + page_size = adjustment.get_page_size() + value = adjustment.get_value() + if row_y < value: + adjustment.set_value(max(adjustment.get_lower(), row_y)) + elif row_y + allocation.height > value + page_size: + adjustment.set_value( + min( + adjustment.get_upper() - page_size, + row_y + allocation.height - page_size, + ) + ) + return False + + GLib.idle_add(scroll_once) + + def _refresh_dictionary_ui(self) -> None: + """Rebuild custom dictionary controls from the live, file-backed state.""" + if not hasattr(self, "dictionary_terms_enabled_switch"): + return + if self.dictionary_manager is None: + self.dictionary_terms_enabled_switch.set_sensitive(False) + self.dictionary_terms_status_label.set_text("Custom dictionary support is unavailable.") + return + transient = self.dictionary_manager.is_transient_terms + enabled = self.dictionary_manager.terms_enabled() + self.dictionary_terms_enabled_switch.set_active(enabled) + self.dictionary_terms_enabled_switch.set_sensitive(not transient) + self.dictionary_terms_path_entry.set_text(self.dictionary_manager.terms_path_text()) + self.dictionary_terms_path_entry.set_sensitive(not transient) + self.dictionary_terms_file_button.set_sensitive(not transient) + self.dictionary_terms_status_label.set_text( + ("Session-only --dictionary-file override is active. " if transient else "") + + self.dictionary_manager.terms_status() + ) + self.dictionary_term_entry.set_sensitive(not transient) + self.dictionary_add_term_button.set_sensitive(not transient) + + for child in list(self.dictionary_terms_listbox.get_children()): + self.dictionary_terms_listbox.remove(child) + terms = self.dictionary_manager.get_terms() + visible_terms = terms if self._show_all_terms else terms[:_MAX_TERMS_DISPLAYED] + # Recently added terms that would fall outside the visible window + # stay pinned to the top while they remain unseen; a pin expires as + # soon as the term disappears or becomes visible on its own. + visible_keys = {term.casefold() for term in visible_terms} + term_keys = {term.casefold() for term in terms} + self._pinned_terms = [ + pinned + for pinned in self._pinned_terms + if pinned.casefold() in term_keys and pinned.casefold() not in visible_keys + ] + self._terms_build_token += 1 + self._build_term_rows( + [*self._pinned_terms, *visible_terms], len(terms), transient, self._terms_build_token + ) + + for child in list(self.dictionary_corrections_listbox.get_children()): + self.dictionary_corrections_listbox.remove(child) + + for entry in self.dictionary_manager.get_corrections(): + row = Gtk.ListBoxRow() + row.set_activatable(False) + row_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + row_box.set_margin_top(6) + row_box.set_margin_bottom(6) + row_box.set_margin_start(16) + row_box.set_margin_end(16) + label = Gtk.Label(label=f"{entry['heard']} → {entry['replacement']}", xalign=0) + label.set_hexpand(True) + row_box.pack_start(label, True, True, 0) + remove_button = Gtk.Button(label="Remove") + remove_button.set_tooltip_text(f"Remove correction for {entry['heard']}") + remove_button.get_accessible().set_name(f"Remove correction for {entry['heard']}") + remove_button.connect("clicked", self._on_dictionary_remove_correction, entry["heard"]) + row_box.pack_start(remove_button, False, False, 0) + row.add(row_box) + self.dictionary_corrections_listbox.add(row) + self.dictionary_terms_listbox.show_all() + self.dictionary_corrections_listbox.show_all() + def _build_auto_pause_section(self): """Build Auto-Pause settings: enable toggle + process name list.""" group = PreferencesGroup( @@ -6670,6 +7297,7 @@ class SettingsDialog(Gtk.Dialog): self.auto_pause_switch.set_active(auto_pause_enabled) self._update_auto_pause_sensitivity(auto_pause_enabled) self._refresh_auto_pause_list() + self._refresh_dictionary_ui() keepalive_settings = self.config_manager.get_settings().get("model_keepalive", {}) keepalive_enabled = bool(keepalive_settings.get("enabled", False)) diff --git a/tests/test_custom_dictionary.py b/tests/test_custom_dictionary.py new file mode 100644 index 0000000..6cba739 --- /dev/null +++ b/tests/test_custom_dictionary.py @@ -0,0 +1,1106 @@ +"""Tests for the file-backed custom dictionary contract.""" + +import json +import re +import stat +import threading +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, Mock, patch + +from vocalinux.custom_dictionary import ( + CORRECTIONS_FILENAME, + DEFAULT_TERMS_PATH, + LEGACY_DEFAULT_TERMS_PATH, + MAX_CORRECTION_CHARACTERS, + MAX_TERMS_FILE_BYTES, + MAX_TERMS_YIELDED, + TERMS_FILENAME, + CustomDictionaryManager, + apply_corrections, + normalize_corrections, +) +from vocalinux.main import parse_arguments +from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager + + +class FakeConfig: + """Minimal ConfigManager substitute with controllable persistence.""" + + def __init__(self, values: dict[str, dict[str, Any]] | None = None, save_result: bool = True): + self.values = values or { + "dictionary": { + "enabled": False, + "file_path": DEFAULT_TERMS_PATH, + "max_words": 200, + } + } + self.save_result = save_result + + def get(self, section: str, key: str, default: Any = None) -> Any: + """Return one configured value.""" + return self.values.get(section, {}).get(key, default) + + def set(self, section: str, key: str, value: Any) -> bool: + """Set one configured value.""" + self.values.setdefault(section, {})[key] = value + return True + + def save_config(self) -> bool: + """Return the requested persistence result.""" + return self.save_result + + +def manager_at( + tmp_path: Path, monkeypatch, config: FakeConfig | None = None +) -> CustomDictionaryManager: + """Create a manager whose configured files live in pytest's temp directory.""" + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path)) + config = config or FakeConfig() + dictionary = config.values.setdefault("dictionary", {}) + if dictionary.get("file_path") in (None, DEFAULT_TERMS_PATH): + dictionary["file_path"] = str(tmp_path / TERMS_FILENAME) + return CustomDictionaryManager(config) + + +def recognition_manager( + engine: str, dictionary: CustomDictionaryManager, advanced_prompt: str = "" +) -> SpeechRecognitionManager: + """Build a recognition manager without loading a real speech model.""" + with patch.object(SpeechRecognitionManager, "_init_vosk"): + with patch.object(SpeechRecognitionManager, "_init_whisper"): + with patch.object(SpeechRecognitionManager, "_init_whispercpp"): + return SpeechRecognitionManager( + engine=engine, + dictionary_manager=dictionary, + whispercpp_initial_prompt=advanced_prompt, + ) + + +def mock_numpy() -> MagicMock: + """Return the minimum NumPy surface used by transcription helpers.""" + numpy = MagicMock() + numpy.frombuffer.return_value = MagicMock(__len__=lambda _: 16_000) + numpy.frombuffer.return_value.astype.return_value = numpy.frombuffer.return_value + numpy.int16 = "int16" + numpy.float32 = "float32" + return numpy + + +def test_terms_are_live_reloaded_and_scanner_friendly(tmp_path: Path, monkeypatch) -> None: + """The UTF-8 line file accepts comments and applies an external replacement.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + terms_file = tmp_path / TERMS_FILENAME + terms_file.write_text("# scanner comment\nVocaLinux\nvocalinux\nSupabase\n", encoding="utf-8") + + assert manager.get_terms() == ["VocaLinux", "Supabase"] + assert manager.build_initial_prompt() == "VocaLinux Supabase" + + terms_file.write_text("PyGObject\n", encoding="utf-8") + assert manager.build_initial_prompt() == "PyGObject" + + +def test_terms_save_is_atomic_shape_and_transient_override_is_not_written( + tmp_path: Path, monkeypatch +) -> None: + """UI saves standard terms safely while CLI terms remain session-only.""" + manager = manager_at(tmp_path, monkeypatch) + assert manager.save_terms(["VocaLinux", "vocalinux", "# ignored", " PyGObject "]) + assert (tmp_path / TERMS_FILENAME).read_text(encoding="utf-8") == "VocaLinux\nPyGObject\n" + + override = tmp_path / "override.txt" + override.write_text("Override\n", encoding="utf-8") + transient = CustomDictionaryManager(FakeConfig(), str(override)) + assert transient.terms_enabled() + assert transient.get_terms() == ["Override"] + assert not transient.save_terms(["No write"]) + + +def test_incremental_term_edits_preserve_comments_and_blank_lines( + tmp_path: Path, monkeypatch +) -> None: + """GTK add/remove operations retain unrelated scanner-managed source lines.""" + manager = manager_at(tmp_path, monkeypatch) + terms_file = tmp_path / TERMS_FILENAME + terms_file.write_text("# scanner note\n\nVocaLinux\n# retain me\nPyGObject\n", encoding="utf-8") + + assert manager.add_term("Supabase") + assert terms_file.read_text(encoding="utf-8") == ( + "# scanner note\n\nVocaLinux\n# retain me\nPyGObject\nSupabase\n" + ) + assert manager.remove_term("PyGObject") + assert terms_file.read_text(encoding="utf-8") == ( + "# scanner note\n\nVocaLinux\n# retain me\nSupabase\n" + ) + + +def test_enable_rollback_when_config_persistence_fails(tmp_path: Path, monkeypatch) -> None: + """A failed config save preserves the previous in-memory setting.""" + config = FakeConfig(save_result=False) + manager = manager_at(tmp_path, monkeypatch, config) + + assert not manager.set_terms_enabled(True) + assert not manager.terms_enabled() + + +def test_prompt_read_stops_at_the_terms_limit(tmp_path: Path, monkeypatch) -> None: + """Prompt assembly stops processing terms after the configured term cap.""" + config = FakeConfig({"dictionary": {"enabled": True, "max_words": 2}}) + manager = manager_at(tmp_path, monkeypatch, config) + (tmp_path / TERMS_FILENAME).write_text("one\ntwo\nthree\n", encoding="utf-8") + + assert manager.build_initial_prompt() == "one two" + + +def test_partially_invalid_terms_file_fails_closed(tmp_path: Path, monkeypatch) -> None: + """Valid lines before corrupt UTF-8 must not reach recognition.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + payload = b"one\ntwo\n" + b"x" * 9000 + b"\xff\nthree\n" + (tmp_path / TERMS_FILENAME).write_bytes(payload) + + assert manager.get_terms() == [] + assert manager.build_initial_prompt() is None + assert manager.terms_status() == "Terms file is not valid UTF-8." + + +def test_oversized_terms_file_supplies_its_leading_lines(tmp_path: Path, monkeypatch) -> None: + """A file beyond the read bound still supplies its complete leading lines.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + line_count = MAX_TERMS_FILE_BYTES // 8 + 2 # 8-byte lines plus a partial tail + (tmp_path / TERMS_FILENAME).write_bytes( + b"".join(f"t{i:06d}\n".encode() for i in range(line_count)) + ) + + terms = manager.get_terms() + assert terms + assert all(re.fullmatch(r"t\d{6}", term) for term in terms) + assert len(terms) == MAX_TERMS_YIELDED + assert manager.build_initial_prompt() is not None + assert "leading lines only" in manager.terms_status() + + +def test_prompt_on_oversized_file_streams_only_the_needed_prefix( + tmp_path: Path, monkeypatch +) -> None: + """Prompt assembly stops after max_terms instead of decoding the bound.""" + config = FakeConfig({"dictionary": {"enabled": True, "max_words": 3}}) + manager = manager_at(tmp_path, monkeypatch, config) + line_count = MAX_TERMS_FILE_BYTES // 8 + 2 + (tmp_path / TERMS_FILENAME).write_bytes( + b"".join(f"t{i:06d}\n".encode() for i in range(line_count)) + ) + + assert manager.build_initial_prompt() == "t000000 t000001 t000002" + + +def test_prompt_on_oversized_file_with_corrupt_prefix_still_fails_closed( + tmp_path: Path, monkeypatch +) -> None: + """Invalid UTF-8 inside the usable prefix supplies no prompt at all.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + line_count = MAX_TERMS_FILE_BYTES // 8 + 2 + (tmp_path / TERMS_FILENAME).write_bytes( + b"one\ntwo\n" + b"\xff\n" + b"".join(f"t{i:06d}\n".encode() for i in range(line_count)) + ) + + assert manager.build_initial_prompt() is None + + +def test_prompt_on_oversized_file_validates_past_the_term_limit( + tmp_path: Path, monkeypatch +) -> None: + """Invalid UTF-8 after the collected prompt terms still supplies none.""" + config = FakeConfig({"dictionary": {"enabled": True, "max_words": 2}}) + manager = manager_at(tmp_path, monkeypatch, config) + line_count = MAX_TERMS_FILE_BYTES // 8 + 2 + (tmp_path / TERMS_FILENAME).write_bytes( + b"one\ntwo\nthree\n" + + b"\xff\n" + + b"".join(f"t{i:06d}\n".encode() for i in range(line_count)) + ) + + assert manager.build_initial_prompt() is None + assert manager.get_terms() == [] + + +def test_prompt_on_oversized_file_with_single_huge_line_stays_bounded( + tmp_path: Path, monkeypatch +) -> None: + """A first line longer than the read bound supplies no prompt terms.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + (tmp_path / TERMS_FILENAME).write_bytes(b"x" * (MAX_TERMS_FILE_BYTES + 10) + b"\nvalid\n") + + assert manager.build_initial_prompt() is None + + +def test_add_term_on_oversized_file_stays_visible(tmp_path: Path, monkeypatch) -> None: + """Prepending past the read window keeps the new term visible while + preserving every scanner-owned line, comment, and blank line.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + line_count = MAX_TERMS_FILE_BYTES // 8 + 2 + original = b"# scanner tail\n" + b"".join(f"t{i:06d}\n".encode() for i in range(line_count)) + (tmp_path / TERMS_FILENAME).write_bytes(original) + + assert manager.add_term("brandnew") + assert (tmp_path / TERMS_FILENAME).read_bytes() == b"brandnew\n" + original + terms = manager.get_terms() + assert terms[0] == "brandnew" + assert len(terms) == MAX_TERMS_YIELDED + + +def test_add_term_on_yield_capped_file_stays_visible(tmp_path: Path, monkeypatch) -> None: + """A file at the yield cap but under the byte bound also needs the new + term prepended so Settings and the prompt can see it.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + original = "".join(f"t{i:05d}\n" for i in range(MAX_TERMS_YIELDED)) + terms_file = tmp_path / TERMS_FILENAME + terms_file.write_text(original, encoding="utf-8") + + assert manager.add_term("brandnew") + assert terms_file.read_text(encoding="utf-8") == f"brandnew\n{original}" + terms = manager.get_terms() + assert terms[0] == "brandnew" + assert len(terms) == MAX_TERMS_YIELDED + + +def test_oversized_terms_file_without_complete_line_yields_nothing( + tmp_path: Path, monkeypatch +) -> None: + """A huge file whose first line never ends supplies no terms.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + (tmp_path / TERMS_FILENAME).write_bytes(b"x" * (MAX_TERMS_FILE_BYTES + 10)) + + assert manager.get_terms() == [] + + +def test_pr_767_dictionary_configuration_keys_and_contract_are_preserved( + tmp_path: Path, monkeypatch +) -> None: + """Existing enabled, file_path, and max_words settings remain effective.""" + configured_path = tmp_path / TERMS_FILENAME + configured_path.write_text("VocaLinux\nPyGObject\n", encoding="utf-8") + config = FakeConfig( + { + "dictionary": { + "enabled": True, + "file_path": str(configured_path), + "max_words": 1, + } + } + ) + manager = manager_at(tmp_path, monkeypatch, config) + + assert TERMS_FILENAME == "dictionary.txt" + assert Path(DEFAULT_TERMS_PATH).name == TERMS_FILENAME + assert manager.terms_enabled() + assert manager.terms_path() == configured_path + assert manager.build_initial_prompt() == "VocaLinux" + + +def test_default_terms_path_follows_xdg_config_home(tmp_path: Path, monkeypatch) -> None: + """A missing saved path uses the same XDG-aware root as corrections.""" + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path)) + manager = CustomDictionaryManager(FakeConfig({"dictionary": {}})) + + assert manager.terms_path() == tmp_path / TERMS_FILENAME + assert manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME + + +def test_legacy_vocalinux_terms_root_keeps_corrections_at_config_dir( + tmp_path: Path, monkeypatch +) -> None: + """Persisted ~/.config/vocalinux terms do not move the corrections file.""" + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path)) + legacy_corrections = Path.home() / ".config" / "vocalinux" / CORRECTIONS_FILENAME + expanded_legacy = str(Path.home() / ".config" / "vocalinux" / TERMS_FILENAME) + assert legacy_corrections != tmp_path / CORRECTIONS_FILENAME + + tilde_manager = CustomDictionaryManager( + FakeConfig({"dictionary": {"file_path": LEGACY_DEFAULT_TERMS_PATH}}) + ) + assert tilde_manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME + + expanded_manager = CustomDictionaryManager( + FakeConfig({"dictionary": {"file_path": expanded_legacy}}) + ) + assert expanded_manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME + + +def test_explicit_legacy_terms_path_keeps_corrections_at_config_dir( + tmp_path: Path, monkeypatch +) -> None: + """An explicit leftover pre-XDG terms path cannot relocate corrections.""" + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path)) + legacy_root = Path.home() / ".config" / "vocalinux" + expanded_legacy = str(legacy_root / TERMS_FILENAME) + assert legacy_root / CORRECTIONS_FILENAME != tmp_path / CORRECTIONS_FILENAME + + tilde_manager = CustomDictionaryManager( + FakeConfig( + { + "dictionary": { + "file_path": LEGACY_DEFAULT_TERMS_PATH, + "file_path_explicit": True, + } + } + ) + ) + assert tilde_manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME + + stamped = FakeConfig({"dictionary": {"file_path": expanded_legacy}}) + stamped_manager = CustomDictionaryManager(stamped) + stamped_manager.terms_path_text() + assert stamped.get("dictionary", "file_path_explicit") is True + assert stamped_manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME + + +def test_corrections_fall_back_from_terms_dir_when_primary_missing( + tmp_path: Path, monkeypatch +) -> None: + """Leftover terms-adjacent corrections are read and copied to the standard file.""" + fake_home = tmp_path / "home" + xdg_dir = tmp_path / "xdg-config" + legacy_root = fake_home / ".config" / "vocalinux" + legacy_root.mkdir(parents=True) + xdg_dir.mkdir() + monkeypatch.setenv("HOME", str(fake_home)) + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir)) + + old_path = Path(LEGACY_DEFAULT_TERMS_PATH).expanduser().parent / CORRECTIONS_FILENAME + payload = { + "version": 1, + "corrections": [{"heard": "super base", "replacement": "Supabase"}], + } + old_path.write_text(json.dumps(payload) + "\n", encoding="utf-8") + + manager = CustomDictionaryManager( + FakeConfig( + { + "dictionary": { + "file_path": LEGACY_DEFAULT_TERMS_PATH, + "file_path_explicit": True, + } + } + ) + ) + primary = xdg_dir / CORRECTIONS_FILENAME + assert manager.corrections_path() == primary + assert primary != old_path + assert not primary.exists() + assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}] + assert primary.is_file() + old_path.unlink() + assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}] + + +def test_corrections_fallback_survives_copy_failure(tmp_path: Path, monkeypatch) -> None: + """A failed copy-once still returns leftover terms-adjacent corrections.""" + terms_root = tmp_path / "legacy-root" + xdg_dir = tmp_path / "xdg-config" + terms_root.mkdir() + xdg_dir.mkdir() + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir)) + old_path = terms_root / CORRECTIONS_FILENAME + payload = { + "version": 1, + "corrections": [{"heard": "super base", "replacement": "Supabase"}], + } + old_path.write_text(json.dumps(payload) + "\n", encoding="utf-8") + + manager = CustomDictionaryManager( + FakeConfig( + { + "dictionary": { + "file_path": str(terms_root / TERMS_FILENAME), + "file_path_explicit": True, + } + } + ) + ) + primary = xdg_dir / CORRECTIONS_FILENAME + + def fail_replace(source: Path, destination: Path) -> None: + """Simulate a filesystem failure during atomic replacement.""" + raise OSError("disk full") + + monkeypatch.setattr("vocalinux.custom_dictionary.os.replace", fail_replace) + + assert manager.corrections_path() == primary + assert not primary.exists() + assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}] + assert not primary.exists() + + +def test_corrections_stay_at_config_dir_with_arbitrary_terms_parent( + tmp_path: Path, monkeypatch +) -> None: + """Corrections ignore the terms file location outside a vocalinux directory.""" + xdg_dir = tmp_path / "xdg-config" + xdg_dir.mkdir() + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir)) + terms = tmp_path / "my-terms.txt" + manager = CustomDictionaryManager( + FakeConfig( + { + "dictionary": { + "file_path": str(terms), + "file_path_explicit": True, + } + } + ) + ) + assert manager.corrections_path() == xdg_dir / CORRECTIONS_FILENAME + assert manager.corrections_path() != tmp_path / CORRECTIONS_FILENAME + + +def test_transient_terms_override_keeps_corrections_at_config_dir( + tmp_path: Path, monkeypatch +) -> None: + """A --dictionary-file session override must not move the corrections file.""" + xdg_dir = tmp_path / "xdg-config" + xdg_dir.mkdir() + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir)) + override = tmp_path / "override.txt" + override.write_text("Override\n", encoding="utf-8") + + manager = CustomDictionaryManager(FakeConfig(), str(override)) + + assert manager.corrections_path() == xdg_dir / CORRECTIONS_FILENAME + assert manager.corrections_path() != tmp_path / CORRECTIONS_FILENAME + + +def test_transient_terms_override_does_not_import_beside_corrections( + tmp_path: Path, monkeypatch +) -> None: + """A --dictionary-file session must not import the override's neighbor file.""" + xdg_dir = tmp_path / "xdg-config" + override_dir = tmp_path / "elsewhere" + xdg_dir.mkdir() + override_dir.mkdir() + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir)) + override = override_dir / "terms.txt" + override.write_text("Override\n", encoding="utf-8") + payload = { + "version": 1, + "corrections": [{"heard": "super base", "replacement": "Supabase"}], + } + (override_dir / CORRECTIONS_FILENAME).write_text(json.dumps(payload) + "\n", encoding="utf-8") + + manager = CustomDictionaryManager(FakeConfig(), str(override)) + + assert manager.get_corrections() == [] + assert manager.get_corrections_for_edit() == [] + assert not (xdg_dir / CORRECTIONS_FILENAME).exists() + + +def test_historical_legacy_looking_path_without_explicit_marker_is_not_migrated( + tmp_path: Path, monkeypatch +) -> None: + """A leftover pre-XDG path stays put when XDG_CONFIG_HOME points elsewhere.""" + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path)) + xdg_path = str(tmp_path / TERMS_FILENAME) + tilde_path = LEGACY_DEFAULT_TERMS_PATH + expanded_legacy = str(Path.home() / ".config" / "vocalinux" / TERMS_FILENAME) + assert expanded_legacy != xdg_path + + config = FakeConfig({"dictionary": {"file_path": tilde_path}}) + manager = CustomDictionaryManager(config) + + assert manager.terms_path_text() == tilde_path + assert manager.terms_path() == Path(tilde_path).expanduser() + assert config.get("dictionary", "file_path") == tilde_path + assert config.get("dictionary", "file_path") != xdg_path + assert config.get("dictionary", "file_path_explicit") is True + + failing = FakeConfig( + {"dictionary": {"file_path": tilde_path}}, + save_result=False, + ) + failing_manager = CustomDictionaryManager(failing) + assert failing_manager.terms_path_text() == tilde_path + assert failing.get("dictionary", "file_path") == tilde_path + assert failing.get("dictionary", "file_path") != xdg_path + assert not failing.get("dictionary", "file_path_explicit", False) + + expanded_config = FakeConfig({"dictionary": {"file_path": expanded_legacy}}) + expanded_manager = CustomDictionaryManager(expanded_config) + assert expanded_manager.terms_path_text() == expanded_legacy + assert expanded_config.get("dictionary", "file_path") == expanded_legacy + assert expanded_config.get("dictionary", "file_path") != xdg_path + assert expanded_config.get("dictionary", "file_path_explicit") is True + + custom = str(tmp_path / "custom-terms.txt") + custom_config = FakeConfig({"dictionary": {"file_path": custom}}) + custom_manager = CustomDictionaryManager(custom_config) + assert custom_manager.terms_path_text() == custom + assert custom_config.get("dictionary", "file_path") == custom + + +def test_explicit_legacy_terms_path_is_not_migrated(tmp_path: Path, monkeypatch) -> None: + """set_terms_path of the pre-XDG string is kept when XDG points elsewhere.""" + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path)) + xdg_path = str(tmp_path / TERMS_FILENAME) + expanded_legacy = str(Path.home() / ".config" / "vocalinux" / TERMS_FILENAME) + assert expanded_legacy != xdg_path + + tilde_config = FakeConfig({"dictionary": {"file_path": xdg_path}}) + tilde_manager = CustomDictionaryManager(tilde_config) + assert tilde_manager.set_terms_path(LEGACY_DEFAULT_TERMS_PATH) + assert tilde_config.get("dictionary", "file_path") == LEGACY_DEFAULT_TERMS_PATH + assert tilde_config.get("dictionary", "file_path_explicit") is True + assert tilde_manager.terms_path_text() == LEGACY_DEFAULT_TERMS_PATH + assert tilde_config.get("dictionary", "file_path") == LEGACY_DEFAULT_TERMS_PATH + + expanded_config = FakeConfig({"dictionary": {"file_path": xdg_path}}) + expanded_manager = CustomDictionaryManager(expanded_config) + assert expanded_manager.set_terms_path(expanded_legacy) + assert expanded_config.get("dictionary", "file_path") == expanded_legacy + assert expanded_config.get("dictionary", "file_path_explicit") is True + assert expanded_manager.terms_path_text() == expanded_legacy + assert expanded_config.get("dictionary", "file_path") == expanded_legacy + + +def test_invalid_or_unreadable_configured_terms_path_is_not_persisted( + tmp_path: Path, monkeypatch +) -> None: + """Invalid paths are harmless and leave the previous configured path intact.""" + previous_path = str(tmp_path / TERMS_FILENAME) + config = FakeConfig({"dictionary": {"file_path": previous_path}}) + manager = manager_at(tmp_path, monkeypatch, config) + + assert not manager.set_terms_path("~vocalinux-user-does-not-exist/dictionary.txt") + assert config.get("dictionary", "file_path") == previous_path + + assert not manager.set_terms_path(str(tmp_path)) + assert config.get("dictionary", "file_path") == previous_path + + +def test_invalid_saved_terms_path_is_safe_to_read_and_report(tmp_path: Path, monkeypatch) -> None: + """A pre-existing unresolved path cannot crash recognition or Settings.""" + config = FakeConfig( + { + "dictionary": { + "enabled": True, + "file_path": "~vocalinux-user-does-not-exist/dictionary.txt", + } + } + ) + manager = manager_at(tmp_path, monkeypatch, config) + + assert manager.terms_path() is None + assert manager.get_terms() == [] + assert manager.build_initial_prompt() is None + assert manager.terms_status() == "Configured terms path is invalid or cannot be expanded." + + +def test_terms_path_save_failure_preserves_the_previous_configured_path( + tmp_path: Path, monkeypatch +) -> None: + """A failed config save never claims a new path persisted.""" + previous_path = str(tmp_path / TERMS_FILENAME) + config = FakeConfig({"dictionary": {"file_path": previous_path}}, save_result=False) + manager = manager_at(tmp_path, monkeypatch, config) + + assert not manager.set_terms_path(str(tmp_path / "new-terms.txt")) + assert config.get("dictionary", "file_path") == previous_path + assert not config.get("dictionary", "file_path_explicit", False) + + +def test_invalid_transient_terms_path_does_not_crash(tmp_path: Path, monkeypatch) -> None: + """An unresolved CLI tilde path disables only its session's terms safely.""" + manager_at(tmp_path, monkeypatch) + manager = CustomDictionaryManager(FakeConfig(), "~vocalinux-user-does-not-exist/dictionary.txt") + + assert manager.terms_enabled() + assert manager.terms_path() is None + assert manager.get_terms() == [] + assert manager.terms_status() == "Configured terms path is invalid or cannot be expanded." + + +def test_atomic_save_failure_preserves_existing_terms_file(tmp_path: Path, monkeypatch) -> None: + """A failed replacement leaves the scanner-visible file intact.""" + manager = manager_at(tmp_path, monkeypatch) + terms_file = tmp_path / TERMS_FILENAME + terms_file.write_text("Existing\n", encoding="utf-8") + + def fail_replace(source: Path, destination: Path) -> None: + """Simulate a filesystem failure during atomic replacement.""" + raise OSError("disk full") + + monkeypatch.setattr("vocalinux.custom_dictionary.os.replace", fail_replace) + + assert not manager.save_terms(["New"]) + assert terms_file.read_text(encoding="utf-8") == "Existing\n" + + +def test_file_edits_preserve_existing_permissions(tmp_path: Path, monkeypatch) -> None: + """Replacing either dictionary file keeps the mode the owner set on it.""" + manager = manager_at(tmp_path, monkeypatch) + + terms_file = tmp_path / TERMS_FILENAME + terms_file.write_text("VocaLinux\n", encoding="utf-8") + terms_file.chmod(0o600) + assert manager.save_terms(["Supabase"]) + assert stat.S_IMODE(terms_file.stat().st_mode) == 0o600 + + corrections_file = tmp_path / CORRECTIONS_FILENAME + corrections_file.write_text('{"version": 1, "corrections": []}\n', encoding="utf-8") + corrections_file.chmod(0o640) + assert manager.save_corrections([{"heard": "a", "replacement": "b"}]) + assert stat.S_IMODE(corrections_file.stat().st_mode) == 0o640 + + +def test_dictionary_file_argument_is_available_for_a_session_override(monkeypatch) -> None: + """The retained CLI option accepts a terms file without changing config.""" + monkeypatch.setattr("sys.argv", ["vocalinux", "--dictionary-file", "/tmp/terms.txt"]) + + assert parse_arguments().dictionary_file == "/tmp/terms.txt" + + +def test_corrections_use_versioned_json_and_live_reload(tmp_path: Path, monkeypatch) -> None: + """Structured corrections use exact replacement spelling on every read.""" + manager = manager_at(tmp_path, monkeypatch) + assert manager.save_corrections([{"heard": "super base", "replacement": "Supabase"}]) + stored = json.loads((tmp_path / CORRECTIONS_FILENAME).read_text(encoding="utf-8")) + assert stored == { + "version": 1, + "corrections": [{"heard": "super base", "replacement": "Supabase"}], + } + assert manager.get_corrections_for_edit() == [ + {"heard": "super base", "replacement": "Supabase"} + ] + assert manager.apply_corrections("super base is useful") == "Supabase is useful" + + assert manager.save_corrections([{"heard": "pie object", "replacement": "PyGObject"}]) + assert manager.apply_corrections("pie object") == "PyGObject" + + +def test_corrections_are_literal_longest_first_and_unicode_safe() -> None: + """Matching is whole-phrase, case-insensitive, and never maps through lower().""" + entries = normalize_corrections( + [ + {"heard": "super", "replacement": "short"}, + {"heard": "super base", "replacement": "Supabase"}, + {"heard": "C++", "replacement": "C plus plus"}, + {"heard": "i", "replacement": "letter"}, + ] + ) + + assert apply_corrections("SUPER BASE and super", entries) == "Supabase and short" + assert apply_corrections("C++ is not C++foo", entries) == "C plus plus is not C++foo" + assert apply_corrections("İ", entries) == "letter" + assert apply_corrections("e\u0301clair", [{"heard": "e", "replacement": "E"}]) == "éclair" + + +def test_invalid_corrections_file_fails_closed(tmp_path: Path, monkeypatch) -> None: + """Malformed JSON never aborts a segment or invents replacements.""" + manager = manager_at(tmp_path, monkeypatch) + (tmp_path / CORRECTIONS_FILENAME).write_text("{not json", encoding="utf-8") + + assert manager.get_corrections() == [] + assert manager.get_corrections_for_edit() is None + assert manager.apply_corrections("super base") == "super base" + + +def test_partially_invalid_corrections_cannot_be_destructively_edited( + tmp_path: Path, monkeypatch +) -> None: + """UI write-back is refused when runtime reads had to filter source entries.""" + manager = manager_at(tmp_path, monkeypatch) + path = tmp_path / CORRECTIONS_FILENAME + original = { + "version": 1, + "corrections": [ + {"heard": "super base", "replacement": "Supabase"}, + {"heard": "missing replacement"}, + ], + } + path.write_text(json.dumps(original), encoding="utf-8") + + assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}] + assert manager.get_corrections_for_edit() is None + assert json.loads(path.read_text(encoding="utf-8")) == original + + +def test_corrections_with_extra_fields_cannot_be_destructively_edited( + tmp_path: Path, monkeypatch +) -> None: + """Unknown fields still apply at runtime but block a UI rewrite that would drop them.""" + manager = manager_at(tmp_path, monkeypatch) + path = tmp_path / CORRECTIONS_FILENAME + original = { + "version": 1, + "corrections": [ + { + "heard": "super base", + "replacement": "Supabase", + "note": "scanner annotation", + } + ], + } + path.write_text(json.dumps(original), encoding="utf-8") + + assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}] + assert manager.apply_corrections("super base is useful") == "Supabase is useful" + assert manager.get_corrections_for_edit() is None + assert json.loads(path.read_text(encoding="utf-8")) == original + + +def test_corrections_with_top_level_metadata_cannot_be_destructively_edited( + tmp_path: Path, monkeypatch +) -> None: + """Unknown top-level keys apply at runtime but block a UI rewrite that would drop them.""" + manager = manager_at(tmp_path, monkeypatch) + path = tmp_path / CORRECTIONS_FILENAME + original = { + "version": 1, + "metadata": {"author": "scanner"}, + "corrections": [{"heard": "super base", "replacement": "Supabase"}], + } + path.write_text(json.dumps(original), encoding="utf-8") + + assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}] + assert manager.apply_corrections("super base is useful") == "Supabase is useful" + assert manager.get_corrections_for_edit() is None + assert json.loads(path.read_text(encoding="utf-8")) == original + + +def test_corrections_with_whitespace_cannot_be_destructively_edited( + tmp_path: Path, monkeypatch +) -> None: + """Leading or trailing whitespace still applies at runtime but blocks a UI rewrite.""" + manager = manager_at(tmp_path, monkeypatch) + path = tmp_path / CORRECTIONS_FILENAME + original = { + "version": 1, + "corrections": [{"heard": " super base ", "replacement": " Supabase "}], + } + path.write_text(json.dumps(original), encoding="utf-8") + + assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}] + assert manager.apply_corrections("super base is useful") == "Supabase is useful" + assert manager.get_corrections_for_edit() is None + assert json.loads(path.read_text(encoding="utf-8")) == original + + +def test_corrections_with_decomposed_unicode_cannot_be_destructively_edited( + tmp_path: Path, monkeypatch +) -> None: + """NFD heard values still apply at runtime but block a UI rewrite to NFC.""" + manager = manager_at(tmp_path, monkeypatch) + path = tmp_path / CORRECTIONS_FILENAME + original = { + "version": 1, + "corrections": [{"heard": "e\u0301clair", "replacement": "éclair"}], + } + path.write_text(json.dumps(original, ensure_ascii=False), encoding="utf-8") + + assert manager.get_corrections() == [{"heard": "éclair", "replacement": "éclair"}] + assert manager.apply_corrections("e\u0301clair") == "éclair" + assert manager.get_corrections_for_edit() is None + assert json.loads(path.read_text(encoding="utf-8")) == original + + +def test_corrections_with_duplicate_keys_cannot_be_destructively_edited( + tmp_path: Path, monkeypatch +) -> None: + """Duplicate JSON keys still apply at runtime but block a UI rewrite that would drop them.""" + manager = manager_at(tmp_path, monkeypatch) + path = tmp_path / CORRECTIONS_FILENAME + cases = [ + ( + "{\n" + ' "version": 1,\n' + ' "corrections": [{"heard": "pie object", "replacement": "PyGObject"}],\n' + ' "corrections": [{"heard": "super base", "replacement": "Supabase"}]\n' + "}\n", + [{"heard": "super base", "replacement": "Supabase"}], + ), + ( + "{\n" + ' "version": 1,\n' + ' "corrections": [' + '{"heard": "super base", "replacement": "Wrong", "replacement": "Supabase"}' + "]\n}\n", + [{"heard": "super base", "replacement": "Supabase"}], + ), + ] + for original, expected_runtime in cases: + path.write_text(original, encoding="utf-8") + before = path.read_bytes() + + assert manager.get_corrections() == expected_runtime + assert manager.apply_corrections("super base is useful") == "Supabase is useful" + assert manager.get_corrections_for_edit() is None + assert path.read_bytes() == before + + +def test_invalid_correction_update_does_not_wipe_existing_entry( + tmp_path: Path, monkeypatch +) -> None: + """A too-long replacement must not delete the previous same-heard correction.""" + import importlib + import sys + + import vocalinux.ui as ui_pkg + + manager = manager_at(tmp_path, monkeypatch) + original = [{"heard": "super base", "replacement": "Supabase"}] + assert manager.save_corrections(original) + corrections_path = tmp_path / CORRECTIONS_FILENAME + before = corrections_path.read_text(encoding="utf-8") + + repository = sys.modules["gi.repository"] + bases = {name: type(name, (), {}) for name in ("Box", "ListBoxRow", "Dialog")} + saved = sys.modules.pop("vocalinux.ui.settings_dialog", None) + try: + with patch.object(repository, "Gtk", MagicMock(**bases)): + settings_dialog = importlib.import_module("vocalinux.ui.settings_dialog") + finally: + sys.modules.pop("vocalinux.ui.settings_dialog", None) + if saved is not None: + sys.modules["vocalinux.ui.settings_dialog"] = saved + ui_pkg.settings_dialog = saved + else: + ui_pkg.__dict__.pop("settings_dialog", None) + + dialog = Mock() + dialog._initializing = False + dialog._applying_settings = False + dialog._dictionary_available.return_value = True + dialog.dictionary_manager = manager + dialog.dictionary_heard_entry.get_text.return_value = "super base" + dialog.dictionary_replacement_entry.get_text.return_value = "x" * ( + MAX_CORRECTION_CHARACTERS + 1 + ) + + settings_dialog.SettingsDialog._on_dictionary_add_correction(dialog, None) + + assert corrections_path.read_text(encoding="utf-8") == before + assert manager.get_corrections() == original + dialog.dictionary_heard_entry.set_text.assert_not_called() + dialog.dictionary_replacement_entry.set_text.assert_not_called() + dialog.dictionary_feedback_label.set_text.assert_called_once_with( + "That correction is not valid for the file." + ) + + +def test_nfd_correction_update_replaces_nfc_entry(tmp_path: Path, monkeypatch) -> None: + """A decomposed heard phrase must update the existing NFC correction.""" + import importlib + import sys + + import vocalinux.ui as ui_pkg + + manager = manager_at(tmp_path, monkeypatch) + original = [{"heard": "éclair", "replacement": "OLD"}] + assert manager.save_corrections(original) + corrections_path = tmp_path / CORRECTIONS_FILENAME + + repository = sys.modules["gi.repository"] + bases = {name: type(name, (), {}) for name in ("Box", "ListBoxRow", "Dialog")} + saved = sys.modules.pop("vocalinux.ui.settings_dialog", None) + try: + with patch.object(repository, "Gtk", MagicMock(**bases)): + settings_dialog = importlib.import_module("vocalinux.ui.settings_dialog") + finally: + sys.modules.pop("vocalinux.ui.settings_dialog", None) + if saved is not None: + sys.modules["vocalinux.ui.settings_dialog"] = saved + ui_pkg.settings_dialog = saved + else: + ui_pkg.__dict__.pop("settings_dialog", None) + + dialog = Mock() + dialog._initializing = False + dialog._applying_settings = False + dialog._dictionary_available.return_value = True + dialog.dictionary_manager = manager + dialog.dictionary_heard_entry.get_text.return_value = "e\u0301clair" + dialog.dictionary_replacement_entry.get_text.return_value = "NEW" + + settings_dialog.SettingsDialog._on_dictionary_add_correction(dialog, None) + + stored = json.loads(corrections_path.read_text(encoding="utf-8")) + assert stored == { + "version": 1, + "corrections": [{"heard": "éclair", "replacement": "NEW"}], + } + assert manager.get_corrections() == [{"heard": "éclair", "replacement": "NEW"}] + dialog.dictionary_feedback_label.set_text.assert_called_once_with("Correction updated.") + + +def test_non_integer_corrections_version_fails_closed(tmp_path: Path, monkeypatch) -> None: + """Boolean, float, and string versions must not be treated as schema 1.""" + manager = manager_at(tmp_path, monkeypatch) + path = tmp_path / CORRECTIONS_FILENAME + cases = ( + '{"version": true, "corrections": [{"heard": "super base", "replacement": "Supabase"}]}\n', + '{"version": 1.0, "corrections": [{"heard": "super base", "replacement": "Supabase"}]}\n', + '{"version": "1", "corrections": [{"heard": "super base", "replacement": "Supabase"}]}\n', + '{"version": 1e0, "corrections": [{"heard": "super base", "replacement": "Supabase"}]}\n', + ) + for original in cases: + path.write_text(original, encoding="utf-8") + before = path.read_bytes() + + assert manager.get_corrections() == [] + assert manager.get_corrections_for_edit() is None + assert path.read_bytes() == before + + +def test_invalid_terms_file_is_ignored_and_explained(tmp_path: Path, monkeypatch) -> None: + """An invalid scanner file leaves prompt bias empty with a clear status.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + (tmp_path / TERMS_FILENAME).write_bytes(b"\xff\xfe") + + assert manager.build_initial_prompt() is None + assert manager.terms_status() == "Terms file is not valid UTF-8." + + +def test_whisper_receives_the_live_custom_terms_prompt(tmp_path: Path, monkeypatch) -> None: + """Whisper gets the current file contents as initial_prompt at transcription time.""" + dictionary = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + (tmp_path / TERMS_FILENAME).write_text("VocaLinux\n", encoding="utf-8") + manager = recognition_manager("whisper", dictionary) + manager.model = MagicMock() + manager.model.transcribe.return_value = {"text": "ok"} + manager.model.device = MagicMock() + + with patch.dict("sys.modules", {"numpy": mock_numpy(), "torch": MagicMock()}): + assert manager._transcribe_with_whisper([b"\x00\x00"]) == "ok" + + assert manager.model.transcribe.call_args.kwargs["initial_prompt"] == "VocaLinux" + + +def test_faster_whisper_receives_the_live_custom_terms_prompt(tmp_path: Path, monkeypatch) -> None: + """Faster Whisper receives the same live vocabulary bias as OpenAI Whisper.""" + dictionary = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + (tmp_path / TERMS_FILENAME).write_text("VocaLinux\n", encoding="utf-8") + manager = object.__new__(SpeechRecognitionManager) + manager.dictionary_manager = dictionary + manager._faster_whisper_engine = MagicMock() + manager._faster_whisper_engine.is_ready.return_value = True + manager._faster_whisper_engine.transcribe.return_value = "ok" + + assert manager._transcribe_with_faster_whisper([b"\x00\x00"]) == "ok" + manager._faster_whisper_engine.transcribe.assert_called_once_with( + [b"\x00\x00"], language="auto", initial_prompt="VocaLinux" + ) + + +def test_whispercpp_composes_and_clears_live_terms_prompt(tmp_path: Path, monkeypatch) -> None: + """whisper.cpp keeps Advanced context and explicitly clears a stale prompt.""" + config = FakeConfig({"dictionary": {"enabled": True}}) + dictionary = manager_at(tmp_path, monkeypatch, config) + terms_file = tmp_path / TERMS_FILENAME + terms_file.write_text("VocaLinux\n", encoding="utf-8") + manager = recognition_manager("whisper_cpp", dictionary, advanced_prompt="Explicit context") + manager._model_lock = threading.Lock() + manager.model = MagicMock() + manager.model.transcribe.return_value = [MagicMock(text="ok")] + + with patch.dict("sys.modules", {"numpy": mock_numpy()}): + assert manager._transcribe_with_whispercpp([b"\x00\x00"]) == "ok" + config.values["dictionary"]["enabled"] = False + assert manager._transcribe_with_whispercpp([b"\x00\x00"]) == "ok" + manager.whispercpp_initial_prompt = "" + assert manager._transcribe_with_whispercpp([b"\x00\x00"]) == "ok" + + prompts = [call.kwargs["initial_prompt"] for call in manager.model.transcribe.call_args_list] + assert prompts == ["Explicit context VocaLinux", "Explicit context", ""] + + +def test_vosk_warns_once_for_bias_while_corrections_remain_available( + tmp_path: Path, monkeypatch +) -> None: + """VOSK rejects vocabulary bias only; its transcript can still be corrected.""" + dictionary = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + (tmp_path / TERMS_FILENAME).write_text("VocaLinux\n", encoding="utf-8") + manager = recognition_manager("vosk", dictionary) + manager._get_vosk_model_path = MagicMock(return_value="/models/vosk") + vosk = MagicMock() + + with patch.dict("sys.modules", {"vosk": vosk}): + with patch( + "vocalinux.speech_recognition.recognition_manager.os.path.exists", return_value=True + ): + with patch( + "vocalinux.speech_recognition.recognition_manager.logger.warning" + ) as warning: + manager._init_vosk() + manager._init_vosk() + + warning.assert_called_once_with( + "Custom terms are ignored by VOSK; transcript corrections still apply." + ) + + +def test_whispercpp_prompt_composes_advanced_and_terms() -> None: + """Advanced prompt remains a prefix and an emptied terms file clears the value.""" + recognition_manager = object.__new__(SpeechRecognitionManager) + recognition_manager.whispercpp_initial_prompt = "Write product names accurately." + recognition_manager.dictionary_manager = type( + "Dictionary", (), {"build_initial_prompt": lambda self: "VocaLinux"} + )() + + assert ( + recognition_manager._get_whispercpp_prompt() == "Write product names accurately. VocaLinux" + ) + recognition_manager.dictionary_manager = type( + "Dictionary", (), {"build_initial_prompt": lambda self: None} + )() + assert recognition_manager._get_whispercpp_prompt() == "Write product names accurately." + + +def test_corrections_run_before_voice_commands() -> None: + """A correction changes raw text before the command processor can act on it.""" + + class Recognizer: + """Minimal VOSK recognizer.""" + + def AcceptWaveform(self, data: bytes) -> None: + """Accept a fake audio chunk.""" + + def FinalResult(self) -> str: + """Return the raw model text.""" + return '{"text": "delete that"}' + + recognition_manager = object.__new__(SpeechRecognitionManager) + recognition_manager.engine = "vosk" + recognition_manager.recognizer = Recognizer() + recognition_manager._model_lock = threading.Lock() + recognition_manager._voice_commands_enabled = True + recognition_manager.dictionary_manager = type( + "Dictionary", (), {"apply_corrections": lambda self, text: "delete note"} + )() + received: list[str] = [] + recognition_manager.command_processor = type( + "Commands", (), {"process_text": lambda self, text: (text, [])} + )() + recognition_manager.text_callbacks = [received.append] + recognition_manager.action_callbacks = [] + recognition_manager.segment_callbacks = [] + + recognition_manager._process_audio_buffer([b"audio"]) + + assert received == ["delete note"] + + +def test_correction_stage_is_shared_by_every_engine() -> None: + """The common post-transcription stage is available to local and remote engines.""" + for engine in ( + "vosk", + "whisper", + "whisper_cpp", + "faster_whisper", + "parakeet", + "remote_api", + ): + manager = object.__new__(SpeechRecognitionManager) + manager.engine = engine + manager.dictionary_manager = type( + "Dictionary", (), {"apply_corrections": lambda self, text: "Supabase"} + )() + + assert manager._apply_dictionary_corrections("super base") == "Supabase" diff --git a/tests/test_faster_whisper_engine.py b/tests/test_faster_whisper_engine.py index 4f2d8c0..a21d6a4 100644 --- a/tests/test_faster_whisper_engine.py +++ b/tests/test_faster_whisper_engine.py @@ -323,6 +323,21 @@ class TestFasterWhisperEngine: assert text == "Hello, world." + def test_transcribe_passes_initial_prompt(self) -> None: + """Test that vocabulary bias is forwarded to faster-whisper.""" + whisper_mock = self._mock_whisper_model([MagicMock(text="VocaLinux")]) + with patch.dict(sys.modules, {"faster_whisper": whisper_mock}): + engine = FasterWhisperEngine(model_size="tiny", device="cpu") + engine.init() + + audio = np.array([0, 1000, -1000, 0], dtype=np.int16) + engine.transcribe([audio.tobytes()], initial_prompt="VocaLinux") + + assert ( + whisper_mock.WhisperModel.return_value.transcribe.call_args.kwargs["initial_prompt"] + == "VocaLinux" + ) + def test_transcribe_empty_audio(self): """Test that empty audio returns empty text.""" whisper_mock = self._mock_whisper_model([]) diff --git a/tests/test_main.py b/tests/test_main.py index bbdc5c2..863c51c 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -3,6 +3,7 @@ Tests for the main module functionality. """ import argparse +import queue import sys import threading import time @@ -34,6 +35,7 @@ class TestMainModule(unittest.TestCase): self.assertIsNone(args.model) # No default set, loaded from config instead self.assertIsNone(args.engine) self.assertIsNone(args.language) + self.assertIsNone(args.dictionary_file) self.assertFalse(args.wayland) self.assertFalse(args.start_minimized) @@ -248,6 +250,7 @@ class TestMainModule(unittest.TestCase): whispercpp_no_speech_thold=0.6, whispercpp_n_threads=0, whispercpp_gpu_device=None, + dictionary_manager=ANY, remote_api_url="", remote_api_key="", remote_api_endpoint="/inference", @@ -1694,25 +1697,91 @@ class TestSessionHistoryRecording(unittest.TestCase): finally: stack.close() + def _worker_deliverer( + self, segment_cb: Callable[[str, float], None] + ) -> Tuple[Callable[[str], None], Callable[[], None]]: + """Deliver segments from one persistent worker thread, like the real engine. + + Returns (deliver, close): ``deliver(text)`` runs the segment callback + on the worker synchronously; ``close()`` stops it. + """ + work: queue.Queue = queue.Queue() + errors: list[BaseException] = [] + + def worker() -> None: + while True: + job = work.get() + try: + if job is not None: + job() + except BaseException as error: + errors.append(error) + finally: + work.task_done() + if job is None: + return + + thread = threading.Thread(target=worker) + thread.start() + + def deliver(text: str) -> None: + work.put(lambda: segment_cb(text, time.monotonic())) + work.join() + if errors: + raise errors.pop(0) + + def close() -> None: + work.put(None) + work.join() + thread.join() + + return deliver, close + def test_late_segment_from_old_worker_during_next_session(self) -> None: """A leftover worker delivering on its own thread stays out of the open session.""" stack, _, segment_cb, state_cb, history = self._boot() + on_old_worker, close_old_worker = self._worker_deliverer(segment_cb) try: state_cb(RecognitionState.LISTENING) - segment_cb("one", time.monotonic()) + on_old_worker("one") state_cb(RecognitionState.IDLE) state_cb(RecognitionState.LISTENING) segment_cb("two", time.monotonic()) # The previous session's worker finally delivers on its own thread. - t = threading.Thread(target=segment_cb, args=("trailing", time.monotonic())) - t.start() - t.join() + on_old_worker("trailing") self.assertEqual(history.get_all(), ["one trailing"]) state_cb(RecognitionState.IDLE) self.assertEqual(history.get_all(), ["two", "one trailing"]) finally: + close_old_worker() + stack.close() + + def test_straggler_after_newer_session_forms_own_snippet(self) -> None: + """An old worker finishing after the next session must not grow its entry.""" + stack, _, segment_cb, state_cb, history = self._boot() + on_old_worker, close_old_worker = self._worker_deliverer(segment_cb) + try: + state_cb(RecognitionState.LISTENING) + on_old_worker("first") + state_cb(RecognitionState.IDLE) + + state_cb(RecognitionState.LISTENING) + segment_cb("second", time.monotonic()) + state_cb(RecognitionState.IDLE) + + # The older session's worker decodes last: its text belongs to + # the first session, not to the currently-newest entry. + on_old_worker("late tail") + self.assertEqual(history.get_all(), ["late tail", "second", "first"]) + + # A later straggler on that same worker still extends the + # snippet it owns. + on_old_worker("more tail") + self.assertEqual(history.get_all(), ["late tail more tail", "second", "first"]) + finally: + close_old_worker() stack.close() def test_session_without_segments_creates_no_snippet(self) -> None: diff --git a/tests/test_record_during_reload.py b/tests/test_record_during_reload.py index ab59325..868bff4 100644 --- a/tests/test_record_during_reload.py +++ b/tests/test_record_during_reload.py @@ -2,6 +2,7 @@ import sys import threading +import time from collections.abc import Generator from unittest.mock import Mock @@ -90,6 +91,46 @@ def test_release_before_reload_finishes_preserves_whole_recording(session: Sessi assert not manager.is_idle_unloaded +def test_worker_exit_during_release_join_still_transcribes_final_buffer( + session: Session, monkeypatch: pytest.MonkeyPatch +) -> None: + """A release whose final enqueue lands after the worker exits is kept. + + stop_recognition joins the audio thread with a bounded wait; while it + waits, the recognition worker can notice ``should_record`` is false and + exit. The final buffer it then enqueues must still be transcribed by + the reload teardown instead of being thrown away with the queue. + """ + manager, captured, loading, release_load, transcribed = session + capture_release = threading.Event() + + def slow_record() -> None: + with manager._buffer_lock: + manager.audio_buffer = [b"final speech"] + manager._recording_segment_has_speech = True + captured.set() + assert capture_release.wait(3) + + monkeypatch.setattr(manager, "_record_audio", slow_record) + assert manager.start_recognition(mode="push_to_talk") + assert captured.wait(1) + assert loading.wait(1) + release_load.set() # Model loads; the worker starts polling an empty queue. + + stopper = threading.Thread(target=manager.stop_recognition) + stopper.start() + # The worker's 0.5s second-chance window elapses while the stop path is + # still joining the audio thread, so the tail buffer is enqueued only + # after the worker has already exited. + time.sleep(1.2) + capture_release.set() + stopper.join(3) + manager.recognition_thread.join(3) + + assert transcribed == [[b"final speech"]] + assert manager.state == RecognitionState.IDLE + + def test_toggle_transcribes_queued_speech_after_reload(session: Session) -> None: manager, captured, loading, release_load, transcribed = session assert manager.start_recognition() diff --git a/tests/test_settings_dialog.py b/tests/test_settings_dialog.py index 9f5c88a..b1e4a62 100644 --- a/tests/test_settings_dialog.py +++ b/tests/test_settings_dialog.py @@ -1081,9 +1081,10 @@ class TestSettingsNavigation(unittest.TestCase): self.assertIn("self.sidebar_listbox = Gtk.ListBox()", self.source_code) self.assertNotIn("Gtk.Notebook()", self.source_code) - def test_topic_pages_exist(self): + def test_topic_pages_exist(self) -> None: for name, title in [ ("dictation", "Dictation"), + ("dictionary", "Custom Dictionary"), ("model", "Speech Model"), ("audio", "Audio"), ("performance", "Performance"), @@ -1092,6 +1093,108 @@ class TestSettingsNavigation(unittest.TestCase): ]: self.assertIn(f'SettingsPage("{name}", "{title}"', self.source_code) + def test_custom_dictionary_page_has_both_accessible_subsections(self) -> None: + """The dictionary page keeps terms and corrections distinct and accessible.""" + body = self.source_code.split("def _build_dictionary_section")[1].split("\n def ")[0] + self.assertIn('title="Custom terms"', body) + self.assertIn('title="Transcript corrections"', body) + self.assertIn("Vocabulary bias works with Whisper, whisper.cpp, and Faster Whisper", body) + self.assertIn("self.dictionary_management_switcher = Gtk.StackSwitcher()", body) + self.assertIn( + "self.dictionary_tab.pack_start(self.dictionary_management_stack, True, True, 0)", + body, + ) + self.assertIn('terms_scroller, "terms", "Custom terms"', body) + self.assertIn('corrections_scroller, "corrections", "Corrections"', body) + for accessible_name in [ + "Custom terms file", + "Custom term", + "Heard phrase", + "Replacement text", + "Custom dictionary status", + ]: + self.assertIn(accessible_name, body) + # Shared status lives on the page, not only in the Corrections stack child. + self.assertIn( + "self.dictionary_tab.pack_start(self.dictionary_feedback_label, False, False, 0)", + body, + ) + switcher_pack = body.find( + "self.dictionary_tab.pack_start(self.dictionary_management_switcher" + ) + feedback_pack = body.find("self.dictionary_tab.pack_start(self.dictionary_feedback_label") + stack_pack = body.find("self.dictionary_tab.pack_start(self.dictionary_management_stack") + self.assertLess(switcher_pack, feedback_pack) + self.assertLess(feedback_pack, stack_pack) + corrections_child = body[ + body.find("corrections_group = PreferencesGroup") : body.find( + 'corrections_scroller, "corrections"' + ) + ] + self.assertNotIn("dictionary_feedback_label", corrections_child) + + def test_custom_dictionary_cards_preserve_rounded_bottom_corners(self) -> None: + """Transparent list backgrounds do not cover the card's lower radius.""" + self.assertIn(".preferences-group-list", self.source_code) + self.assertIn('add_class("preferences-group-list")', self.source_code) + self.assertIn("border-radius: 0 0 11px 11px", self.source_code) + + def test_custom_dictionary_path_and_persistence_handlers_are_present(self) -> None: + """Path chooser and failure feedback keep existing settings safe.""" + self.assertIn('Gtk.FileChooserButton(title="Choose Terms File")', self.source_code) + self.assertIn("def _on_dictionary_terms_path_changed", self.source_code) + self.assertIn("set_terms_path", self.source_code) + self.assertIn("Could not save that custom terms path", self.source_code) + self.assertIn("add_term(term)", self.source_code) + self.assertIn("remove_term(term)", self.source_code) + + def test_correction_add_validates_candidate_before_save(self) -> None: + """Invalid replacements must be rejected before the list is rewritten.""" + body = self.source_code.split("def _on_dictionary_add_correction")[1].split("\n def ")[0] + self.assertIn("normalize_corrections", body) + self.assertLess(body.find("normalize_corrections"), body.find("save_corrections")) + self.assertLess(body.find("normalize_corrections"), body.find("entries.append")) + + def test_terms_list_builds_in_bounded_idle_slices_with_pinned_adds(self) -> None: + """Large term files never build all their GTK rows synchronously. + + The stall finding requires two guards to hold together: the row + build is sliced through GLib idle callbacks, and a saved term that + lands past the display cap is pinned so success is never reported + for an invisible term. + """ + self.assertIn("_TERMS_ROWS_PER_IDLE", self.source_code) + refresh_body = self.source_code.split("def _refresh_dictionary_ui")[1].split("\n def ")[ + 0 + ] + self.assertNotIn("for term in visible_terms:", refresh_body) + self.assertIn("self._build_term_rows(", refresh_body) + build_body = self.source_code.split("def _build_term_rows")[1].split("\n def ")[0] + self.assertIn("GLib.idle_add", build_body) + self.assertIn("_TERMS_ROWS_PER_IDLE", build_body) + self.assertIn("self._terms_build_token", build_body) + add_body = self.source_code.split("def _on_dictionary_add_term")[1].split("\n def ")[0] + self.assertIn("self._pinned_terms.append(term)", add_body) + self.assertIn("self._pinned_terms", self.source_code) + + def test_show_all_add_scrolls_to_the_landed_term(self) -> None: + """Under "show all" the rebuild scrolls to the new term's own row. + + Oversized or yield-capped files prepend the term, so scrolling to + the bottom would hide it; the target is the term's index in the + refreshed list and is verified against the row's label. + """ + add_body = self.source_code.split("def _on_dictionary_add_term")[1].split("\n def ")[0] + self.assertIn("self._show_all_terms", add_body) + self.assertIn("self._scroll_terms_to_row", add_body) + self.assertIn("terms_now[new_index]", add_body) + finish_body = self.source_code.split("def _finish_term_rows")[1].split("\n def ")[0] + self.assertIn("self._scroll_terms_to_row", finish_body) + self.assertIn("get_row_at_index", finish_body) + self.assertIn("_term_row_text", finish_body) + self.assertIn("_scroll_terms_row_into_view", finish_body) + self.assertNotIn("_scroll_terms_to_end", self.source_code) + def test_application_page_has_tray_warning_toggle(self): self.assertIn('PreferencesGroup(title="General")', self.source_code) self.assertIn("self.missing_tray_warning_switch = Gtk.Switch()", self.source_code)