diff --git a/README.md b/README.md index 5517d69..b2cc0f9 100644 --- a/README.md +++ b/README.md @@ -5,9 +5,9 @@ machine by default, a model cleans it up (dropping the *uh*s, the restarts, the missing punctuation), and the result lands in your clipboard and is pasted into whatever window you were typing in. -Built for KDE Plasma 6 on Wayland, and runs on GNOME X11 and macOS too. No -dependencies beyond system packages: just the Python standard library, 3.11 or -newer, and PyQt6. +Built for KDE Plasma 6 on Wayland, and runs on GNOME X11, macOS and +[Windows](README.windows.md) too. No dependencies beyond system packages: just +the Python standard library, 3.11 or newer, and PyQt6. *[Türkçe README](README.tr.md)* @@ -61,6 +61,12 @@ that catches the keys is the mechanism there and there is nothing to run: needs BlackHole or Loopback, because nothing else offers what the speakers are playing. +Windows works the same way, holding the keys through the system's own hotkey +service while Dikte runs: `winget install Gyan.FFmpeg`, `pip install PyQt6`, +then `python dikte.py`, with an optional `install.ps1` for the Start Menu entry +and the `dikte` command. Meetings are not supported there yet; the details are +in the [Windows README](README.windows.md). + `install.sh` adds the `dikte` command, a menu entry, an autostart entry and the two global shortcuts, whose keys are its two arguments. `./update.sh` pulls and puts all of that back, keeping the keys you chose; `./uninstall.sh` takes it away diff --git a/README.windows.md b/README.windows.md new file mode 100644 index 0000000..2a9f0c3 --- /dev/null +++ b/README.windows.md @@ -0,0 +1,72 @@ +# Dikte on Windows + +Press `Ctrl+Space`, talk, press again: what you said is transcribed, cleaned +up and pasted where your cursor is. + +## Requirements + +- **Windows 10/11** +- **Python 3.11+** with **PyQt6** (`pip install PyQt6`; install.ps1 installs + it when it is missing) +- **ffmpeg** for microphone capture: `winget install Gyan.FFmpeg` + +## Installing + +```powershell +powershell -ExecutionPolicy Bypass -File install.ps1 +``` + +This adds a **Dikte** entry to the Start Menu and a **`dikte`** command to the +terminal. Add `-Autostart` to also start it at sign-in; `-Uninstall` removes +all of it and leaves the repository and your settings alone. + +To try it without installing anything: + +```sh +python dikte.py +``` + +## First run + +1. The tray icon appears and the Settings window opens. +2. Under **API and models**, download a local whisper model (the whisper.cpp + Windows build is fetched automatically) or enter an OpenAI, Groq or + OpenRouter key. +3. The shortcut defaults to `Ctrl+Space` and is changed under Shortcuts. + While Dikte runs, Windows' own hotkey service (RegisterHotKey) listens for + it: nothing to install and no permission to grant. + +## What is different from Linux and macOS + +- **Meeting recording (microphone + speakers) is not supported yet.** Windows + does not offer what the speakers are playing as a capture device, so there + is nothing to record the far side from. Everything else works, including + transcribing audio and video files. +- **The shortcut is swallowed**: while Dikte holds `Ctrl+Space`, the focused + application does not see it. This is how macOS behaves too, and unlike the + Linux listener, which shares the key. +- No external tools for the clipboard or the key press: both go straight + through the Windows API (the clipboard, SendInput). +- Settings live under `%APPDATA%\Dikte`, models and recordings under + `%LOCALAPPDATA%\Dikte`. + +## Performance + +- The local install fetches whisper.cpp's **OpenBLAS build**, which + transcribes about twice as fast as the stock one on a plain CPU. There is + no GPU build to fetch for machines without an NVIDIA card. +- Setting Settings → API and models → **Threads** near your physical core + count helps noticeably; the server's own default is 4. +- If speed matters more than accuracy, `ggml-small` and `ggml-base` are much + faster; `ggml-large-v3-turbo-q5_0` transcribes best. + +## Troubleshooting + +- **Recording does not start:** does `ffmpeg -version` run? Does + `dikte devices` list your microphone? +- **Nothing is pasted:** a normal-privilege process cannot type into an + elevated (administrator) window; run Dikte elevated too, or paste by hand. + The text lands on the clipboard either way. +- **The shortcut does nothing:** another application already holds the + combination. Dikte says so in a tray notification when it asks for the key; + pick a different one under Settings → Shortcuts. diff --git a/api.py b/api.py index 775fe31..ee0222e 100644 --- a/api.py +++ b/api.py @@ -18,6 +18,7 @@ import mimetypes import os import secrets import socket +import sys import threading import urllib.error import urllib.request @@ -148,6 +149,12 @@ def _stop_using(conn): if sock is not None: with contextlib.suppress(OSError): sock.shutdown(socket.SHUT_RDWR) + if sys.platform == "win32": + # On Windows the shutdown leaves a blocked recv exactly where it + # was; only closing the OS handle ends it, and close() on the + # object would wait for the blocked reader to let go of it first. + with contextlib.suppress(OSError): + socket.close(sock.detach()) with contextlib.suppress(OSError): conn.close() @@ -258,7 +265,12 @@ def _multipart(fields, file_field, file_path): out += str(value).encode("utf-8") + b"\r\n" filename = os.path.basename(file_path) - ctype = mimetypes.guess_type(filename)[0] or "application/octet-stream" + # The two types a dictation actually sends are pinned: on Windows, + # guess_type answers from the registry and differs machine to machine. + known = {".wav": "audio/x-wav", ".mp3": "audio/mpeg"} + extension = os.path.splitext(filename)[1].lower() + ctype = (known.get(extension) or mimetypes.guess_type(filename)[0] + or "application/octet-stream") with open(file_path, "rb") as fh: payload = fh.read() out += f"--{boundary}\r\n".encode() diff --git a/assistant.py b/assistant.py index f708ab3..0dcf180 100644 --- a/assistant.py +++ b/assistant.py @@ -378,6 +378,7 @@ def _stream(cmd, conf, on_event, should_stop): cmd, cwd=working_dir(conf), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", bufsize=1, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) except OSError as exc: raise AssistantError(t("Could not run {binary}: {error}", diff --git a/audio.py b/audio.py index 55beb50..5108043 100644 --- a/audio.py +++ b/audio.py @@ -30,6 +30,10 @@ from PyQt6.QtCore import QObject, pyqtSignal from i18n import t +# Console programs started from a windowless process would otherwise each open +# a console window of their own on Windows. +NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0) if sys.platform == "win32" else 0 + RATE = 16000 CHANNELS = 1 SAMPLE_WIDTH = 2 # s16 @@ -39,6 +43,19 @@ CHUNK_LATENCY_MS = round(CHUNK_FRAMES / RATE * 1000) MIN_FRAMES = int(RATE * 0.25) +def _interrupt(proc): + """Ask a recorder process to end. + + SIGINT is the polite way everywhere it exists; Windows has no equivalent a + child can be sent, so the process is terminated outright. The captured + audio is not lost either way: it has already been read from the pipe. + """ + if sys.platform == "win32": + proc.terminate() + else: + proc.send_signal(signal.SIGINT) + + class Recorder(QObject): """Runs the available sound-server recorder and reads raw PCM from stdout.""" @@ -70,7 +87,8 @@ class Recorder(QObject): try: self._proc = subprocess.Popen( - cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0 + cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0, + creationflags=NO_WINDOW, ) except OSError as exc: self.failed.emit(t("Could not start recording: {error}", error=exc)) @@ -125,7 +143,7 @@ class Recorder(QObject): proc = self._proc if proc and proc.poll() is None: try: - proc.send_signal(signal.SIGINT) + _interrupt(proc) proc.wait(timeout=1.5) except (subprocess.TimeoutExpired, OSError): try: @@ -246,7 +264,8 @@ class MeetingRecorder(QObject): # nobody drains would eventually block it, so it writes to a file. self._log = tempfile.TemporaryFile() self._proc = subprocess.Popen( - cmd, stdout=subprocess.PIPE, stderr=self._log, bufsize=0 + cmd, stdout=subprocess.PIPE, stderr=self._log, bufsize=0, + creationflags=NO_WINDOW, ) except (OSError, wave.Error) as exc: self._close_file() @@ -298,7 +317,7 @@ class MeetingRecorder(QObject): proc = self._proc if proc and proc.poll() is None: try: - proc.send_signal(signal.SIGINT) + _interrupt(proc) proc.wait(timeout=2) except (subprocess.TimeoutExpired, OSError): try: @@ -603,6 +622,71 @@ def _avfoundation_default_output(): return "" +# Windows records through DirectShow, the one capture API ffmpeg's Windows +# builds all ship with. What the speakers are playing is not offered as a +# device at all, so a meeting has nothing to record the far side from yet. + + +def _dshow_devices(): + """[(name, name)] for every DirectShow audio capture device. + + The list comes out on stderr of a command that then fails, the same + documented trick AVFoundation uses above. Names are the only stable handle + dshow offers a user; they are what the recorder is given back. + """ + if not shutil.which("ffmpeg"): + return [] + try: + result = subprocess.run( + ["ffmpeg", "-hide_banner", "-list_devices", "true", + "-f", "dshow", "-i", "dummy"], + capture_output=True, timeout=8, check=False, creationflags=NO_WINDOW, + ) + except (subprocess.SubprocessError, OSError): + return [] + + devices = [] + for line in result.stderr.decode("utf-8", "replace").splitlines(): + if "(audio)" not in line: + continue + match = re.search(r'"([^"]+)"\s*\([^)]*audio[^)]*\)', line) + if match: + devices.append((match.group(1), match.group(1))) + return devices + + +def _dshow_record(target): + if not shutil.which("ffmpeg"): + return [] + # dshow has no "default" device: an unset target means the first one listed. + device = target + if not device: + inputs = _dshow_devices() + if not inputs: + return [] + device = inputs[0][0] + return [ + "ffmpeg", "-hide_banner", "-nostdin", "-loglevel", "error", + # dshow holds half a second of audio before handing anything over; + # asked for the chunk the level meter is measured in instead. + "-f", "dshow", "-audio_buffer_size", str(CHUNK_LATENCY_MS), + "-i", f"audio={device}", + "-ac", str(CHANNELS), "-ar", str(RATE), "-f", "s16le", "-", + ] + + +def _dshow_meeting(mic_target, system_target): + return [] # no monitor devices to record the far side from + + +def _dshow_no_outputs(): + return [] + + +def _dshow_no_default_output(): + return "" + + Sound = collections.namedtuple( "Sound", # How to capture one source and two at once, the two device lists, which @@ -633,9 +717,24 @@ COREAUDIO = Sound( ) +DSHOW = Sound( + record=_dshow_record, + meeting=_dshow_meeting, + inputs=_dshow_devices, + outputs=_dshow_no_outputs, + default_output=_dshow_no_default_output, + missing="ffmpeg or a microphone was not found. Install ffmpeg with: " + "winget install Gyan.FFmpeg", +) + + def sound(): """The programs this machine records through.""" - return COREAUDIO if sys.platform == "darwin" else PULSE + if sys.platform == "darwin": + return COREAUDIO + if sys.platform == "win32": + return DSHOW + return PULSE def list_sources(): diff --git a/cleanup.py b/cleanup.py index b6485b3..d59eb36 100644 --- a/cleanup.py +++ b/cleanup.py @@ -199,6 +199,7 @@ def _output(cmd, timeout, service): cmd, cwd=os.path.expanduser("~"), stdin=subprocess.DEVNULL, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) except subprocess.TimeoutExpired: raise CleanupError(t("{service} did not finish within {seconds} seconds.", diff --git a/cli.py b/cli.py index 7992bd0..d282b2b 100644 --- a/cli.py +++ b/cli.py @@ -17,6 +17,7 @@ import json import os import shutil import signal +import subprocess import sys import time @@ -134,6 +135,16 @@ def launch_gui(verb=""): if verb: args.append(verb) args.append("--gui") + if sys.platform == "win32": + # execv on Windows mangles arguments with spaces and would leave the + # application tied to this console; start it detached instead. + subprocess.Popen( + args, + creationflags=(subprocess.DETACHED_PROCESS + | subprocess.CREATE_NEW_PROCESS_GROUP), + close_fds=True, + ) + sys.exit(0) os.execv(sys.executable, args) @@ -297,6 +308,9 @@ def cmd_transcribe(opts): return fail(opts, f"no such file: {path}") conf = cfg.Config() + # This runs here rather than in the instance, so the local servers have to + # be handed their settings here too; the GUI does this at startup. + conf.apply_local() timestamps = opts.srt or _pick(opts.timestamps, conf["file_timestamps"]) worker = filetranscribe.FileTranscriber(conf) diff --git a/config.py b/config.py index c80bbfb..529c3c1 100644 --- a/config.py +++ b/config.py @@ -27,6 +27,12 @@ def _directories(platform=None): if (platform or sys.platform) == "darwin": support = pathlib.Path.home() / "Library/Application Support/Dikte" return support, support + if (platform or sys.platform) == "win32": + roaming = pathlib.Path( + os.environ.get("APPDATA") or pathlib.Path.home() / "AppData/Roaming") + local = pathlib.Path( + os.environ.get("LOCALAPPDATA") or pathlib.Path.home() / "AppData/Local") + return roaming / "Dikte", local / "Dikte" return (_xdg("XDG_CONFIG_HOME", "~/.config") / "dikte", _xdg("XDG_DATA_HOME", "~/.local/share") / "dikte") diff --git a/dikte.py b/dikte.py index cda7f54..d76a82a 100755 --- a/dikte.py +++ b/dikte.py @@ -12,6 +12,7 @@ import json import os import signal import socket +import subprocess import sys import threading @@ -881,6 +882,17 @@ class Dikte: self.settings_window.close() self.shutdown() QLocalServer.removeServer(SERVER_NAME) + if sys.platform == "win32": + # execv on Windows mangles arguments with spaces and leaves the two + # processes sharing a console; a detached start does neither. + subprocess.Popen( + [sys.executable, ipc.script_path(), "--gui"], + creationflags=(subprocess.DETACHED_PROCESS + | subprocess.CREATE_NEW_PROCESS_GROUP), + close_fds=True, + ) + QApplication.instance().quit() + return os.execv(sys.executable, [sys.executable, ipc.script_path(), "--gui"]) def shutdown(self): @@ -951,7 +963,11 @@ def install_signal_handlers(app): app.quit() # aboutToQuit runs shutdown() notifier.activated.connect(woken) - for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP): + # SIGHUP does not exist on Windows, and neither does a session to hang up. + signals = [signal.SIGINT, signal.SIGTERM] + if hasattr(signal, "SIGHUP"): + signals.append(signal.SIGHUP) + for sig in signals: # A handler that does nothing, so that the default action, stopping the # process where it stands, is replaced by the wakeup above. signal.signal(sig, lambda *_: None) diff --git a/filetranscribe.py b/filetranscribe.py index 4c8b4b7..d6401b4 100644 --- a/filetranscribe.py +++ b/filetranscribe.py @@ -283,6 +283,7 @@ def _ffmpeg(args, out, aborter=None): ["ffmpeg", "-nostdin", "-y", *args], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) # A two hour film is a minute of ffmpeg, which is a minute of a Stop button # doing nothing unless the abort reaches the process itself. diff --git a/ggml.py b/ggml.py index 5ba85a7..890b641 100644 --- a/ggml.py +++ b/ggml.py @@ -43,6 +43,7 @@ import threading import time import urllib.error import urllib.request +import zipfile import hub from i18n import t @@ -51,8 +52,15 @@ HOST = "127.0.0.1" # The path api.py asks for, so its URL and the server's line up. INFERENCE_PATH = "/v1/audio/transcriptions" -DATA_DIR = (pathlib.Path(os.environ.get("XDG_DATA_HOME") - or os.path.expanduser("~/.local/share")) / "dikte") +def _data_dir(): + if sys.platform == "win32": + return pathlib.Path(os.environ.get("LOCALAPPDATA") + or os.path.expanduser("~/AppData/Local")) / "Dikte" + return pathlib.Path(os.environ.get("XDG_DATA_HOME") + or os.path.expanduser("~/.local/share")) / "dikte" + + +DATA_DIR = _data_dir() BIN_DIR = DATA_DIR / "bin" MODELS_DIR = DATA_DIR / "models" @@ -151,11 +159,14 @@ def download(item, target, on_progress=None, should_stop=None, require_hash=True try: with urllib.request.urlopen(request, timeout=60) as response: total = int(response.headers.get("Content-Length") or item.size or 0) + # Windows refuses to delete a file that is open, so nothing is + # unlinked until the handle is closed again. + stopped = overlong = False with open(part, "wb") as out: while True: if should_stop is not None and should_stop(): - part.unlink(missing_ok=True) - return False + stopped = True + break block = response.read(DOWNLOAD_CHUNK) if not block: break @@ -165,11 +176,17 @@ def download(item, target, on_progress=None, should_stop=None, require_hash=True # More than was announced: a body that does not end is the # one way this loop could run until the disk is full. if total and done > total: - part.unlink(missing_ok=True) - raise LocalError(t("{name} is longer than it said it " - "would be.", name=item.name)) + overlong = True + break if on_progress is not None: on_progress(done, total) + if stopped: + part.unlink(missing_ok=True) + return False + if overlong: + part.unlink(missing_ok=True) + raise LocalError(t("{name} is longer than it said it " + "would be.", name=item.name)) # A proxy notice or an error page that came back as 200 would otherwise # be renamed into place and only fail when something tries to read it. if total and done != total: @@ -215,9 +232,11 @@ def _has_vulkan(): llama.cpp publishes no CUDA build for Linux, so Vulkan is what a graphics card gets here. The build without it is smaller and runs on the CPU, and fetching the Vulkan one for a machine that cannot load it would only make - the download bigger. + the download bigger. Windows spells the loader vulkan-1.dll. """ - return bool(ctypes.util.find_library("vulkan")) + return bool(ctypes.util.find_library("vulkan") + or (sys.platform == "win32" + and ctypes.util.find_library("vulkan-1"))) def _wanted_assets(program): @@ -230,6 +249,16 @@ def _wanted_assets(program): arch = _arch() if sys.platform == "darwin": return () if program is WHISPER else (f"bin-macos-{arch}.tar.gz",) + if sys.platform == "win32": + if program is WHISPER: + # The BLAS build first: on a plain CPU it transcribes about twice + # as fast as the stock one, and it carries everything it needs. + # Full names, because "bin-x64.zip" alone would also match the + # CUDA archives, whichever the release happened to list first. + return ("whisper-blas-bin-x64.zip", "whisper-bin-x64.zip") + if _has_vulkan() and arch == "x64": + return ("bin-win-vulkan-x64.zip", f"bin-win-cpu-{arch}.zip") + return (f"bin-win-cpu-{arch}.zip",) if program is LLAMA and _has_vulkan(): return (f"bin-ubuntu-vulkan-{arch}.tar.gz", f"bin-ubuntu-{arch}.tar.gz") return (f"bin-ubuntu-{arch}.tar.gz",) @@ -275,6 +304,11 @@ def system_program(program): return bool(shutil.which(program.binary)) +def _binary_file(program): + """What the program's file is called on disk here.""" + return f"{program.binary}.exe" if sys.platform == "win32" else program.binary + + def _find_binary(root, name): for path in sorted(pathlib.Path(root).rglob(name)): if path.is_file(): @@ -283,19 +317,24 @@ def _find_binary(root, name): def _extract(archive, into): - """Unpack a release tarball, refusing anything that reaches outside `into`. + """Unpack a release archive, refusing anything that reaches outside `into`. The archives lay their libraries next to their binaries and are linked with an $ORIGIN runpath, so a whole directory is what has to survive the trip and - the binary cannot be lifted out of it. + the binary cannot be lifted out of it. Linux and macOS releases come as + tarballs, Windows ones as zips; zipfile never writes outside its target. """ try: + if str(archive).endswith(".zip"): + with zipfile.ZipFile(archive) as bundle: + bundle.extractall(into) + return with tarfile.open(archive, "r:gz") as tar: try: tar.extractall(into, filter="data") except TypeError: # Python without the extraction filters tar.extractall(into) - except (tarfile.TarError, OSError) as exc: + except (tarfile.TarError, zipfile.BadZipFile, OSError) as exc: raise LocalError(t("Could not unpack {name}: {error}", name=os.path.basename(str(archive)), error=exc)) from exc @@ -334,7 +373,7 @@ def install_program(program, tag="", on_progress=None, should_stop=None, if not download(item, archive, on_progress, should_stop): return "" _extract(archive, into) - binary = _find_binary(into, program.binary) + binary = _find_binary(into, _binary_file(program)) if binary is None: raise LocalError(t("{name} was not in the download.", name=program.binary)) @@ -491,6 +530,26 @@ def _tail(path, lines=3): return " | ".join(found[-lines:]) +def _win_image_name(pid): + """The lower-cased file name of the process's executable, or ''.""" + import ctypes + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + kernel32.OpenProcess.restype = ctypes.c_void_p + kernel32.OpenProcess.argtypes = [ctypes.c_uint32, ctypes.c_int, ctypes.c_uint32] + kernel32.CloseHandle.argtypes = [ctypes.c_void_p] + handle = kernel32.OpenProcess(0x1000, False, pid) # QUERY_LIMITED_INFORMATION + if not handle: + return "" + try: + buffer = ctypes.create_unicode_buffer(260) + size = ctypes.c_uint32(len(buffer)) + ok = kernel32.QueryFullProcessImageNameW( + ctypes.c_void_p(handle), 0, buffer, ctypes.byref(size)) + return os.path.basename(buffer.value).lower() if ok else "" + finally: + kernel32.CloseHandle(handle) + + class Server: """One process, started when something needs it and stopped when nothing does. @@ -592,6 +651,8 @@ class Server: args + ["--host", HOST, "--port", str(port)], stdout=sink, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, + # No console window of its own on Windows. + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) except OSError as exc: raise LocalError(t("Could not start {name}: {error}", @@ -689,8 +750,11 @@ class Server: number could belong to something else entirely, and killing it would be a good deal worse than the leak being cleaned up. The program name alone could be somebody else's copy; the name together with Dikte's own data - directory on the command line could not. + directory on the command line could not. Windows offers no command line + to read, so the executable's name is the whole of the answer there. """ + if sys.platform == "win32": + return _win_image_name(pid) == _binary_file(self.program).lower() try: blob = pathlib.Path(f"/proc/{pid}/cmdline").read_bytes() except OSError: diff --git a/hotkey.py b/hotkey.py index e7e10ea..89a6898 100644 --- a/hotkey.py +++ b/hotkey.py @@ -418,12 +418,175 @@ def _carbon(): return carbon +# --- Windows: RegisterHotKey ------------------------------------------------ + +# Windows virtual-key codes: where a key sits, not what a layout prints on it. +WIN_KEYS = { + "space": 0x20, "tab": 0x09, "enter": 0x0D, "return": 0x0D, + "esc": 0x1B, "escape": 0x1B, "backspace": 0x08, "insert": 0x2D, + "delete": 0x2E, "home": 0x24, "end": 0x23, "pgup": 0x21, "pgdown": 0x22, + "up": 0x26, "down": 0x28, "left": 0x25, "right": 0x27, + **{str(digit): 0x30 + digit for digit in range(10)}, + **{chr(ord("a") + i): 0x41 + i for i in range(26)}, + **{f"f{n}": 0x6F + n for n in range(1, 13)}, +} +WIN_MODS = { + "alt": 0x0001, "ctrl": 0x0002, "control": 0x0002, "shift": 0x0004, + "meta": 0x0008, "super": 0x0008, "win": 0x0008, +} +WIN_MOD_NOREPEAT = 0x4000 # holding the combination fires it once +WM_HOTKEY = 0x0312 +WM_QUIT = 0x0012 + + +def _win_input(): + """user32 and kernel32, which is all the listener talks to. + + Loaded on the first start rather than at import: this module is read on + every system, and these two libraries exist on one of them. + """ + return ctypes.windll.user32, ctypes.windll.kernel32 + + +def parse_windows_shortcut(text): + """'Ctrl+Space' -> (2, 32), or (None, None) when unusable.""" + parts = [part.strip().lower() for part in str(text).split("+") if part.strip()] + modifiers, key = 0, None + for part in parts: + if part in WIN_MODS: + modifiers |= WIN_MODS[part] + elif key is None and part in WIN_KEYS: + key = WIN_KEYS[part] + else: + return None, None + if key is None: + return None, None + return modifiers, key + + +class WinHotkey(QObject): + """Catches global shortcuts through Windows' own hotkey service. + + RegisterHotKey asks for one combination rather than reading the keyboard, + so it needs no permission at all. Like Carbon's and unlike the evdev + listener it swallows the key: while Dikte holds a combination, nothing + else on the machine receives it. + + RegisterHotKey only fires on the thread that called it, so registration + and the message loop live together on one worker thread; start() hands the + bindings over and waits for it to report what Windows actually gave us. + """ + + triggered = pyqtSignal(str) # the name the binding was registered under + failed = pyqtSignal(str) + + def __init__(self, parent=None): + super().__init__(parent) + self._user32 = None + self._kernel32 = None + self._thread = None + self._thread_id = None + self._count = 0 + + @property + def running(self): + return self._count > 0 and self._thread is not None and self._thread.is_alive() + + def start(self, bindings): + """`bindings` is {name: 'Ctrl+Space'}; an empty combination is skipped.""" + self.stop() + try: + self._user32, self._kernel32 = _win_input() + except (AttributeError, OSError) as exc: + self.failed.emit(t("Could not reach the Windows shortcut service: " + "{error}", error=exc)) + return False + wanted = [] + for identifier, (name, shortcut) in enumerate(bindings.items(), 1): + if not shortcut: + continue + modifiers, key = parse_windows_shortcut(shortcut) + if key is None: + self.failed.emit( + t("Could not parse the shortcut: {shortcut}", shortcut=shortcut) + ) + continue + wanted.append((identifier, name, shortcut, modifiers, key)) + if not wanted: + return False + + ready = threading.Event() + outcome = {"count": 0, "thread_id": None} + self._thread = threading.Thread( + target=self._loop, args=(wanted, ready, outcome), daemon=True + ) + self._thread.start() + ready.wait(timeout=5) + self._thread_id = outcome["thread_id"] + self._count = outcome["count"] + if not self._count: + self._thread = None + return self._count > 0 + + def stop(self): + if self._thread and self._thread_id and self._user32: + self._user32.PostThreadMessageW(self._thread_id, WM_QUIT, 0, 0) + self._thread.join(timeout=1.5) + self._thread = None + self._thread_id = None + self._count = 0 + _REGISTERED.clear() + + def _loop(self, wanted, ready, outcome): + import ctypes.wintypes + user32, kernel32 = self._user32, self._kernel32 + outcome["thread_id"] = kernel32.GetCurrentThreadId() + + # The message queue a PostThreadMessage needs only exists once the + # thread has asked for messages; peek once before reporting ready. + message = ctypes.wintypes.MSG() + user32.PeekMessageW(ctypes.byref(message), None, WM_QUIT, WM_QUIT, 0) + + names = {} + for identifier, name, shortcut, modifiers, key in wanted: + if user32.RegisterHotKey(None, identifier, + modifiers | WIN_MOD_NOREPEAT, key): + names[identifier] = name + spec = SHORTCUTS.get(name) + if spec: + _REGISTERED[spec.desktop_id] = shortcut + else: + # This is the conflict warning on Windows: there is no list to + # read beforehand, the answer comes from asking for the key. + self.failed.emit(t( + "Windows would not give Dikte {shortcut}; another " + "application already holds it.", shortcut=shortcut)) + outcome["count"] = len(names) + ready.set() + if not names: + return + + try: + while user32.GetMessageW(ctypes.byref(message), None, 0, 0) > 0: + if message.message == WM_HOTKEY: + name = names.get(int(message.wParam)) + if name: + self.triggered.emit(name) + finally: + for identifier in names: + user32.UnregisterHotKey(None, identifier) + + # --- the desktop's own shortcut ------------------------------------------- def _macos(): return sys.platform == "darwin" +def _windows(): + return sys.platform == "win32" + + def _gnome(): desktop = os.environ.get("XDG_CURRENT_DESKTOP", "").lower() return "gnome" in desktop and shutil.which("gsettings") is not None @@ -548,37 +711,44 @@ def gnome_shortcut_status(desktop_id=DESKTOP_ID): def listener(parent=None): """The thing that hears the key, for whichever system this is.""" - return CarbonHotkey(parent) if _macos() else EvdevHotkey(parent) + if _macos(): + return CarbonHotkey(parent) + if _windows(): + return WinHotkey(parent) + return EvdevHotkey(parent) def valid_shortcut(text): """Whether this machine can bind the combination as it was typed.""" - parse = parse_macos_shortcut if _macos() else parse_shortcut - return parse(text)[1] is not None + if _macos(): + return parse_macos_shortcut(text)[1] is not None + if _windows(): + return parse_windows_shortcut(text)[1] is not None + return parse_shortcut(text)[1] is not None def installs_shortcuts(): """Whether this system keeps a shortcut registry to write into. KDE and GNOME do, and something outside Dikte reads it, so the combination - survives Dikte being closed. macOS does not: there is nothing to install, - nothing to remove, and Settings should not offer either. + survives Dikte being closed. macOS and Windows do not: there is nothing to + install, nothing to remove, and Settings should not offer either. """ - return not _macos() + return not _macos() and not _windows() def shortcut_needs_restart(): """Whether an installed shortcut waits for the next login before it works. KWin reads kglobalshortcutsrc once, when it starts. GNOME picks a binding - up as it is written, and macOS never had one to write. + up as it is written, and macOS and Windows never had one to write. """ - return not _macos() and not _gnome() + return not _macos() and not _windows() and not _gnome() def install_shortcut(shortcut, exec_command, name="Dikte: start/stop recording", desktop_id=DESKTOP_ID): - if _macos(): + if _macos() or _windows(): _REGISTERED[desktop_id] = shortcut return True, t( "Shortcut saved: {shortcut}\nDikte holds this one itself while it " @@ -591,7 +761,7 @@ def install_shortcut(shortcut, exec_command, name="Dikte: start/stop recording", def remove_shortcut(desktop_id=DESKTOP_ID): - if _macos(): + if _macos() or _windows(): _REGISTERED.pop(desktop_id, None) elif _gnome(): remove_gnome_shortcut(desktop_id) @@ -600,7 +770,7 @@ def remove_shortcut(desktop_id=DESKTOP_ID): def shortcut_status(desktop_id=DESKTOP_ID): - if _macos(): + if _macos() or _windows(): return _REGISTERED.get(desktop_id) return (gnome_shortcut_status(desktop_id) if _gnome() else kde_shortcut_status(desktop_id)) @@ -609,6 +779,8 @@ def shortcut_status(desktop_id=DESKTOP_ID): def desktop_name(): if _macos(): return "macOS" + if _windows(): + return "Windows" return "GNOME" if _gnome() else "KDE" @@ -693,9 +865,9 @@ def kde_shortcut_status(desktop_id=DESKTOP_ID): def conflicting_shortcuts(shortcut, desktop_id=DESKTOP_ID): """Names of other KDE entries bound to the same combination.""" - if _macos(): - # There is no list to read: macOS answers the question by refusing the - # registration, which CarbonHotkey reports when it asks for the key. + if _macos() or _windows(): + # There is no list to read: both answer the question by refusing the + # registration, which their listeners report when they ask for the key. return [] try: text = SHORTCUTS_FILE.read_text(encoding="utf-8") diff --git a/i18n.py b/i18n.py index 436f917..761416d 100644 --- a/i18n.py +++ b/i18n.py @@ -106,6 +106,10 @@ TR = { "Ses kayıt aracı bulunamadı. pulseaudio-utils ya da pipewire-audio kur.", "ffmpeg not found. Install it with: brew install ffmpeg": "ffmpeg bulunamadı. Şununla kur: brew install ffmpeg", + "ffmpeg or a microphone was not found. Install ffmpeg with: " + "winget install Gyan.FFmpeg": + "ffmpeg ya da bir mikrofon bulunamadı. ffmpeg'i şununla kur: " + "winget install Gyan.FFmpeg", "Audio recorder stopped before receiving sound: {error}": "Ses kayıt aracı veri alamadan kapandı: {error}", "Could not copy to clipboard: {error}": "Panoya kopyalanamadı: {error}", @@ -355,6 +359,12 @@ TR = { "macOS would not give Dikte {shortcut}; another application already holds it.": "macOS {shortcut} kombinasyonunu Dikte'ye vermedi; başka bir uygulama " "onu şimdiden tutuyor.", + "Could not reach the Windows shortcut service: {error}": + "Windows kısayol servisine ulaşılamadı: {error}", + "Windows would not give Dikte {shortcut}; another application already " + "holds it.": + "Windows {shortcut} kombinasyonunu Dikte'ye vermedi; başka bir uygulama " + "onu şimdiden tutuyor.", "Cannot read /dev/input. Your user needs to be in the 'input' group:\n" " sudo usermod -aG input $USER (then log out and back in)": "/dev/input okunamıyor. Kullanıcının 'input' grubunda olması gerekir:\n" diff --git a/install.ps1 b/install.ps1 new file mode 100644 index 0000000..cbd86de --- /dev/null +++ b/install.ps1 @@ -0,0 +1,74 @@ +# Dikte'yi bu Windows kullanicisi icin kurar: Baslat Menusu kisayolu, istege +# bagli otomatik baslangic ve her yerden calisan bir `dikte` komutu. +# +# powershell -ExecutionPolicy Bypass -File install.ps1 # kur +# powershell -ExecutionPolicy Bypass -File install.ps1 -Autostart # + oturum acilisinda baslat +# powershell -ExecutionPolicy Bypass -File install.ps1 -Uninstall # kaldir +param( + [switch]$Autostart, + [switch]$Uninstall +) + +$ErrorActionPreference = "Stop" +$repo = $PSScriptRoot +$startMenu = [Environment]::GetFolderPath("Programs") +$startup = [Environment]::GetFolderPath("Startup") +$shortcut = Join-Path $startMenu "Dikte.lnk" +$autostartLink = Join-Path $startup "Dikte.lnk" +# WindowsApps kullanici PATH'inde hazir durur; oraya birakilan dikte.cmd her +# terminalden calisir. +$cmdShim = Join-Path $env:LOCALAPPDATA "Microsoft\WindowsApps\dikte.cmd" + +if ($Uninstall) { + foreach ($path in @($shortcut, $autostartLink, $cmdShim)) { + if (Test-Path $path) { Remove-Item $path -Force; Write-Host "silindi: $path" } + } + Write-Host "Dikte kisayollari kaldirildi. Depo klasoru ve ayarlar duruyor." + exit 0 +} + +# --- gereksinimler ---------------------------------------------------------- +$python = Get-Command python -ErrorAction SilentlyContinue +if (-not $python) { + Write-Error "Python bulunamadi. Kurun: winget install Python.Python.3.12" +} +$version = & python -c "import sys; print('%d.%d' % sys.version_info[:2])" +if ([version]$version -lt [version]"3.11") { + Write-Error "Python 3.11+ gerekli, bulunan: $version" +} +& python -c "import PyQt6.QtWidgets" 2>$null +if ($LASTEXITCODE -ne 0) { + Write-Host "PyQt6 kuruluyor..." + & python -m pip install PyQt6 + if ($LASTEXITCODE -ne 0) { Write-Error "PyQt6 kurulamadi." } +} +if (-not (Get-Command ffmpeg -ErrorAction SilentlyContinue)) { + Write-Warning "ffmpeg bulunamadi. Ses kaydi icin gerekli: winget install Gyan.FFmpeg" +} + +# pythonw.exe konsol penceresi acmadan calistirir. +$pythonw = Join-Path (Split-Path $python.Source) "pythonw.exe" +if (-not (Test-Path $pythonw)) { $pythonw = $python.Source } + +# --- Baslat Menusu kisayolu ------------------------------------------------- +$shell = New-Object -ComObject WScript.Shell +foreach ($path in @($shortcut) + $(if ($Autostart) { @($autostartLink) } else { @() })) { + $link = $shell.CreateShortcut($path) + $link.TargetPath = $pythonw + $link.Arguments = "`"$repo\dikte.py`" --gui" + $link.WorkingDirectory = $repo + $link.Description = "Dikte: sesli dikte" + $link.Save() + Write-Host "kisayol: $path" +} + +# --- dikte komutu ----------------------------------------------------------- +$shimDir = Split-Path $cmdShim +if (Test-Path $shimDir) { + "@echo off`r`npython `"$repo\dikte.py`" %*" | Out-File $cmdShim -Encoding ascii + Write-Host "komut: dikte ($cmdShim)" +} + +Write-Host "" +Write-Host "Kurulum tamam. Baslat Menusu'nden 'Dikte' ile ya da terminalden 'dikte' yazarak baslatin." +Write-Host "Ilk acilista Ayarlar penceresi acilir: oradan model indirin ve kisayolu secin (varsayilan Ctrl+Space)." diff --git a/ipc.py b/ipc.py index 541e067..659811c 100644 --- a/ipc.py +++ b/ipc.py @@ -14,7 +14,9 @@ import sys from PyQt6.QtNetwork import QLocalSocket -SERVER_NAME = "dikte-" + str(os.getuid()) +SERVER_NAME = "dikte-" + ( + str(os.getuid()) if hasattr(os, "getuid") + else os.environ.get("USERNAME", "user")) # Long enough for a process that is already running to answer, short enough that # "nothing is running" is not a noticeable pause in front of a key press. diff --git a/overlay.py b/overlay.py index f843306..b0cc39e 100644 --- a/overlay.py +++ b/overlay.py @@ -61,9 +61,10 @@ class Overlay(QWidget): | Qt.WindowType.Tool | Qt.WindowType.WindowDoesNotAcceptFocus ) - if sys.platform != "darwin": + if sys.platform not in ("darwin", "win32"): # It is the window manager that would otherwise move this out of - # the corner. macOS has no such hint, and Qt warns about it. + # the corner. macOS has no such hint, and Qt warns about it; + # Windows places tool windows where they ask to be anyway. flags |= Qt.WindowType.X11BypassWindowManagerHint # One that can be clicked away has to receive the click, which means it # also swallows one aimed at whatever is underneath it. The rest stay diff --git a/paste.py b/paste.py index 030f203..1ca172b 100644 --- a/paste.py +++ b/paste.py @@ -278,6 +278,159 @@ def _macos_press(shortcut, delay): core.CFRelease(up) +def _win_keys(shortcut): + """'Ctrl+V' -> [0x11, 0x56]: Windows virtual-key codes, modifiers first.""" + codes = [] + for key in _keys(shortcut): + if key not in WIN_KEYCODES: + raise PasteError(t("Unknown key: {key}", key=key)) + codes.append(WIN_KEYCODES[key]) + return codes + + +# Windows virtual-key codes (winuser.h). Like Apple's, they say where the key +# sits rather than what a layout prints on it. +WIN_KEYCODES = { + "ctrl": 0x11, "control": 0x11, "shift": 0x10, "alt": 0x12, + "super": 0x5B, "meta": 0x5B, + "v": 0x56, "insert": 0x2D, "enter": 0x0D, "return": 0x0D, +} +_WIN_KEYUP = 0x0002 # KEYEVENTF_KEYUP +_WIN_CF_UNICODETEXT = 13 # what the clipboard calls UTF-16 text +_WIN_GMEM_MOVEABLE = 0x0002 + + +@functools.lru_cache(maxsize=1) +def _win_api(): + """user32 and kernel32 with their prototypes spelled out. + + The default return type is a 32-bit int, which silently truncates the + 64-bit handles and pointers every one of these calls trades in. + """ + user32 = ctypes.WinDLL("user32", use_last_error=True) + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + user32.OpenClipboard.argtypes = [ctypes.c_void_p] + user32.GetClipboardData.restype = ctypes.c_void_p + user32.GetClipboardData.argtypes = [ctypes.c_uint] + user32.SetClipboardData.restype = ctypes.c_void_p + user32.SetClipboardData.argtypes = [ctypes.c_uint, ctypes.c_void_p] + kernel32.GlobalAlloc.restype = ctypes.c_void_p + kernel32.GlobalAlloc.argtypes = [ctypes.c_uint, ctypes.c_size_t] + kernel32.GlobalLock.restype = ctypes.c_void_p + kernel32.GlobalLock.argtypes = [ctypes.c_void_p] + kernel32.GlobalUnlock.argtypes = [ctypes.c_void_p] + kernel32.GlobalFree.argtypes = [ctypes.c_void_p] + return user32, kernel32 + + +def _win_error(): + """GetLastError where it exists, so the failure paths run under any test.""" + return getattr(ctypes, "get_last_error", lambda: 0)() + + +def _win_open_clipboard(user32): + """The clipboard is a lock another program may hold for a moment.""" + for _ in range(10): + if user32.OpenClipboard(None): + return True + time.sleep(0.01) + return False + + +def _win_read_text(): + """The clipboard's text, '' when it holds none, None when it cannot be read.""" + user32, kernel32 = _win_api() + if not _win_open_clipboard(user32): + return None + try: + handle = user32.GetClipboardData(_WIN_CF_UNICODETEXT) + if not handle: + return "" + pointer = kernel32.GlobalLock(handle) + if not pointer: + return None + try: + return ctypes.wstring_at(pointer) + finally: + kernel32.GlobalUnlock(handle) + finally: + user32.CloseClipboard() + + +def _win_write_text(text): + user32, kernel32 = _win_api() + payload = str(text).encode("utf-16-le") + b"\x00\x00" + if not _win_open_clipboard(user32): + raise PasteError(t("Could not copy to clipboard: {error}", + error="the clipboard is held by another program")) + handle = None + try: + user32.EmptyClipboard() + handle = kernel32.GlobalAlloc(_WIN_GMEM_MOVEABLE, len(payload)) + pointer = kernel32.GlobalLock(handle) if handle else None + if not pointer: + raise PasteError(t("Could not copy to clipboard: {error}", + error="out of memory")) + ctypes.memmove(pointer, payload, len(payload)) + kernel32.GlobalUnlock(handle) + if not user32.SetClipboardData(_WIN_CF_UNICODETEXT, handle): + raise PasteError(t("Could not copy to clipboard: {error}", + error=f"error {_win_error()}")) + handle = None # the clipboard owns it now + finally: + if handle: + kernel32.GlobalFree(handle) + user32.CloseClipboard() + + +class _WinKeybdInput(ctypes.Structure): + _fields_ = [("wVk", ctypes.c_ushort), ("wScan", ctypes.c_ushort), + ("dwFlags", ctypes.c_ulong), ("time", ctypes.c_ulong), + ("dwExtraInfo", ctypes.c_size_t)] + + +class _WinMouseInput(ctypes.Structure): + _fields_ = [("dx", ctypes.c_long), ("dy", ctypes.c_long), + ("mouseData", ctypes.c_ulong), ("dwFlags", ctypes.c_ulong), + ("time", ctypes.c_ulong), ("dwExtraInfo", ctypes.c_size_t)] + + +class _WinInputUnion(ctypes.Union): + _fields_ = [("mi", _WinMouseInput), ("ki", _WinKeybdInput)] + + +class _WinInput(ctypes.Structure): + # The union carries the mouse shape too: SendInput sizes its argument by + # the biggest member whether or not it is the one being sent. + _fields_ = [("type", ctypes.c_ulong), ("union", _WinInputUnion)] + + +def _win_press(shortcut, delay): + """Post the presses and releases straight into the input queue. + + No permission stands in front of SendInput the way Accessibility does on + macOS: whatever window has focus receives the combination. + """ + codes = _win_keys(shortcut) + user32, _ = _win_api() + time.sleep(delay) # let the selection settle and focus come back + + events = ([(code, 0) for code in codes] + + [(code, _WIN_KEYUP) for code in reversed(codes)]) + inputs = (_WinInput * len(events))() + for entry, (code, flags) in zip(inputs, events): + entry.type = 1 # INPUT_KEYBOARD + entry.union.ki = _WinKeybdInput(code, 0, flags, 0, 0) + sent = user32.SendInput(len(inputs), inputs, ctypes.sizeof(_WinInput)) + if sent != len(inputs): + raise PasteError(t("Could not run {tool}: {error}", tool="SendInput", + error=f"error {_win_error()}")) + + +def _win_ready(): + return True + + # --- which of them is here ------------------------------------------------- Desktop = collections.namedtuple( @@ -310,6 +463,17 @@ X11 = Desktop( **_program_keyboard("xdotool", _xdotool_command), ) +WINDOWS = Desktop( + clipboard="", # no program: both directions are calls into the system + packages="", + read_command=[], + copy_command=[], + shortcuts=["ctrl+v", "ctrl+shift+v", "shift+insert"], + keyboard="", + ready=_win_ready, + press=_win_press, +) + MACOS = Desktop( clipboard="pbcopy", packages="", # both are part of macOS; there is nothing to install @@ -331,6 +495,8 @@ def desktop(): """ if sys.platform == "darwin": return MACOS + if sys.platform == "win32": + return WINDOWS if os.environ.get("XDG_SESSION_TYPE") == "x11": return X11 if os.environ.get("DISPLAY") and not os.environ.get("WAYLAND_DISPLAY"): @@ -375,6 +541,9 @@ def _macos_restore(snapshot): def read_clipboard(): here = desktop() + if here is WINDOWS: + text = _win_read_text() + return None if text is None else text.encode("utf-8") if here is MACOS and shutil.which("osascript"): snapshot = _macos_snapshot() if snapshot is not None: @@ -402,6 +571,9 @@ def _run_copy(payload): def copy(text): here = desktop() + if here is WINDOWS: + _win_write_text(text) + return if not shutil.which(here.clipboard): raise PasteError( t("{tool} not found. Install {packages}.", @@ -421,7 +593,15 @@ def copy_bytes(data): if isinstance(data, _MAC_SNAPSHOT): _macos_restore(data) return - if data is None or not shutil.which(desktop().clipboard): + if data is None: + return + if desktop() is WINDOWS: + try: + _win_write_text(data.decode("utf-8", "replace")) + except PasteError: + pass + return + if not shutil.which(desktop().clipboard): return try: _run_copy(data) diff --git a/tests/test_audio.py b/tests/test_audio.py index 22d4f2a..0f9582f 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -606,5 +606,83 @@ class MacRecordingCommand(OnMacOS, DikteTest): self.assertFalse(recorder.active) +class OnWindows: + """A test that runs as if the machine ran Windows.""" + + def setUp(self): + super().setUp() + self.enterContext(mock.patch.object(sys, "platform", "win32")) + + +class WindowsDevices(OnWindows, DikteTest): + """The one ffmpeg listing the device questions are answered from. + + dshow names devices rather than numbering them, and the names carry + whatever alphabet the machine speaks, so the listing here does too. + """ + + LISTING = ( + '[dshow @ 0000020c] "Integrated Camera" (video)\n' + '[dshow @ 0000020c] Alternative name "@device_pnp_\\...."\n' + '[dshow @ 0000020c] "Mikrofon Dizisi (Intel Smart Sound)" (audio)\n' + '[dshow @ 0000020c] Alternative name "@device_cm_{33D9A762}...."\n' + '[dshow @ 0000020c] "Kulaklık (Soundcore Life Q30)" (audio)\n' + "dummy: Immediate exit requested\n" + ).encode("utf-8") + + @contextlib.contextmanager + def listing(self, stderr=None, tools=("ffmpeg",)): + completed = FakeCompleted( + returncode=1, stderr=self.LISTING if stderr is None else stderr) + with only_these_tools(*tools), \ + mock.patch.object(subprocess, "run", return_value=completed): + yield + + def test_windows_records_through_dshow(self): + self.assertIs(audio.sound(), audio.DSHOW) + + def test_the_audio_lines_are_the_only_ones_read(self): + with self.listing(): + self.assertEqual(audio.list_sources(), [ + ("Mikrofon Dizisi (Intel Smart Sound)", + "Mikrofon Dizisi (Intel Smart Sound)"), + ("Kulaklık (Soundcore Life Q30)", + "Kulaklık (Soundcore Life Q30)"), + ]) + + def test_no_ffmpeg_installed(self): + with only_these_tools(): + self.assertEqual(audio.list_sources(), []) + self.assertEqual(audio.recording_command(), []) + + def test_the_name_is_what_the_recorder_is_given_back(self): + with self.listing(): + cmd = audio.recording_command("Kulaklık (Soundcore Life Q30)") + self.assertEqual(cmd[cmd.index("-f") + 1], "dshow") + self.assertIn("audio=Kulaklık (Soundcore Life Q30)", cmd) + + def test_no_microphone_named_means_the_first_one_listed(self): + """dshow has no default device for an empty target to mean.""" + with self.listing(): + self.assertIn("audio=Mikrofon Dizisi (Intel Smart Sound)", + audio.recording_command()) + + def test_a_machine_with_no_microphone_at_all(self): + with self.listing(stderr=b'[dshow @ 0] "Integrated Camera" (video)\n'): + self.assertEqual(audio.recording_command(), []) + + def test_an_ffmpeg_that_will_not_run(self): + with only_these_tools("ffmpeg"), \ + mock.patch.object(subprocess, "run", side_effect=OSError("nope")): + self.assertEqual(audio.list_sources(), []) + + def test_nothing_offers_the_far_side_of_a_meeting(self): + """What the speakers play is not a capture device Windows hands out.""" + with self.listing(): + self.assertEqual(audio.list_monitors(), []) + self.assertEqual(audio.default_monitor(), "") + self.assertEqual(audio.meeting_command("mic", "sys"), []) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_cli.py b/tests/test_cli.py index 9547df2..af42b2e 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -14,6 +14,7 @@ from unittest import mock import cli import config as cfg +import ggml import hotkey import ipc from tests.support import DikteTest, fake_urlopen @@ -564,5 +565,27 @@ class Replies(DikteTest): self.assertEqual(cli.run(["cancel"]), 0) +class TranscribeRunsHere(DikteTest): + """`dikte transcribe` runs in this process, not in the instance.""" + + def test_the_local_servers_are_handed_the_settings_first(self): + # The GUI does this at startup; a CLI run has no GUI to have done it, + # and without it the whisper server holds an empty model name. + wav = self.path("clip.wav") + wav.write_bytes(b"RIFF not really audio") + self.write_config({"local_model": "ggml-base.bin"}) + self.addCleanup(ggml.whisper.configure, + model="", threads=0, gpu=True, binary="") + + opts = cli.build_parser().parse_args(["transcribe", str(wav)]) + with mock.patch.object(cli.filetranscribe, "FileTranscriber"), \ + mock.patch.object(cli, "_headless", + return_value={"error": "stopped"}), \ + captured(): + cli.cmd_transcribe(opts) + + self.assertEqual(ggml.whisper.settings()["model"], "ggml-base.bin") + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_config.py b/tests/test_config.py index 31e153c..fec1417 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -8,6 +8,7 @@ config and now shadows the default. import json import os +import sys import unittest from unittest import mock @@ -89,6 +90,8 @@ class Saving(DikteTest): cfg.Config().save() self.assertTrue(cfg.CONFIG_FILE.exists()) + @unittest.skipIf(sys.platform == "win32", + "NTFS access is decided by ACLs, not by the mode bits") def test_the_file_is_readable_by_nobody_else(self): """It holds two API keys.""" cfg.Config().save() @@ -467,25 +470,33 @@ class Directories(unittest.TestCase): with mock.patch.dict(os.environ, {"XDG_CONFIG_HOME": "/c", "XDG_DATA_HOME": "/d"}): config_dir, data_dir = cfg._directories("linux") - self.assertEqual(str(config_dir), "/c/dikte") - self.assertEqual(str(data_dir), "/d/dikte") + self.assertEqual(config_dir.as_posix(), "/c/dikte") + self.assertEqual(data_dir.as_posix(), "/d/dikte") def test_linux_without_the_variables_set(self): with mock.patch.dict(os.environ, {}, clear=True): config_dir, data_dir = cfg._directories("linux") - self.assertTrue(str(config_dir).endswith("/.config/dikte")) - self.assertTrue(str(data_dir).endswith("/.local/share/dikte")) + self.assertTrue(config_dir.as_posix().endswith("/.config/dikte")) + self.assertTrue(data_dir.as_posix().endswith("/.local/share/dikte")) def test_a_mac_keeps_both_in_application_support(self): config_dir, data_dir = cfg._directories("darwin") self.assertEqual(config_dir, data_dir) - self.assertTrue(str(config_dir).endswith("/Library/Application Support/Dikte")) + self.assertTrue(config_dir.as_posix() + .endswith("/Library/Application Support/Dikte")) def test_a_mac_does_not_read_the_xdg_variables(self): """A Mac with them set from some other tool still stores in one place.""" with mock.patch.dict(os.environ, {"XDG_CONFIG_HOME": "/c"}): config_dir, _ = cfg._directories("darwin") - self.assertNotIn("/c", str(config_dir)) + self.assertNotIn("/c", config_dir.as_posix()) + + def test_windows_keeps_settings_and_data_apart(self): + with mock.patch.dict(os.environ, {"APPDATA": r"C:\roam", + "LOCALAPPDATA": r"C:\local"}): + config_dir, data_dir = cfg._directories("win32") + self.assertEqual(config_dir.as_posix(), "C:/roam/Dikte") + self.assertEqual(data_dir.as_posix(), "C:/local/Dikte") if __name__ == "__main__": @@ -519,6 +530,9 @@ class ReadyToRun(DikteTest): def setUp(self): super().setUp() self.patch_attr(ggml, "MODELS_DIR", self.path("models")) + # A machine Dikte is actually installed on would otherwise answer for + # the "missing program" below through the real install record. + self.patch_attr(ggml, "BIN_DIR", self.path("bin")) def install(self, name): path = ggml.whisper_model_path(name) diff --git a/tests/test_ggml.py b/tests/test_ggml.py index 161e688..2e3075f 100644 --- a/tests/test_ggml.py +++ b/tests/test_ggml.py @@ -15,6 +15,7 @@ import tarfile import textwrap import threading import time +import zipfile from unittest import mock import ggml @@ -77,6 +78,15 @@ def tarball(entries): return buf.getvalue() +def zipball(entries): + """A .zip laid out the way the Windows releases are.""" + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as bundle: + for name, content in entries.items(): + bundle.writestr(name, content) + return buf.getvalue() + + class Local(DikteTest): """A test with its own bin, models and cache directories.""" @@ -681,3 +691,78 @@ class Sizes(DikteTest): self.assertEqual(ggml.human_size(512), "512 B") self.assertEqual(ggml.human_size(574041195), "547.4 MB") self.assertEqual(ggml.human_size(3_095_033_483), "2.9 GB") + + +# --- Windows ---------------------------------------------------------------- + + +class WindowsAssets(Local): + """Which archive a Windows machine is handed.""" + + def setUp(self): + super().setUp() + self.patch_attr(sys, "platform", "win32") + self.patch_attr(ggml, "_arch", lambda: "x64") + + def test_whisper_prefers_the_blas_build(self): + # On a plain CPU it transcribes about twice as fast as the stock one. + self.assertEqual(ggml._wanted_assets(ggml.WHISPER), + ("whisper-blas-bin-x64.zip", "whisper-bin-x64.zip")) + + def test_llama_takes_the_vulkan_build_when_there_is_a_loader(self): + self.patch_attr(ggml, "_has_vulkan", lambda: True) + self.assertEqual(ggml._wanted_assets(ggml.LLAMA), + ("bin-win-vulkan-x64.zip", "bin-win-cpu-x64.zip")) + + def test_llama_falls_back_to_the_cpu_build_without_one(self): + self.patch_attr(ggml, "_has_vulkan", lambda: False) + self.assertEqual(ggml._wanted_assets(ggml.LLAMA), + ("bin-win-cpu-x64.zip",)) + + def test_an_arm_machine_is_not_handed_the_x64_build(self): + self.patch_attr(ggml, "_arch", lambda: "arm64") + self.patch_attr(ggml, "_has_vulkan", lambda: True) + self.assertEqual(ggml._wanted_assets(ggml.LLAMA), + ("bin-win-cpu-arm64.zip",)) + + +class InstallOnWindows(Local): + """The Windows releases are zips, and the binary carries .exe.""" + + def setUp(self): + super().setUp() + self.patch_attr(sys, "platform", "win32") + self.patch_attr(ggml, "_arch", lambda: "x64") + self.archive = zipball({ + "Release/whisper-server.exe": b"MZ not really a program", + "Release/whisper.dll": b"not really a library", + }) + + def release(self, *names): + digest = hashlib.sha256(self.archive) + return {"tag_name": "v1.9.1", "assets": [ + {"name": name, "browser_download_url": f"https://example.invalid/{name}", + "size": 10, "digest": "sha256:" + digest.hexdigest()} + for name in names]} + + def test_the_zip_lands_and_the_exe_inside_it_is_found(self): + with serving(self.release("whisper-blas-bin-x64.zip"), self.archive): + path = ggml.install_program(ggml.WHISPER) + self.assertTrue(path.endswith("whisper-server.exe")) + self.assertTrue(os.path.isfile(path)) + self.assertTrue(os.path.isfile(os.path.join(os.path.dirname(path), + "whisper.dll"))) + self.assertEqual(ggml.installed_program(ggml.WHISPER), path) + + def test_the_blas_build_is_the_one_fetched_when_both_are_offered(self): + listing = self.release("whisper-bin-x64.zip", "whisper-blas-bin-x64.zip") + with serving(listing, self.archive) as calls: + ggml.install_program(ggml.WHISPER) + urls = [call.args[0].full_url for call in calls.call_args_list] + self.assertTrue(urls[1].endswith("whisper-blas-bin-x64.zip")) + + def test_a_release_with_nothing_for_windows_says_so(self): + with fake_urlopen(json_body(self.release("whisper-bin-ubuntu-x64.tar.gz"))): + with self.assertRaises(ggml.LocalError) as caught: + ggml.install_program(ggml.WHISPER) + self.assertIn("this machine", str(caught.exception)) diff --git a/tests/test_hotkey.py b/tests/test_hotkey.py index c6ec50f..d466b89 100644 --- a/tests/test_hotkey.py +++ b/tests/test_hotkey.py @@ -2,10 +2,14 @@ import contextlib import os +import queue import subprocess +import time import unittest from unittest import mock +from PyQt6.QtCore import Qt + import config as cfg import hotkey from tests.support import DikteTest, FakeCompleted, linux_only @@ -681,5 +685,204 @@ class MacChooser(DikteTest): self.assertFalse(hotkey.valid_shortcut("Cmd+Space")) +# --- Windows ---------------------------------------------------------------- + +class ParseWindowsShortcut(unittest.TestCase): + def test_the_default(self): + self.assertEqual(hotkey.parse_windows_shortcut("Ctrl+Space"), + (hotkey.WIN_MODS["ctrl"], 0x20)) + + def test_case_and_spacing_do_not_matter(self): + self.assertEqual(hotkey.parse_windows_shortcut(" ctrl + SPACE "), + hotkey.parse_windows_shortcut("Ctrl+Space")) + + def test_several_modifiers_are_one_number(self): + modifiers, key = hotkey.parse_windows_shortcut("Ctrl+Shift+M") + self.assertEqual(modifiers, + hotkey.WIN_MODS["ctrl"] | hotkey.WIN_MODS["shift"]) + self.assertEqual(key, hotkey.WIN_KEYS["m"]) + + def test_the_synonyms_land_on_one_number(self): + for name in ("meta", "super", "win"): + with self.subTest(name=name): + self.assertEqual(hotkey.parse_windows_shortcut(f"{name}+space"), + (hotkey.WIN_MODS["win"], 0x20)) + self.assertEqual(hotkey.parse_windows_shortcut("Control+Space"), + hotkey.parse_windows_shortcut("Ctrl+Space")) + + def test_a_key_on_its_own(self): + self.assertEqual(hotkey.parse_windows_shortcut("F9"), + (0, hotkey.WIN_KEYS["f9"])) + + def test_modifiers_with_no_key(self): + self.assertEqual(hotkey.parse_windows_shortcut("Ctrl+Alt"), (None, None)) + + def test_a_key_nobody_mapped(self): + self.assertEqual(hotkey.parse_windows_shortcut("Ctrl+F13"), (None, None)) + + def test_something_that_is_not_even_a_string(self): + self.assertEqual(hotkey.parse_windows_shortcut(None), (None, None)) + + +class FakeWinHotkeys: + """user32 and kernel32, as much of both as the listener calls. + + The message queue is a real queue: GetMessageW blocks on it the way the + real one blocks on the thread's, so the listener runs its actual loop and + a test presses the key by posting the message a press would. + """ + + def __init__(self): + self.registered = {} # identifier -> (modifiers, key) + self.refused = set() # (modifiers, key) another program holds + self.unregistered = [] + self.queue = queue.Queue() + + # --- user32 + def RegisterHotKey(self, hwnd, identifier, modifiers, key): + if (modifiers & ~hotkey.WIN_MOD_NOREPEAT, key) in self.refused: + return 0 + self.registered[identifier] = (modifiers, key) + return 1 + + def UnregisterHotKey(self, hwnd, identifier): + self.unregistered.append(identifier) + self.registered.pop(identifier, None) + return 1 + + def PeekMessageW(self, reference, hwnd, low, high, remove): + return 0 + + def GetMessageW(self, reference, hwnd, low, high): + kind, wparam = self.queue.get() + if kind == hotkey.WM_QUIT: + return 0 + message = reference._obj + message.message = kind + message.wParam = wparam + return 1 + + def PostThreadMessageW(self, thread_id, message, wparam, lparam): + self.queue.put((message, wparam)) + return 1 + + # --- kernel32 + def GetCurrentThreadId(self): + return 1 + + # --- the keyboard + def press(self, identifier): + self.queue.put((hotkey.WM_HOTKEY, identifier)) + + +class WinListener(DikteTest): + """What the listener asks Windows for, without a Windows to ask.""" + + def setUp(self): + super().setUp() + self.api = FakeWinHotkeys() + self.patch_attr(hotkey, "_win_input", lambda: (self.api, self.api)) + self.addCleanup(hotkey._REGISTERED.clear) + self.listener = hotkey.WinHotkey() + self.addCleanup(self.listener.stop) + self.failures = [] + # Direct, because the emits come from the listener's own thread and + # there is no event loop here to carry a queued one across. + self.listener.failed.connect(self.failures.append, + Qt.ConnectionType.DirectConnection) + + @staticmethod + def settles(seen, count=1): + """The signals arrive from the listener's own thread, not this one.""" + deadline = time.monotonic() + 2 + while len(seen) < count and time.monotonic() < deadline: + time.sleep(0.01) + return seen + + def test_every_binding_is_registered_with_its_modifiers(self): + self.assertTrue(self.listener.start({"toggle": "Ctrl+Space", + "cancel": "Ctrl+Shift+Space"})) + norepeat = hotkey.WIN_MOD_NOREPEAT + self.assertEqual(self.api.registered, { + 1: (hotkey.WIN_MODS["ctrl"] | norepeat, 0x20), + 2: (hotkey.WIN_MODS["ctrl"] | hotkey.WIN_MODS["shift"] | norepeat, 0x20), + }) + + def test_what_landed_is_what_the_status_line_shows(self): + self.listener.start({"toggle": "Ctrl+Space"}) + self.assertEqual(hotkey._REGISTERED, + {hotkey.DESKTOP_ID: "Ctrl+Space"}) + + def test_a_press_arrives_under_the_name_it_was_registered_as(self): + seen = [] + self.listener.triggered.connect(seen.append, + Qt.ConnectionType.DirectConnection) + self.listener.start({"toggle": "Ctrl+Space", "cancel": "Ctrl+Shift+Space"}) + self.api.press(2) + self.assertEqual(self.settles(seen), ["cancel"]) + + def test_a_held_combination_is_reported_and_the_rest_still_land(self): + self.api.refused = {(hotkey.WIN_MODS["ctrl"], 0x20)} + started = self.listener.start({"toggle": "Ctrl+Space", + "cancel": "Ctrl+Shift+Space"}) + self.assertTrue(started) + self.assertIn("Ctrl+Space", self.settles(self.failures)[0]) + self.assertEqual(list(self.api.registered), [2]) + + def test_an_unparsable_binding_is_reported(self): + self.assertFalse(self.listener.start({"toggle": "Ctrl+F13"})) + self.assertIn("Ctrl+F13", self.failures[0]) + + def test_nothing_but_empty_bindings_does_not_start(self): + self.assertFalse(self.listener.start({"toggle": "", "cancel": ""})) + self.assertFalse(self.listener.running) + + def test_stop_lets_go_of_everything(self): + self.listener.start({"toggle": "Ctrl+Space", "cancel": "Ctrl+Shift+Space"}) + self.listener.stop() + self.assertEqual(self.api.registered, {}) + self.assertEqual(hotkey._REGISTERED, {}) + self.assertFalse(self.listener.running) + + def test_a_second_start_is_a_clean_slate(self): + self.listener.start({"toggle": "Ctrl+Space"}) + self.assertTrue(self.listener.start({"toggle": "Ctrl+Shift+Space"})) + self.assertEqual(self.api.registered, + {1: (hotkey.WIN_MODS["ctrl"] | hotkey.WIN_MODS["shift"] + | hotkey.WIN_MOD_NOREPEAT, 0x20)}) + + +class WindowsChooser(DikteTest): + def setUp(self): + super().setUp() + self.enterContext(mock.patch.object(hotkey.sys, "platform", "win32")) + self.addCleanup(hotkey._REGISTERED.clear) + + def test_the_listener_is_the_windows_hotkey_service(self): + self.assertIsInstance(hotkey.listener(), hotkey.WinHotkey) + + def test_a_combination_is_checked_against_the_windows_table(self): + self.assertTrue(hotkey.valid_shortcut("Ctrl+Space")) + self.assertFalse(hotkey.valid_shortcut("Ctrl+F13")) + + def test_no_registry_to_write_into_and_no_restart_to_wait_for(self): + self.assertFalse(hotkey.installs_shortcuts()) + self.assertFalse(hotkey.shortcut_needs_restart()) + self.assertEqual(hotkey.desktop_name(), "Windows") + + def test_installing_records_it_rather_than_writing_anything(self): + with mock.patch.object(hotkey.subprocess, "run") as run: + ok, message = hotkey.install_shortcut("Ctrl+Space", "dikte toggle") + run.assert_not_called() + self.assertTrue(ok) + self.assertEqual(hotkey.shortcut_status(), "Ctrl+Space") + hotkey.remove_shortcut() + self.assertIsNone(hotkey.shortcut_status()) + + def test_no_list_of_conflicts_to_read(self): + """Not even KDE's file, which a dual-boot home directory could hold.""" + self.assertEqual(hotkey.conflicting_shortcuts("Ctrl+Space"), []) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_paste.py b/tests/test_paste.py index 4bfe99f..7e80c0e 100644 --- a/tests/test_paste.py +++ b/tests/test_paste.py @@ -13,6 +13,7 @@ is checked on a Mac and the macOS half on Linux, and a change to the chooser cannot quietly break the platform nobody is sitting at. """ +import ctypes import os import pathlib import subprocess @@ -56,6 +57,9 @@ class Chooser(DikteTest): def test_a_mac(self): self.assertIs(self.under("darwin"), paste.MACOS) + def test_windows(self): + self.assertIs(self.under("win32"), paste.WINDOWS) + def test_a_mac_running_an_x_server_is_still_a_mac(self): """XQuartz sets DISPLAY, and none of X's programs are what pastes here.""" self.assertIs(self.under("darwin", DISPLAY=":0"), paste.MACOS) @@ -429,5 +433,128 @@ class MacClipboardSnapshot(DikteTest): self.assertFalse(os.path.exists(directory)) +class FakeWin32: + """user32 and kernel32, as much of both as paste.py calls. + + The clipboard is a string held here. A read materialises it as this + machine's own wide characters, which is what wstring_at reads wherever the + test runs; a write arrives as the UTF-16 the real clipboard is handed, so + what the code sent is exactly what is checked. + """ + + def __init__(self): + self.text = None + self.buffers = {} + self.next_handle = 1 + self.pressed = [] # (virtual key, flags), in the order sent + self.send_result = None # None: report every event as delivered + + def _keep(self, buffer): + handle = self.next_handle + self.next_handle += 1 + self.buffers[handle] = buffer + return handle + + # --- user32 + def OpenClipboard(self, owner): + return 1 + + def CloseClipboard(self): + return 1 + + def EmptyClipboard(self): + self.text = None + return 1 + + def GetClipboardData(self, fmt): + if self.text is None: + return 0 + return self._keep(ctypes.create_unicode_buffer(self.text)) + + def SetClipboardData(self, fmt, handle): + raw = self.buffers[handle].raw + self.text = raw.decode("utf-16-le").split("\x00", 1)[0] + return handle + + def SendInput(self, count, inputs, size): + self.pressed.extend((entry.union.ki.wVk, entry.union.ki.dwFlags) + for entry in inputs) + return count if self.send_result is None else self.send_result + + # --- kernel32 + def GlobalAlloc(self, flags, size): + return self._keep(ctypes.create_string_buffer(size)) + + def GlobalLock(self, handle): + buffer = self.buffers.get(handle) + return ctypes.addressof(buffer) if buffer else 0 + + def GlobalUnlock(self, handle): + return 1 + + def GlobalFree(self, handle): + self.buffers.pop(handle, None) + return 1 + + +class Windows(Standing, DikteTest): + """Windows shells out to nothing: both halves are calls into the system.""" + + platform = "win32" + here = paste.WINDOWS + + def setUp(self): + super().setUp() + self.api = FakeWin32() + self.patch_attr(paste, "_win_api", lambda: (self.api, self.api)) + self.patch_attr(paste.time, "sleep", lambda seconds: None) + + def test_what_is_copied_is_what_reads_back(self): + paste.copy("ığüşöç İ") + self.assertEqual(paste.read_clipboard(), "ığüşöç İ".encode("utf-8")) + + def test_an_empty_clipboard_reads_as_empty_text(self): + self.assertEqual(paste.read_clipboard(), b"") + + def test_what_was_saved_goes_back_after_the_paste(self): + paste.copy("mine") + saved = paste.read_clipboard() + paste.copy("the dictation") + paste.copy_bytes(saved) + self.assertEqual(self.api.text, "mine") + + def test_readiness_asks_for_no_program_and_no_permission(self): + with only_these_tools(): + self.assertTrue(paste.paste_ready()) + + def test_the_keys_go_down_in_order_and_up_in_reverse(self): + paste.press("ctrl+v") + keyup = 0x0002 + self.assertEqual(self.api.pressed, + [(0x11, 0), (0x56, 0), (0x56, keyup), (0x11, keyup)]) + + def test_three_keys(self): + paste.press("ctrl+shift+v") + self.assertEqual([code for code, _ in self.api.pressed], + [0x11, 0x10, 0x56, 0x56, 0x10, 0x11]) + + def test_the_other_spellings_land_on_the_same_keys(self): + paste.press("super+enter") + first, self.api.pressed = self.api.pressed, [] + paste.press("meta+return") + self.assertEqual(self.api.pressed, first) + + def test_a_key_nobody_mapped_is_refused_before_anything_is_sent(self): + with self.assertRaises(paste.PasteError): + paste.press("ctrl+f13") + self.assertEqual(self.api.pressed, []) + + def test_a_press_the_system_did_not_take_says_so(self): + self.api.send_result = 0 + with self.assertRaises(paste.PasteError) as caught: + paste.press("ctrl+v") + self.assertIn("SendInput", str(caught.exception)) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_ui.py b/tests/test_ui.py index 518166f..790390c 100644 --- a/tests/test_ui.py +++ b/tests/test_ui.py @@ -15,6 +15,7 @@ from PyQt6.QtWidgets import QApplication, QMessageBox import cleanup import config as cfg +import ggml import hotkey import overlay as overlay_module import paste @@ -449,6 +450,13 @@ if __name__ == "__main__": class LocalModels(DikteTest): """The download boxes, without a network and without either program.""" + def setUp(self): + super().setUp() + # A machine Dikte is actually installed on would otherwise answer the + # "nothing can transcribe" question from its real binary and model. + self.patch_attr(ggml, "BIN_DIR", self.path("bin")) + self.patch_attr(ggml, "MODELS_DIR", self.path("models")) + def window(self, conf): window = settings_ui.SettingsWindow(conf) self.addCleanup(window.deleteLater)