From d64ed8968a01e9bbd4613143dff8febaf208edb0 Mon Sep 17 00:00:00 2001 From: Muzaffer Emre Date: Fri, 31 Jul 2026 18:47:49 +0300 Subject: [PATCH 1/5] Add GNOME X11 and PulseAudio support --- README.md | 11 ++- README.tr.md | 11 ++- audio.py | 59 ++++++++++---- hotkey.py | 150 ++++++++++++++++++++++++++++++++++- i18n.py | 21 +++++ install.sh | 27 +++++-- paste.py | 61 ++++++++++---- tests/test_desktop_compat.py | 55 +++++++++++++ 8 files changed, 353 insertions(+), 42 deletions(-) create mode 100644 tests/test_desktop_compat.py diff --git a/README.md b/README.md index de518bd..03e8dd0 100644 --- a/README.md +++ b/README.md @@ -29,8 +29,15 @@ systemctl --user enable --now ydotool # needed for auto-paste dikte # the settings window opens on first run ``` -`install.sh` adds the `dikte` command, a menu entry, an autostart entry and the -KDE shortcut. +On Ubuntu/GNOME X11, recording uses PulseAudio and clipboard/paste use the X11 +tools instead: + +```sh +sudo apt install pulseaudio-utils xclip xdotool ffmpeg +``` + +`install.sh` adds the `dikte` command, a menu entry and an autostart entry. The +settings window installs a GNOME or KDE global shortcut. Two keys go in the settings window: **OpenAI** and **OpenRouter**. Speech to text runs on either one (`gpt-4o-transcribe` by default), cleanup always on diff --git a/README.tr.md b/README.tr.md index 3e8a877..c620ebf 100644 --- a/README.tr.md +++ b/README.tr.md @@ -29,8 +29,15 @@ systemctl --user enable --now ydotool # otomatik yapıştırma için dikte # ilk açılışta ayarlar penceresi gelir ``` -`install.sh` `dikte` komutunu, menü girdisini, oturum açılışında otomatik -başlatmayı ve KDE kısayolunu kurar. +Ubuntu/GNOME X11 için kayıt PulseAudio üzerinden, pano ve yapıştırma ise X11 +araçlarıyla çalışır: + +```sh +sudo apt install pulseaudio-utils xclip xdotool ffmpeg +``` + +`install.sh` `dikte` komutunu, menü girdisini ve oturum açılışında otomatik +başlatmayı kurar. Ayarlar penceresi GNOME veya KDE global kısayolunu kurar. Ayarlar penceresinde iki anahtar istenir: **OpenAI** ve **OpenRouter**. Sesi yazıya çevirme ikisinden birinde çalışır (varsayılan `gpt-4o-transcribe`), diff --git a/audio.py b/audio.py index a9f47ed..0200058 100644 --- a/audio.py +++ b/audio.py @@ -31,7 +31,7 @@ MIN_FRAMES = int(RATE * 0.25) class Recorder(QObject): - """Runs pw-record as a child process and reads raw PCM from its stdout.""" + """Runs the available sound-server recorder and reads raw PCM from stdout.""" level = pyqtSignal(float) # 0.0 - 1.0, for the waveform stopped = pyqtSignal(str, float, object) # wav path, duration (s), per-chunk RMS @@ -53,21 +53,13 @@ class Recorder(QObject): def start(self, target="", max_seconds=300): if self.active: return - if not shutil.which("pw-record"): - self.failed.emit(t("pw-record not found. Is pipewire-audio installed?")) + cmd = recording_command(target) + if not cmd: + self.failed.emit(t( + "No audio recorder found. Install pulseaudio-utils or pipewire-audio." + )) return - cmd = [ - "pw-record", - "--raw", - f"--rate={RATE}", - f"--channels={CHANNELS}", - "--format=s16", - ] - if target: - cmd.append(f"--target={target}") - cmd.append("-") - try: self._proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0 @@ -84,7 +76,8 @@ class Recorder(QObject): self._thread.start() def _pump(self): - stdout = self._proc.stdout + proc = self._proc + stdout = proc.stdout try: while True: chunk = stdout.read(CHUNK_BYTES) @@ -101,6 +94,15 @@ class Recorder(QObject): break except (OSError, ValueError): pass + if not self._cancelled and not self._buffer and proc.poll() is not None: + try: + detail = proc.stderr.read().decode("utf-8", "replace").strip() + except (AttributeError, OSError): + detail = "" + self.failed.emit(t( + "Audio recorder stopped before receiving sound: {error}", + error=detail or f"exit code {proc.returncode}", + )) def _terminate(self): proc = self._proc @@ -161,6 +163,33 @@ def write_wav(pcm, rate=RATE, channels=CHANNELS, width=SAMPLE_WIDTH): return path +def recording_command(target=""): + """Return a raw-s16 capture command for the sound server on this desktop. + + parec works with both PulseAudio and PipeWire's PulseAudio compatibility + service, and its source names are the same ones shown by list_sources(). + Keep pw-record as the fallback for minimal native-PipeWire installations. + """ + if shutil.which("parec"): + cmd = [ + "parec", "--record", "--raw", f"--rate={RATE}", + f"--channels={CHANNELS}", "--format=s16le", + ] + if target: + cmd.append(f"--device={target}") + return cmd + if shutil.which("pw-record"): + cmd = [ + "pw-record", "--raw", f"--rate={RATE}", + f"--channels={CHANNELS}", "--format=s16", + ] + if target: + cmd.append(f"--target={target}") + cmd.append("-") + return cmd + return [] + + class MeetingRecorder(QObject): """Microphone and speaker output into one stereo file: left is you, right is everyone else. diff --git a/hotkey.py b/hotkey.py index 77f1163..8347f50 100644 --- a/hotkey.py +++ b/hotkey.py @@ -1,10 +1,12 @@ -"""Global shortcut: KDE custom-shortcut installation plus a built-in evdev listener.""" +"""GNOME/KDE global-shortcut installation plus a built-in evdev listener.""" +import ast import glob import os import pathlib import re import select +import shutil import struct import subprocess import threading @@ -19,6 +21,8 @@ ASK_DESKTOP_ID = "dikte-ask.desktop" APPLICATIONS_DIR = pathlib.Path.home() / ".local/share/applications" DESKTOP_FILE = APPLICATIONS_DIR / DESKTOP_ID SHORTCUTS_FILE = pathlib.Path.home() / ".config/kglobalshortcutsrc" +GNOME_MEDIA_SCHEMA = "org.gnome.settings-daemon.plugins.media-keys" +GNOME_BINDING_SCHEMA = "org.gnome.settings-daemon.plugins.media-keys.custom-keybinding" # --- evdev key codes (linux/input-event-codes.h) -------------------------- @@ -174,6 +178,150 @@ class EvdevHotkey(QObject): # --- KDE custom shortcut -------------------------------------------------- +def _gnome(): + desktop = os.environ.get("XDG_CURRENT_DESKTOP", "").lower() + return "gnome" in desktop and shutil.which("gsettings") is not None + + +def _gnome_path(desktop_id): + name = re.sub(r"[^a-zA-Z0-9_-]+", "-", desktop_id.removesuffix(".desktop")) + return f"/org/gnome/settings-daemon/plugins/media-keys/custom-keybindings/{name}/" + + +def gnome_accelerator(shortcut): + """Translate Qt-style Ctrl+Alt+A into GNOME's a syntax.""" + parts = [part.strip() for part in str(shortcut).split("+") if part.strip()] + modifiers = [] + key = "" + names = { + "ctrl": "", "control": "", + "alt": "", "shift": "", + "super": "", "meta": "", + } + for part in parts: + modifier = names.get(part.lower()) + if modifier: + if modifier not in modifiers: + modifiers.append(modifier) + else: + key = part.lower() if len(part) == 1 else part + return "".join(modifiers) + key if key else "" + + +def display_accelerator(accelerator): + """Translate a GNOME accelerator back to the form shown in Dikte.""" + text = str(accelerator) + parts = [] + for token, label in (("", "Ctrl"), ("", "Ctrl"), + ("", "Alt"), ("", "Shift"), + ("", "Super")): + if token.lower() in text.lower(): + parts.append(label) + text = re.sub(re.escape(token), "", text, flags=re.IGNORECASE) + key = text.strip() + if len(key) == 1: + key = key.upper() + if key: + parts.append(key) + return "+".join(parts) + + +def _gsettings(*args, check=True): + return subprocess.run( + ["gsettings", *args], capture_output=True, text=True, timeout=10, check=check, + ) + + +def _gsettings_array(value): + """Parse a gsettings string-array, including the empty `@as []` form.""" + text = str(value).strip() + if text.startswith("@as "): + text = text[4:].strip() + parsed = ast.literal_eval(text) if text else [] + if not isinstance(parsed, (list, tuple)): + raise ValueError(f"not a string array: {value}") + return list(parsed) + + +def install_gnome_shortcut(shortcut, exec_command, + name="Dikte: start/stop recording", + desktop_id=DESKTOP_ID): + path = _gnome_path(desktop_id) + try: + current = _gsettings( + "get", GNOME_MEDIA_SCHEMA, "custom-keybindings" + ).stdout.strip() + paths = _gsettings_array(current) + if path not in paths: + paths.append(path) + _gsettings("set", GNOME_MEDIA_SCHEMA, "custom-keybindings", repr(paths)) + schema = f"{GNOME_BINDING_SCHEMA}:{path}" + _gsettings("set", schema, "name", repr(name)) + _gsettings("set", schema, "command", repr(exec_command)) + accelerator = gnome_accelerator(shortcut) + if not accelerator: + raise ValueError(t("Could not parse the shortcut: {shortcut}", + shortcut=shortcut)) + _gsettings("set", schema, "binding", repr(accelerator)) + except (ValueError, SyntaxError, subprocess.SubprocessError, OSError) as exc: + return False, t("Could not register the GNOME shortcut: {error}", error=exc) + return True, t("Shortcut saved: {shortcut}", shortcut=shortcut) + + +def remove_gnome_shortcut(desktop_id=DESKTOP_ID): + path = _gnome_path(desktop_id) + try: + current = _gsettings( + "get", GNOME_MEDIA_SCHEMA, "custom-keybindings" + ).stdout.strip() + paths = _gsettings_array(current) + if path in paths: + paths.remove(path) + _gsettings("set", GNOME_MEDIA_SCHEMA, "custom-keybindings", repr(paths)) + except (ValueError, SyntaxError, subprocess.SubprocessError, OSError): + pass + + +def gnome_shortcut_status(desktop_id=DESKTOP_ID): + path = _gnome_path(desktop_id) + try: + current = _gsettings( + "get", GNOME_MEDIA_SCHEMA, "custom-keybindings" + ).stdout.strip() + paths = _gsettings_array(current) + if path not in paths: + return None + value = _gsettings( + "get", f"{GNOME_BINDING_SCHEMA}:{path}", "binding" + ).stdout.strip() + accelerator = ast.literal_eval(value) + return display_accelerator(accelerator) if accelerator else None + except (ValueError, SyntaxError, subprocess.SubprocessError, OSError): + return None + + +def install_shortcut(shortcut, exec_command, name="Dikte: start/stop recording", + desktop_id=DESKTOP_ID): + if _gnome(): + return install_gnome_shortcut(shortcut, exec_command, name, desktop_id) + return install_kde_shortcut(shortcut, exec_command, name, desktop_id) + + +def remove_shortcut(desktop_id=DESKTOP_ID): + if _gnome(): + remove_gnome_shortcut(desktop_id) + else: + remove_kde_shortcut(desktop_id) + + +def shortcut_status(desktop_id=DESKTOP_ID): + return (gnome_shortcut_status(desktop_id) if _gnome() + else kde_shortcut_status(desktop_id)) + + +def desktop_name(): + return "GNOME" if _gnome() else "KDE" + def install_kde_shortcut(shortcut, exec_command, name="Dikte: start/stop recording", desktop_id=DESKTOP_ID): """Write the desktop file and the kglobalshortcutsrc entry. diff --git a/i18n.py b/i18n.py index 13ec3d2..ef80f4e 100644 --- a/i18n.py +++ b/i18n.py @@ -101,9 +101,20 @@ TR = { "pw-record not found. Is pipewire-audio installed?": "pw-record bulunamadı. pipewire-audio kurulu mu?", "Could not start recording: {error}": "Kayıt başlatılamadı: {error}", + "No audio recorder found. Install pulseaudio-utils or pipewire-audio.": + "Ses kayıt aracı bulunamadı. pulseaudio-utils ya da pipewire-audio kur.", + "Audio recorder stopped before receiving sound: {error}": + "Ses kayıt aracı veri alamadan kapandı: {error}", "wl-copy not found. Install wl-clipboard.": "wl-copy bulunamadı. wl-clipboard paketini kur.", "Could not copy to clipboard: {error}": "Panoya kopyalanamadı: {error}", + "{tool} not found; clipboard copy is unavailable.": + "{tool} bulunamadı; panoya kopyalama kullanılamıyor.", + "{tool} exited with code {code}.": "{tool} {code} koduyla çıktı.", + "{tool} not found, cannot paste automatically.": + "{tool} bulunamadı, otomatik yapıştırma yapılamıyor.", + "Could not run xdotool: {error}": "xdotool çalıştırılamadı: {error}", + "xdotool failed: {error}": "xdotool hatası: {error}", "wl-copy exited with code {code}.": "wl-copy {code} koduyla çıktı.", "ydotool not found, cannot paste automatically.": "ydotool bulunamadı, otomatik yapıştırma yapılamıyor.", @@ -265,9 +276,19 @@ TR = { # --- settings: shortcut ------------------------------------------------ "Install as a KDE shortcut": "KDE kısayolu olarak kur", + "Install as a global shortcut": "Global kısayol olarak kur", "Remove": "Kaldır", "Registered in KDE: {shortcut}": "KDE'de kayıtlı: {shortcut}", "No KDE shortcut installed.": "KDE kısayolu kurulu değil.", + "Registered in {desktop}: {shortcut}": "{desktop}'da kayıtlı: {shortcut}", + "No global shortcut installed.": "Global kısayol kurulu değil.", + "No global shortcut installed. The tray menu starts a meeting too.": + "Global kısayol kurulu değil. Toplantı tepsi menüsünden de başlatılabilir.", + "No global shortcut installed. The tray menu asks it too.": + "Global kısayol kurulu değil. Tepsi menüsünden de soru sorulabilir.", + "Shortcut saved: {shortcut}": "Kısayol kaydedildi: {shortcut}", + "Could not register the GNOME shortcut: {error}": + "GNOME kısayolu kaydedilemedi: {error}", "Use the built-in listener (/dev/input), for when the KDE shortcut is not active yet": "Yerleşik dinleyici kullan (/dev/input), KDE kısayolu henüz etkin değilken", "Works immediately, no session restart. The only difference: the key " diff --git a/install.sh b/install.sh index 8aef023..2cbaaee 100755 --- a/install.sh +++ b/install.sh @@ -19,21 +19,33 @@ echo "────────────────" # 1. Dependencies ---------------------------------------------------------- missing=() -for cmd in pw-record wl-copy wl-paste ydotool ffmpeg; do +audio_cmds=(ffmpeg) +if command -v parec >/dev/null || command -v pw-record >/dev/null; then + : +else + missing+=("pulseaudio-utils-or-pipewire-audio") +fi +if [[ "${XDG_SESSION_TYPE:-}" == "x11" ]]; then + desktop_cmds=(xclip xdotool) +else + desktop_cmds=(wl-copy wl-paste ydotool) +fi +for cmd in "${audio_cmds[@]}" "${desktop_cmds[@]}"; do command -v "$cmd" >/dev/null || missing+=("$cmd") done python3 -c 'import PyQt6.QtWidgets' 2>/dev/null || missing+=("python-pyqt6") if ((${#missing[@]})); then warn "Missing: ${missing[*]}" - say "Arch/CachyOS: sudo pacman -S --needed pipewire-audio wl-clipboard ydotool ffmpeg python-pyqt6" + say "Ubuntu X11: sudo apt install pulseaudio-utils xclip xdotool ffmpeg" + say "Arch Wayland: sudo pacman -S --needed pipewire-audio wl-clipboard ydotool ffmpeg python-pyqt6" echo else ok "All dependencies present" fi # 2. ydotoold -------------------------------------------------------------- -if command -v ydotool >/dev/null; then +if [[ "${XDG_SESSION_TYPE:-}" != "x11" ]] && command -v ydotool >/dev/null; then if systemctl --user is-active --quiet ydotool 2>/dev/null \ || systemctl --user is-active --quiet ydotoold 2>/dev/null; then ok "ydotoold is running (auto-paste ready)" @@ -87,7 +99,10 @@ Type=Application X-KDE-GlobalAccel-CommandShortcut=true EOF -if command -v kwriteconfig6 >/dev/null; then +if [[ "${XDG_CURRENT_DESKTOP:-}" == *GNOME* || "${XDG_CURRENT_DESKTOP:-}" == *gnome* ]]; then + ok "GNOME detected" + say "Open Dikte Settings > Shortcut to install the global shortcut." +elif command -v kwriteconfig6 >/dev/null; then kwriteconfig6 --notify --file kglobalshortcutsrc \ --group services --group dikte-toggle.desktop \ --key _launch "$SHORTCUT" @@ -96,10 +111,10 @@ if command -v kwriteconfig6 >/dev/null; then say "next login. Until then open Settings → Shortcut and turn on the" say "built-in listener to use it right away." else - warn "kwriteconfig6 not found. Add the shortcut via System Settings > Shortcuts" + warn "No supported shortcut manager found. Add the shortcut in desktop settings." fi echo ok "Done. Start it with: dikte" -say "The settings window opens on first run; add your OpenAI and OpenRouter keys." +say "The settings window opens on first run; add an OpenAI, Groq or OpenRouter key." echo diff --git a/paste.py b/paste.py index 2065d56..b87d71d 100644 --- a/paste.py +++ b/paste.py @@ -1,5 +1,6 @@ -"""Clipboard (wl-clipboard) and key injection (ydotool).""" +"""Clipboard and key injection for Wayland and X11.""" +import os import shutil import subprocess import time @@ -18,20 +19,23 @@ class PasteError(Exception): def read_clipboard(): - if not shutil.which("wl-paste"): + command = (["xclip", "-selection", "clipboard", "-out"] if _x11() + else ["wl-paste", "--no-newline"]) + if not shutil.which(command[0]): return None try: - res = subprocess.run(["wl-paste", "--no-newline"], capture_output=True, timeout=5) + res = subprocess.run(command, capture_output=True, timeout=5) except (subprocess.SubprocessError, OSError): return None return res.stdout if res.returncode == 0 else None -def _run_wl_copy(payload): - """wl-copy forks to keep owning the selection; leaving its pipes open makes - subprocess.run wait for EOF forever, hence DEVNULL.""" +def _run_copy(payload): + """The clipboard owner may fork; do not leave inherited pipes open.""" + command = (["xclip", "-selection", "clipboard", "-in"] if _x11() + else ["wl-copy"]) return subprocess.run( - ["wl-copy"], + command, input=payload, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, @@ -40,33 +44,58 @@ def _run_wl_copy(payload): def copy(text): - if not shutil.which("wl-copy"): - raise PasteError(t("wl-copy not found. Install wl-clipboard.")) + tool = "xclip" if _x11() else "wl-copy" + if not shutil.which(tool): + raise PasteError(t("{tool} not found; clipboard copy is unavailable.", tool=tool)) try: - res = _run_wl_copy(text.encode("utf-8")) + res = _run_copy(text.encode("utf-8")) except (subprocess.SubprocessError, OSError) as exc: raise PasteError(t("Could not copy to clipboard: {error}", error=exc)) from exc if res.returncode != 0: - raise PasteError(t("wl-copy exited with code {code}.", code=res.returncode)) + raise PasteError(t("{tool} exited with code {code}.", + tool=tool, code=res.returncode)) def copy_bytes(data): - if data is None or not shutil.which("wl-copy"): + tool = "xclip" if _x11() else "wl-copy" + if data is None or not shutil.which(tool): return try: - _run_wl_copy(data) + _run_copy(data) except (subprocess.SubprocessError, OSError): pass def ydotool_ready(): - return shutil.which("ydotool") is not None + tool = "xdotool" if _x11() else "ydotool" + return shutil.which(tool) is not None + + +def _x11(): + return (os.environ.get("XDG_SESSION_TYPE") == "x11" + or bool(os.environ.get("DISPLAY") and not os.environ.get("WAYLAND_DISPLAY"))) def press(shortcut="ctrl+v", delay=0.12): - """Press a key combination through ydotool, e.g. 'ctrl+v'.""" + """Press a key combination through xdotool or ydotool.""" if not ydotool_ready(): - raise PasteError(t("ydotool not found, cannot paste automatically.")) + tool = "xdotool" if _x11() else "ydotool" + raise PasteError(t("{tool} not found, cannot paste automatically.", tool=tool)) + + if _x11(): + key = shortcut.lower().replace("control", "ctrl") + time.sleep(delay) + try: + res = subprocess.run( + ["xdotool", "key", "--clearmodifiers", key], + capture_output=True, text=True, timeout=10, + ) + except (subprocess.SubprocessError, OSError) as exc: + raise PasteError(t("Could not run xdotool: {error}", error=exc)) from exc + if res.returncode != 0: + raise PasteError(t("xdotool failed: {error}", + error=res.stderr.strip() or "unknown error")) + return codes = [] for key in (k.strip().lower() for k in shortcut.split("+") if k.strip()): diff --git a/tests/test_desktop_compat.py b/tests/test_desktop_compat.py new file mode 100644 index 0000000..150944b --- /dev/null +++ b/tests/test_desktop_compat.py @@ -0,0 +1,55 @@ +import os +import unittest +from unittest import mock + +import audio +import hotkey +import paste + + +class AudioBackendTests(unittest.TestCase): + def test_parec_is_preferred(self): + with mock.patch.object(audio.shutil, "which", side_effect=lambda cmd: f"/usr/bin/{cmd}"): + self.assertEqual(audio.recording_command()[0], "parec") + + def test_pw_record_remains_the_fallback(self): + with mock.patch.object( + audio.shutil, "which", side_effect=lambda cmd: "/usr/bin/pw-record" + if cmd == "pw-record" else None, + ): + self.assertEqual(audio.recording_command()[0], "pw-record") + + +class DesktopBackendTests(unittest.TestCase): + def test_x11_uses_xclip(self): + result = mock.Mock(returncode=0) + with mock.patch.dict(os.environ, {"XDG_SESSION_TYPE": "x11"}), \ + mock.patch.object(paste.shutil, "which", return_value="/usr/bin/xclip"), \ + mock.patch.object(paste.subprocess, "run", return_value=result) as run: + paste.copy("hello") + self.assertEqual(run.call_args.args[0][:3], + ["xclip", "-selection", "clipboard"]) + + def test_x11_uses_xdotool(self): + result = mock.Mock(returncode=0, stderr="") + with mock.patch.dict(os.environ, {"XDG_SESSION_TYPE": "x11"}), \ + mock.patch.object(paste.shutil, "which", return_value="/usr/bin/xdotool"), \ + mock.patch.object(paste.time, "sleep"), \ + mock.patch.object(paste.subprocess, "run", return_value=result) as run: + paste.press("ctrl+shift+v") + self.assertEqual(run.call_args.args[0], + ["xdotool", "key", "--clearmodifiers", "ctrl+shift+v"]) + + +class GnomeShortcutTests(unittest.TestCase): + def test_accelerator_round_trip(self): + accelerator = hotkey.gnome_accelerator("Ctrl+Alt+A") + self.assertEqual(accelerator, "a") + self.assertEqual(hotkey.display_accelerator(accelerator), "Ctrl+Alt+A") + + def test_empty_gsettings_array(self): + self.assertEqual(hotkey._gsettings_array("@as []"), []) + + +if __name__ == "__main__": + unittest.main() From 0e6ec3aea9d2a9b96edcba38a178458f1e2387ed Mon Sep 17 00:00:00 2001 From: yusufipk Date: Sat, 1 Aug 2026 20:32:07 +0700 Subject: [PATCH 2/5] Choose the clipboard's programs once, instead of in every function Which pair of tools a session uses was being asked inside copy(), read(), copy_bytes() and press(), four times, in a conditional each. A third desktop means a third branch in each of them, and macOS is already open as a pull request doing exactly that in a different style. Wayland and X11 are two entries at the top of the file now, each naming its programs and how to build a key press from them. The functions below take whichever entry the session gets. Adding a desktop is an entry and a line in desktop(), and it touches nothing else. Two things fall out of it. The key press is checked against the same table on both: xdotool was being handed the shortcut as typed, so ctrl+f13 was refused on Wayland and passed through on X11. And a missing tool names the package to install again, which the shared message had dropped. The tests say the promises once and run them against both desktops, so a third one inherits the list rather than needing a copy of it. --- i18n.py | 20 ++-- paste.py | 168 +++++++++++++++++++---------- tests/test_desktop_compat.py | 55 ---------- tests/test_paste.py | 200 ++++++++++++++++++++++++----------- 4 files changed, 256 insertions(+), 187 deletions(-) delete mode 100644 tests/test_desktop_compat.py diff --git a/i18n.py b/i18n.py index 3ce777f..dd70837 100644 --- a/i18n.py +++ b/i18n.py @@ -101,30 +101,22 @@ TR = { "Unexpected error: {error}": "Beklenmeyen hata: {error}", # --- audio / paste errors ----------------------------------------- - "pw-record not found. Is pipewire-audio installed?": - "pw-record bulunamadı. pipewire-audio kurulu mu?", "Could not start recording: {error}": "Kayıt başlatılamadı: {error}", "No audio recorder found. Install pulseaudio-utils or pipewire-audio.": "Ses kayıt aracı bulunamadı. pulseaudio-utils ya da pipewire-audio kur.", "Audio recorder stopped before receiving sound: {error}": "Ses kayıt aracı veri alamadan kapandı: {error}", - "wl-copy not found. Install wl-clipboard.": - "wl-copy bulunamadı. wl-clipboard paketini kur.", "Could not copy to clipboard: {error}": "Panoya kopyalanamadı: {error}", - "{tool} not found; clipboard copy is unavailable.": - "{tool} bulunamadı; panoya kopyalama kullanılamıyor.", + "{tool} not found. Install {packages}.": + "{tool} bulunamadı. {packages} paketlerini kur.", "{tool} exited with code {code}.": "{tool} {code} koduyla çıktı.", "{tool} not found, cannot paste automatically.": "{tool} bulunamadı, otomatik yapıştırma yapılamıyor.", - "Could not run xdotool: {error}": "xdotool çalıştırılamadı: {error}", - "xdotool failed: {error}": "xdotool hatası: {error}", - "wl-copy exited with code {code}.": "wl-copy {code} koduyla çıktı.", - "ydotool not found, cannot paste automatically.": - "ydotool bulunamadı, otomatik yapıştırma yapılamıyor.", "Unknown key: {key}": "Bilinmeyen tuş: {key}", - "Could not run ydotool: {error}": "ydotool çalıştırılamadı: {error}", - "ydotool failed: {error}\nIs ydotoold running? (systemctl --user status ydotool)": - "ydotool hatası: {error}\nydotoold çalışıyor mu? (systemctl --user status ydotool)", + "Could not run {tool}: {error}": "{tool} çalıştırılamadı: {error}", + "{tool} failed: {error}": "{tool} hatası: {error}", + "Is ydotoold running? (systemctl --user status ydotool)": + "ydotoold çalışıyor mu? (systemctl --user status ydotool)", # --- api errors ---------------------------------------------------- "{service} API key is empty. Add it in Settings.": diff --git a/paste.py b/paste.py index b87d71d..29b3a5b 100644 --- a/paste.py +++ b/paste.py @@ -1,5 +1,13 @@ -"""Clipboard and key injection for Wayland and X11.""" +"""Clipboard and key injection, through whichever pair of programs is here. +A Wayland session has wl-clipboard and ydotool, an X11 one has xclip and +xdotool, and a session is one or the other. Which it is gets decided in one +place, and each desktop is a small group of functions below it: another desktop, +or another operating system, adds a group and a line to the chooser rather than +a branch inside every function here. +""" + +import collections import os import shutil import subprocess @@ -7,35 +15,105 @@ import time from i18n import t -# Linux input event codes (linux/input-event-codes.h) +# Linux input event codes (linux/input-event-codes.h), which is what ydotool +# takes. They are also the list of keys a paste shortcut may be built from, so +# xdotool is held to the same table rather than being handed the text as typed. KEYCODES = { "ctrl": 29, "control": 29, "shift": 42, "alt": 56, "super": 125, "meta": 125, "v": 47, "insert": 110, "enter": 28, "return": 28, } +# xdotool speaks X keysyms, which spell some of those differently. +KEYSYMS = {"control": "ctrl", "meta": "super", "insert": "Insert", + "enter": "Return", "return": "Return"} + class PasteError(Exception): pass +def _keys(shortcut): + """'Ctrl+V' -> ['ctrl', 'v'], every one of them a key we know.""" + parts = [key.strip().lower() for key in str(shortcut).split("+") if key.strip()] + for key in parts: + if key not in KEYCODES: + raise PasteError(t("Unknown key: {key}", key=key)) + return parts + + +def _ydotool_command(shortcut): + """ydotool wants a press event per key, then a release in reverse.""" + codes = [KEYCODES[key] for key in _keys(shortcut)] + return ["ydotool", "key", *[f"{code}:1" for code in codes], + *[f"{code}:0" for code in reversed(codes)]] + + +def _xdotool_command(shortcut): + """xdotool takes the whole combination as one argument.""" + keys = [KEYSYMS.get(key, key) for key in _keys(shortcut)] + return ["xdotool", "key", "--clearmodifiers", "+".join(keys)] + + +Desktop = collections.namedtuple( + "Desktop", + # The two programs, the packages to install them from, how to build the key + # press, and what else to say when the key press fails. + "clipboard keyboard packages read_command copy_command key_command key_hint", +) + +WAYLAND = Desktop( + clipboard="wl-copy", + keyboard="ydotool", + packages="wl-clipboard and ydotool", + read_command=["wl-paste", "--no-newline"], + copy_command=["wl-copy"], + key_command=_ydotool_command, + key_hint="Is ydotoold running? (systemctl --user status ydotool)", +) + +X11 = Desktop( + clipboard="xclip", + keyboard="xdotool", + packages="xclip and xdotool", + read_command=["xclip", "-selection", "clipboard", "-out"], + copy_command=["xclip", "-selection", "clipboard", "-in"], + key_command=_xdotool_command, + key_hint="", +) + + +def desktop(): + """The pair of programs this session's clipboard and keyboard go through. + + Read every time rather than settled at import: a session started before the + display server was up would otherwise be stuck with the wrong answer, and a + test would have nowhere to say which one it means. + """ + if os.environ.get("XDG_SESSION_TYPE") == "x11": + return X11 + if os.environ.get("DISPLAY") and not os.environ.get("WAYLAND_DISPLAY"): + return X11 + return WAYLAND + + +# --- the clipboard --------------------------------------------------------- + def read_clipboard(): - command = (["xclip", "-selection", "clipboard", "-out"] if _x11() - else ["wl-paste", "--no-newline"]) - if not shutil.which(command[0]): + here = desktop() + if not shutil.which(here.read_command[0]): return None try: - res = subprocess.run(command, capture_output=True, timeout=5) + res = subprocess.run(here.read_command, capture_output=True, timeout=5) except (subprocess.SubprocessError, OSError): return None return res.stdout if res.returncode == 0 else None def _run_copy(payload): - """The clipboard owner may fork; do not leave inherited pipes open.""" - command = (["xclip", "-selection", "clipboard", "-in"] if _x11() - else ["wl-copy"]) + """The clipboard owner forks to keep holding the selection; leaving its + pipes open makes subprocess.run wait for EOF forever, hence DEVNULL.""" return subprocess.run( - command, + desktop().copy_command, input=payload, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, @@ -44,21 +122,21 @@ def _run_copy(payload): def copy(text): - tool = "xclip" if _x11() else "wl-copy" - if not shutil.which(tool): - raise PasteError(t("{tool} not found; clipboard copy is unavailable.", tool=tool)) + here = desktop() + if not shutil.which(here.clipboard): + raise PasteError(t("{tool} not found. Install {packages}.", + tool=here.clipboard, packages=here.packages)) try: res = _run_copy(text.encode("utf-8")) except (subprocess.SubprocessError, OSError) as exc: raise PasteError(t("Could not copy to clipboard: {error}", error=exc)) from exc if res.returncode != 0: raise PasteError(t("{tool} exited with code {code}.", - tool=tool, code=res.returncode)) + tool=here.clipboard, code=res.returncode)) def copy_bytes(data): - tool = "xclip" if _x11() else "wl-copy" - if data is None or not shutil.which(tool): + if data is None or not shutil.which(desktop().clipboard): return try: _run_copy(data) @@ -66,54 +144,28 @@ def copy_bytes(data): pass -def ydotool_ready(): - tool = "xdotool" if _x11() else "ydotool" - return shutil.which(tool) is not None +# --- the key press --------------------------------------------------------- - -def _x11(): - return (os.environ.get("XDG_SESSION_TYPE") == "x11" - or bool(os.environ.get("DISPLAY") and not os.environ.get("WAYLAND_DISPLAY"))) +def paste_ready(): + return shutil.which(desktop().keyboard) is not None def press(shortcut="ctrl+v", delay=0.12): - """Press a key combination through xdotool or ydotool.""" - if not ydotool_ready(): - tool = "xdotool" if _x11() else "ydotool" - raise PasteError(t("{tool} not found, cannot paste automatically.", tool=tool)) + """Press a key combination, e.g. 'ctrl+v'.""" + here = desktop() + if not paste_ready(): + raise PasteError(t("{tool} not found, cannot paste automatically.", + tool=here.keyboard)) - if _x11(): - key = shortcut.lower().replace("control", "ctrl") - time.sleep(delay) - try: - res = subprocess.run( - ["xdotool", "key", "--clearmodifiers", key], - capture_output=True, text=True, timeout=10, - ) - except (subprocess.SubprocessError, OSError) as exc: - raise PasteError(t("Could not run xdotool: {error}", error=exc)) from exc - if res.returncode != 0: - raise PasteError(t("xdotool failed: {error}", - error=res.stderr.strip() or "unknown error")) - return - - codes = [] - for key in (k.strip().lower() for k in shortcut.split("+") if k.strip()): - code = KEYCODES.get(key) - if code is None: - raise PasteError(t("Unknown key: {key}", key=key)) - codes.append(code) - - seq = [f"{c}:1" for c in codes] + [f"{c}:0" for c in reversed(codes)] + command = here.key_command(shortcut) time.sleep(delay) # let the selection settle and focus come back try: - res = subprocess.run(["ydotool", "key", *seq], capture_output=True, - text=True, timeout=10) + res = subprocess.run(command, capture_output=True, text=True, timeout=10) except (subprocess.SubprocessError, OSError) as exc: - raise PasteError(t("Could not run ydotool: {error}", error=exc)) from exc + raise PasteError(t("Could not run {tool}: {error}", + tool=here.keyboard, error=exc)) from exc if res.returncode != 0: - raise PasteError(t( - "ydotool failed: {error}\nIs ydotoold running? " - "(systemctl --user status ydotool)", - error=res.stderr.strip() or "unknown error", - )) + message = t("{tool} failed: {error}", tool=here.keyboard, + error=res.stderr.strip() or "unknown error") + raise PasteError(f"{message}\n{t(here.key_hint)}" if here.key_hint + else message) diff --git a/tests/test_desktop_compat.py b/tests/test_desktop_compat.py deleted file mode 100644 index 150944b..0000000 --- a/tests/test_desktop_compat.py +++ /dev/null @@ -1,55 +0,0 @@ -import os -import unittest -from unittest import mock - -import audio -import hotkey -import paste - - -class AudioBackendTests(unittest.TestCase): - def test_parec_is_preferred(self): - with mock.patch.object(audio.shutil, "which", side_effect=lambda cmd: f"/usr/bin/{cmd}"): - self.assertEqual(audio.recording_command()[0], "parec") - - def test_pw_record_remains_the_fallback(self): - with mock.patch.object( - audio.shutil, "which", side_effect=lambda cmd: "/usr/bin/pw-record" - if cmd == "pw-record" else None, - ): - self.assertEqual(audio.recording_command()[0], "pw-record") - - -class DesktopBackendTests(unittest.TestCase): - def test_x11_uses_xclip(self): - result = mock.Mock(returncode=0) - with mock.patch.dict(os.environ, {"XDG_SESSION_TYPE": "x11"}), \ - mock.patch.object(paste.shutil, "which", return_value="/usr/bin/xclip"), \ - mock.patch.object(paste.subprocess, "run", return_value=result) as run: - paste.copy("hello") - self.assertEqual(run.call_args.args[0][:3], - ["xclip", "-selection", "clipboard"]) - - def test_x11_uses_xdotool(self): - result = mock.Mock(returncode=0, stderr="") - with mock.patch.dict(os.environ, {"XDG_SESSION_TYPE": "x11"}), \ - mock.patch.object(paste.shutil, "which", return_value="/usr/bin/xdotool"), \ - mock.patch.object(paste.time, "sleep"), \ - mock.patch.object(paste.subprocess, "run", return_value=result) as run: - paste.press("ctrl+shift+v") - self.assertEqual(run.call_args.args[0], - ["xdotool", "key", "--clearmodifiers", "ctrl+shift+v"]) - - -class GnomeShortcutTests(unittest.TestCase): - def test_accelerator_round_trip(self): - accelerator = hotkey.gnome_accelerator("Ctrl+Alt+A") - self.assertEqual(accelerator, "a") - self.assertEqual(hotkey.display_accelerator(accelerator), "Ctrl+Alt+A") - - def test_empty_gsettings_array(self): - self.assertEqual(hotkey._gsettings_array("@as []"), []) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_paste.py b/tests/test_paste.py index 96c258d..1d6806b 100644 --- a/tests/test_paste.py +++ b/tests/test_paste.py @@ -1,12 +1,18 @@ """The clipboard and the key press, which is where a dictation actually lands. Everything here shells out, so the tools are faked. What the tests hold onto is -the command line: a paste that presses the wrong codes, or in the wrong order, +the command line: a paste that presses the wrong keys, or in the wrong order, types nothing and looks like a hang. + +Both desktops owe the same promises, so those are written once and run against +each of them. A third one added to paste.py inherits the same list rather than +needing its own copy of it. """ +import os import subprocess import unittest +from typing import ClassVar from unittest import mock import paste @@ -14,47 +20,86 @@ from tests.support import DikteTest, FakeCompleted, linux_only, only_these_tools @linux_only -class ReadClipboard(DikteTest): - def test_no_wl_paste_installed(self): +class Chooser(DikteTest): + """Which pair of programs this session's clipboard goes through.""" + + def under(self, **env): + with mock.patch.dict(os.environ, env, clear=True): + return paste.desktop() + + def test_a_wayland_session(self): + self.assertIs(self.under(XDG_SESSION_TYPE="wayland", + WAYLAND_DISPLAY="wayland-0"), paste.WAYLAND) + + def test_an_x11_session(self): + self.assertIs(self.under(XDG_SESSION_TYPE="x11", DISPLAY=":0"), paste.X11) + + def test_a_display_with_no_wayland_beside_it(self): + self.assertIs(self.under(DISPLAY=":0"), paste.X11) + + def test_an_x11_display_under_wayland_is_still_wayland(self): + """XWayland sets DISPLAY too; the session type is the one to believe.""" + self.assertIs(self.under(XDG_SESSION_TYPE="wayland", + DISPLAY=":0", WAYLAND_DISPLAY="wayland-0"), + paste.WAYLAND) + + def test_nothing_set_at_all(self): + self.assertIs(self.under(), paste.WAYLAND) + + +class DesktopContract: + """What both desktops owe. Each of them subclasses this once, below.""" + + env: ClassVar[dict] = {} + here = None + + def setUp(self): + super().setUp() + self.enterContext(mock.patch.dict(os.environ, self.env, clear=True)) + + # ---- reading the clipboard ------------------------------------------- + + def test_no_reader_installed(self): with only_these_tools(): self.assertIsNone(paste.read_clipboard()) def test_what_is_on_the_clipboard_comes_back_as_bytes(self): - with only_these_tools("wl-paste"), \ + with only_these_tools(self.here.read_command[0]), \ mock.patch.object(subprocess, "run", return_value=FakeCompleted(stdout=b"hello")) as run: self.assertEqual(paste.read_clipboard(), b"hello") - self.assertEqual(run.call_args.args[0], ["wl-paste", "--no-newline"]) + self.assertEqual(run.call_args.args[0], self.here.read_command) def test_an_empty_clipboard_is_not_an_error(self): - with only_these_tools("wl-paste"), \ + with only_these_tools(self.here.read_command[0]), \ mock.patch.object(subprocess, "run", return_value=FakeCompleted(returncode=1)): self.assertIsNone(paste.read_clipboard()) - def test_a_tool_that_will_not_run(self): - with only_these_tools("wl-paste"), \ + def test_a_reader_that_will_not_run(self): + with only_these_tools(self.here.read_command[0]), \ mock.patch.object(subprocess, "run", side_effect=OSError("nope")): self.assertIsNone(paste.read_clipboard()) + # ---- copying ---------------------------------------------------------- -@linux_only -class Copy(DikteTest): - def test_no_wl_copy_installed(self): + def test_no_clipboard_tool_installed_says_what_to_install(self): with only_these_tools(), self.assertRaises(paste.PasteError) as caught: paste.copy("hello") - self.assertIn("wl-clipboard", str(caught.exception)) + self.assertIn(self.here.clipboard, str(caught.exception)) + self.assertIn(self.here.packages.split(" and ")[0], str(caught.exception)) def test_the_text_goes_in_as_utf8(self): - with only_these_tools("wl-copy"), \ + with only_these_tools(self.here.clipboard), \ mock.patch.object(subprocess, "run", return_value=FakeCompleted()) as run: paste.copy("günaydın") + self.assertEqual(run.call_args.args[0], self.here.copy_command) self.assertEqual(run.call_args.kwargs["input"], "günaydın".encode()) def test_the_pipes_are_closed_so_the_call_can_return(self): - """wl-copy forks and holds the selection; a pipe nobody drains hangs.""" - with only_these_tools("wl-copy"), \ + """The clipboard owner forks; a pipe nobody drains hangs the caller.""" + with only_these_tools(self.here.clipboard), \ mock.patch.object(subprocess, "run", return_value=FakeCompleted()) as run: paste.copy("hello") @@ -62,95 +107,130 @@ class Copy(DikteTest): self.assertEqual(run.call_args.kwargs["stderr"], subprocess.DEVNULL) def test_a_non_zero_exit_is_reported(self): - with only_these_tools("wl-copy"), \ + with only_these_tools(self.here.clipboard), \ mock.patch.object(subprocess, "run", return_value=FakeCompleted(returncode=1)), \ self.assertRaises(paste.PasteError): paste.copy("hello") - def test_a_tool_that_will_not_run(self): - with only_these_tools("wl-copy"), \ + def test_a_clipboard_tool_that_will_not_run(self): + with only_these_tools(self.here.clipboard), \ mock.patch.object(subprocess, "run", side_effect=OSError("nope")), \ self.assertRaises(paste.PasteError): paste.copy("hello") - -@linux_only -class CopyBytes(DikteTest): - def test_nothing_to_restore(self): - with only_these_tools("wl-copy"), \ + def test_there_is_nothing_to_restore(self): + with only_these_tools(self.here.clipboard), \ mock.patch.object(subprocess, "run") as run: paste.copy_bytes(None) run.assert_not_called() def test_restoring_never_raises(self): """It runs after the paste went in; failing here must not undo that.""" - with only_these_tools("wl-copy"), \ + with only_these_tools(self.here.clipboard), \ mock.patch.object(subprocess, "run", side_effect=OSError("nope")): paste.copy_bytes(b"whatever was there before") def test_the_bytes_go_back_untouched(self): - with only_these_tools("wl-copy"), \ + with only_these_tools(self.here.clipboard), \ mock.patch.object(subprocess, "run", return_value=FakeCompleted()) as run: paste.copy_bytes(b"\x89PNG\r\n") self.assertEqual(run.call_args.kwargs["input"], b"\x89PNG\r\n") + # ---- pressing the key ------------------------------------------------- -@linux_only -class Press(DikteTest): - def setUp(self): - super().setUp() - # The settle delay is real time nobody needs to spend in a test. - self.patch_attr(paste.time, "sleep", lambda seconds: None) - - def run_press(self, shortcut, result=None): - with only_these_tools("ydotool"), \ + def press(self, shortcut, result=None): + with only_these_tools(self.here.keyboard), \ + mock.patch.object(paste.time, "sleep", lambda seconds: None), \ mock.patch.object(subprocess, "run", return_value=result or FakeCompleted()) as run: paste.press(shortcut) return run.call_args.args[0] - def test_no_ydotool_installed(self): + def test_no_keyboard_tool_installed(self): with only_these_tools(): - self.assertFalse(paste.ydotool_ready()) - with self.assertRaises(paste.PasteError): + self.assertFalse(paste.paste_ready()) + with self.assertRaises(paste.PasteError) as caught: paste.press() - - def test_ctrl_v_presses_down_then_lets_go_in_reverse(self): - self.assertEqual(self.run_press("ctrl+v"), - ["ydotool", "key", "29:1", "47:1", "47:0", "29:0"]) - - def test_three_keys(self): - self.assertEqual(self.run_press("ctrl+shift+v"), - ["ydotool", "key", "29:1", "42:1", "47:1", - "47:0", "42:0", "29:0"]) + self.assertIn(self.here.keyboard, str(caught.exception)) def test_case_and_spacing_do_not_matter(self): - self.assertEqual(self.run_press(" Ctrl + V "), self.run_press("ctrl+v")) + self.assertEqual(self.press(" Ctrl + V "), self.press("ctrl+v")) - def test_the_synonyms_land_on_the_same_codes(self): - self.assertEqual(self.run_press("control+insert"), - ["ydotool", "key", "29:1", "110:1", "110:0", "29:0"]) - self.assertEqual(self.run_press("super+enter"), self.run_press("meta+return")) - - def test_a_key_nobody_mapped(self): - with only_these_tools("ydotool"), mock.patch.object(subprocess, "run"), \ + def test_a_key_nobody_mapped_is_refused_before_the_tool_runs(self): + """Whichever desktop it is, the shortcut is held to one table.""" + with only_these_tools(self.here.keyboard), \ + mock.patch.object(subprocess, "run") as run, \ self.assertRaises(paste.PasteError) as caught: paste.press("ctrl+f13") self.assertIn("f13", str(caught.exception)) - - def test_ydotoold_not_running_says_so(self): - with self.assertRaises(paste.PasteError) as caught: - self.run_press("ctrl+v", FakeCompleted(returncode=1, stderr="no socket")) - self.assertIn("ydotoold", str(caught.exception)) + run.assert_not_called() def test_a_tool_that_will_not_run(self): - with only_these_tools("ydotool"), \ + with only_these_tools(self.here.keyboard), \ + mock.patch.object(paste.time, "sleep", lambda seconds: None), \ mock.patch.object(subprocess, "run", side_effect=OSError("nope")), \ self.assertRaises(paste.PasteError): paste.press("ctrl+v") + def test_a_failed_key_press_names_the_tool_and_what_it_said(self): + with self.assertRaises(paste.PasteError) as caught: + self.press("ctrl+v", FakeCompleted(returncode=1, stderr="no socket")) + self.assertIn(self.here.keyboard, str(caught.exception)) + self.assertIn("no socket", str(caught.exception)) + + +@linux_only +class Wayland(DesktopContract, DikteTest): + env: ClassVar[dict] = {"XDG_SESSION_TYPE": "wayland", + "WAYLAND_DISPLAY": "wayland-0"} + here = paste.WAYLAND + + def test_ydotool_presses_down_then_lets_go_in_reverse(self): + self.assertEqual(self.press("ctrl+v"), + ["ydotool", "key", "29:1", "47:1", "47:0", "29:0"]) + + def test_three_keys(self): + self.assertEqual(self.press("ctrl+shift+v"), + ["ydotool", "key", "29:1", "42:1", "47:1", + "47:0", "42:0", "29:0"]) + + def test_the_synonyms_land_on_the_same_codes(self): + self.assertEqual(self.press("control+insert"), + ["ydotool", "key", "29:1", "110:1", "110:0", "29:0"]) + self.assertEqual(self.press("super+enter"), self.press("meta+return")) + + def test_a_failure_asks_after_the_daemon(self): + """ydotool needs ydotoold, and says nothing useful when it is not up.""" + with self.assertRaises(paste.PasteError) as caught: + self.press("ctrl+v", FakeCompleted(returncode=1, stderr="no socket")) + self.assertIn("ydotoold", str(caught.exception)) + + +@linux_only +class X11(DesktopContract, DikteTest): + env: ClassVar[dict] = {"XDG_SESSION_TYPE": "x11", "DISPLAY": ":0"} + here = paste.X11 + + def test_xdotool_takes_the_combination_as_one_argument(self): + self.assertEqual(self.press("ctrl+v"), + ["xdotool", "key", "--clearmodifiers", "ctrl+v"]) + + def test_three_keys(self): + self.assertEqual(self.press("ctrl+shift+v"), + ["xdotool", "key", "--clearmodifiers", "ctrl+shift+v"]) + + def test_the_keys_x_spells_differently(self): + """xdotool wants keysyms, not the names the code table is keyed by.""" + self.assertEqual(self.press("control+insert")[-1], "ctrl+Insert") + self.assertEqual(self.press("meta+enter")[-1], "super+Return") + + def test_no_daemon_to_ask_after(self): + with self.assertRaises(paste.PasteError) as caught: + self.press("ctrl+v", FakeCompleted(returncode=1, stderr="bad keysym")) + self.assertNotIn("ydotoold", str(caught.exception)) + if __name__ == "__main__": unittest.main() From 1544cca15c84bb37508fc074082ef962f58b7796 Mon Sep 17 00:00:00 2001 From: yusufipk Date: Sat, 1 Aug 2026 20:32:07 +0700 Subject: [PATCH 3/5] Install the GNOME shortcut this already knew how to install install_shortcut(), remove_shortcut(), shortcut_status() and desktop_name() were added and nothing called them: the settings window and the command line both still went straight to the KDE ones. On GNOME, pressing Install ran kwriteconfig6, which is not there, and the three interface strings written for this never appeared. README said the settings window installs a GNOME or KDE shortcut, and it installed neither. Both callers go through the chooser now, and the status line names whichever desktop answered. --- cli.py | 6 +- hotkey.py | 5 +- settings_ui.py | 33 ++++---- tests/test_hotkey.py | 191 ++++++++++++++++++++++++++++++++++++++++++- 4 files changed, 215 insertions(+), 20 deletions(-) diff --git a/cli.py b/cli.py index b584458..facb667 100644 --- a/cli.py +++ b/cli.py @@ -702,7 +702,7 @@ def cmd_shortcut(opts): if opts.shortcut == "status": rows = {} for name, (desktop_id, _label, key) in SHORTCUTS.items(): - rows[name] = {"registered": hotkey.kde_shortcut_status(desktop_id), + rows[name] = {"registered": hotkey.shortcut_status(desktop_id), "configured": conf[key]} lines = [f"{name:8} {row['registered'] or '(not installed)':16} " f"setting: {row['configured'] or '(none)'}" @@ -713,7 +713,7 @@ def cmd_shortcut(opts): desktop_id, label, key = SHORTCUTS[opts.which] if opts.shortcut == "remove": - hotkey.remove_kde_shortcut(desktop_id) + hotkey.remove_shortcut(desktop_id) return out(opts, {"ok": True, "removed": opts.which}, f"Removed the {opts.which} shortcut.") @@ -727,7 +727,7 @@ def cmd_shortcut(opts): return fail(opts, f"{combo} is also used by: {', '.join(clashes[:6])}. " "Pass --force to install it anyway.", 1, conflicts=clashes) - ok, message = hotkey.install_kde_shortcut( + ok, message = hotkey.install_shortcut( combo, ipc.command_for(opts.which), name=label, desktop_id=desktop_id, ) if not ok: diff --git a/hotkey.py b/hotkey.py index 8347f50..b6b3ed1 100644 --- a/hotkey.py +++ b/hotkey.py @@ -176,7 +176,7 @@ class EvdevHotkey(QObject): return True -# --- KDE custom shortcut -------------------------------------------------- +# --- the desktop's own shortcut ------------------------------------------- def _gnome(): desktop = os.environ.get("XDG_CURRENT_DESKTOP", "").lower() @@ -322,6 +322,9 @@ def shortcut_status(desktop_id=DESKTOP_ID): def desktop_name(): return "GNOME" if _gnome() else "KDE" + +# --- KDE ------------------------------------------------------------------ + def install_kde_shortcut(shortcut, exec_command, name="Dikte: start/stop recording", desktop_id=DESKTOP_ID): """Write the desktop file and the kglobalshortcutsrc entry. diff --git a/settings_ui.py b/settings_ui.py index 9638ae8..1690a6b 100644 --- a/settings_ui.py +++ b/settings_ui.py @@ -1289,7 +1289,7 @@ class SettingsWindow(QDialog): ) if answer != QMessageBox.StandardButton.Yes: return - ok, message = hotkey.install_kde_shortcut(combo, self.launch_command) + ok, message = hotkey.install_shortcut(combo, self.launch_command) QMessageBox.information(self, t("Shortcut"), message) if ok: self.conf["shortcut"] = combo @@ -1297,14 +1297,15 @@ class SettingsWindow(QDialog): self._refresh_shortcut_status() def _remove_shortcut(self): - hotkey.remove_kde_shortcut() + hotkey.remove_shortcut() self._refresh_shortcut_status() def _refresh_shortcut_status(self): - current = hotkey.kde_shortcut_status() + current = hotkey.shortcut_status() self.shortcut_status.setText( - t("Registered in KDE: {shortcut}", shortcut=current) if current - else t("No KDE shortcut installed.") + t("Registered in {desktop}: {shortcut}", + desktop=hotkey.desktop_name(), shortcut=current) if current + else t("No global shortcut installed.") ) def _install_meeting_shortcut(self): @@ -1322,7 +1323,7 @@ class SettingsWindow(QDialog): ) if answer != QMessageBox.StandardButton.Yes: return - ok, message = hotkey.install_kde_shortcut( + ok, message = hotkey.install_shortcut( combo, self.meeting_command, name="Dikte: start/end a meeting recording", desktop_id=hotkey.MEETING_DESKTOP_ID, ) @@ -1333,14 +1334,15 @@ class SettingsWindow(QDialog): self._refresh_meeting_shortcut_status() def _remove_meeting_shortcut(self): - hotkey.remove_kde_shortcut(hotkey.MEETING_DESKTOP_ID) + hotkey.remove_shortcut(hotkey.MEETING_DESKTOP_ID) self._refresh_meeting_shortcut_status() def _refresh_meeting_shortcut_status(self): - current = hotkey.kde_shortcut_status(hotkey.MEETING_DESKTOP_ID) + current = hotkey.shortcut_status(hotkey.MEETING_DESKTOP_ID) self.meeting_shortcut_status.setText( - t("Registered in KDE: {shortcut}", shortcut=current) if current - else t("No KDE shortcut installed. The tray menu starts a meeting too.") + t("Registered in {desktop}: {shortcut}", + desktop=hotkey.desktop_name(), shortcut=current) if current + else t("No global shortcut installed. The tray menu starts a meeting too.") ) # ---- Claude ---------------------------------------------------------- @@ -1360,7 +1362,7 @@ class SettingsWindow(QDialog): ) if answer != QMessageBox.StandardButton.Yes: return - ok, message = hotkey.install_kde_shortcut( + ok, message = hotkey.install_shortcut( combo, self.ask_command, name="Dikte: ask Claude Code", desktop_id=hotkey.ASK_DESKTOP_ID, ) @@ -1371,14 +1373,15 @@ class SettingsWindow(QDialog): self._refresh_ask_shortcut_status() def _remove_ask_shortcut(self): - hotkey.remove_kde_shortcut(hotkey.ASK_DESKTOP_ID) + hotkey.remove_shortcut(hotkey.ASK_DESKTOP_ID) self._refresh_ask_shortcut_status() def _refresh_ask_shortcut_status(self): - current = hotkey.kde_shortcut_status(hotkey.ASK_DESKTOP_ID) + current = hotkey.shortcut_status(hotkey.ASK_DESKTOP_ID) self.assistant_shortcut_status.setText( - t("Registered in KDE: {shortcut}", shortcut=current) if current - else t("No KDE shortcut installed. The tray menu asks it too.") + t("Registered in {desktop}: {shortcut}", + desktop=hotkey.desktop_name(), shortcut=current) if current + else t("No global shortcut installed. The tray menu asks it too.") ) def _assistant_provider_changed(self): diff --git a/tests/test_hotkey.py b/tests/test_hotkey.py index b3e0d70..4297724 100644 --- a/tests/test_hotkey.py +++ b/tests/test_hotkey.py @@ -1,5 +1,7 @@ -"""Parsing a shortcut, and the KDE entry it is installed as.""" +"""Parsing a shortcut, and the entry the desktop is asked to register it as.""" +import contextlib +import os import subprocess import unittest from unittest import mock @@ -121,6 +123,193 @@ class Bindings(DikteTest): self.assertEqual(len(listener._bindings[57]), 2) +@linux_only +class Chooser(DikteTest): + """Which desktop is asked to register the shortcut.""" + + @contextlib.contextmanager + def under(self, desktop, has_gsettings=True): + """A session that says it is this desktop, with or without gsettings.""" + with mock.patch.dict(os.environ, {"XDG_CURRENT_DESKTOP": desktop}), \ + mock.patch.object(hotkey.shutil, "which", + return_value="/usr/bin/gsettings" + if has_gsettings else None): + yield + + def test_gnome_when_the_session_says_so_and_gsettings_is_there(self): + with self.under("GNOME"): + self.assertEqual(hotkey.desktop_name(), "GNOME") + + def test_kde_otherwise(self): + with self.under("KDE"): + self.assertEqual(hotkey.desktop_name(), "KDE") + + def test_a_gnome_session_with_no_gsettings_falls_back(self): + """Nothing to write the binding with, so KDE's file is the only try.""" + with self.under("GNOME", has_gsettings=False): + self.assertEqual(hotkey.desktop_name(), "KDE") + + def test_the_desktop_is_matched_loosely(self): + for desktop in ("GNOME", "ubuntu:GNOME", "gnome"): + with self.subTest(desktop=desktop), self.under(desktop): + self.assertEqual(hotkey.desktop_name(), "GNOME") + + def test_installing_goes_to_whichever_it_is(self): + with self.under("GNOME"), \ + mock.patch.object(hotkey, "install_gnome_shortcut", + return_value=(True, "ok")) as gnome: + hotkey.install_shortcut("Ctrl+Space", "dikte toggle") + gnome.assert_called_once() + + with self.under("KDE"), \ + mock.patch.object(hotkey, "install_kde_shortcut", + return_value=(True, "ok")) as kde: + hotkey.install_shortcut("Ctrl+Space", "dikte toggle") + kde.assert_called_once() + + def test_removing_and_reading_back_go_to_the_same_one(self): + with self.under("GNOME"), \ + mock.patch.object(hotkey, "remove_gnome_shortcut") as remove, \ + mock.patch.object(hotkey, "gnome_shortcut_status", + return_value="Ctrl+Space") as status: + hotkey.remove_shortcut() + self.assertEqual(hotkey.shortcut_status(), "Ctrl+Space") + remove.assert_called_once() + status.assert_called_once() + + +@linux_only +class GnomeAccelerator(DikteTest): + """Qt spells a combination one way, GNOME another.""" + + def test_the_default_shortcut(self): + self.assertEqual(hotkey.gnome_accelerator("Ctrl+Space"), "Space") + + def test_several_modifiers_keep_their_order(self): + self.assertEqual(hotkey.gnome_accelerator("Ctrl+Alt+A"), "a") + + def test_the_synonyms(self): + self.assertEqual(hotkey.gnome_accelerator("Meta+A"), + hotkey.gnome_accelerator("Super+A")) + self.assertEqual(hotkey.gnome_accelerator("Control+A"), + hotkey.gnome_accelerator("Ctrl+A")) + + def test_a_modifier_repeated_is_written_once(self): + self.assertEqual(hotkey.gnome_accelerator("Ctrl+Control+A"), "a") + + def test_modifiers_with_no_key_are_not_a_shortcut(self): + self.assertEqual(hotkey.gnome_accelerator("Ctrl+Alt"), "") + self.assertEqual(hotkey.gnome_accelerator(""), "") + + def test_what_goes_out_comes_back_the_way_dikte_writes_it(self): + for shortcut in ("Ctrl+Space", "Ctrl+Alt+A", "Shift+F9", "Super+M"): + with self.subTest(shortcut=shortcut): + accelerator = hotkey.gnome_accelerator(shortcut) + self.assertEqual(hotkey.display_accelerator(accelerator), shortcut) + + def test_the_control_spelling_gnome_also_uses(self): + self.assertEqual(hotkey.display_accelerator("a"), "Ctrl+A") + + def test_an_empty_binding(self): + self.assertEqual(hotkey.display_accelerator(""), "") + + +@linux_only +class GsettingsArray(DikteTest): + def test_a_list_of_paths(self): + self.assertEqual( + hotkey._gsettings_array("['/org/gnome/one/', '/org/gnome/two/']"), + ["/org/gnome/one/", "/org/gnome/two/"]) + + def test_the_empty_form_gsettings_prints(self): + self.assertEqual(hotkey._gsettings_array("@as []"), []) + + def test_nothing_at_all(self): + self.assertEqual(hotkey._gsettings_array(""), []) + + def test_something_that_is_not_an_array(self): + with self.assertRaises(ValueError): + hotkey._gsettings_array("'just a string'") + + +@linux_only +class GnomeShortcut(DikteTest): + """The gsettings calls, without a session bus to make them against.""" + + def setUp(self): + super().setUp() + self.enterContext(mock.patch.dict(os.environ, + {"XDG_CURRENT_DESKTOP": "GNOME"})) + self.enterContext(mock.patch.object(hotkey.shutil, "which", + return_value="/usr/bin/gsettings")) + + def gsettings(self, listed="@as []", binding="'Space'"): + def run(cmd, **kwargs): + if cmd[1] == "get" and cmd[3] == "custom-keybindings": + return FakeCompleted(stdout=listed) + return FakeCompleted(stdout=binding) + return mock.patch.object(subprocess, "run", side_effect=run) + + def written(self, run, key): + """The value the last `gsettings set ... ` was given.""" + for call in reversed(run.mock_calls): + cmd = call.args[0] if call.args else [] + if len(cmd) > 4 and cmd[1] == "set" and cmd[3] == key: + return cmd[4] + return None + + def test_installing_registers_the_path_the_name_and_the_binding(self): + with self.gsettings() as run: + ok, message = hotkey.install_shortcut("Ctrl+Space", "dikte toggle") + self.assertTrue(ok) + self.assertIn("Ctrl+Space", message) + self.assertIn(hotkey.DESKTOP_ID.removesuffix(".desktop"), + self.written(run, "custom-keybindings")) + self.assertEqual(self.written(run, "command"), repr("dikte toggle")) + self.assertEqual(self.written(run, "binding"), repr("Space")) + + def test_installing_twice_does_not_list_the_path_twice(self): + path = hotkey._gnome_path(hotkey.DESKTOP_ID) + with self.gsettings(listed=repr([path])) as run: + hotkey.install_shortcut("Ctrl+Space", "dikte toggle") + self.assertIsNone(self.written(run, "custom-keybindings")) + + def test_each_verb_gets_its_own_path(self): + self.assertNotEqual(hotkey._gnome_path(hotkey.DESKTOP_ID), + hotkey._gnome_path(hotkey.ASK_DESKTOP_ID)) + + def test_a_shortcut_gnome_cannot_express(self): + with self.gsettings(): + ok, message = hotkey.install_shortcut("Ctrl+Alt", "dikte toggle") + self.assertFalse(ok) + self.assertIn("Ctrl+Alt", message) + + def test_no_session_bus_to_talk_to(self): + with mock.patch.object(subprocess, "run", side_effect=OSError("no bus")): + ok, _ = hotkey.install_shortcut("Ctrl+Space", "dikte toggle") + self.assertFalse(ok) + + def test_reading_back_a_shortcut_that_is_registered(self): + path = hotkey._gnome_path(hotkey.DESKTOP_ID) + with self.gsettings(listed=repr([path])): + self.assertEqual(hotkey.shortcut_status(), "Ctrl+Space") + + def test_reading_back_one_that_is_not(self): + with self.gsettings(listed="@as []"): + self.assertIsNone(hotkey.shortcut_status()) + + def test_removing_takes_the_path_off_the_list(self): + path = hotkey._gnome_path(hotkey.DESKTOP_ID) + with self.gsettings(listed=repr([path, "/org/gnome/other/"])) as run: + hotkey.remove_shortcut() + self.assertEqual(self.written(run, "custom-keybindings"), + repr(["/org/gnome/other/"])) + + def test_removing_one_that_was_never_installed(self): + with self.gsettings(listed="@as []"): + hotkey.remove_shortcut() # must not raise + + @linux_only class KdeShortcut(DikteTest): def setUp(self): From 45a064e545eeb6f66f494cbfd890d22a483906c4 Mon Sep 17 00:00:00 2001 From: yusufipk Date: Sat, 1 Aug 2026 20:34:15 +0700 Subject: [PATCH 4/5] Tell a recorder that died from one that was asked to stop The pump now says when the capture ended with nothing captured, which is worth saying: parec refusing the device looks like silence otherwise. But stop() ends it the same way, so a recording shorter than 0.3 s raised that alarm first and "Recording too short" second, sending the user after a sound server that is fine. It follows the flag MeetingRecorder already carries for this. test_desktop_compat.py moves into the files for the modules it covers, so a test is where the next person looking at that module will find it. --- audio.py | 28 ++++++++---- tests/test_audio.py | 106 +++++++++++++++++++++++++++++++++++++++----- 2 files changed, 115 insertions(+), 19 deletions(-) diff --git a/audio.py b/audio.py index 0200058..39bceb2 100644 --- a/audio.py +++ b/audio.py @@ -44,6 +44,7 @@ class Recorder(QObject): self._buffer = bytearray() self._rms = [] self._cancelled = False + self._stopping = False self._lock = threading.Lock() @property @@ -71,6 +72,7 @@ class Recorder(QObject): self._buffer = bytearray() self._rms = [] self._cancelled = False + self._stopping = False self._max_bytes = int(max_seconds * RATE * SAMPLE_WIDTH * CHANNELS) self._thread = threading.Thread(target=self._pump, daemon=True) self._thread.start() @@ -94,17 +96,25 @@ class Recorder(QObject): break except (OSError, ValueError): pass - if not self._cancelled and not self._buffer and proc.poll() is not None: - try: - detail = proc.stderr.read().decode("utf-8", "replace").strip() - except (AttributeError, OSError): - detail = "" - self.failed.emit(t( - "Audio recorder stopped before receiving sound: {error}", - error=detail or f"exit code {proc.returncode}", - )) + # Nobody asked it to end and it captured nothing: the recorder is not + # installed properly, or the device was refused. Said out loud here, + # because stop() would otherwise report it as a recording that was too + # short, which sends the user looking in the wrong place. + with self._lock: + captured = bool(self._buffer) + if self._stopping or self._cancelled or captured: + return + try: + detail = proc.stderr.read().decode("utf-8", "replace").strip() + except (AttributeError, OSError): + detail = "" + self.failed.emit(t( + "Audio recorder stopped before receiving sound: {error}", + error=detail or f"exit code {proc.returncode}", + )) def _terminate(self): + self._stopping = True proc = self._proc if proc and proc.poll() is None: try: diff --git a/tests/test_audio.py b/tests/test_audio.py index 9a8dbaa..52f2f54 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -7,6 +7,7 @@ speakers, and neither list may go missing when pactl is absent. import array import contextlib +import io import json import os import subprocess @@ -194,10 +195,10 @@ class FakeProcess: """A pw-record that hands over a fixed buffer and then ends.""" def __init__(self, data): - import io self.stdout = io.BytesIO(data) self.stderr = io.BytesIO(b"") self.signals = [] + self.returncode = 0 self._alive = True def poll(self): @@ -215,6 +216,48 @@ class FakeProcess: self._alive = False +@linux_only +class RecordingCommand(DikteTest): + """Which program captures the microphone, and how it is asked to.""" + + def test_parec_is_preferred(self): + """It speaks to PulseAudio and to PipeWire's compatibility service, so + it is the one that works on both desktops.""" + with only_these_tools("parec", "pw-record"): + self.assertEqual(audio.recording_command()[0], "parec") + + def test_pw_record_is_the_fallback(self): + with only_these_tools("pw-record"): + self.assertEqual(audio.recording_command()[0], "pw-record") + + def test_neither_is_installed(self): + with only_these_tools(): + self.assertEqual(audio.recording_command(), []) + + def test_both_capture_the_format_the_rest_of_the_code_expects(self): + for tool in ("parec", "pw-record"): + with self.subTest(tool=tool), only_these_tools(tool): + cmd = audio.recording_command() + joined = " ".join(cmd) + self.assertIn(str(audio.RATE), joined) + self.assertIn(str(audio.CHANNELS), joined) + self.assertIn("s16", joined) + + def test_a_chosen_microphone_reaches_either_one(self): + with only_these_tools("parec"): + self.assertIn("--device=alsa_input.usb", audio.recording_command( + "alsa_input.usb")) + with only_these_tools("pw-record"): + self.assertIn("--target=alsa_input.usb", audio.recording_command( + "alsa_input.usb")) + + def test_no_microphone_named_means_no_device_flag(self): + for tool, flag in (("parec", "--device="), ("pw-record", "--target=")): + with self.subTest(tool=tool), only_these_tools(tool): + self.assertFalse([arg for arg in audio.recording_command() + if arg.startswith(flag)]) + + @linux_only class RecorderChain(DikteTest): """Start to WAV, with pw-record faked out.""" @@ -233,15 +276,6 @@ class RecorderChain(DikteTest): recorder.stop() return recorder, results, failures, popen - def test_pw_record_is_not_installed(self): - recorder = audio.Recorder() - failures = [] - recorder.failed.connect(failures.append) - with only_these_tools(): - recorder.start() - self.assertEqual(len(failures), 1) - self.assertIn("pipewire", failures[0]) - def test_the_capture_format_is_what_the_rest_of_the_code_expects(self): _, _, _, popen = self.record(silence(1.0)) cmd = popen.call_args.args[0] @@ -295,6 +329,58 @@ class RecorderChain(DikteTest): self.addCleanup(os.unlink, path) self.assertLessEqual(duration, 1.1) + def test_a_recorder_that_is_not_installed_at_all(self): + recorder = audio.Recorder() + failures = [] + recorder.failed.connect(failures.append) + with only_these_tools(): + recorder.start() + self.assertEqual(len(failures), 1) + self.assertIn("pulseaudio-utils", failures[0]) + + def pump(self, data=b"", stderr=b"", stopping=False, cancelled=False): + """Run the pump in this thread, where a queued signal would need an + event loop nobody is running here.""" + recorder = audio.Recorder() + failures = [] + recorder.failed.connect(failures.append) + proc = FakeProcess(data) + proc.stderr = io.BytesIO(stderr) + proc._alive = False + recorder._proc = proc + recorder._max_bytes = 10 ** 9 + recorder._stopping = stopping + recorder._cancelled = cancelled + recorder._pump() + return failures + + def test_a_recorder_that_died_on_its_own_says_so(self): + """parec refused the device, or the sound server went away.""" + failures = self.pump(stderr=b"connection refused\n") + self.assertEqual(len(failures), 1) + self.assertIn("connection refused", failures[0]) + + def test_a_death_with_nothing_on_stderr_still_names_the_exit_code(self): + failures = self.pump() + self.assertIn("exit code", failures[0]) + + def test_a_recording_we_ended_ourselves_is_not_a_death(self): + """Otherwise a stray keypress produces two errors, and the first one + sends the user looking for a broken sound server.""" + self.assertEqual(self.pump(stopping=True), []) + + def test_a_cancelled_recording_is_not_a_death(self): + self.assertEqual(self.pump(cancelled=True), []) + + def test_a_recorder_that_captured_something_first_is_not_a_death(self): + self.assertEqual(self.pump(data=silence(0.5)), []) + + def test_a_short_recording_reports_only_that(self): + _, results, failures, _ = self.record(silence(0.1)) + self.assertEqual(results, []) + self.assertEqual(len(failures), 1) + self.assertIn("0.3", failures[0]) + def test_a_recorder_that_could_not_start(self): recorder = audio.Recorder() failures = [] From 5c429ab077e9da133e96028d081884a15163fa89 Mon Sep 17 00:00:00 2001 From: yusufipk Date: Sat, 1 Aug 2026 20:52:43 +0700 Subject: [PATCH 5/5] Ask parec for sound as it arrives, not two seconds of it at a time Left alone, parec buffers about two seconds and then hands the lot over at once. Measured against the chunk the level meter reads: parec, as it was longest gap 2005 ms parec --latency-msec=64 longest gap 86 ms pw-record longest gap 129 ms On a machine that has parec, which is every PulseAudio one and every PipeWire one with the compatibility service, the waveform stood still and then jumped, looking like a microphone that was not picking anything up. The recording itself was fine, but the tail of one can be lost that way too: stop() ends the process, and whatever is still held back goes with it. The number asked for is the meter's own chunk, which is the unit the rest of this file is measured in. --- audio.py | 7 +++++++ tests/test_audio.py | 12 ++++++++++++ 2 files changed, 19 insertions(+) diff --git a/audio.py b/audio.py index 39bceb2..0ef5038 100644 --- a/audio.py +++ b/audio.py @@ -27,6 +27,7 @@ CHANNELS = 1 SAMPLE_WIDTH = 2 # s16 CHUNK_FRAMES = 1024 CHUNK_BYTES = CHUNK_FRAMES * SAMPLE_WIDTH * CHANNELS +CHUNK_LATENCY_MS = round(CHUNK_FRAMES / RATE * 1000) MIN_FRAMES = int(RATE * 0.25) @@ -184,6 +185,12 @@ def recording_command(target=""): cmd = [ "parec", "--record", "--raw", f"--rate={RATE}", f"--channels={CHANNELS}", "--format=s16le", + # Left alone, parec holds about two seconds before handing anything + # over, and then hands over all of it at once: the level meter sits + # still and jumps, and the tail of a recording can be lost on the + # way out. A chunk of the meter is the unit the rest of this file + # is measured in, so ask for that. + f"--latency-msec={CHUNK_LATENCY_MS}", ] if target: cmd.append(f"--device={target}") diff --git a/tests/test_audio.py b/tests/test_audio.py index 52f2f54..7595d1b 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -243,6 +243,18 @@ class RecordingCommand(DikteTest): self.assertIn(str(audio.CHANNELS), joined) self.assertIn("s16", joined) + def test_parec_is_asked_for_the_level_meter_s_own_chunk(self): + """Left alone it buffers about two seconds, which the waveform shows as + a still bar that jumps once a second, and which can cost the tail of a + recording when the process is asked to stop.""" + with only_these_tools("parec"): + self.assertIn(f"--latency-msec={audio.CHUNK_LATENCY_MS}", + audio.recording_command()) + + def test_the_latency_asked_for_is_the_chunk_the_meter_reads(self): + self.assertEqual(audio.CHUNK_LATENCY_MS, + round(audio.CHUNK_FRAMES / audio.RATE * 1000)) + def test_a_chosen_microphone_reaches_either_one(self): with only_these_tools("parec"): self.assertIn("--device=alsa_input.usb", audio.recording_command(