From eef2554f7c1c4453d2d7e104c4e4916caa0ae8d6 Mon Sep 17 00:00:00 2001 From: Jatin K Malik Date: Tue, 29 Sep 2026 14:24:28 -0700 Subject: [PATCH] feat(audio): lower other audio while dictating (#861) * feat(audio): lower other audio while dictating Opt-in setting lowers the default PipeWire or PulseAudio sink while the microphone is open, then restores that exact level. It stays off by default. * fix(test): keep pytest.approx working when numpy is mocked Other test modules replace sys.modules["numpy"] with a MagicMock at import and leave it there. pytest.approx then crashes on isinstance(..., numpy.bool_). * fix(audio): restore every channel and do not duck if it cannot be undone A failed restore-point write no longer lowers the sink. PulseAudio puts each channel back, and a dead microphone restores playback instead of waiting for an explicit stop. The recognition test patches the real audio module when an older test left a mock behind. * fix(test): put the audio-feedback mock back after the ducking test The recognition test has to patch the real module, but later tests still expect the mock that test_text_injector installs. Restore that mock when the test ends, and annotate the new helpers. * fix(test): annotate the playback-duck tests AGENTS.md requires a type hint on every new function. These tests were the ones still missing them. * fix(audio): skip arming playback duck after recording already failed * chore: retrigger review after rebase --- AGENTS.md | 2 + docs/USER_GUIDE.md | 2 + pyproject.toml | 1 + src/vocalinux/audio/__init__.py | 1 + src/vocalinux/audio/playback_ducker.py | 804 ++++++++++++++++++ src/vocalinux/common_types.py | 28 + .../speech_recognition/recognition_manager.py | 117 ++- src/vocalinux/ui/config_manager.py | 54 ++ src/vocalinux/ui/settings_dialog.py | 66 ++ src/vocalinux/ui/tray_indicator.py | 11 + tests/test_playback_ducker.py | 747 ++++++++++++++++ 11 files changed, 1832 insertions(+), 1 deletion(-) create mode 100644 src/vocalinux/audio/__init__.py create mode 100644 src/vocalinux/audio/playback_ducker.py create mode 100644 tests/test_playback_ducker.py diff --git a/AGENTS.md b/AGENTS.md index f4c1399..dc9cd39 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -135,6 +135,8 @@ Optional extras: `vosk`, `whisper`, `vad`, `dev`. ``` src/vocalinux/ ├── main.py, version.py, common_types.py +├── audio/ +│ └── playback_ducker.py # lower default sink while dictating ├── single_instance.py # $XDG_DATA_HOME/vocalinux/instance.lock ├── auto_pause_monitor.py # unload model while configured apps run ├── model_keepalive.py # idle unload diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index dbef203..266cb49 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -34,6 +34,8 @@ When Vocalinux injects through the clipboard (the usual Wayland / ydotool path), On non-US layouts such as German Neo, that chord uses the key that types **v** on the active layout (not physical KEY_V). Nested terminal panels inside an IDE are often invisible to window-class detection. If paste lands as a literal `^V` or does nothing, open **Settings → Dictation → Clipboard Paste Shortcut** and choose **Ctrl+Shift+V**. +**Other audio.** Settings → Audio → Other audio can lower speakers and headphones while the microphone is open, then put that volume back. It stays off until you turn it on. Level while dictating is a percent of the current volume (0 is silent). Quitting mid-dictation puts the volume back; if the app crashes first, the next launch does. + ## Shortcuts Configure under **Settings → Shortcuts**: diff --git a/pyproject.toml b/pyproject.toml index 2edc96a..e74e101 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -124,6 +124,7 @@ required-version = ">=0.12,<0.13" # Resource dirs carry no .py files, but setuptools still treats them as packages. packages = [ "vocalinux", + "vocalinux.audio", "vocalinux.resources", "vocalinux.resources.icons", "vocalinux.resources.icons.scalable", diff --git a/src/vocalinux/audio/__init__.py b/src/vocalinux/audio/__init__.py new file mode 100644 index 0000000..16a882b --- /dev/null +++ b/src/vocalinux/audio/__init__.py @@ -0,0 +1 @@ +"""Playback helpers that are not the dictation microphone.""" diff --git a/src/vocalinux/audio/playback_ducker.py b/src/vocalinux/audio/playback_ducker.py new file mode 100644 index 0000000..4a07b3b --- /dev/null +++ b/src/vocalinux/audio/playback_ducker.py @@ -0,0 +1,804 @@ +"""Lower the default speaker volume while dictating, then put it back. + +PipeWire (``wpctl``) and PulseAudio (``pactl``) can read a sink and restore +that exact level. The setting stays off until the user opts in, so an upgrade +does not change anyone's volume. A file in the config directory remembers an +in-progress duck: if the process quits or crashes with the microphone open, +the next launch puts the same sink back and then deletes the file. +""" + +from __future__ import annotations + +import json +import logging +import math +import os +import re +import shutil +import subprocess +import threading +from dataclasses import dataclass +from typing import Callable, Optional + +from ..common_types import CancelableTimer, SinkVolumeControl +from ..utils.host_process import host_env +from ..utils.paths import config_dir + +logger = logging.getLogger(__name__) + +# wpctl prints volume with two decimal places (``%.2f``). A restore that +# required an exact match would skip itself on that rounding. One and a half +# percent is enough for the printed value and for ordinary float noise, and +# still smaller than a normal volume-key step. +VOLUME_MATCH_TOLERANCE = 0.015 + +# How long the start cue is allowed to finish before other audio is lowered. +# The tail is a little longer than the WAV so the last samples are not cut. +CUE_TAIL_SECONDS = 0.05 +MAX_DUCK_DELAY_SECONDS = 1.5 + +PENDING_RECORD_NAME = "playback-duck.json" +_COMMAND_TIMEOUT_SECONDS = 2.0 +_MAX_LINEAR_VOLUME = 10.0 # 1000%; anything past this is not a real sink level +_SINK_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:+~-]{0,200}$") +_WPCTL_VOLUME_RE = re.compile(r"Volume:\s*(\d+(?:\.\d+)?)") +_WPCTL_ID_RE = re.compile(r"(?m)^id\s+(\d+)\b") +_PACTL_PERCENT_RE = re.compile(r"(\d+(?:\.\d+)?)\s*%") + +CommandResult = tuple[int, str, str] +CommandRunner = Callable[[list[str]], CommandResult] +Which = Callable[[str], Optional[str]] +Channels = tuple[float, ...] + + +@dataclass(frozen=True) +class PendingDuck: + """Per-channel volume to put back, and the level we left the sink at.""" + + sink_id: str + original_channels: Channels + ducked_channels: Channels + + @property + def original_volume(self) -> float: + """First channel, kept so a one-channel record stays easy to log.""" + return self.original_channels[0] + + @property + def ducked_volume(self) -> float: + """First channel of the level we applied.""" + return self.ducked_channels[0] + + +def duck_delay_seconds( + *, + sound_effects_enabled: bool, + tone: str, + cue_duration_seconds: float, +) -> float: + """Seconds to wait before ducking so the start cue can be heard. + + Sound effects off, or the Off tone, duck immediately. Otherwise the cue's + WAV duration plus a short tail, and never longer than 1.5 seconds. + """ + if not sound_effects_enabled or tone == "off": + return 0.0 + try: + duration = float(cue_duration_seconds) + except (TypeError, ValueError): + duration = 0.0 + if math.isnan(duration) or duration < 0.0: + duration = 0.0 + return min(MAX_DUCK_DELAY_SECONDS, duration + CUE_TAIL_SECONDS) + + +def volumes_match(left: float, right: float) -> bool: + """Whether two linear volumes are the same aside from readout noise.""" + return math.isclose(left, right, rel_tol=0.0, abs_tol=VOLUME_MATCH_TOLERANCE) + + +def channels_match(left: Channels, right: Channels) -> bool: + """Whether every channel matches, so a stereo balance change is visible.""" + if len(left) != len(right): + return False + return all(volumes_match(a, b) for a, b in zip(left, right)) + + +def parse_wpctl_volume(text: str) -> Optional[float]: + """Parse ``wpctl get-volume`` stdout. None when it is not a volume line.""" + match = _WPCTL_VOLUME_RE.search(text or "") + if match is None: + return None + return _finite_volume(float(match.group(1))) + + +def parse_wpctl_sink_id(text: str) -> Optional[str]: + """Parse the leading ``id N`` line from ``wpctl inspect``.""" + match = _WPCTL_ID_RE.search(text or "") + if match is None: + return None + sink_id = match.group(1) + if not _valid_sink_id(sink_id): + return None + return sink_id + + +def parse_pactl_channels(text: str) -> Optional[Channels]: + """Every channel percent on the ``Volume:`` line. Ignores Base Volume. + + Left and right are often different. Saving only the first one and writing + it back to the whole sink would flatten that balance. + """ + for line in (text or "").splitlines(): + stripped = line.strip() + if not stripped.lower().startswith("volume:"): + continue + matches = _PACTL_PERCENT_RE.findall(stripped) + if not matches: + return None + channels: list[float] = [] + for raw in matches: + value = _finite_volume(float(raw) / 100.0) + if value is None: + return None + channels.append(value) + return tuple(channels) + return None + + +def parse_pactl_volume(text: str) -> Optional[float]: + """First channel of :func:`parse_pactl_channels`, or None.""" + channels = parse_pactl_channels(text) + if not channels: + return None + return channels[0] + + +def parse_pactl_short_sink_ids(text: str) -> set[str]: + """Sink indexes and names from ``pactl list short sinks``.""" + found: set[str] = set() + for line in (text or "").splitlines(): + parts = line.split() + if not parts: + continue + if _valid_sink_id(parts[0]): + found.add(parts[0]) + if len(parts) > 1 and _valid_sink_id(parts[1]): + found.add(parts[1]) + return found + + +def _finite_volume(value: float) -> Optional[float]: + if math.isnan(value) or math.isinf(value) or value < 0.0 or value > _MAX_LINEAR_VOLUME: + return None + return value + + +def _valid_sink_id(sink_id: str) -> bool: + """Reject ids that are empty, a flag, or not a single argv token.""" + return bool(_SINK_ID_RE.fullmatch(sink_id)) + + +def _valid_volume(value: float) -> bool: + return _finite_volume(value) is not None + + +def _finite_channels(channels: Channels) -> Optional[Channels]: + """A non-empty tuple of finite volumes, or None.""" + if not channels: + return None + checked: list[float] = [] + for channel in channels: + value = _finite_volume(channel) + if value is None: + return None + checked.append(value) + return tuple(checked) + + +def _output_says_missing(stdout: str, stderr: str) -> bool: + text = f"{stdout}\n{stderr}".lower() + return any( + needle in text + for needle in ("not found", "no such", "does not exist", "no sink", "unknown entity") + ) + + +def _run_host_command(args: list[str]) -> CommandResult: + """Run a host volume tool. The caller handles a non-zero status.""" + completed = subprocess.run( + args, + capture_output=True, + text=True, + timeout=_COMMAND_TIMEOUT_SECONDS, + check=False, + env=host_env(), + ) + return completed.returncode, completed.stdout or "", completed.stderr or "" + + +def _default_playback_duck_settings() -> tuple[bool, int]: + """Read the saved switch and percent without creating a config on first import. + + Constructing ConfigManager on a machine with no config file seeds the + language and shells out. Dictation tests, and a first run that has not + opened Settings, should see the packaged default (off, 20%) instead. + """ + from ..ui.config_manager import ( + CONFIG_FILE, + DEFAULT_PLAYBACK_DUCK_PERCENT, + get_shared_config_manager, + peek_shared_config_manager, + ) + + manager = peek_shared_config_manager() + if manager is None and os.path.isfile(CONFIG_FILE): + manager = get_shared_config_manager() + if manager is None: + return False, DEFAULT_PLAYBACK_DUCK_PERCENT + return manager.is_playback_duck_enabled(), manager.get_playback_duck_percent() + + +def _default_enabled() -> bool: + enabled, _percent = _default_playback_duck_settings() + return enabled + + +def _default_percent() -> int: + _enabled, percent = _default_playback_duck_settings() + return percent + + +class SystemSinkVolume: + """Default-sink volume through ``wpctl``, or ``pactl`` when ``wpctl`` is absent. + + The sink id captured at duck time is the only id later commands receive. + ``@DEFAULT_*`` is used to find that sink, not to restore it: the default + may have moved, and a missing sink must not spill onto another output. + """ + + def __init__( + self, + runner: Optional[CommandRunner] = None, + which: Optional[Which] = None, + ) -> None: + self._runner = runner or _run_host_command + self._which = which or shutil.which + if self._which("wpctl"): + self._backend: Optional[str] = "wpctl" + elif self._which("pactl"): + self._backend = "pactl" + else: + self._backend = None + + def default_sink(self) -> Optional[tuple[str, Channels]]: + """Identify the default sink and read each channel's linear volume.""" + if self._backend == "wpctl": + return self._wpctl_default() + if self._backend == "pactl": + return self._pactl_default() + logger.warning("Neither wpctl nor pactl is available; not changing playback volume") + return None + + def volume_of(self, sink_id: str) -> Optional[Channels]: + """Read ``sink_id``. None on a bad id, a missing sink, or a parse failure.""" + if not _valid_sink_id(sink_id): + logger.warning("Refusing to read sink id %r", sink_id) + return None + if self._backend == "wpctl": + return self._wpctl_volume(sink_id) + if self._backend == "pactl": + return self._pactl_volume(sink_id) + return None + + def sink_exists(self, sink_id: str) -> Optional[bool]: + """Whether ``sink_id`` is still this machine's sink.""" + if not _valid_sink_id(sink_id) or self._backend is None: + return False if self._backend is not None else None + if self._backend == "wpctl": + result = self._run(["wpctl", "inspect", sink_id]) + if result is None: + return None + code, stdout, stderr = result + if code == 0 and parse_wpctl_sink_id(stdout) == sink_id: + return True + if _output_says_missing(stdout, stderr): + return False + return None + result = self._run(["pactl", "list", "short", "sinks"]) + if result is None: + return None + code, stdout, stderr = result + if code != 0: + if _output_says_missing(stdout, stderr): + return False + return None + return sink_id in parse_pactl_short_sink_ids(stdout) + + def set_volume(self, sink_id: str, channels: Channels) -> bool: + """Set one sink, one argument per channel. A failure leaves the server as it was.""" + if not _valid_sink_id(sink_id): + logger.warning("Refusing to set volume on sink id %r", sink_id) + return False + checked = _finite_channels(channels) + if checked is None: + logger.warning("Refusing to set sink %s to an invalid volume %r", sink_id, channels) + return False + if self._backend == "wpctl": + # wpctl reports one sink volume. Per-channel balance is not in that reading. + result = self._run(["wpctl", "set-volume", sink_id, f"{checked[0]:.6f}"]) + elif self._backend == "pactl": + percents = [f"{channel * 100.0:.4f}%" for channel in checked] + result = self._run(["pactl", "set-sink-volume", sink_id, *percents]) + else: + logger.warning("Neither wpctl nor pactl is available; not changing playback volume") + return False + if result is None or result[0] != 0: + detail = "" + if result is not None: + detail = (result[2] or result[1]).strip() + logger.warning( + "Could not set volume on sink %s%s", sink_id, f": {detail}" if detail else "" + ) + return False + return True + + def _wpctl_default(self) -> Optional[tuple[str, Channels]]: + inspected = self._run(["wpctl", "inspect", "@DEFAULT_AUDIO_SINK@"]) + if inspected is None or inspected[0] != 0: + logger.warning("Could not identify the default audio sink") + return None + sink_id = parse_wpctl_sink_id(inspected[1]) + if sink_id is None: + logger.warning("Could not parse the default sink id from wpctl inspect") + return None + volume = self._wpctl_volume(sink_id) + if volume is None: + return None + return sink_id, volume + + def _wpctl_volume(self, sink_id: str) -> Optional[Channels]: + result = self._run(["wpctl", "get-volume", sink_id]) + if result is None or result[0] != 0: + logger.warning("wpctl get-volume failed for sink %s", sink_id) + return None + volume = parse_wpctl_volume(result[1]) + if volume is None: + logger.warning("Could not parse wpctl volume %r", result[1].strip()[:200]) + return None + return (volume,) + + def _pactl_default(self) -> Optional[tuple[str, Channels]]: + named = self._run(["pactl", "get-default-sink"]) + if named is None or named[0] != 0: + logger.warning("Could not identify the default PulseAudio sink") + return None + lines = [line.strip() for line in named[1].splitlines() if line.strip()] + sink_id = lines[-1] if lines else "" + if not _valid_sink_id(sink_id): + logger.warning("Unexpected default sink name %r", sink_id) + return None + volume = self._pactl_volume(sink_id) + if volume is None: + return None + return sink_id, volume + + def _pactl_volume(self, sink_id: str) -> Optional[Channels]: + result = self._run(["pactl", "get-sink-volume", sink_id]) + if result is None or result[0] != 0: + logger.warning("pactl get-sink-volume failed for sink %s", sink_id) + return None + channels = parse_pactl_channels(result[1]) + if channels is None: + logger.warning("Could not parse pactl volume %r", result[1].strip()[:200]) + return None + return channels + + def _run(self, args: list[str]) -> Optional[CommandResult]: + try: + return self._runner(args) + except (OSError, subprocess.SubprocessError) as exc: + logger.warning("Volume command %s failed: %s", args[0] if args else "?", exc) + return None + + +class PlaybackDucker: + """Duck the default sink for one dictation and restore that same sink. + + A second ``duck`` while one is active does nothing, so the percent is not + applied twice. ``restore`` writes the saved volume only when the sink is + still at the ducked level; if the user moved it, their level is kept. + Either way a finished restore deletes the on-disk record. + """ + + def __init__( + self, + control: Optional[SinkVolumeControl] = None, + directory: Optional[str] = None, + *, + enabled: Optional[Callable[[], bool]] = None, + percent: Optional[Callable[[], int]] = None, + recover: bool = True, + ) -> None: + self._control = control if control is not None else SystemSinkVolume() + self._record_path = os.path.join( + directory if directory is not None else config_dir(), + PENDING_RECORD_NAME, + ) + self._enabled_fn = enabled if enabled is not None else _default_enabled + self._percent_fn = percent if percent is not None else _default_percent + self._lock = threading.Lock() + self._pending: Optional[PendingDuck] = None + if recover: + # A previous process may have crashed between duck and restore. + try: + self.restore() + except Exception: + logger.error("Could not restore playback volume from a previous run", exc_info=True) + + def is_enabled(self) -> bool: + """Whether the user asked to lower other audio. Failures count as off.""" + try: + return bool(self._enabled_fn()) + except Exception: + logger.warning("Could not read whether playback ducking is enabled", exc_info=True) + return False + + def duck(self) -> None: + """Lower the default sink to the configured percent of its current volume.""" + with self._lock: + if not self.is_enabled(): + return + if self._load_pending() is not None: + logger.info("Playback is already lowered; not stacking another duck") + return + try: + snapshot = self._control.default_sink() + except Exception: + logger.warning("Could not read the default sink", exc_info=True) + return + if snapshot is None: + logger.warning("Not lowering playback: the default sink volume could not be read") + return + sink_id, original = snapshot + original_channels = _finite_channels(original) + if not _valid_sink_id(sink_id) or original_channels is None: + logger.warning("Not lowering playback: the default sink reading was unusable") + return + percent = self._current_percent() + target = _finite_channels( + tuple(max(0.0, channel * (percent / 100.0)) for channel in original_channels) + ) + if target is None: + logger.warning("Not lowering playback: computed volume %r is unusable", target) + return + # Persist the restore point before changing the sink. If that write + # fails, do not lower anything: a crash would otherwise forget the + # original level. A crash after a successful write can still put it back. + record = PendingDuck(sink_id, original_channels, target) + try: + self._write_pending(record) + except OSError: + logger.warning( + "Could not save the volume to restore for sink %s; not lowering it", + sink_id, + ) + return + self._pending = record + # Percent 100 (or an already-matching level) must not touch the sink. + if channels_match(original_channels, target): + logger.info( + "Playback duck level is %d%%; leaving sink %s at %.3f", + percent, + sink_id, + original_channels[0], + ) + return + try: + applied = self._control.set_volume(sink_id, target) + except Exception: + logger.warning("Could not lower sink %s", sink_id, exc_info=True) + return + if not applied: + # The sink was not changed, so the record would restore nothing + # useful and might later skip a real user volume. + self._forget() + return + logger.info( + "Lowered sink %s from %s to %s (%d%%) while dictating", + sink_id, + original_channels, + target, + percent, + ) + + def restore(self) -> None: + """Put the ducked sink back, unless the user changed it or it is gone.""" + with self._lock: + record = self._load_pending() + if record is None: + return + try: + exists = self._control.sink_exists(record.sink_id) + except Exception: + logger.warning("Could not check sink %s", record.sink_id, exc_info=True) + return + if exists is False: + logger.info( + "Ducked sink %s is gone; dropping the saved volume without touching another", + record.sink_id, + ) + self._forget() + return + if exists is None: + logger.warning( + "Could not check sink %s; leaving the saved volume in place", record.sink_id + ) + return + try: + current = self._control.volume_of(record.sink_id) + except Exception: + logger.warning("Could not read sink %s", record.sink_id, exc_info=True) + return + if current is None: + logger.warning( + "Could not read sink %s; leaving the saved volume in place", record.sink_id + ) + return + if not channels_match(current, record.ducked_channels): + logger.info( + "Playback volume changed during dictation (now %s, ducked %s); " + "not overwriting it", + current, + record.ducked_channels, + ) + self._forget() + return + if not channels_match(current, record.original_channels): + try: + restored = self._control.set_volume(record.sink_id, record.original_channels) + except Exception: + logger.warning("Could not restore sink %s", record.sink_id, exc_info=True) + return + if not restored: + return + logger.info("Restored sink %s to %s", record.sink_id, record.original_channels) + else: + logger.info("Sink %s is already at the saved volume", record.sink_id) + self._forget() + + def _current_percent(self) -> int: + from ..ui.config_manager import ( + DEFAULT_PLAYBACK_DUCK_PERCENT, + clamp_playback_duck_percent, + ) + + try: + return clamp_playback_duck_percent(self._percent_fn()) + except Exception: + logger.warning( + "Could not read the playback-duck level; using %d", + DEFAULT_PLAYBACK_DUCK_PERCENT, + exc_info=True, + ) + return DEFAULT_PLAYBACK_DUCK_PERCENT + + def _load_pending(self) -> Optional[PendingDuck]: + if self._pending is not None: + return self._pending + record = self._read_pending() + self._pending = record + return record + + def _read_pending(self) -> Optional[PendingDuck]: + try: + with open(self._record_path, encoding="utf-8") as handle: + payload = json.load(handle) + except FileNotFoundError: + return None + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + logger.warning("Could not read the playback-duck record: %s", exc) + return None + record = _record_from_payload(payload) + if record is None: + logger.warning("Dropping an unreadable playback-duck record") + self._clear_file() + return None + return record + + def _write_pending(self, record: PendingDuck) -> None: + directory = os.path.dirname(self._record_path) + os.makedirs(directory, exist_ok=True) + temporary = self._record_path + ".tmp" + payload = { + "sink_id": record.sink_id, + "original_volume": record.original_volume, + "ducked_volume": record.ducked_volume, + "original_channels": list(record.original_channels), + "ducked_channels": list(record.ducked_channels), + } + try: + with open(temporary, "w", encoding="utf-8") as handle: + json.dump(payload, handle) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, self._record_path) + except OSError: + try: + os.unlink(temporary) + except OSError: + pass + raise + + def _forget(self) -> None: + self._pending = None + self._clear_file() + + def _clear_file(self) -> None: + try: + os.remove(self._record_path) + except FileNotFoundError: + return + except OSError as exc: + logger.warning("Could not remove the playback-duck record: %s", exc) + + +def _record_from_payload(payload: object) -> Optional[PendingDuck]: + if not isinstance(payload, dict): + return None + sink_id = payload.get("sink_id") + original_raw = payload.get("original_volume") + ducked_raw = payload.get("ducked_volume") + if ( + isinstance(original_raw, bool) + or isinstance(ducked_raw, bool) + or not isinstance(original_raw, (int, float)) + or not isinstance(ducked_raw, (int, float)) + ): + return None + if not isinstance(sink_id, str) or not _valid_sink_id(sink_id): + return None + original = float(original_raw) + ducked = float(ducked_raw) + if not _valid_volume(original) or not _valid_volume(ducked): + return None + original_channels = _channels_from_payload(payload.get("original_channels"), original) + ducked_channels = _channels_from_payload(payload.get("ducked_channels"), ducked) + if original_channels is None or ducked_channels is None: + return None + return PendingDuck(sink_id, original_channels, ducked_channels) + + +def _channels_from_payload(raw: object, fallback: float) -> Optional[Channels]: + """Use a saved channel list, or the single legacy volume when it is absent.""" + if raw is None: + checked = _finite_channels((fallback,)) + return checked + if not isinstance(raw, list) or isinstance(raw, bool): + return None + numbers: list[float] = [] + for item in raw: + if isinstance(item, bool) or not isinstance(item, (int, float)): + return None + numbers.append(float(item)) + return _finite_channels(tuple(numbers)) + + +Scheduler = Callable[[float, Callable[[], None]], CancelableTimer] + + +def _daemon_timer(delay: float, callback: Callable[[], None]) -> threading.Timer: + """Fire ``callback`` on a daemon timer so a missed cancel cannot pin the process.""" + timer = threading.Timer(delay, callback) + timer.daemon = True + timer.start() + return timer + + +class DictationDuckSession: + """When a dictation starts, duck after the start cue; stop cancels or restores. + + The scheduler is injectable so tests can fire or skip the callback without + sleeping. ``cancel`` invalidates a callback that is already running too: + the timer thread and ``cancel`` share a lock, and the duck itself happens + inside that lock, so a stop cannot be followed by a late duck. + """ + + def __init__( + self, + ducker: PlaybackDucker, + *, + enabled: Callable[[], bool], + schedule: Optional[Scheduler] = None, + ) -> None: + self._ducker = ducker + self._enabled_fn = enabled + self._schedule = schedule or _daemon_timer + self._lock = threading.Lock() + self._generation = 0 + self._armed = False + self._timer: Optional[CancelableTimer] = None + + def enabled(self) -> bool: + """Whether this dictation should duck. A settings failure means no.""" + try: + return bool(self._enabled_fn()) + except Exception: + logger.warning("Could not read the playback-duck setting", exc_info=True) + return False + + def start(self, delay_seconds: float) -> None: + """Arm a duck. ``delay_seconds`` of zero still goes through the scheduler.""" + if not self.enabled(): + return + try: + delay = float(delay_seconds) + except (TypeError, ValueError): + delay = 0.0 + if math.isnan(delay) or delay < 0.0: + delay = 0.0 + delay = min(MAX_DUCK_DELAY_SECONDS, delay) + + with self._lock: + self._generation += 1 + generation = self._generation + self._armed = True + previous = self._timer + self._timer = None + self._cancel_timer(previous) + try: + timer = self._schedule(delay, lambda: self._fire(generation)) + except Exception: + logger.error("Could not schedule playback duck", exc_info=True) + with self._lock: + if self._generation == generation: + self._armed = False + return + with self._lock: + if self._generation != generation or not self._armed: + self._cancel_timer(timer) + return + self._timer = timer + + def cancel(self) -> None: + """Drop a duck that has not run. Safe if nothing was armed.""" + with self._lock: + self._generation += 1 + self._armed = False + timer = self._timer + self._timer = None + self._cancel_timer(timer) + + def restore(self) -> None: + """Restore now. Does not cancel; call ``cancel`` first on the stop path.""" + try: + self._ducker.restore() + except Exception: + logger.error("Could not restore playback volume", exc_info=True) + + def _fire(self, generation: int) -> None: + with self._lock: + if generation != self._generation or not self._armed: + return + self._armed = False + self._timer = None + try: + self._ducker.duck() + except Exception: + logger.error("Could not lower playback volume", exc_info=True) + + @staticmethod + def _cancel_timer(timer: Optional[CancelableTimer]) -> None: + if timer is None: + return + try: + timer.cancel() + except Exception: + logger.warning("Could not cancel the playback-duck timer", exc_info=True) + + +def default_dictation_duck_session() -> DictationDuckSession: + """The session the recognition manager uses, including crash recovery.""" + ducker = PlaybackDucker() + return DictationDuckSession(ducker, enabled=ducker.is_enabled) diff --git a/src/vocalinux/common_types.py b/src/vocalinux/common_types.py index 1bdfc4b..5a4cae9 100644 --- a/src/vocalinux/common_types.py +++ b/src/vocalinux/common_types.py @@ -99,3 +99,31 @@ class _EvdevCaptureDevice(Protocol): def active_keys(self) -> Iterable[int]: """Return evdev codes currently down on this device.""" ... + + +class SinkVolumeControl(Protocol): + """Default playback sink. Tests supply a fake; production uses wpctl or pactl.""" + + def default_sink(self) -> Optional[tuple[str, tuple[float, ...]]]: + """``(sink_id, per_channel_linear_volume)``, or None if it cannot be read.""" + ... + + def volume_of(self, sink_id: str) -> Optional[tuple[float, ...]]: + """Per-channel linear volume of ``sink_id``, or None if it cannot be read.""" + ... + + def sink_exists(self, sink_id: str) -> Optional[bool]: + """True if present, False if gone, None if presence could not be checked.""" + ... + + def set_volume(self, sink_id: str, channels: tuple[float, ...]) -> bool: + """Set ``sink_id`` only. False on failure. Never substitute another sink.""" + ... + + +class CancelableTimer(Protocol): + """A scheduled call that can be dropped before it runs.""" + + def cancel(self) -> None: + """Drop the scheduled call if it has not started.""" + ... diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index 2f0e24a..4730acd 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -23,6 +23,7 @@ from typing import TYPE_CHECKING, Callable, Optional if TYPE_CHECKING: import numpy as np +from ..audio.playback_ducker import default_dictation_duck_session, duck_delay_seconds from ..common_types import RecognitionState from ..ui.audio_feedback import play_error_sound, play_start_sound, play_stop_sound from ..ui.config_manager import _multilingual_sibling @@ -1221,6 +1222,16 @@ class SpeechRecognitionManager: self._capture_channels = 1 # Default, updated when device is opened self._capture_downmix_channel = None # Speech-gated sticky N>=3 channel for open stream + # Recover a sink left quiet by a crash before doing anything slow. + # Tests inject a session so this never touches a real audio server. + injected_duck = kwargs.get("playback_duck") + self._playback_duck = ( + default_dictation_duck_session() if injected_duck is None else injected_duck + ) + # Serializes arm vs release: a failure path that already released the + # duck must not be followed by an arm that re-lowers the sink. + self._playback_duck_lock = threading.Lock() + # Create models directory if it doesn't exist os.makedirs(MODELS_DIR, exist_ok=True) @@ -3329,6 +3340,85 @@ class SpeechRecognitionManager: chunk_duration_ms = (1024 / 16000) * 1000 return int(guard_ms / chunk_duration_ms) + def _playback_duck_delay_seconds(self) -> float: + """How long to let the start cue play before lowering other audio. + + Sound effects off, or the Off tone, duck immediately. A mocked or + broken feedback module also ducks immediately rather than guessing. + """ + try: + from ..ui import audio_feedback + + enabled_fn = getattr(audio_feedback, "_is_sound_effects_enabled", None) + tone_fn = getattr(audio_feedback, "_resolved_tone", None) + duration_fn = getattr(audio_feedback, "_wav_duration_seconds", None) + path_fn = getattr(audio_feedback, "tone_sound_path", None) + # Separate checks so the type checker treats each as callable. + if not callable(enabled_fn) or not callable(tone_fn): + return 0.0 + if not callable(duration_fn) or not callable(path_fn): + return 0.0 + enabled = enabled_fn() + tone = tone_fn() + if not isinstance(enabled, bool) or not isinstance(tone, str): + return 0.0 + if not enabled or tone == "off": + return duck_delay_seconds( + sound_effects_enabled=False, tone=tone, cue_duration_seconds=0.0 + ) + path = path_fn(tone, "start") + if not isinstance(path, str): + return 0.0 + duration = duration_fn(path) + if isinstance(duration, bool) or not isinstance(duration, (int, float)): + return 0.0 + return duck_delay_seconds( + sound_effects_enabled=True, + tone=tone, + cue_duration_seconds=float(duration), + ) + except Exception: + logger.warning( + "Could not measure the start cue; ducking other audio immediately", + exc_info=True, + ) + return 0.0 + + def _arm_playback_duck(self) -> None: + """Schedule a duck once dictation is actually listening.""" + try: + if not self._playback_duck.enabled(): + return + # The capture thread may already have failed and released the duck; + # arming now would lower playback in the error state with nothing + # left to put it back. + with self._playback_duck_lock: + if not self.should_record or self.state != RecognitionState.LISTENING: + return + self._playback_duck.start(self._playback_duck_delay_seconds()) + except Exception: + logger.error("Could not schedule playback duck", exc_info=True) + + def _cancel_pending_playback_duck(self) -> None: + """Drop a duck whose timer has not fired. Does not change the volume.""" + try: + self._playback_duck.cancel() + except Exception: + logger.error("Could not cancel playback duck", exc_info=True) + + def release_playback_duck(self) -> None: + """Cancel a not-yet-applied duck and restore the sink. Safe to call twice. + + Used when dictation ends, including error exits and quitting while the + microphone is still open. Failures are logged and never raised. + """ + with self._playback_duck_lock: + self._cancel_pending_playback_duck() + try: + self._playback_duck.restore() + except Exception: + logger.error("Could not restore playback volume", exc_info=True) + def start_recognition(self, mode: str = "toggle") -> bool: """Start the speech recognition process. @@ -3406,6 +3496,8 @@ class SpeechRecognitionManager: self.recognition_thread = threading.Thread(target=self._perform_recognition) self.recognition_thread.daemon = True self.recognition_thread.start() + # After the threads exist: a start that returned early must not duck. + self._arm_playback_duck() return True def stop_recognition(self): @@ -3415,8 +3507,11 @@ class SpeechRecognitionManager: logger.info("Stopping speech recognition") - # Stop recording FIRST to prevent capturing the stop sound + # Mark recording over before touching the duck so a concurrent arm + # sees the dictation as ended. The volume goes back only once the + # microphone thread has left the device, and before the stop cue. self.should_record = False + self._cancel_pending_playback_duck() # Wait for audio thread to finish recording and enqueue any pending audio # This is critical to prevent race condition where recognition thread exits @@ -3424,6 +3519,8 @@ class SpeechRecognitionManager: if self.audio_thread and self.audio_thread.is_alive(): self.audio_thread.join(timeout=2.0) + self.release_playback_duck() + # Play stop sound now that the audio thread is done and cannot capture it. # Kept before buffer processing so the cue still feels immediate. play_stop_sound() @@ -3476,6 +3573,8 @@ class SpeechRecognitionManager: except ImportError as e: logger.error(f"Failed to import required audio libraries: {e}") logger.error("Please install required dependencies: pip install pyaudio numpy") + self.should_record = False + self.release_playback_duck() play_error_sound() self._update_state(RecognitionState.ERROR) return @@ -3582,6 +3681,8 @@ class SpeechRecognitionManager: stream = self._audio_stream CHANNELS = self._capture_channels else: + self.should_record = False + self.release_playback_duck() play_error_sound() audio.terminate() self._update_state(RecognitionState.ERROR) @@ -3595,6 +3696,7 @@ class SpeechRecognitionManager: self._recording_segment_has_speech = False log_level_interval = 0 # Counter for periodic level logging max_level_seen = 0.0 + capture_failed = False # Accumulator for 512-sample Silero chunks. When the capture rate # is higher than 16 kHz (e.g. 48 kHz), resampling produces fewer # than 1024 samples per read (~341 at 48 kHz), so the buffer may @@ -3761,16 +3863,27 @@ class SpeechRecognitionManager: continue # Continue recording with new stream else: logger.error("Audio reconnection failed, stopping recording") + capture_failed = True break else: logger.warning( "Audio error occurred too soon after last error, stopping recording" ) + capture_failed = True break except Exception as e: logger.error(f"Unexpected error reading audio data: {e}") + capture_failed = True break + # A dead microphone must not leave other audio lowered, or the + # session stuck in LISTENING, until the user happens to stop. + if capture_failed: + self.should_record = False + self.release_playback_duck() + play_error_sound() + self._update_state(RecognitionState.ERROR) + # Clean up _safe_close_stream(stream) @@ -3799,6 +3912,8 @@ class SpeechRecognitionManager: except Exception as e: logger.error(f"Error in audio recording: {e}") + self.should_record = False + self.release_playback_duck() play_error_sound() self._update_state(RecognitionState.ERROR) diff --git a/src/vocalinux/ui/config_manager.py b/src/vocalinux/ui/config_manager.py index 5b7d089..75377da 100644 --- a/src/vocalinux/ui/config_manager.py +++ b/src/vocalinux/ui/config_manager.py @@ -41,6 +41,7 @@ SOUND_EFFECT_TONES: tuple[tuple[str, str], ...] = ( ) SOUND_EFFECT_TONE_IDS = frozenset(tone_id for tone_id, _label in SOUND_EFFECT_TONES) DEFAULT_SOUND_EFFECT_TONE = "voca" +DEFAULT_PLAYBACK_DUCK_PERCENT = 20 PASTE_SHORTCUTS: tuple[tuple[str, str], ...] = ( ("auto", "Auto-detect"), @@ -85,6 +86,21 @@ def _backend_from_config(config: Any) -> tuple[bool, Any]: return True, text_injection.get("backend", DEFAULT_TEXT_INJECTION_BACKEND) +def clamp_playback_duck_percent(value: Any) -> int: + """Return ``value`` as an integer percent in 0..100. + + Missing or unusable values become the default (20). Booleans are rejected + because ``bool`` is an ``int`` subclass and must not become 0 or 1. + """ + if isinstance(value, bool) or value is None: + return DEFAULT_PLAYBACK_DUCK_PERCENT + try: + percent = int(round(float(value))) + except (TypeError, ValueError, OverflowError): + return DEFAULT_PLAYBACK_DUCK_PERCENT + return max(0, min(100, percent)) + + # Default configuration DEFAULT_CONFIG = { "speech_recognition": { # Changed section name @@ -124,6 +140,11 @@ DEFAULT_CONFIG = { "enabled": True, # Master mute for start/stop/error cues "tone": "voca", # Family catalog id; missing/unknown also resolve to voca }, + # Off by default so an upgrade does not change the user's speaker volume. + "playback_duck": { + "enabled": False, # Lower the default sink while the microphone is open + "percent": 20, # Percent of the current volume; 0 is silent + }, "shortcuts": { "toggle_recognition": "right_alt+right_alt", "mode": "push_to_talk", # "toggle" or "push_to_talk" @@ -731,6 +752,29 @@ class ConfigManager: self.config["sound_effects"] = {} self.config["sound_effects"]["tone"] = normalize_sound_effect_tone(tone) + def is_playback_duck_enabled(self) -> bool: + """Whether other audio is lowered while dictating. Off when unset.""" + enabled = self.config.get("playback_duck", {}).get("enabled", False) + return enabled if isinstance(enabled, bool) else False + + def set_playback_duck_enabled(self, enabled: bool) -> None: + """Enable or disable lowering other audio while dictating.""" + if "playback_duck" not in self.config: + self.config["playback_duck"] = {} + self.config["playback_duck"]["enabled"] = bool(enabled) + + def get_playback_duck_percent(self) -> int: + """Percent of the current volume to use while dictating, clamped 0–100.""" + return clamp_playback_duck_percent( + self.config.get("playback_duck", {}).get("percent", DEFAULT_PLAYBACK_DUCK_PERCENT) + ) + + def set_playback_duck_percent(self, percent: int) -> None: + """Save the duck level. Values outside 0–100, and non-numbers, are clamped.""" + if "playback_duck" not in self.config: + self.config["playback_duck"] = {} + self.config["playback_duck"]["percent"] = clamp_playback_duck_percent(percent) + def get_paste_shortcut(self) -> str: """Return the clipboard-paste shortcut id (auto when unset or unknown).""" return normalize_paste_shortcut(self.config.get("text_injection", {}).get("paste_shortcut")) @@ -765,6 +809,16 @@ _shared_instance: Optional[ConfigManager] = None _shared_instance_lock = threading.Lock() +def peek_shared_config_manager() -> Optional[ConfigManager]: + """Return the process-wide ConfigManager, or None if nothing has created it. + + Callers that only need a setting during dictation use this so a missing + config file does not construct a manager (that seeds language and can + shell out) just to discover the packaged default. + """ + return _shared_instance + + def get_shared_config_manager() -> ConfigManager: """Return the process-wide ConfigManager, creating it on first use. diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index 498f1dd..449d010 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -2954,6 +2954,50 @@ class SettingsDialog(Gtk.Dialog): self.sound_tone_combo.connect("changed", self._on_sound_tone_changed) self.preview_tone_btn.connect("clicked", self._on_preview_tone_clicked) + # Other audio: opt-in duck of the default sink for the length of a dictation. + other_audio = PreferencesGroup( + title="Other audio", + keywords=("duck", "volume", "speakers", "music"), + ) + self.duck_playback_switch = Gtk.Switch() + self.duck_playback_switch.set_tooltip_text( + "Turn speakers and headphones down while the microphone is open, " + "then put the volume back" + ) + duck_switch_row = PreferenceRow( + title="Lower other audio while dictating", + subtitle=( + "Turn speakers and headphones down while the microphone is open, " + "then put the volume back." + ), + widget=self.duck_playback_switch, + keywords=("duck", "volume", "speakers", "music", "headphones"), + ) + other_audio.add_row(duck_switch_row) + + self.duck_level_scale = Gtk.Scale.new_with_range(Gtk.Orientation.HORIZONTAL, 0, 100, 1) + self.duck_level_scale.set_digits(0) + self.duck_level_scale.set_round_digits(0) + self.duck_level_scale.set_value_pos(Gtk.PositionType.RIGHT) + self.duck_level_scale.set_value(20) + self.duck_level_scale.set_draw_value(True) + self.duck_level_scale.set_size_request(180, -1) + self.duck_level_scale.set_hexpand(False) + self.duck_level_scale.set_halign(Gtk.Align.END) + self.duck_level_scale.set_sensitive(False) + self.duck_level_scale.set_tooltip_text("Percent of the current volume. 0 is silent.") + _prevent_scroll_on_hover(self.duck_level_scale) + duck_level_row = PreferenceRow( + title="Level while dictating", + subtitle="Percent of the current volume. 0 is silent.", + widget=self.duck_level_scale, + keywords=("duck", "volume", "speakers", "music"), + ) + other_audio.add_row(duck_level_row) + self.audio_tab.pack_start(other_audio, False, False, 0) + self.duck_playback_switch.connect("state-set", self._on_duck_playback_toggled) + self.duck_level_scale.connect("value-changed", self._on_duck_level_changed) + # Populate devices self._populate_audio_devices() self.audio_device_combo.connect("changed", self._on_audio_device_changed) @@ -3435,6 +3479,24 @@ class SettingsDialog(Gtk.Dialog): logger.info(f"Sound effects {'enabled' if enabled else 'disabled'}") return False + def _on_duck_playback_toggled(self, _widget: Gtk.Widget, state: bool) -> bool: + enabled = bool(state) + self.duck_level_scale.set_sensitive(enabled) + if self._initializing or self._applying_settings: + return False + logger.info("Lower other audio while dictating toggled: %s", enabled) + self.config_manager.set_playback_duck_enabled(enabled) + self.config_manager.save_settings() + return False + + def _on_duck_level_changed(self, scale: Gtk.Scale) -> None: + if self._initializing or self._applying_settings: + return + percent = int(round(scale.get_value())) + logger.info("Playback duck level set to %s%%", percent) + self.config_manager.set_playback_duck_percent(percent) + self.config_manager.save_settings() + def _sync_tone_preview_button(self, tone_id: str) -> None: """Update the two-stage preview button next to the tone combo.""" if tone_id == "off": @@ -5519,6 +5581,10 @@ class SettingsDialog(Gtk.Dialog): if not self.paste_shortcut_combo.set_active_id(paste_shortcut): self.paste_shortcut_combo.set_active_id(DEFAULT_PASTE_SHORTCUT) self.sound_effects_switch.set_active(self.config_manager.is_sound_effects_enabled()) + duck_enabled = self.config_manager.is_playback_duck_enabled() + self.duck_level_scale.set_value(self.config_manager.get_playback_duck_percent()) + self.duck_playback_switch.set_active(duck_enabled) + self.duck_level_scale.set_sensitive(duck_enabled) tone_id = self.config_manager.get_sound_effects_tone() if not self.sound_tone_combo.set_active_id(tone_id): self.sound_tone_combo.set_active_id(DEFAULT_SOUND_EFFECT_TONE) diff --git a/src/vocalinux/ui/tray_indicator.py b/src/vocalinux/ui/tray_indicator.py index e5483b6..21d6a81 100644 --- a/src/vocalinux/ui/tray_indicator.py +++ b/src/vocalinux/ui/tray_indicator.py @@ -1136,6 +1136,17 @@ class TrayIndicator: """Quit the application.""" logger.info("Quitting application") + # stop_recognition is not called here (it would play the stop cue and + # join the capture thread). Put the speakers back before the process + # exits; a crash that skips this still restores on the next launch. + engine = getattr(self, "speech_engine", None) + release = getattr(engine, "release_playback_duck", None) + if callable(release): + try: + release() + except Exception: + logger.error("Could not restore playback volume while quitting", exc_info=True) + if self._suspend_handler is not None: self._suspend_handler.shutdown() diff --git a/tests/test_playback_ducker.py b/tests/test_playback_ducker.py new file mode 100644 index 0000000..b1061b4 --- /dev/null +++ b/tests/test_playback_ducker.py @@ -0,0 +1,747 @@ +"""Playback ducking: lower the default sink while dictating, then restore it. + +These tests never spawn wpctl or pactl. The system backend is driven with a +fake command runner, and the recognition hook with a fake clock. +""" + +import json +import sys +from pathlib import Path +from typing import Any, Callable, Iterator, Optional, Union +from unittest.mock import MagicMock, patch + +import pytest + +from vocalinux.audio import playback_ducker as duck +from vocalinux.audio.playback_ducker import ( + DictationDuckSession, + PlaybackDucker, + SystemSinkVolume, + duck_delay_seconds, +) +from vocalinux.ui.config_manager import ( + DEFAULT_CONFIG, + DEFAULT_PLAYBACK_DUCK_PERCENT, + ConfigManager, + clamp_playback_duck_percent, +) + + +@pytest.fixture(autouse=True) +def _numpy_is_real_while_comparing() -> Iterator[None]: + """Keep pytest.approx working in the full suite. + + Older tests assign ``sys.modules["numpy"] = MagicMock()`` at import and + never put the real module back. ``pytest.approx`` then does + ``isinstance(value, numpy.bool_)`` and raises TypeError. + """ + current = sys.modules.get("numpy") + real = getattr(sys, "_vocalinux_real_numpy", None) + if isinstance(current, MagicMock) and real is not None: + sys.modules["numpy"] = real + try: + yield + finally: + sys.modules["numpy"] = current + else: + yield + + +class FakeSink: + """In-memory default sink. ``present`` False means that sink is gone.""" + + def __init__( + self, + sink_id: str = "sink-a", + volume: float = 0.5, + channels: Optional[tuple[float, ...]] = None, + ) -> None: + self.sink_id = sink_id + self._channels = channels if channels is not None else (float(volume),) + self.present = True + self.fail_read = False + self.sets: list[tuple[str, Union[float, tuple[float, ...]]]] = [] + + @property + def volume(self) -> float: + return self._channels[0] + + @volume.setter + def volume(self, value: float) -> None: + # A single knob moves every channel together. Tests that assign + # ``sink.volume`` are describing that. + self._channels = tuple(float(value) for _ in self._channels) + + def default_sink(self) -> Optional[tuple[str, tuple[float, ...]]]: + if not self.present or self.fail_read: + return None + return self.sink_id, self._channels + + def volume_of(self, sink_id: str) -> Optional[tuple[float, ...]]: + if sink_id != self.sink_id or not self.present or self.fail_read: + return None + return self._channels + + def sink_exists(self, sink_id: str) -> bool: + return sink_id == self.sink_id and self.present + + def set_volume(self, sink_id: str, linear: tuple[float, ...]) -> bool: + recorded: Union[float, tuple[float, ...]] = linear[0] if len(linear) == 1 else linear + self.sets.append((sink_id, recorded)) + if sink_id != self.sink_id or not self.present: + return False + self._channels = tuple(linear) + return True + + +def _ducker( + tmp_path: Path, + sink: FakeSink, + *, + enabled: bool = True, + percent: int = 20, + recover: bool = True, +) -> PlaybackDucker: + return PlaybackDucker( + sink, + str(tmp_path), + enabled=lambda: enabled, + percent=lambda: percent, + recover=recover, + ) + + +def _record(tmp_path: Path) -> dict: + return json.loads((tmp_path / duck.PENDING_RECORD_NAME).read_text(encoding="utf-8")) + + +def _real_audio_feedback(monkeypatch: pytest.MonkeyPatch) -> Any: + """Module object the recognition manager's ``import audio_feedback`` returns. + + ``test_audio_feedback`` leaves a MagicMock in ``sys.modules``. The package + attribute can still be the real module, and that is the one a later import + uses. Patching the mock measures the real cue file instead of the stub. + """ + import importlib + + import vocalinux.ui as ui_pkg + + key = "vocalinux.ui.audio_feedback" + bound = getattr(ui_pkg, "audio_feedback", None) + leaked = sys.modules.get(key) + if isinstance(bound, MagicMock) or isinstance(leaked, MagicMock): + # Import the real module, then put the leak back so monkeypatch + # restores it. Leaving the real module in place breaks later tests + # that expect the mock (test_text_injector). + if key in sys.modules: + del sys.modules[key] + if isinstance(bound, MagicMock): + delattr(ui_pkg, "audio_feedback") + real = importlib.import_module(key) + if leaked is None: + sys.modules.pop(key, None) + else: + sys.modules[key] = leaked + if isinstance(bound, MagicMock): + setattr(ui_pkg, "audio_feedback", bound) + elif bound is not None: + real = bound + else: + real = importlib.import_module(key) + monkeypatch.setattr(ui_pkg, "audio_feedback", real, raising=False) + monkeypatch.setitem(sys.modules, key, real) + return real + + +class _Timer: + def __init__(self, delay: float, callback: Callable[[], None]) -> None: + self.delay = delay + self.callback = callback + self.cancelled = False + + def cancel(self) -> None: + self.cancelled = True + + def fire(self) -> None: + # Invoke even after cancel so the test proves the generation guard, + # not merely that cancel dropped the callback. + self.callback() + + +class _Clock: + def __init__(self) -> None: + self.timers: list[_Timer] = [] + + def __call__(self, delay: float, callback: Callable[[], None]) -> "_Timer": + timer = _Timer(delay, callback) + self.timers.append(timer) + return timer + + +class _FakeDucker: + def __init__(self) -> None: + self.ducks = 0 + self.restores = 0 + + def duck(self) -> None: + self.ducks += 1 + + def restore(self) -> None: + self.restores += 1 + + +class _RecordingSession: + def __init__(self, enabled: bool = True) -> None: + self._enabled = enabled + self.events: list = [] + + def enabled(self) -> bool: + return self._enabled + + def start(self, delay_seconds: float) -> None: + self.events.append(("start", delay_seconds)) + + def cancel(self) -> None: + self.events.append("cancel") + + def restore(self) -> None: + self.events.append("restore") + + +def test_duck_scales_saved_volume_and_does_not_stack(tmp_path: Path) -> None: + sink = FakeSink(volume=0.5) + ducker = _ducker(tmp_path, sink, percent=20) + + ducker.duck() + + assert sink.volume == pytest.approx(0.1) + assert sink.sets == [("sink-a", pytest.approx(0.1))] + saved = _record(tmp_path) + assert saved["sink_id"] == "sink-a" + assert saved["original_volume"] == pytest.approx(0.5) + assert saved["ducked_volume"] == pytest.approx(0.1) + assert saved["original_channels"] == pytest.approx((0.5,)) + assert saved["ducked_channels"] == pytest.approx((0.1,)) + + sink.volume = 0.9 + ducker.duck() + + assert sink.volume == pytest.approx(0.9) + assert len(sink.sets) == 1 + + +def test_zero_percent_silences_and_disabled_does_nothing(tmp_path: Path) -> None: + sink = FakeSink(volume=0.8) + silent = _ducker(tmp_path, sink, percent=0) + silent.duck() + assert sink.volume == 0.0 + assert sink.sets == [("sink-a", 0.0)] + + untouched = FakeSink(volume=0.8) + disabled = _ducker(tmp_path / "off", untouched, enabled=False, percent=0) + (tmp_path / "off").mkdir() + disabled.duck() + assert untouched.volume == 0.8 + assert untouched.sets == [] + assert not (tmp_path / "off" / duck.PENDING_RECORD_NAME).exists() + + +def test_unwritable_restore_point_does_not_lower_the_sink( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + sink = FakeSink(volume=0.5) + ducker = _ducker(tmp_path, sink, percent=20) + + def refuse(_record: duck.PendingDuck) -> None: + raise OSError("read-only config") + + monkeypatch.setattr(ducker, "_write_pending", refuse) + ducker.duck() + + assert sink.volume == pytest.approx(0.5) + assert sink.sets == [] + assert ducker._pending is None + assert not (tmp_path / duck.PENDING_RECORD_NAME).exists() + + +def test_percent_100_does_not_change_the_sink(tmp_path: Path) -> None: + sink = FakeSink(volume=0.5) + ducker = _ducker(tmp_path, sink, percent=100) + + ducker.duck() + + assert sink.volume == pytest.approx(0.5) + assert sink.sets == [] + assert _record(tmp_path)["ducked_volume"] == pytest.approx(0.5) + ducker.duck() + assert sink.sets == [] + + +def test_restore_writes_saved_volume_unless_the_user_moved_it(tmp_path: Path) -> None: + sink = FakeSink(volume=0.5) + ducker = _ducker(tmp_path, sink, percent=20) + ducker.duck() + + sink.volume = 0.1 + 0.01 # readout noise, still our ducked level + ducker.restore() + assert sink.volume == pytest.approx(0.5) + assert not (tmp_path / duck.PENDING_RECORD_NAME).exists() + + ducker.duck() + sink.volume = 0.42 # user moved it while dictating + sets_before = len(sink.sets) + ducker.restore() + assert sink.volume == pytest.approx(0.42) + assert len(sink.sets) == sets_before + assert not (tmp_path / duck.PENDING_RECORD_NAME).exists() + + +def test_unreadable_volume_keeps_the_record_and_a_missing_sink_drops_it(tmp_path: Path) -> None: + sink = FakeSink(volume=0.5) + ducker = _ducker(tmp_path, sink, percent=20) + ducker.duck() + sink.fail_read = True + + ducker.restore() + + assert (tmp_path / duck.PENDING_RECORD_NAME).exists() + assert sink.sets == [("sink-a", pytest.approx(0.1))] + + sink.fail_read = False + sink.present = False + ducker.restore() + + assert sink.volume == pytest.approx(0.1) + assert [target for target, _level in sink.sets] == ["sink-a"] + assert len(sink.sets) == 1 + assert not (tmp_path / duck.PENDING_RECORD_NAME).exists() + + +def test_new_instance_restores_pending_record_once(tmp_path: Path) -> None: + sink = FakeSink(volume=0.8) + first = _ducker(tmp_path, sink, percent=25) + first.duck() + assert sink.volume == pytest.approx(0.2) + assert (tmp_path / duck.PENDING_RECORD_NAME).is_file() + + # Setting off must not skip crash recovery: the previous process already ducked. + second = _ducker(tmp_path, sink, enabled=False, percent=25) + assert sink.volume == pytest.approx(0.8) + assert not (tmp_path / duck.PENDING_RECORD_NAME).exists() + sets_after_restore = len(sink.sets) + + third = _ducker(tmp_path, sink, percent=25) + assert sink.volume == pytest.approx(0.8) + assert len(sink.sets) == sets_after_restore + assert third is not None + + +def test_corrupt_record_is_dropped_without_touching_volume(tmp_path: Path) -> None: + sink = FakeSink(volume=0.5) + path = tmp_path / duck.PENDING_RECORD_NAME + path.write_text( + json.dumps({"sink_id": "-rf", "original_volume": 0.5, "ducked_volume": 0.1}), + encoding="utf-8", + ) + + _ducker(tmp_path, sink) + + assert sink.sets == [] + assert sink.volume == pytest.approx(0.5) + assert not path.exists() + + +def test_duck_delay_waits_for_the_cue_and_caps() -> None: + assert ( + duck_delay_seconds(sound_effects_enabled=False, tone="voca", cue_duration_seconds=0.4) + == 0.0 + ) + assert ( + duck_delay_seconds(sound_effects_enabled=True, tone="off", cue_duration_seconds=0.4) == 0.0 + ) + assert duck_delay_seconds( + sound_effects_enabled=True, tone="voca", cue_duration_seconds=0.4 + ) == pytest.approx(0.45) + assert ( + duck_delay_seconds(sound_effects_enabled=True, tone="voca", cue_duration_seconds=10) == 1.5 + ) + assert duck_delay_seconds( + sound_effects_enabled=True, tone="voca", cue_duration_seconds=-1 + ) == pytest.approx(0.05) + + +def test_stop_before_the_timer_does_not_duck_and_stop_after_restores() -> None: + clock = _Clock() + backend = _FakeDucker() + session = DictationDuckSession(backend, enabled=lambda: True, schedule=clock) + + session.start(0.45) + assert len(clock.timers) == 1 + assert clock.timers[0].delay == pytest.approx(0.45) + assert backend.ducks == 0 + + session.cancel() + session.restore() + clock.timers[0].fire() + + assert backend.ducks == 0 + assert backend.restores == 1 + + clock = _Clock() + backend = _FakeDucker() + session = DictationDuckSession(backend, enabled=lambda: True, schedule=clock) + session.start(0.2) + clock.timers[0].fire() + assert backend.ducks == 1 + session.cancel() + session.restore() + clock.timers[0].fire() + assert backend.ducks == 1 + assert backend.restores == 1 + + +def test_disabled_session_does_not_schedule() -> None: + clock = _Clock() + session = DictationDuckSession(_FakeDucker(), enabled=lambda: False, schedule=clock) + session.start(0.4) + assert clock.timers == [] + + +def test_daemon_timer_is_a_daemon_and_can_be_cancelled(monkeypatch: pytest.MonkeyPatch) -> None: + created = {} + + class FakeTimer: + def __init__(self, delay: float, callback: Callable[[], None]) -> None: + created["delay"] = delay + created["callback"] = callback + self.daemon = False + + def start(self) -> None: + created["started"] = True + + def cancel(self) -> None: + created["cancelled"] = True + + monkeypatch.setattr(duck.threading, "Timer", FakeTimer) + timer = duck._daemon_timer(0.2, lambda: None) + assert created["delay"] == pytest.approx(0.2) + assert created["started"] is True + assert timer.daemon is True + timer.cancel() + assert created["cancelled"] is True + + +def test_wpctl_parse_and_commands_do_not_set_on_parse_failure() -> None: + assert duck.parse_wpctl_volume("Volume: 0.40\n") == pytest.approx(0.40) + assert duck.parse_wpctl_volume("Volume: 0.40 [MUTED]\n") == pytest.approx(0.40) + assert duck.parse_wpctl_volume("Volume: 1.25\n") == pytest.approx(1.25) + assert duck.parse_wpctl_volume("not a volume") is None + assert duck.parse_wpctl_sink_id("id 48, type PipeWire:Interface:Node\n") == "48" + + pactl = ( + "Volume: front-left: 32768 / 50% / -18.06 dB, " + "front-right: 32768 / 40% / -23.88 dB\n" + " balance 0.00\n" + "Base Volume: 65536 / 100% / 0.00 dB\n" + ) + assert duck.parse_pactl_volume(pactl) == pytest.approx(0.50) + + +def test_wpctl_is_preferred_and_parse_failure_does_not_set(tmp_path: Path) -> None: + calls = [] + + def runner(args: list[str]) -> tuple[int, str, str]: + calls.append(list(args)) + if args[:3] == ["wpctl", "inspect", "@DEFAULT_AUDIO_SINK@"]: + return 0, "id 7, type PipeWire:Interface:Node\n", "" + if args[:3] == ["wpctl", "get-volume", "7"]: + return 0, "nope\n", "" + if args[1] == "set-volume": + raise AssertionError(f"set-volume must not run after a parse failure: {args}") + return 1, "", "unexpected" + + control = SystemSinkVolume( + runner=runner, + which=lambda name: "/usr/bin/wpctl" if name == "wpctl" else "/usr/bin/pactl", + ) + PlaybackDucker(control, str(tmp_path), enabled=lambda: True, percent=lambda: 20).duck() + + assert calls[0][0] == "wpctl" + assert not any(args[1] == "set-volume" for args in calls) + assert not (tmp_path / duck.PENDING_RECORD_NAME).exists() + + +def test_wpctl_sets_the_resolved_sink_and_a_missing_sink_is_not_replaced(tmp_path: Path) -> None: + calls = [] + + def runner(args: list[str]) -> tuple[int, str, str]: + calls.append(list(args)) + if args[:3] == ["wpctl", "inspect", "@DEFAULT_AUDIO_SINK@"]: + return 0, "id 7, type PipeWire:Interface:Node\n", "" + if args[:3] == ["wpctl", "get-volume", "7"]: + return 0, "Volume: 0.50\n", "" + if args[:3] == ["wpctl", "set-volume", "7"]: + return 0, "", "" + return 1, "", "unexpected" + + control = SystemSinkVolume(runner=runner, which=lambda name: "/usr/bin/" + name) + PlaybackDucker(control, str(tmp_path), enabled=lambda: True, percent=lambda: 20).duck() + assert ["wpctl", "set-volume", "7", "0.100000"] in calls + + def missing(args: list[str]) -> tuple[int, str, str]: + if args[:3] == ["wpctl", "inspect", "7"]: + return 3, "", "Object '7' not found\n" + if args[1] == "set-volume": + raise AssertionError(args) + return 1, "", "unexpected" + + SystemSink = SystemSinkVolume(runner=missing, which=lambda name: "/usr/bin/" + name) + PlaybackDucker(SystemSink, str(tmp_path), enabled=lambda: True, percent=lambda: 20) + assert not (tmp_path / duck.PENDING_RECORD_NAME).exists() + + +def test_pactl_is_used_when_wpctl_is_absent(tmp_path: Path) -> None: + calls = [] + state = {"volume": 0.50} + sink_name = "alsa_output.pci-0000_00_1f.3.analog-stereo" + + def runner(args: list[str]) -> tuple[int, str, str]: + calls.append(list(args)) + if args[:2] == ["pactl", "get-default-sink"]: + return 0, sink_name + "\n", "" + if args[:2] == ["pactl", "get-sink-volume"]: + percent = state["volume"] * 100.0 + text = ( + f"Volume: front-left: 1 / {percent:.0f}% / 0 dB, " + f"front-right: 1 / {percent:.0f}% / 0 dB\n" + "Base Volume: 65536 / 100% / 0.00 dB\n" + ) + return 0, text, "" + if args[:2] == ["pactl", "set-sink-volume"]: + state["volume"] = float(args[3].rstrip("%")) / 100.0 + return 0, "", "" + if args[:3] == ["pactl", "list", "short"]: + return 0, f"7\t{sink_name}\tPipeWire\n", "" + return 1, "", "unexpected" + + control = SystemSinkVolume( + runner=runner, + which=lambda name: None if name == "wpctl" else "/usr/bin/pactl", + ) + ducker = PlaybackDucker( + control, str(tmp_path), enabled=lambda: True, percent=lambda: 20, recover=False + ) + ducker.duck() + assert ["pactl", "set-sink-volume", sink_name, "10.0000%", "10.0000%"] in calls + ducker.restore() + assert ["pactl", "set-sink-volume", sink_name, "50.0000%", "50.0000%"] in calls + assert state["volume"] == pytest.approx(0.50) + + +def test_pactl_restores_each_channel(tmp_path: Path) -> None: + calls = [] + channels = [0.50, 0.40] + + def runner(args: list[str]) -> tuple[int, str, str]: + calls.append(list(args)) + if args[:2] == ["pactl", "get-default-sink"]: + return 0, "analog-stereo\n", "" + if args[:2] == ["pactl", "get-sink-volume"]: + text = ( + "Volume: front-left: 1 / " + f"{channels[0] * 100:.0f}% / 0 dB, front-right: 1 / " + f"{channels[1] * 100:.0f}% / 0 dB\n" + "Base Volume: 65536 / 100% / 0.00 dB\n" + ) + return 0, text, "" + if args[:2] == ["pactl", "set-sink-volume"]: + channels[0] = float(args[3].rstrip("%")) / 100.0 + channels[1] = float(args[4].rstrip("%")) / 100.0 + return 0, "", "" + if args[:3] == ["pactl", "list", "short"]: + return 0, "7\tanalog-stereo\tPipeWire\n", "" + return 1, "", "unexpected" + + control = SystemSinkVolume( + runner=runner, + which=lambda name: None if name == "wpctl" else "/usr/bin/pactl", + ) + ducker = PlaybackDucker( + control, str(tmp_path), enabled=lambda: True, percent=lambda: 20, recover=False + ) + ducker.duck() + assert ["pactl", "set-sink-volume", "analog-stereo", "10.0000%", "8.0000%"] in calls + ducker.restore() + assert ["pactl", "set-sink-volume", "analog-stereo", "50.0000%", "40.0000%"] in calls + assert channels == pytest.approx([0.50, 0.40]) + + +def test_playback_duck_percent_clamps_and_round_trips( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + config_dir = tmp_path / "vocalinux" + config_dir.mkdir() + config_file = config_dir / "config.json" + monkeypatch.setattr("vocalinux.ui.config_manager.CONFIG_DIR", str(config_dir)) + monkeypatch.setattr("vocalinux.ui.config_manager.CONFIG_FILE", str(config_file)) + monkeypatch.setattr( + "vocalinux.utils.system_language.detect_system_language", lambda *_a, **_k: None + ) + + assert DEFAULT_CONFIG["playback_duck"] == {"enabled": False, "percent": 20} + assert DEFAULT_PLAYBACK_DUCK_PERCENT == 20 + assert clamp_playback_duck_percent(True) == 20 + assert clamp_playback_duck_percent(None) == 20 + assert clamp_playback_duck_percent("nope") == 20 + assert clamp_playback_duck_percent(-4) == 0 + assert clamp_playback_duck_percent(150) == 100 + assert clamp_playback_duck_percent(20.4) == 20 + + fresh = ConfigManager() + assert fresh.is_playback_duck_enabled() is False + assert fresh.get_playback_duck_percent() == 20 + + config_file.write_text( + json.dumps({"playback_duck": {"enabled": True, "percent": 250}}), + encoding="utf-8", + ) + loaded = ConfigManager() + assert loaded.is_playback_duck_enabled() is True + assert loaded.get_playback_duck_percent() == 100 + loaded.set_playback_duck_percent(-10) + assert loaded.config["playback_duck"]["percent"] == 0 + loaded.set_playback_duck_percent(101) + assert loaded.config["playback_duck"]["percent"] == 100 + loaded.set_playback_duck_enabled(True) + loaded.set_playback_duck_percent(35) + loaded.save_config() + + again = ConfigManager() + assert again.is_playback_duck_enabled() is True + assert again.get_playback_duck_percent() == 35 + assert again.config["playback_duck"]["percent"] == 35 + + +def test_recognition_hook_schedules_after_start_and_restores_before_stop_cue( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Dictation that never starts does not duck; stop restores before the cue.""" + from vocalinux.common_types import RecognitionState + from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager + + timeline = [] + session = _RecordingSession(enabled=True) + session.start = lambda delay: timeline.append(("start", delay)) + session.cancel = lambda: timeline.append("cancel") + session.restore = lambda: timeline.append("restore") + + feedback = _real_audio_feedback(monkeypatch) + monkeypatch.setattr(feedback, "_is_sound_effects_enabled", lambda: True) + monkeypatch.setattr(feedback, "_resolved_tone", lambda: "voca") + monkeypatch.setattr(feedback, "tone_sound_path", lambda _tone, _kind: "/cue.wav") + monkeypatch.setattr(feedback, "_wav_duration_seconds", lambda _path: 0.4) + + model_dir = tmp_path / "vosk-model" + model_dir.mkdir() + with patch.dict(sys.modules, {"vosk": MagicMock()}): + with patch.object( + SpeechRecognitionManager, "_get_vosk_model_path", return_value=str(model_dir) + ): + manager = SpeechRecognitionManager(engine="vosk", playback_duck=session) + + assert manager._playback_duck_delay_seconds() == pytest.approx(0.45) + monkeypatch.setattr(feedback, "_resolved_tone", lambda: "off") + assert manager._playback_duck_delay_seconds() == 0.0 + monkeypatch.setattr(feedback, "_is_sound_effects_enabled", lambda: False) + assert manager._playback_duck_delay_seconds() == 0.0 + monkeypatch.setattr(feedback, "_is_sound_effects_enabled", lambda: True) + monkeypatch.setattr(feedback, "_resolved_tone", lambda: "voca") + monkeypatch.setattr(feedback, "_wav_duration_seconds", lambda _path: 2.0) + assert manager._playback_duck_delay_seconds() == 1.5 + monkeypatch.setattr(feedback, "_wav_duration_seconds", lambda _path: 0.4) + + manager.stop_recognition() + assert timeline == [] + + manager.state = RecognitionState.LISTENING + assert manager.start_recognition() is False + assert timeline == [] + + manager._model_initialized = False + manager.model = None + manager.state = RecognitionState.IDLE + assert manager.start_recognition() is False + assert timeline == [] + + manager._model_initialized = True + manager.model = object() + with ( + patch( + "vocalinux.speech_recognition.recognition_manager.play_start_sound", + side_effect=lambda: timeline.append("start_sound"), + ), + patch( + "vocalinux.speech_recognition.recognition_manager.play_stop_sound", + side_effect=lambda: timeline.append("stop_sound"), + ), + patch("vocalinux.speech_recognition.recognition_manager.threading.Thread") as thread_cls, + ): + thread_cls.return_value = MagicMock() + assert manager.start_recognition() is True + manager.stop_recognition() + + start_at = timeline.index("start_sound") + scheduled = next(item for item in timeline if isinstance(item, tuple)) + assert scheduled[0] == "start" + assert scheduled[1] == pytest.approx(0.45) + assert start_at < timeline.index(scheduled) + assert timeline.index("restore") < timeline.index("stop_sound") + assert "cancel" in timeline[: timeline.index("restore")] + + session._enabled = False + timeline.clear() + with patch("vocalinux.speech_recognition.recognition_manager.threading.Thread") as thread_cls: + thread_cls.return_value = MagicMock() + assert manager.start_recognition() is True + assert not any(isinstance(item, tuple) for item in timeline) + + +def test_arm_after_capture_failure_does_not_schedule( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """A capture thread that already failed must not be followed by a duck.""" + from vocalinux.common_types import RecognitionState + from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager + + session = _RecordingSession(enabled=True) + feedback = _real_audio_feedback(monkeypatch) + monkeypatch.setattr(feedback, "_is_sound_effects_enabled", lambda: False) + monkeypatch.setattr(feedback, "_resolved_tone", lambda: "voca") + monkeypatch.setattr(feedback, "tone_sound_path", lambda _tone, _kind: "/cue.wav") + monkeypatch.setattr(feedback, "_wav_duration_seconds", lambda _path: 0.4) + + model_dir = tmp_path / "vosk-model" + model_dir.mkdir() + with patch.dict(sys.modules, {"vosk": MagicMock()}): + with patch.object( + SpeechRecognitionManager, "_get_vosk_model_path", return_value=str(model_dir) + ): + manager = SpeechRecognitionManager(engine="vosk", playback_duck=session) + + # The capture thread failed before start_recognition reached the arm: it + # released the session and ended the dictation. The late arm must not + # schedule a duck that nothing will restore. + manager.should_record = False + manager.release_playback_duck() + manager._arm_playback_duck() + assert session.events == ["cancel", "restore"] + + manager.state = RecognitionState.ERROR + manager.should_record = True + session.events.clear() + manager._arm_playback_duck() + assert session.events == [] + + manager.state = RecognitionState.LISTENING + manager._arm_playback_duck() + assert session.events == [("start", 0.0)]