Merge pull request #5 from muzafferemre06/feature/ubuntu-x11-gnome

Support GNOME X11 and PulseAudio without changing Wayland
This commit is contained in:
Yusuf İpek
2026-08-01 16:58:18 +03:00
committed by GitHub
12 changed files with 851 additions and 161 deletions
+9 -2
View File
@@ -29,8 +29,15 @@ systemctl --user enable --now ydotool # needed for auto-paste
dikte # the settings window opens on first run dikte # the settings window opens on first run
``` ```
`install.sh` adds the `dikte` command, a menu entry, an autostart entry and the On Ubuntu/GNOME X11, recording uses PulseAudio and clipboard/paste use the X11
KDE shortcut. 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 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 runs on either one (`gpt-4o-transcribe` by default), cleanup always on
+9 -2
View File
@@ -29,8 +29,15 @@ systemctl --user enable --now ydotool # otomatik yapıştırma için
dikte # ilk açılışta ayarlar penceresi gelir dikte # ilk açılışta ayarlar penceresi gelir
``` ```
`install.sh` `dikte` komutunu, menü girdisini, oturum açılışında otomatik Ubuntu/GNOME X11 için kayıt PulseAudio üzerinden, pano ve yapıştırma ise X11
başlatmayı ve KDE kısayolunu kurar. 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 Ayarlar penceresinde iki anahtar istenir: **OpenAI** ve **OpenRouter**. Sesi
yazıya çevirme ikisinden birinde çalışır (varsayılan `gpt-4o-transcribe`), yazıya çevirme ikisinden birinde çalışır (varsayılan `gpt-4o-transcribe`),
+61 -15
View File
@@ -27,11 +27,12 @@ CHANNELS = 1
SAMPLE_WIDTH = 2 # s16 SAMPLE_WIDTH = 2 # s16
CHUNK_FRAMES = 1024 CHUNK_FRAMES = 1024
CHUNK_BYTES = CHUNK_FRAMES * SAMPLE_WIDTH * CHANNELS CHUNK_BYTES = CHUNK_FRAMES * SAMPLE_WIDTH * CHANNELS
CHUNK_LATENCY_MS = round(CHUNK_FRAMES / RATE * 1000)
MIN_FRAMES = int(RATE * 0.25) MIN_FRAMES = int(RATE * 0.25)
class Recorder(QObject): 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 level = pyqtSignal(float) # 0.0 - 1.0, for the waveform
stopped = pyqtSignal(str, float, object) # wav path, duration (s), per-chunk RMS stopped = pyqtSignal(str, float, object) # wav path, duration (s), per-chunk RMS
@@ -44,6 +45,7 @@ class Recorder(QObject):
self._buffer = bytearray() self._buffer = bytearray()
self._rms = [] self._rms = []
self._cancelled = False self._cancelled = False
self._stopping = False
self._lock = threading.Lock() self._lock = threading.Lock()
@property @property
@@ -53,21 +55,13 @@ class Recorder(QObject):
def start(self, target="", max_seconds=300): def start(self, target="", max_seconds=300):
if self.active: if self.active:
return return
if not shutil.which("pw-record"): cmd = recording_command(target)
self.failed.emit(t("pw-record not found. Is pipewire-audio installed?")) if not cmd:
self.failed.emit(t(
"No audio recorder found. Install pulseaudio-utils or pipewire-audio."
))
return return
cmd = [
"pw-record",
"--raw",
f"--rate={RATE}",
f"--channels={CHANNELS}",
"--format=s16",
]
if target:
cmd.append(f"--target={target}")
cmd.append("-")
try: try:
self._proc = subprocess.Popen( self._proc = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0 cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0
@@ -79,12 +73,14 @@ class Recorder(QObject):
self._buffer = bytearray() self._buffer = bytearray()
self._rms = [] self._rms = []
self._cancelled = False self._cancelled = False
self._stopping = False
self._max_bytes = int(max_seconds * RATE * SAMPLE_WIDTH * CHANNELS) self._max_bytes = int(max_seconds * RATE * SAMPLE_WIDTH * CHANNELS)
self._thread = threading.Thread(target=self._pump, daemon=True) self._thread = threading.Thread(target=self._pump, daemon=True)
self._thread.start() self._thread.start()
def _pump(self): def _pump(self):
stdout = self._proc.stdout proc = self._proc
stdout = proc.stdout
try: try:
while True: while True:
chunk = stdout.read(CHUNK_BYTES) chunk = stdout.read(CHUNK_BYTES)
@@ -101,8 +97,25 @@ class Recorder(QObject):
break break
except (OSError, ValueError): except (OSError, ValueError):
pass pass
# 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): def _terminate(self):
self._stopping = True
proc = self._proc proc = self._proc
if proc and proc.poll() is None: if proc and proc.poll() is None:
try: try:
@@ -161,6 +174,39 @@ def write_wav(pcm, rate=RATE, channels=CHANNELS, width=SAMPLE_WIDTH):
return path 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",
# 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}")
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): class MeetingRecorder(QObject):
"""Microphone and speaker output into one stereo file: left is you, right is """Microphone and speaker output into one stereo file: left is you, right is
everyone else. everyone else.
+3 -3
View File
@@ -702,7 +702,7 @@ def cmd_shortcut(opts):
if opts.shortcut == "status": if opts.shortcut == "status":
rows = {} rows = {}
for name, (desktop_id, _label, key) in SHORTCUTS.items(): 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]} "configured": conf[key]}
lines = [f"{name:8} {row['registered'] or '(not installed)':16} " lines = [f"{name:8} {row['registered'] or '(not installed)':16} "
f"setting: {row['configured'] or '(none)'}" f"setting: {row['configured'] or '(none)'}"
@@ -713,7 +713,7 @@ def cmd_shortcut(opts):
desktop_id, label, key = SHORTCUTS[opts.which] desktop_id, label, key = SHORTCUTS[opts.which]
if opts.shortcut == "remove": if opts.shortcut == "remove":
hotkey.remove_kde_shortcut(desktop_id) hotkey.remove_shortcut(desktop_id)
return out(opts, {"ok": True, "removed": opts.which}, return out(opts, {"ok": True, "removed": opts.which},
f"Removed the {opts.which} shortcut.") 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])}. " return fail(opts, f"{combo} is also used by: {', '.join(clashes[:6])}. "
"Pass --force to install it anyway.", 1, conflicts=clashes) "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, combo, ipc.command_for(opts.which), name=label, desktop_id=desktop_id,
) )
if not ok: if not ok:
+153 -2
View File
@@ -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 glob
import os import os
import pathlib import pathlib
import re import re
import select import select
import shutil
import struct import struct
import subprocess import subprocess
import threading import threading
@@ -19,6 +21,8 @@ ASK_DESKTOP_ID = "dikte-ask.desktop"
APPLICATIONS_DIR = pathlib.Path.home() / ".local/share/applications" APPLICATIONS_DIR = pathlib.Path.home() / ".local/share/applications"
DESKTOP_FILE = APPLICATIONS_DIR / DESKTOP_ID DESKTOP_FILE = APPLICATIONS_DIR / DESKTOP_ID
SHORTCUTS_FILE = pathlib.Path.home() / ".config/kglobalshortcutsrc" 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) -------------------------- # --- evdev key codes (linux/input-event-codes.h) --------------------------
@@ -172,7 +176,154 @@ class EvdevHotkey(QObject):
return True return True
# --- KDE custom shortcut -------------------------------------------------- # --- the desktop's own 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 <Primary><Alt>a syntax."""
parts = [part.strip() for part in str(shortcut).split("+") if part.strip()]
modifiers = []
key = ""
names = {
"ctrl": "<Primary>", "control": "<Primary>",
"alt": "<Alt>", "shift": "<Shift>",
"super": "<Super>", "meta": "<Super>",
}
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 (("<Primary>", "Ctrl"), ("<Control>", "Ctrl"),
("<Alt>", "Alt"), ("<Shift>", "Shift"),
("<Super>", "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"
# --- KDE ------------------------------------------------------------------
def install_kde_shortcut(shortcut, exec_command, name="Dikte: start/stop recording", def install_kde_shortcut(shortcut, exec_command, name="Dikte: start/stop recording",
desktop_id=DESKTOP_ID): desktop_id=DESKTOP_ID):
+23 -10
View File
@@ -101,19 +101,22 @@ TR = {
"Unexpected error: {error}": "Beklenmeyen hata: {error}", "Unexpected error: {error}": "Beklenmeyen hata: {error}",
# --- audio / paste errors ----------------------------------------- # --- 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}", "Could not start recording: {error}": "Kayıt başlatılamadı: {error}",
"wl-copy not found. Install wl-clipboard.": "No audio recorder found. Install pulseaudio-utils or pipewire-audio.":
"wl-copy bulunamadı. wl-clipboard paketini kur.", "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}",
"Could not copy to clipboard: {error}": "Panoya kopyalanamadı: {error}", "Could not copy to clipboard: {error}": "Panoya kopyalanamadı: {error}",
"wl-copy exited with code {code}.": "wl-copy {code} koduyla çıktı.", "{tool} not found. Install {packages}.":
"ydotool not found, cannot paste automatically.": "{tool} bulunamadı. {packages} paketlerini kur.",
"ydotool bulunamadı, otomatik yapıştırma yapı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.",
"Unknown key: {key}": "Bilinmeyen tuş: {key}", "Unknown key: {key}": "Bilinmeyen tuş: {key}",
"Could not run ydotool: {error}": "ydotool çalıştırılamadı: {error}", "Could not run {tool}: {error}": "{tool} çalıştırılamadı: {error}",
"ydotool failed: {error}\nIs ydotoold running? (systemctl --user status ydotool)": "{tool} failed: {error}": "{tool} hatası: {error}",
"ydotool hatası: {error}\nydotoold çalışıyor mu? (systemctl --user status ydotool)", "Is ydotoold running? (systemctl --user status ydotool)":
"ydotoold çalışıyor mu? (systemctl --user status ydotool)",
# --- api errors ---------------------------------------------------- # --- api errors ----------------------------------------------------
"{service} API key is empty. Add it in Settings.": "{service} API key is empty. Add it in Settings.":
@@ -268,9 +271,19 @@ TR = {
# --- settings: shortcut ------------------------------------------------ # --- settings: shortcut ------------------------------------------------
"Install as a KDE shortcut": "KDE kısayolu olarak kur", "Install as a KDE shortcut": "KDE kısayolu olarak kur",
"Install as a global shortcut": "Global kısayol olarak kur",
"Remove": "Kaldır", "Remove": "Kaldır",
"Registered in KDE: {shortcut}": "KDE'de kayıtlı: {shortcut}", "Registered in KDE: {shortcut}": "KDE'de kayıtlı: {shortcut}",
"No KDE shortcut installed.": "KDE kısayolu kurulu değil.", "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": "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", "Yerleşik dinleyici kullan (/dev/input), KDE kısayolu henüz etkin değilken",
"Works immediately, no session restart. The only difference: the key " "Works immediately, no session restart. The only difference: the key "
+21 -6
View File
@@ -19,21 +19,33 @@ echo "────────────────"
# 1. Dependencies ---------------------------------------------------------- # 1. Dependencies ----------------------------------------------------------
missing=() 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") command -v "$cmd" >/dev/null || missing+=("$cmd")
done done
python3 -c 'import PyQt6.QtWidgets' 2>/dev/null || missing+=("python-pyqt6") python3 -c 'import PyQt6.QtWidgets' 2>/dev/null || missing+=("python-pyqt6")
if ((${#missing[@]})); then if ((${#missing[@]})); then
warn "Missing: ${missing[*]}" 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 echo
else else
ok "All dependencies present" ok "All dependencies present"
fi fi
# 2. ydotoold -------------------------------------------------------------- # 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 \ if systemctl --user is-active --quiet ydotool 2>/dev/null \
|| systemctl --user is-active --quiet ydotoold 2>/dev/null; then || systemctl --user is-active --quiet ydotoold 2>/dev/null; then
ok "ydotoold is running (auto-paste ready)" ok "ydotoold is running (auto-paste ready)"
@@ -87,7 +99,10 @@ Type=Application
X-KDE-GlobalAccel-CommandShortcut=true X-KDE-GlobalAccel-CommandShortcut=true
EOF 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 \ kwriteconfig6 --notify --file kglobalshortcutsrc \
--group services --group dikte-toggle.desktop \ --group services --group dikte-toggle.desktop \
--key _launch "$SHORTCUT" --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 "next login. Until then open Settings → Shortcut and turn on the"
say "built-in listener to use it right away." say "built-in listener to use it right away."
else 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 fi
echo echo
ok "Done. Start it with: dikte" 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 echo
+116 -35
View File
@@ -1,37 +1,119 @@
"""Clipboard (wl-clipboard) and key injection (ydotool).""" """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 shutil
import subprocess import subprocess
import time import time
from i18n import t 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 = { KEYCODES = {
"ctrl": 29, "control": 29, "shift": 42, "alt": 56, "super": 125, "meta": 125, "ctrl": 29, "control": 29, "shift": 42, "alt": 56, "super": 125, "meta": 125,
"v": 47, "insert": 110, "enter": 28, "return": 28, "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): class PasteError(Exception):
pass 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(): def read_clipboard():
if not shutil.which("wl-paste"): here = desktop()
if not shutil.which(here.read_command[0]):
return None return None
try: try:
res = subprocess.run(["wl-paste", "--no-newline"], capture_output=True, timeout=5) res = subprocess.run(here.read_command, capture_output=True, timeout=5)
except (subprocess.SubprocessError, OSError): except (subprocess.SubprocessError, OSError):
return None return None
return res.stdout if res.returncode == 0 else None return res.stdout if res.returncode == 0 else None
def _run_wl_copy(payload): def _run_copy(payload):
"""wl-copy forks to keep owning the selection; leaving its pipes open makes """The clipboard owner forks to keep holding the selection; leaving its
subprocess.run wait for EOF forever, hence DEVNULL.""" pipes open makes subprocess.run wait for EOF forever, hence DEVNULL."""
return subprocess.run( return subprocess.run(
["wl-copy"], desktop().copy_command,
input=payload, input=payload,
stdout=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
@@ -40,51 +122,50 @@ def _run_wl_copy(payload):
def copy(text): def copy(text):
if not shutil.which("wl-copy"): here = desktop()
raise PasteError(t("wl-copy not found. Install wl-clipboard.")) if not shutil.which(here.clipboard):
raise PasteError(t("{tool} not found. Install {packages}.",
tool=here.clipboard, packages=here.packages))
try: try:
res = _run_wl_copy(text.encode("utf-8")) res = _run_copy(text.encode("utf-8"))
except (subprocess.SubprocessError, OSError) as exc: except (subprocess.SubprocessError, OSError) as exc:
raise PasteError(t("Could not copy to clipboard: {error}", error=exc)) from exc raise PasteError(t("Could not copy to clipboard: {error}", error=exc)) from exc
if res.returncode != 0: 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=here.clipboard, code=res.returncode))
def copy_bytes(data): def copy_bytes(data):
if data is None or not shutil.which("wl-copy"): if data is None or not shutil.which(desktop().clipboard):
return return
try: try:
_run_wl_copy(data) _run_copy(data)
except (subprocess.SubprocessError, OSError): except (subprocess.SubprocessError, OSError):
pass pass
def ydotool_ready(): # --- the key press ---------------------------------------------------------
return shutil.which("ydotool") is not None
def paste_ready():
return shutil.which(desktop().keyboard) is not None
def press(shortcut="ctrl+v", delay=0.12): def press(shortcut="ctrl+v", delay=0.12):
"""Press a key combination through ydotool, e.g. 'ctrl+v'.""" """Press a key combination, e.g. 'ctrl+v'."""
if not ydotool_ready(): here = desktop()
raise PasteError(t("ydotool not found, cannot paste automatically.")) if not paste_ready():
raise PasteError(t("{tool} not found, cannot paste automatically.",
tool=here.keyboard))
codes = [] command = here.key_command(shortcut)
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)]
time.sleep(delay) # let the selection settle and focus come back time.sleep(delay) # let the selection settle and focus come back
try: try:
res = subprocess.run(["ydotool", "key", *seq], capture_output=True, res = subprocess.run(command, capture_output=True, text=True, timeout=10)
text=True, timeout=10)
except (subprocess.SubprocessError, OSError) as exc: 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: if res.returncode != 0:
raise PasteError(t( message = t("{tool} failed: {error}", tool=here.keyboard,
"ydotool failed: {error}\nIs ydotoold running? " error=res.stderr.strip() or "unknown error")
"(systemctl --user status ydotool)", raise PasteError(f"{message}\n{t(here.key_hint)}" if here.key_hint
error=res.stderr.strip() or "unknown error", else message)
))
+18 -15
View File
@@ -1289,7 +1289,7 @@ class SettingsWindow(QDialog):
) )
if answer != QMessageBox.StandardButton.Yes: if answer != QMessageBox.StandardButton.Yes:
return 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) QMessageBox.information(self, t("Shortcut"), message)
if ok: if ok:
self.conf["shortcut"] = combo self.conf["shortcut"] = combo
@@ -1297,14 +1297,15 @@ class SettingsWindow(QDialog):
self._refresh_shortcut_status() self._refresh_shortcut_status()
def _remove_shortcut(self): def _remove_shortcut(self):
hotkey.remove_kde_shortcut() hotkey.remove_shortcut()
self._refresh_shortcut_status() self._refresh_shortcut_status()
def _refresh_shortcut_status(self): def _refresh_shortcut_status(self):
current = hotkey.kde_shortcut_status() current = hotkey.shortcut_status()
self.shortcut_status.setText( self.shortcut_status.setText(
t("Registered in KDE: {shortcut}", shortcut=current) if current t("Registered in {desktop}: {shortcut}",
else t("No KDE shortcut installed.") desktop=hotkey.desktop_name(), shortcut=current) if current
else t("No global shortcut installed.")
) )
def _install_meeting_shortcut(self): def _install_meeting_shortcut(self):
@@ -1322,7 +1323,7 @@ class SettingsWindow(QDialog):
) )
if answer != QMessageBox.StandardButton.Yes: if answer != QMessageBox.StandardButton.Yes:
return return
ok, message = hotkey.install_kde_shortcut( ok, message = hotkey.install_shortcut(
combo, self.meeting_command, name="Dikte: start/end a meeting recording", combo, self.meeting_command, name="Dikte: start/end a meeting recording",
desktop_id=hotkey.MEETING_DESKTOP_ID, desktop_id=hotkey.MEETING_DESKTOP_ID,
) )
@@ -1333,14 +1334,15 @@ class SettingsWindow(QDialog):
self._refresh_meeting_shortcut_status() self._refresh_meeting_shortcut_status()
def _remove_meeting_shortcut(self): 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() self._refresh_meeting_shortcut_status()
def _refresh_meeting_shortcut_status(self): 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( self.meeting_shortcut_status.setText(
t("Registered in KDE: {shortcut}", shortcut=current) if current t("Registered in {desktop}: {shortcut}",
else t("No KDE shortcut installed. The tray menu starts a meeting too.") desktop=hotkey.desktop_name(), shortcut=current) if current
else t("No global shortcut installed. The tray menu starts a meeting too.")
) )
# ---- Claude ---------------------------------------------------------- # ---- Claude ----------------------------------------------------------
@@ -1360,7 +1362,7 @@ class SettingsWindow(QDialog):
) )
if answer != QMessageBox.StandardButton.Yes: if answer != QMessageBox.StandardButton.Yes:
return return
ok, message = hotkey.install_kde_shortcut( ok, message = hotkey.install_shortcut(
combo, self.ask_command, name="Dikte: ask Claude Code", combo, self.ask_command, name="Dikte: ask Claude Code",
desktop_id=hotkey.ASK_DESKTOP_ID, desktop_id=hotkey.ASK_DESKTOP_ID,
) )
@@ -1371,14 +1373,15 @@ class SettingsWindow(QDialog):
self._refresh_ask_shortcut_status() self._refresh_ask_shortcut_status()
def _remove_ask_shortcut(self): 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() self._refresh_ask_shortcut_status()
def _refresh_ask_shortcut_status(self): 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( self.assistant_shortcut_status.setText(
t("Registered in KDE: {shortcut}", shortcut=current) if current t("Registered in {desktop}: {shortcut}",
else t("No KDE shortcut installed. The tray menu asks it too.") 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): def _assistant_provider_changed(self):
+108 -10
View File
@@ -7,6 +7,7 @@ speakers, and neither list may go missing when pactl is absent.
import array import array
import contextlib import contextlib
import io
import json import json
import os import os
import subprocess import subprocess
@@ -194,10 +195,10 @@ class FakeProcess:
"""A pw-record that hands over a fixed buffer and then ends.""" """A pw-record that hands over a fixed buffer and then ends."""
def __init__(self, data): def __init__(self, data):
import io
self.stdout = io.BytesIO(data) self.stdout = io.BytesIO(data)
self.stderr = io.BytesIO(b"") self.stderr = io.BytesIO(b"")
self.signals = [] self.signals = []
self.returncode = 0
self._alive = True self._alive = True
def poll(self): def poll(self):
@@ -215,6 +216,60 @@ class FakeProcess:
self._alive = False 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_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(
"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 @linux_only
class RecorderChain(DikteTest): class RecorderChain(DikteTest):
"""Start to WAV, with pw-record faked out.""" """Start to WAV, with pw-record faked out."""
@@ -233,15 +288,6 @@ class RecorderChain(DikteTest):
recorder.stop() recorder.stop()
return recorder, results, failures, popen 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): def test_the_capture_format_is_what_the_rest_of_the_code_expects(self):
_, _, _, popen = self.record(silence(1.0)) _, _, _, popen = self.record(silence(1.0))
cmd = popen.call_args.args[0] cmd = popen.call_args.args[0]
@@ -295,6 +341,58 @@ class RecorderChain(DikteTest):
self.addCleanup(os.unlink, path) self.addCleanup(os.unlink, path)
self.assertLessEqual(duration, 1.1) 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): def test_a_recorder_that_could_not_start(self):
recorder = audio.Recorder() recorder = audio.Recorder()
failures = [] failures = []
+190 -1
View File
@@ -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 subprocess
import unittest import unittest
from unittest import mock from unittest import mock
@@ -121,6 +123,193 @@ class Bindings(DikteTest):
self.assertEqual(len(listener._bindings[57]), 2) 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"), "<Primary>Space")
def test_several_modifiers_keep_their_order(self):
self.assertEqual(hotkey.gnome_accelerator("Ctrl+Alt+A"), "<Primary><Alt>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"), "<Primary>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("<Control>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="'<Primary>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 ... <key>` 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("<Primary>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 @linux_only
class KdeShortcut(DikteTest): class KdeShortcut(DikteTest):
def setUp(self): def setUp(self):
+140 -60
View File
@@ -1,12 +1,18 @@
"""The clipboard and the key press, which is where a dictation actually lands. """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 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. 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 subprocess
import unittest import unittest
from typing import ClassVar
from unittest import mock from unittest import mock
import paste import paste
@@ -14,47 +20,86 @@ from tests.support import DikteTest, FakeCompleted, linux_only, only_these_tools
@linux_only @linux_only
class ReadClipboard(DikteTest): class Chooser(DikteTest):
def test_no_wl_paste_installed(self): """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(): with only_these_tools():
self.assertIsNone(paste.read_clipboard()) self.assertIsNone(paste.read_clipboard())
def test_what_is_on_the_clipboard_comes_back_as_bytes(self): 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", mock.patch.object(subprocess, "run",
return_value=FakeCompleted(stdout=b"hello")) as run: return_value=FakeCompleted(stdout=b"hello")) as run:
self.assertEqual(paste.read_clipboard(), b"hello") 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): 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", mock.patch.object(subprocess, "run",
return_value=FakeCompleted(returncode=1)): return_value=FakeCompleted(returncode=1)):
self.assertIsNone(paste.read_clipboard()) self.assertIsNone(paste.read_clipboard())
def test_a_tool_that_will_not_run(self): def test_a_reader_that_will_not_run(self):
with only_these_tools("wl-paste"), \ with only_these_tools(self.here.read_command[0]), \
mock.patch.object(subprocess, "run", side_effect=OSError("nope")): mock.patch.object(subprocess, "run", side_effect=OSError("nope")):
self.assertIsNone(paste.read_clipboard()) self.assertIsNone(paste.read_clipboard())
# ---- copying ----------------------------------------------------------
@linux_only def test_no_clipboard_tool_installed_says_what_to_install(self):
class Copy(DikteTest):
def test_no_wl_copy_installed(self):
with only_these_tools(), self.assertRaises(paste.PasteError) as caught: with only_these_tools(), self.assertRaises(paste.PasteError) as caught:
paste.copy("hello") 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): 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", mock.patch.object(subprocess, "run",
return_value=FakeCompleted()) as run: return_value=FakeCompleted()) as run:
paste.copy("günaydın") 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()) self.assertEqual(run.call_args.kwargs["input"], "günaydın".encode())
def test_the_pipes_are_closed_so_the_call_can_return(self): def test_the_pipes_are_closed_so_the_call_can_return(self):
"""wl-copy forks and holds the selection; a pipe nobody drains hangs.""" """The clipboard owner forks; a pipe nobody drains hangs the caller."""
with only_these_tools("wl-copy"), \ with only_these_tools(self.here.clipboard), \
mock.patch.object(subprocess, "run", mock.patch.object(subprocess, "run",
return_value=FakeCompleted()) as run: return_value=FakeCompleted()) as run:
paste.copy("hello") paste.copy("hello")
@@ -62,95 +107,130 @@ class Copy(DikteTest):
self.assertEqual(run.call_args.kwargs["stderr"], subprocess.DEVNULL) self.assertEqual(run.call_args.kwargs["stderr"], subprocess.DEVNULL)
def test_a_non_zero_exit_is_reported(self): 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", mock.patch.object(subprocess, "run",
return_value=FakeCompleted(returncode=1)), \ return_value=FakeCompleted(returncode=1)), \
self.assertRaises(paste.PasteError): self.assertRaises(paste.PasteError):
paste.copy("hello") paste.copy("hello")
def test_a_tool_that_will_not_run(self): def test_a_clipboard_tool_that_will_not_run(self):
with only_these_tools("wl-copy"), \ with only_these_tools(self.here.clipboard), \
mock.patch.object(subprocess, "run", side_effect=OSError("nope")), \ mock.patch.object(subprocess, "run", side_effect=OSError("nope")), \
self.assertRaises(paste.PasteError): self.assertRaises(paste.PasteError):
paste.copy("hello") paste.copy("hello")
def test_there_is_nothing_to_restore(self):
@linux_only with only_these_tools(self.here.clipboard), \
class CopyBytes(DikteTest):
def test_nothing_to_restore(self):
with only_these_tools("wl-copy"), \
mock.patch.object(subprocess, "run") as run: mock.patch.object(subprocess, "run") as run:
paste.copy_bytes(None) paste.copy_bytes(None)
run.assert_not_called() run.assert_not_called()
def test_restoring_never_raises(self): def test_restoring_never_raises(self):
"""It runs after the paste went in; failing here must not undo that.""" """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")): mock.patch.object(subprocess, "run", side_effect=OSError("nope")):
paste.copy_bytes(b"whatever was there before") paste.copy_bytes(b"whatever was there before")
def test_the_bytes_go_back_untouched(self): 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", mock.patch.object(subprocess, "run",
return_value=FakeCompleted()) as run: return_value=FakeCompleted()) as run:
paste.copy_bytes(b"\x89PNG\r\n") paste.copy_bytes(b"\x89PNG\r\n")
self.assertEqual(run.call_args.kwargs["input"], b"\x89PNG\r\n") self.assertEqual(run.call_args.kwargs["input"], b"\x89PNG\r\n")
# ---- pressing the key -------------------------------------------------
@linux_only def press(self, shortcut, result=None):
class Press(DikteTest): with only_these_tools(self.here.keyboard), \
def setUp(self): mock.patch.object(paste.time, "sleep", lambda seconds: None), \
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"), \
mock.patch.object(subprocess, "run", mock.patch.object(subprocess, "run",
return_value=result or FakeCompleted()) as run: return_value=result or FakeCompleted()) as run:
paste.press(shortcut) paste.press(shortcut)
return run.call_args.args[0] return run.call_args.args[0]
def test_no_ydotool_installed(self): def test_no_keyboard_tool_installed(self):
with only_these_tools(): with only_these_tools():
self.assertFalse(paste.ydotool_ready()) self.assertFalse(paste.paste_ready())
with self.assertRaises(paste.PasteError): with self.assertRaises(paste.PasteError) as caught:
paste.press() paste.press()
self.assertIn(self.here.keyboard, str(caught.exception))
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"])
def test_case_and_spacing_do_not_matter(self): 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): def test_a_key_nobody_mapped_is_refused_before_the_tool_runs(self):
self.assertEqual(self.run_press("control+insert"), """Whichever desktop it is, the shortcut is held to one table."""
["ydotool", "key", "29:1", "110:1", "110:0", "29:0"]) with only_these_tools(self.here.keyboard), \
self.assertEqual(self.run_press("super+enter"), self.run_press("meta+return")) mock.patch.object(subprocess, "run") as run, \
def test_a_key_nobody_mapped(self):
with only_these_tools("ydotool"), mock.patch.object(subprocess, "run"), \
self.assertRaises(paste.PasteError) as caught: self.assertRaises(paste.PasteError) as caught:
paste.press("ctrl+f13") paste.press("ctrl+f13")
self.assertIn("f13", str(caught.exception)) self.assertIn("f13", str(caught.exception))
run.assert_not_called()
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))
def test_a_tool_that_will_not_run(self): 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")), \ mock.patch.object(subprocess, "run", side_effect=OSError("nope")), \
self.assertRaises(paste.PasteError): self.assertRaises(paste.PasteError):
paste.press("ctrl+v") 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__": if __name__ == "__main__":
unittest.main() unittest.main()