diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index cf7aea1..5a5ff42 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -77,3 +77,41 @@ jobs: bash -n scripts/release.sh bash -n packaging/build-appimage.sh bash -n packaging/build-dmg.sh + + # The same job again for the same reason. The Windows backends are faked at + # the one function that loads user32 and kernel32, so every line of them is + # already read on the Linux above; what only this job can catch is the half + # that reads the real system. %APPDATA% and %LOCALAPPDATA% have to be the + # directories Windows actually hands out, a path spelled with a backslash has + # to be one the tests can still read, and the config-permission test has to + # skip rather than fail on a file system that decides by ACL. + windows: + runs-on: windows-latest + strategy: + fail-fast: false + matrix: + python: ["3.11", "3.13"] + + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python }} + + # No apt step: PyQt6's wheel carries the Qt DLLs it needs on Windows. + - name: Install PyQt6 + run: python -m pip install --quiet PyQt6 + + - name: Run the tests + run: python -m unittest discover --verbose + + # What the Mac does for its installer, in the language this one is in. + # Parsing only: install.ps1 writes into the Start Menu and the user PATH. + - name: Check the installer parses + shell: pwsh + run: | + $problems = $null + [System.Management.Automation.Language.Parser]::ParseFile( + "$PWD/install.ps1", [ref]$null, [ref]$problems) > $null + if ($problems) { $problems; exit 1 } diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index c23508a..c364c89 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -53,19 +53,28 @@ forgets fails rather than hangs. ## Another platform -Three systems are supported: Wayland, X11 and macOS. Each one is a named entry -in a table, and one chooser picks between them, so a fourth adds an entry and a -line rather than a branch inside every function. The three tables are -`paste.Desktop` (clipboard and key press), `audio.Sound` (capture and the device -lists) and the `_macos()`/`_gnome()` pair in `hotkey.py`. Keep `sys.platform` +Four systems are supported: Wayland, X11, macOS and Windows. Each one is a named +entry in a table, and one chooser picks between them, so a fifth adds an entry +and a line rather than a branch inside every function. The tables are +`paste.Desktop` (clipboard and key press), `audio.Sound` (capture, the device +lists, and whether the far side of a meeting can be recorded at all), +`paths.directories()` (where the settings and the data live) and `hotkey.backend()`, +which names the one mechanism a session has for holding a key. Keep `sys.platform` inside the chooser and read it there every time: a constant settled at import is one no test can stand somewhere else. +Where a platform cannot do something, say so in its table entry rather than in +the code that asks. `audio.Sound.meetings` is the shape of it: Windows offers no +capture device for what the speakers are playing, and a caller reading a False +there can tell that apart from an empty device list, which only means the tool +that lists them is not installed. + The tests are split along the same line, and almost none of them are skipped. -892 of the 935 run on any machine, including every line of the Wayland, X11 and -macOS backends: the programs are faked at `shutil.which`, the frameworks at the -one function that loads them. A test class says which system it is standing on -rather than avoiding the question: +1084 of the 1147 run on any machine, including every line of the Wayland, X11, +macOS and Windows backends: the programs are faked at `shutil.which`, the +frameworks and system libraries at the one function that loads them +(`paste._win_api`, `hotkey._win_input`). A test class says which system it is +standing on rather than avoiding the question: ```python class MacOS(ClipboardContract, DikteTest): @@ -73,14 +82,17 @@ class MacOS(ClipboardContract, DikteTest): here = paste.MACOS ``` -so the Linux half is checked on a Mac and the macOS half on Linux, and a change -to a chooser cannot quietly break the platform nobody is sitting at. What the -systems owe in common is written once as a contract class and subclassed by each -of them. +so the Linux half is checked on a Mac, the macOS half on Linux and the Windows +half on both, and a change to a chooser cannot quietly break the platform nobody +is sitting at. What the systems owe in common is written once as a contract +class and subclassed by each of them. The 43 that do carry `@linux_only` are the ones that would need the real thing: -the `/dev/input` listener, KDE's shortcut file, GNOME's gsettings. Mark a test -that way only when faking it would leave nothing to test. A test that quietly +the `/dev/input` listener, KDE's shortcut file, GNOME's gsettings. The 20 with +`@posix_only` are `integrate.py`, the menu entry and the login item a downloaded +build writes for itself: there are two downloads, an AppImage and a disk image, +so that module has no Windows half for a Windows host to check. Mark a test +either way only when faking it would leave nothing to test. A test that quietly stops running on the platform you are porting to protects nothing. ## What a pull request should carry diff --git a/README.md b/README.md index 5498f52..75e40de 100644 --- a/README.md +++ b/README.md @@ -5,10 +5,10 @@ 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, macOS and any other -Linux desktop that will let it read the keyboard. 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, +[Windows](README.windows.md) and any other Linux desktop that will let it read +the keyboard. No dependencies beyond system packages: just the Python standard +library, 3.11 or newer, and PyQt6. *[Türkçe README](README.tr.md)* @@ -93,6 +93,12 @@ Dikte talks to. Build it (`cmake -B build -DWHISPER_BUILD_SERVER=ON transcribe in the cloud. A meeting needs BlackHole or Loopback (`brew install blackhole-2ch`); dictation does not. +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 -m dikte`, 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, or the ones already in your settings when it is given none. `./scripts/update.sh` pulls and puts all of diff --git a/README.tr.md b/README.tr.md index 25329f0..642e37f 100644 --- a/README.tr.md +++ b/README.tr.md @@ -4,10 +4,10 @@ çevrilir, bir model transkripti temizler (ıı'lar, tekrarlar, eksik noktalama), sonuç panoya kopyalanır ve o an yazdığın pencereye yapıştırılır. -KDE Plasma 6 / Wayland için yazıldı; GNOME X11'de, macOS'ta ve klavyeyi -okumasına izin veren diğer Linux masaüstlerinde de çalışır. Sistem -paketleri dışında bağımlılığı yok: sadece Python standart kütüphanesi (3.11 veya -üstü) ve PyQt6. +KDE Plasma 6 / Wayland için yazıldı; GNOME X11'de, macOS'ta, +[Windows](README.windows.md)'ta ve klavyeyi okumasına izin veren diğer Linux +masaüstlerinde de çalışır. Sistem paketleri dışında bağımlılığı yok: sadece +Python standart kütüphanesi (3.11 veya üstü) ve PyQt6. *[English README](README.md)* @@ -92,6 +92,12 @@ build -j`) ve yolunu Ayarlar → API'ye yaz, ya da buluta çevir. Toplantı içi BlackHole veya Loopback gerekiyor (`brew install blackhole-2ch`); dikte için gerekmiyor. +Windows da aynı şekilde çalışıyor, Dikte açıkken kombinasyonu sistemin kendi +kısayol servisi üzerinden tutuyor: `winget install Gyan.FFmpeg`, `pip install +PyQt6`, sonra `python -m dikte`; Başlat Menüsü girdisi ve `dikte` komutu için +isteğe bağlı `install.ps1`. Orada toplantı kaydı henüz yok, ayrıntılar +[Windows README](README.windows.md)'sinde. + `install.sh` `dikte` komutunu, menü girdisini, oturum açılışında otomatik başlatmayı ve iki global kısayolu kurar; tuşları iki argümanı, argüman verilmezse ayarlarında duranlar. `./scripts/update.sh` son sürümü çeker ve diff --git a/README.windows.md b/README.windows.md new file mode 100644 index 0000000..05adcf6 --- /dev/null +++ b/README.windows.md @@ -0,0 +1,78 @@ +# 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 + +From a checkout: the releases page carries an AppImage and a disk image, and no +Windows build yet. + +```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 -m dikte +``` + +## 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, and none for + Windows on ARM either: whisper.cpp publishes x64 only, so a Snapdragon + machine runs it under emulation and the cloud is the faster option there. +- 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 `dikte doctor` find ffmpeg, and does + `dikte devices` list your microphone? `devices` also takes a fresh listing, + which is what to run after plugging one in. +- **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/dikte/api.py b/dikte/api.py index fa89af1..0c76b80 100644 --- a/dikte/api.py +++ b/dikte/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/dikte/app.py b/dikte/app.py index a098199..842ee51 100644 --- a/dikte/app.py +++ b/dikte/app.py @@ -15,6 +15,7 @@ import json import os import signal import socket +import subprocess import sys import threading @@ -97,6 +98,9 @@ class Dikte: self.meeting_base = "" self.meeting_message = "" self.settings_window = None + # The single-instance server, handed over once run_app has opened it, so + # that a restart can stop answering before the replacement starts. + self.server = None self._quitting = False # A request that asked to be told how its run ended waits in here until # the run gets there, keyed by which of the three it was waiting on. @@ -966,8 +970,29 @@ class Dikte: if self.settings_window is not None: self.settings_window.close() self.shutdown() + # Stop answering before the replacement is started, not just afterwards. + # execv leaves nothing behind to answer, but a Windows restart is two + # processes for a moment, and removeServer does nothing about a name + # another process is holding. The new one then either opens a second + # server on a name the old one is still answering on, so that a command + # arriving in that moment reaches the process that is going away, or + # fails to open one at all and says so to a console nobody is watching. + # Closing first leaves neither. + if self.server is not None: + self.server.close() QLocalServer.removeServer(SERVER_NAME) args = ipc.launcher() + ["--gui"] + 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( + args, + creationflags=(subprocess.DETACHED_PROCESS + | subprocess.CREATE_NEW_PROCESS_GROUP), + close_fds=True, + ) + QApplication.instance().quit() + return os.execv(args[0], args) def shutdown(self): @@ -1038,7 +1063,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) @@ -1128,6 +1157,7 @@ def run_app(args): QLocalServer.removeServer(SERVER_NAME) if not server.listen(SERVER_NAME): print(f"dikte: could not open the IPC socket: {server.errorString()}") + dikte.server = server def on_connection(): conn = server.nextPendingConnection() diff --git a/dikte/assistant.py b/dikte/assistant.py index 89c2ec1..b57fd06 100644 --- a/dikte/assistant.py +++ b/dikte/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/dikte/audio.py b/dikte/audio.py index 7221625..ef24461 100644 --- a/dikte/audio.py +++ b/dikte/audio.py @@ -33,6 +33,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 @@ -57,6 +61,19 @@ QUIET_MIC_SECONDS = 10 QUIET_MIC_SHARE = 0.5 +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.""" @@ -111,7 +128,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)) @@ -172,7 +190,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: @@ -284,6 +302,14 @@ class MeetingRecorder(QObject): def start(self, path, mic_target="", system_target="", max_seconds=14400): if self.active: return + # Before ffmpeg is looked for, because installing it would not help: a + # system with no way to capture what the speakers are playing has none + # whatever else is on the machine. + if not sound().meetings: + self.failed.emit(t("This system offers nothing that records what " + "the speakers are playing, so a meeting cannot " + "be recorded on it.")) + return if not shutil.which("ffmpeg"): self.failed.emit(t("ffmpeg not found. Install it to record a meeting.")) return @@ -312,7 +338,8 @@ class MeetingRecorder(QObject): self._procs = [] for command, log in zip(commands, self._logs): self._procs.append(subprocess.Popen( - command, stdout=subprocess.PIPE, stderr=log, bufsize=0 + command, stdout=subprocess.PIPE, stderr=log, bufsize=0, + creationflags=NO_WINDOW, )) except (OSError, wave.Error) as exc: # One of two capture processes may already be running, and a Mac @@ -411,7 +438,7 @@ class MeetingRecorder(QObject): running = [proc for proc in self._procs if proc.poll() is None] for proc in running: try: - proc.send_signal(signal.SIGINT) + _interrupt(proc) except OSError: pass for proc in running: @@ -838,13 +865,131 @@ 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. + + +# A device entry and the line under it, in the two shapes ffmpeg has printed +# this listing in. Newer builds mark each device `(audio)` or `(video)`; older +# ones print no marker and group the devices under a heading instead. Both are +# anchored at each end, so that the error lines the command ends with, which +# quote the device name that was not found, are not read as devices. The +# bracketed prefix is not pinned to a spelling: ffmpeg 8 writes `[in#0 @ ...]` +# where the versions before it wrote `[dshow @ ...]`. +_DSHOW_ENTRY = re.compile( + r'^(?:\[[^\]]*\]\s*)?"([^"]+)"\s*(?:\(([^)]*)\))?\s*$') +_DSHOW_ALTERNATIVE = re.compile( + r'^(?:\[[^\]]*\]\s*)?Alternative name\s+"([^"]+)"\s*$') +_DSHOW_HEADING = re.compile(r'DirectShow (audio|video) devices') + +# The last listing taken, so that a dictation does not pay for one of its own. +_DSHOW_SEEN = [] + + +def _parse_dshow_listing(text): + """[(id, name)] for the audio devices in one ffmpeg device listing. + + Two friendly names on one machine are routinely identical: a laptop with a + headset plugged in shows two microphones called the same thing, and + `audio=` would reach only the first of them either way. The + alternative name ffmpeg prints under each device is unique and is what the + recorder is given back, while the friendly name is what a user picks from. + """ + devices = [] + heading = "" + for line in text.splitlines(): + found = _DSHOW_HEADING.search(line) + if found: + heading = found.group(1) + continue + found = _DSHOW_ALTERNATIVE.match(line.strip()) + if found: + if devices: + devices[-1][0] = found.group(1) + continue + found = _DSHOW_ENTRY.match(line.strip()) + if found: + kind = (found.group(2) or heading).lower() + devices.append([found.group(1), found.group(1), kind]) + return [(identifier, name) for identifier, name, kind in devices + if "audio" in kind] + + +def _dshow_devices(): + """[(id, name)] for every DirectShow audio capture device, freshly asked. + + The list comes out on stderr of a command that then fails, the same + documented trick AVFoundation uses above. + """ + 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 = _parse_dshow_listing(result.stderr.decode("utf-8", "replace")) + _DSHOW_SEEN[:] = devices + return devices + + +def _dshow_first_device(): + """The device an unset target stands for, without a listing per dictation. + + dshow has no "default" for an empty target to mean, so it has to be turned + into a name, and asking ffmpeg for one costs a process every time the key + is pressed. The last listing is used when there is one: opening Settings or + running `dikte devices` takes a fresh one, which is what somebody who has + just plugged a microphone in does anyway. + """ + devices = _DSHOW_SEEN or _dshow_devices() + return devices[0][0] if devices else "" + + +def _dshow_record(target): + if not shutil.which("ffmpeg"): + return [] + device = target or _dshow_first_device() + if not device: + return [] + 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 how to capture two at once, that one as the # list of processes it takes, the two device lists, which device a meeting - # records the far side from, and what to say when the programs for any of - # it are not installed. - "record meeting inputs outputs default_output missing", + # records the far side from, whether this system can record one at all, and + # what to say when the programs for any of it are not installed. + # + # `meetings` is the sound system's own answer, not this machine's: an empty + # output list means the tool that lists them is missing, which is a thing a + # user can go and fix, while False here is a thing they cannot. + "record meeting inputs outputs default_output meetings missing", ) PULSE = Sound( @@ -853,6 +998,7 @@ PULSE = Sound( inputs=_pulse_inputs, outputs=_pulse_outputs, default_output=_pulse_default_output, + meetings=True, missing="No audio recorder found. Install pulseaudio-utils or pipewire-audio.", ) @@ -865,13 +1011,33 @@ COREAUDIO = Sound( # empty list would leave nothing to pick. outputs=_avfoundation_named_inputs, default_output=_avfoundation_default_output, + # With a loopback driver installed, which is what the Settings note is for. + meetings=True, missing="ffmpeg not found. Install it with: brew install ffmpeg", ) +DSHOW = Sound( + record=_dshow_record, + meeting=_dshow_meeting, + inputs=_dshow_devices, + outputs=_dshow_no_outputs, + default_output=_dshow_no_default_output, + # Windows offers no capture device for what the speakers are playing, and + # there is no driver to install that would add one. + meetings=False, + 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/dikte/cleanup.py b/dikte/cleanup.py index 22d31ec..96eab1d 100644 --- a/dikte/cleanup.py +++ b/dikte/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/dikte/cli.py b/dikte/cli.py index a7da1cd..78efeef 100644 --- a/dikte/cli.py +++ b/dikte/cli.py @@ -17,6 +17,7 @@ import json import os import shutil import signal +import subprocess import sys import time @@ -136,6 +137,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(args[0], args) @@ -299,6 +310,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) @@ -643,7 +657,10 @@ def cmd_devices(opts): "default": name == default} for name, desc in audio.list_monitors()] if not mics and not monitors: - return fail(opts, "pactl found nothing; is PipeWire running?") + # Which program was asked, and so which one to go and look at, is not + # the same on all four systems: naming pactl on Windows sends somebody + # after a program that was never going to be there. + return fail(opts, audio.sound().missing) lines = ["Microphones:"] lines += [f" {'*' if item['chosen'] else ' '} {item['name']}\n" @@ -802,9 +819,20 @@ def cmd_status(opts): def cmd_doctor(opts): """What the settings window checks behind its buttons, in one pass.""" conf = cfg.Config() - wanted = ["pw-record", "wl-copy", "ydotool", "ffmpeg", "pactl", "kwriteconfig6", - assistant.executable(assistant.provider(conf)) or "claude", - cleanup.executable(cleanup.provider(conf))] + # The two the clipboard and the key press go through come out of the table + # rather than being spelled here, because they are not the same pair on all + # four systems: X11 pastes with xclip where Wayland pastes with wl-copy, a + # Mac shells out for one half and Windows for neither. A row saying ydotool + # is missing on a machine that would never have run it is not a diagnosis, + # it is a red mark to explain away. + here = paste.desktop() + wanted = [here.clipboard, here.keyboard] + if sys.platform.startswith("linux"): + # Recording, the device list, and KDE's shortcut registry. + wanted += ["pw-record", "pactl", "kwriteconfig6"] + wanted += ["ffmpeg", + assistant.executable(assistant.provider(conf)) or "claude", + cleanup.executable(cleanup.provider(conf))] programs = {name: shutil.which(name) or "" for name in wanted if name} target = conf.transcribe_target() cleaner = cleanup.provider(conf) diff --git a/dikte/filetranscribe.py b/dikte/filetranscribe.py index 5ac97e3..8415752 100644 --- a/dikte/filetranscribe.py +++ b/dikte/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/dikte/ggml.py b/dikte/ggml.py index 6805337..9256f0f 100644 --- a/dikte/ggml.py +++ b/dikte/ggml.py @@ -43,6 +43,7 @@ import threading import time import urllib.error import urllib.request +import zipfile from . import hub from . import paths @@ -154,11 +155,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 @@ -168,11 +172,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: @@ -218,9 +228,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): @@ -233,6 +245,20 @@ 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. + # + # x64 whatever this machine is, because whisper.cpp publishes no + # arm64 build for Windows: a Snapdragon runs this one emulated, + # which is slow but is the only local option there is. + 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",) @@ -278,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(): @@ -286,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 @@ -344,7 +380,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)) @@ -501,6 +537,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. @@ -602,6 +658,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}", @@ -699,8 +757,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/dikte/hotkey.py b/dikte/hotkey.py index 16c9a1e..649ef36 100644 --- a/dikte/hotkey.py +++ b/dikte/hotkey.py @@ -440,15 +440,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 ------------------------------------------- -# The four ways a combination can reach Dikte. Everything below asks backend() +# The five ways a combination can reach Dikte. Everything below asks backend() # rather than looking at the session itself, so the name shown, the status read # back, what Install writes and what the installer promises cannot disagree # about which one this session got. KDE = "kde" GNOME = "gnome" MACOS = "macos" +WINDOWS = "windows" LISTENER = "listener" @@ -456,6 +616,10 @@ def _macos(): return sys.platform == "darwin" +def _windows(): + return sys.platform == "win32" + + def backend(): """Which shortcut mechanism this session has. @@ -467,6 +631,8 @@ def backend(): """ if _macos(): return MACOS + if _windows(): + return WINDOWS names = os.environ.get("XDG_CURRENT_DESKTOP", "").lower().split(":") names = [name.strip() for name in names if name.strip()] if any("gnome" in name for name in names) and shutil.which("gsettings"): @@ -596,7 +762,11 @@ 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 default_combo(which): @@ -613,16 +783,20 @@ def default_combo(which): 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 and the plain listener do not: there is - nothing to install, nothing to remove, and Settings should not offer either. + survives Dikte being closed. macOS, Windows and the plain listener do not: + there is nothing to install, nothing to remove, and Settings should not + offer either. """ return backend() in (KDE, GNOME) @@ -631,7 +805,7 @@ 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 the other two never had one to write. + up as it is written, and the others never had one to write. """ return backend() == KDE @@ -644,7 +818,7 @@ def install_shortcut(shortcut, exec_command, name="Dikte: start/stop recording", if which == KDE: return install_kde_shortcut(shortcut, exec_command, name, desktop_id) _REGISTERED[desktop_id] = shortcut - if which == MACOS: + if which in (MACOS, WINDOWS): return True, t( "Shortcut saved: {shortcut}\nDikte holds this one itself while it " "is running, so it works as soon as the settings are saved.", @@ -686,6 +860,8 @@ def desktop_name(): which = backend() if which == MACOS: return "macOS" + if which == WINDOWS: + return "Windows" if which == GNOME: return "GNOME" if which == KDE: @@ -776,11 +952,11 @@ 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 backend() != KDE: - # Nowhere else has a list to read. macOS answers the question by - # refusing the registration, which CarbonHotkey reports when it asks - # for the key; the other two would only be reading a file their session - # never looks at, and a leftover one from a Plasma install the user has - # since left would refuse perfectly good combinations. + # Nowhere else has a list to read. macOS and Windows answer the question + # by refusing the registration, which their listeners report when they + # ask for the key; the other two would only be reading a file their + # session never looks at, and a leftover one from a Plasma install the + # user has since left would refuse perfectly good combinations. return [] try: text = SHORTCUTS_FILE.read_text(encoding="utf-8") diff --git a/dikte/i18n.py b/dikte/i18n.py index 6a29fec..9b4826c 100644 --- a/dikte/i18n.py +++ b/dikte/i18n.py @@ -109,6 +109,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}", @@ -353,6 +357,11 @@ TR = { "meantime.": "Dikte bu kombinasyonları çalışırken macOS'tan kendisi ister. Hiçbir şey " "kurulmaz ve o sırada başka hiçbir uygulama bu tuşları almaz.", + "Dikte asks Windows for these combinations itself, while it is running. " + "Nothing is installed, and no other application receives them in the " + "meantime.": + "Dikte bu kombinasyonları çalışırken Windows'tan kendisi ister. Hiçbir şey " + "kurulmaz ve o sırada başka hiçbir uygulama bu tuşları almaz.", "{desktop} keeps no shortcut registry, so Dikte listens for these " "combinations itself while it is running. Your user has to be able to read " "/dev/input for that, and the focused application receives the keys as " @@ -392,6 +401,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" @@ -643,6 +658,16 @@ TR = { "macOS, hoparlörden çıkan sesi kaydedilebilir bir kaynak olarak sunmaz. " "BlackHole ya da Loopback kur, toplantının sesini oradan geçir ve " "yukarıdan onu seç.", + "This system offers nothing that records what the speakers are playing, " + "so a meeting cannot be recorded on it. Dictation and transcribing a file " + "are unaffected.": + "Bu sistem, hoparlörden çıkan sesi kaydeden hiçbir şey sunmuyor; " + "burada toplantı kaydedilemez. Dikte ve dosya deşifresi bundan " + "etkilenmez.", + "This system offers nothing that records what the speakers are playing, " + "so a meeting cannot be recorded on it.": + "Bu sistem, hoparlörden çıkan sesi kaydeden hiçbir şey sunmuyor; " + "burada toplantı kaydedilemez.", "Wear headphones if you can. Through speakers your microphone hears the " "other side as well, and although a line that lands on both channels at " "once is dropped again, the repair is never as clean as not needing it.": diff --git a/dikte/ipc.py b/dikte/ipc.py index 8d9e367..823799b 100644 --- a/dikte/ipc.py +++ b/dikte/ipc.py @@ -15,7 +15,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/dikte/overlay.py b/dikte/overlay.py index a46cb6d..4f4d3f1 100644 --- a/dikte/overlay.py +++ b/dikte/overlay.py @@ -68,9 +68,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/dikte/paste.py b/dikte/paste.py index fb1a9e4..62fad76 100644 --- a/dikte/paste.py +++ b/dikte/paste.py @@ -330,6 +330,166 @@ 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" + # Filled before the clipboard is opened at all. EmptyClipboard is what + # throws away whatever was there, and a failure after it and before the + # SetClipboardData would leave the clipboard holding nothing: the one way + # this function could lose what it was called to put back. + handle = kernel32.GlobalAlloc(_WIN_GMEM_MOVEABLE, len(payload)) + pointer = kernel32.GlobalLock(handle) if handle else None + if not pointer: + if handle: + kernel32.GlobalFree(handle) + raise PasteError(t("Could not copy to clipboard: {error}", + error="out of memory")) + ctypes.memmove(pointer, payload, len(payload)) + kernel32.GlobalUnlock(handle) + + if not _win_open_clipboard(user32): + kernel32.GlobalFree(handle) + raise PasteError(t("Could not copy to clipboard: {error}", + error="the clipboard is held by another program")) + try: + user32.EmptyClipboard() + 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( @@ -362,6 +522,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 @@ -383,6 +554,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"): @@ -427,6 +600,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: @@ -454,6 +630,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}.", @@ -473,7 +652,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/dikte/paths.py b/dikte/paths.py index 8493746..7670c8b 100644 --- a/dikte/paths.py +++ b/dikte/paths.py @@ -16,7 +16,8 @@ import pathlib import sys -def _xdg(var, default): +def _env(var, default): + """The directory a variable names, or the one it stands in for.""" return pathlib.Path(os.environ.get(var) or os.path.expanduser(default)) @@ -24,13 +25,20 @@ def directories(platform=None): """(settings, data), in the two places this system keeps them. macOS keeps both in the one directory a Mac user's backup already knows - about. Everywhere else they are separate and follow the XDG variables. + about. Windows keeps them apart on purpose: settings roam with the account, + and several gigabytes of models are exactly what a roaming profile must not + carry. Everywhere else they are separate and follow the XDG variables. """ - if (platform or sys.platform) == "darwin": + here = platform or sys.platform + if here == "darwin": support = pathlib.Path.home() / "Library/Application Support/Dikte" return support, support - return (_xdg("XDG_CONFIG_HOME", "~/.config") / "dikte", - _xdg("XDG_DATA_HOME", "~/.local/share") / "dikte") + if here == "win32": + roaming = _env("APPDATA", "~/AppData/Roaming") + local = _env("LOCALAPPDATA", "~/AppData/Local") + return roaming / "Dikte", local / "Dikte" + return (_env("XDG_CONFIG_HOME", "~/.config") / "dikte", + _env("XDG_DATA_HOME", "~/.local/share") / "dikte") CONFIG_DIR, DATA_DIR = directories() diff --git a/dikte/settings_ui.py b/dikte/settings_ui.py index 1a73968..4563d1a 100644 --- a/dikte/settings_ui.py +++ b/dikte/settings_ui.py @@ -1100,6 +1100,18 @@ class SettingsWindow(QDialog): )) mac_note.setWordWrap(True) sources_form.addRow(mac_note) + elif not audio.sound().meetings: + # Windows is the system this is written for: it offers nothing that + # captures what the speakers are playing, and there is no driver to + # install that would put an entry in the list above. Left unsaid, + # the box is simply empty and the Record button fails at the press. + nothing_note = QLabel(t( + "This system offers nothing that records what the speakers are " + "playing, so a meeting cannot be recorded on it. Dictation and " + "transcribing a file are unaffected." + )) + nothing_note.setWordWrap(True) + sources_form.addRow(nothing_note) note = QLabel(t( "Wear headphones if you can. Through speakers your microphone hears " @@ -1369,6 +1381,12 @@ class SettingsWindow(QDialog): "running. Nothing is installed, and no other application receives " "them in the meantime." ) + elif hotkey.backend() == hotkey.WINDOWS: + explanation = t( + "Dikte asks Windows for these combinations itself, while it is " + "running. Nothing is installed, and no other application receives " + "them in the meantime." + ) else: # The desktops nobody writes a backend for. Saying "installed" here # would be the old bug in words: there is no registry, the listener diff --git a/install.ps1 b/install.ps1 new file mode 100644 index 0000000..c847131 --- /dev/null +++ b/install.ps1 @@ -0,0 +1,82 @@ +# Installs Dikte for this Windows user: a Start Menu entry, an optional +# autostart entry, and a `dikte` command that works from any terminal. +# +# powershell -ExecutionPolicy Bypass -File install.ps1 # install +# powershell -ExecutionPolicy Bypass -File install.ps1 -Autostart # + start at sign-in +# powershell -ExecutionPolicy Bypass -File install.ps1 -Uninstall # remove +param( + [switch]$Autostart, + [switch]$Uninstall +) + +$ErrorActionPreference = "Stop" +$repo = $PSScriptRoot +# The one file that starts the application, whoever is asking: the Start Menu +# entry, the autostart entry and the dikte command all name it. +$entry = Join-Path $repo "dikte\__main__.py" +$startMenu = [Environment]::GetFolderPath("Programs") +$startup = [Environment]::GetFolderPath("Startup") +$shortcut = Join-Path $startMenu "Dikte.lnk" +$autostartLink = Join-Path $startup "Dikte.lnk" +# WindowsApps is already on the user PATH, so a dikte.cmd left there runs from +# any terminal without a PATH edit and without an administrator. +$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 "removed: $path" } + } + Write-Host "Dikte's shortcuts are gone. The repository and your settings are not." + exit 0 +} + +# --- what it needs ---------------------------------------------------------- +$python = Get-Command python -ErrorAction SilentlyContinue +if (-not $python) { + Write-Error "No python found. Install it with: 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 or newer is needed, and this one is $version." +} +& python -c "import PyQt6.QtWidgets" 2>$null +if ($LASTEXITCODE -ne 0) { + Write-Host "Installing PyQt6..." + & python -m pip install PyQt6 + if ($LASTEXITCODE -ne 0) { Write-Error "PyQt6 would not install." } +} +if (-not (Get-Command ffmpeg -ErrorAction SilentlyContinue)) { + Write-Warning "No ffmpeg found. Recording needs it: winget install Gyan.FFmpeg" +} + +# pythonw.exe runs the same program without a console window behind it. +$pythonw = Join-Path (Split-Path $python.Source) "pythonw.exe" +if (-not (Test-Path $pythonw)) { $pythonw = $python.Source } + +# --- the Start Menu entry --------------------------------------------------- +$shell = New-Object -ComObject WScript.Shell +foreach ($path in @($shortcut) + $(if ($Autostart) { @($autostartLink) } else { @() })) { + $link = $shell.CreateShortcut($path) + $link.TargetPath = $pythonw + $link.Arguments = "`"$entry`" --gui" + $link.WorkingDirectory = $repo + $link.Description = "Dikte: dictation" + $link.Save() + Write-Host "shortcut: $path" +} + +# --- the dikte command ------------------------------------------------------ +# The interpreter by its full path rather than by name: the one checked above is +# the one the command line should run, whatever a later PATH change puts first. +$shimDir = Split-Path $cmdShim +if (Test-Path $shimDir) { + "@echo off`r`n`"$($python.Source)`" `"$entry`" %*" | + Out-File $cmdShim -Encoding ascii + Write-Host "command: dikte ($cmdShim)" +} else { + Write-Warning "No $shimDir on this machine, so there is no dikte command. Run it as: python `"$entry`"" +} + +Write-Host "" +Write-Host "Installed. Start it from the Start Menu as 'Dikte', or type 'dikte' in a terminal." +Write-Host "The Settings window opens on the first run: download a model there and pick the shortcut (Ctrl+Space by default)." diff --git a/tests/support.py b/tests/support.py index 18c0a06..f6ec508 100644 --- a/tests/support.py +++ b/tests/support.py @@ -40,6 +40,18 @@ linux_only = unittest.skipUnless( "covers the Linux desktop stack (PipeWire, wl-clipboard, ydotool, KDE)", ) +# The launchers a downloaded build writes for itself. There are two downloads, +# an AppImage and a disk image, so `integrate` has a Linux half and a macOS half +# and no third one, and the tests that pin them stand in a home laid out the way +# those two systems lay one out: paths that start at the root, a $HOME the +# library reads, a symlink for the command. None of that is a Windows machine, +# where the same code never runs. A Windows build would add an entry there and +# take the mark off these. +posix_only = unittest.skipIf( + sys.platform == "win32", + "covers what an AppImage and a .app write into the desktop they landed on", +) + def _no_network(*args, **kwargs): raise AssertionError( diff --git a/tests/test_audio.py b/tests/test_audio.py index 3f7777d..a7ea353 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -887,5 +887,201 @@ class MacRecordingCommand(OnMacOS, DikteTest): self.assertFalse(recorder.active) +class NoFarSideToRecord(DikteTest): + """Two different answers, and the table is what tells them apart. + + A sound system that records the far side has a device this machine could + not pick out, and Settings is where to choose one. A sound system that does + not had nothing to offer there in the first place, and "pick one" would + send somebody to an empty box and an installation that cannot help. + """ + + def failure(self, meetings): + recorder = audio.MeetingRecorder() + failures = [] + recorder.failed.connect(failures.append) + with only_these_tools("ffmpeg"), \ + mock.patch.object(audio, "default_monitor", return_value=""), \ + mock.patch.object(audio, "sound", + return_value=audio.PULSE._replace( + meetings=meetings)): + recorder.start(str(self.path("meeting.wav"))) + self.assertFalse(recorder.active) + return failures[0] + + def test_a_system_that_records_the_far_side_sends_you_to_settings(self): + self.assertIn("Settings", self.failure(True)) + + def test_a_system_that_does_not_says_that_instead(self): + message = self.failure(False) + self.assertIn("nothing that records what the speakers", message) + self.assertNotIn("Settings", message) + + def test_the_three_sound_systems_each_answer_the_question(self): + self.assertTrue(audio.PULSE.meetings) + self.assertTrue(audio.COREAUDIO.meetings) + self.assertFalse(audio.DSHOW.meetings) + + +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. + """ + + MIC = "@device_cm_{33D9A762}\\wave_{B1C2}" + LISTING = ( + '[dshow @ 0000020c] "Integrated Camera" (video)\n' + '[dshow @ 0000020c] Alternative name "@device_pnp_\\...."\n' + '[dshow @ 0000020c] "Mikrofon Dizisi (Intel Smart Sound)" (audio)\n' + f'[dshow @ 0000020c] Alternative name "{MIC}"\n' + '[dshow @ 0000020c] "Kulaklık (Soundcore Life Q30)" (audio)\n' + '[dshow @ 0000020c] Could not find audio only device with name ' + '"dummy" among source devices of type audio.\n' + "dummy: Immediate exit requested\n" + ).encode("utf-8") + + def setUp(self): + super().setUp() + # The listing is remembered between calls, so that a dictation does not + # run ffmpeg of its own. It cannot be remembered between tests. + audio._DSHOW_SEEN.clear() + self.addCleanup(audio._DSHOW_SEEN.clear) + + @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) as run: + yield run + + def test_windows_records_through_dshow(self): + self.assertIs(audio.sound(), audio.DSHOW) + + def test_the_audio_devices_are_the_only_ones_read(self): + with self.listing(): + self.assertEqual(audio.list_sources(), [ + (self.MIC, "Mikrofon Dizisi (Intel Smart Sound)"), + ("Kulaklık (Soundcore Life Q30)", + "Kulaklık (Soundcore Life Q30)"), + ]) + + def test_the_device_ffmpeg_could_not_open_is_not_one_of_them(self): + """The command ends by quoting the name it was sent to look for.""" + with self.listing(): + self.assertNotIn("dummy", [name for _, name in audio.list_sources()]) + + def test_a_listing_from_ffmpeg_8_which_renamed_the_prefix(self): + """ffmpeg 8 writes `[in#0 @ ...]` where older builds wrote `[dshow @ ...]`.""" + listing = ( + '[in#0 @ 00000238c3300ac0] "Integrated Camera" (video)\n' + '[in#0 @ 00000238c3300ac0] Alternative name "@device_pnp_\\..."\n' + '[in#0 @ 00000238c3300ac0] "OBS Virtual Camera" (none)\n' + '[in#0 @ 00000238c3300ac0] Alternative name "@device_sw_{860B}"\n' + '[in#0 @ 00000238c3300ac0] "Mikrofon Dizisi (Intel® Smart Sound)" (audio)\n' + '[in#0 @ 00000238c3300ac0] Alternative name "@device_cm_{33D9}"\n' + "Error opening input file dummy.\n" + ).encode("utf-8") + with self.listing(stderr=listing): + self.assertEqual(audio.list_sources(), + [("@device_cm_{33D9}", + "Mikrofon Dizisi (Intel® Smart Sound)")]) + + def test_a_listing_from_an_ffmpeg_that_marks_nothing(self): + """Older builds print a heading instead of an (audio) on every line.""" + listing = ( + '[dshow @ 0] DirectShow video devices\n' + '[dshow @ 0] "Integrated Camera"\n' + '[dshow @ 0] Alternative name "@device_pnp_\\..."\n' + '[dshow @ 0] DirectShow audio devices\n' + '[dshow @ 0] "Microphone (Realtek Audio)"\n' + '[dshow @ 0] Alternative name "@device_cm_{ABCD}"\n' + ).encode("utf-8") + with self.listing(stderr=listing): + self.assertEqual(audio.list_sources(), + [("@device_cm_{ABCD}", "Microphone (Realtek Audio)")]) + + def test_two_devices_called_the_same_thing_stay_apart(self): + """The normal state of a laptop with a headset plugged into it.""" + listing = ( + '[dshow @ 0] "Microphone" (audio)\n' + '[dshow @ 0] Alternative name "@device_cm_{ONE}"\n' + '[dshow @ 0] "Microphone" (audio)\n' + '[dshow @ 0] Alternative name "@device_cm_{TWO}"\n' + ).encode("utf-8") + with self.listing(stderr=listing): + sources = audio.list_sources() + self.assertEqual([identifier for identifier, _ in sources], + ["@device_cm_{ONE}", "@device_cm_{TWO}"]) + self.assertEqual({name for _, name in sources}, {"Microphone"}) + + def test_no_ffmpeg_installed(self): + with only_these_tools(): + self.assertEqual(audio.list_sources(), []) + self.assertEqual(audio.recording_command(), []) + + def test_the_identifier_is_what_the_recorder_is_given_back(self): + with self.listing(): + cmd = audio.recording_command(self.MIC) + self.assertEqual(cmd[cmd.index("-f") + 1], "dshow") + self.assertIn(f"audio={self.MIC}", 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(f"audio={self.MIC}", audio.recording_command()) + + def test_a_dictation_does_not_run_a_listing_of_its_own(self): + """Two hundred milliseconds of ffmpeg in front of every key press.""" + with self.listing() as run: + audio.list_sources() + audio.recording_command() + audio.recording_command() + self.assertEqual(run.call_count, 1) + + def test_opening_the_device_list_asks_again(self): + """Which is what somebody who has just plugged one in does.""" + with self.listing() as run: + audio.list_sources() + audio.list_sources() + self.assertEqual(run.call_count, 2) + + 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_commands("mic", "sys"), []) + + def test_a_meeting_says_what_is_wrong_rather_than_where_to_look(self): + recorder = audio.MeetingRecorder() + failures = [] + recorder.failed.connect(failures.append) + with self.listing(): + recorder.start(str(self.path("meeting.wav"))) + self.assertIn("nothing that records what the speakers", failures[0]) + self.assertFalse(recorder.active) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_cli.py b/tests/test_cli.py index ef8f6f5..3c2de6d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -12,11 +12,14 @@ import json import unittest from unittest import mock +from dikte import audio from dikte import cli from dikte import config as cfg +from dikte import ggml from dikte import hotkey from dikte import ipc -from tests.support import DikteTest, fake_urlopen +from dikte import paste +from tests.support import DikteTest, fake_urlopen, only_these_tools class Options: @@ -434,6 +437,30 @@ class Doctor(DikteTest): self.assertIn("OpenRouter key, cleaning up on some/model", self.run_doctor(as_json=False, cleanup_model="some/model")) + def test_it_asks_after_the_programs_this_desktop_actually_uses(self): + """A missing ydotool on a Mac is a red mark with nothing behind it.""" + with mock.patch.object(cli.paste, "desktop", return_value=paste.MACOS): + mac = self.run_doctor()["programs"] + with mock.patch.object(cli.paste, "desktop", return_value=paste.WAYLAND): + wayland = self.run_doctor()["programs"] + self.assertIn("pbcopy", mac) + self.assertNotIn("ydotool", mac) + self.assertIn("ydotool", wayland) + self.assertIn("ffmpeg", mac) # the one every system records through + + def test_a_system_that_shells_out_for_neither_half_is_asked_for_neither(self): + # shutil.which is faked as well as the platform: the real one reads + # sys.platform too, and reaches for a Windows API this machine has not + # got the moment it is told it is on Windows. + with mock.patch.object(cli.paste, "desktop", return_value=paste.WINDOWS), \ + only_these_tools("ffmpeg"), \ + mock.patch.object(cli.sys, "platform", "win32"): + programs = self.run_doctor()["programs"] + self.assertNotIn("", programs) + self.assertEqual([name for name in ("wl-copy", "ydotool", "pactl", + "pw-record", "kwriteconfig6") + if name in programs], []) + def test_cleanup_on_a_cli_is_a_question_about_the_program(self): reply = self.run_doctor(cleanup_provider="codex", cleanup_codex_model="gpt-5.4") @@ -445,6 +472,25 @@ class Doctor(DikteTest): cleanup_codex_model="gpt-5.4")) +class Devices(DikteTest): + def test_a_machine_with_nothing_names_its_own_missing_program(self): + """The Windows README sends people here, and pactl is not on it.""" + for here, expected in ((audio.DSHOW, "ffmpeg"), + (audio.PULSE, "pulseaudio-utils")): + with self.subTest(sound=expected): + with mock.patch.object(cli.audio, "sound", return_value=here), \ + mock.patch.object(cli.audio, "list_sources", + return_value=[]), \ + mock.patch.object(cli.audio, "list_monitors", + return_value=[]), \ + mock.patch.object(cli.audio, "default_monitor", + return_value=""), \ + captured() as (out, _err): + code = cli.cmd_devices(Options(json=True)) + self.assertEqual(code, 1) + self.assertIn(expected, json.loads(out.getvalue())["error"]) + + class Finding(DikteTest): def test_no_history_at_all(self): self.assertIsNone(cli._find_history("last")) @@ -623,5 +669,27 @@ class Replies(DikteTest): self.assertFalse(launched.called) +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 aa15c91..9d8e372 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() @@ -491,6 +494,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 951654b..14a8ddf 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 from dikte 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.""" @@ -691,3 +701,87 @@ 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",)) + + def test_an_arm_machine_is_handed_the_x64_whisper_anyway(self): + """whisper.cpp publishes no arm64 build for Windows: the release has + Win32 and x64 and nothing else, so emulated is the only local option + a Snapdragon has. Pinned here so that a release which does start + publishing one is noticed rather than quietly ignored.""" + self.patch_attr(ggml, "_arch", lambda: "arm64") + self.assertEqual(ggml._wanted_assets(ggml.WHISPER), + ("whisper-blas-bin-x64.zip", "whisper-bin-x64.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 3ac7b52..55006fa 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 + from dikte import config as cfg from dikte import hotkey from tests.support import DikteTest, FakeCompleted, linux_only @@ -753,5 +757,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_integrate.py b/tests/test_integrate.py index 6bc43b6..9721dd4 100644 --- a/tests/test_integrate.py +++ b/tests/test_integrate.py @@ -15,6 +15,7 @@ import unittest from unittest import mock from dikte import integrate +from tests.support import posix_only class Frozen: @@ -75,6 +76,7 @@ class WhatToStart(unittest.TestCase): def test_a_checkout_names_this_interpreter_and_the_entry_point(self): self.assertFalse(integrate.packaged()) + @posix_only def test_an_appimage_names_the_file_and_not_the_mount(self): """The mount is a fresh /tmp path every run; a shortcut written to it would work until the next login and never again.""" @@ -83,6 +85,7 @@ class WhatToStart(unittest.TestCase): self.assertEqual(str(integrate.target()), "/home/someone/Downloads/Dikte.AppImage") + @posix_only def test_a_mac_names_the_bundle_and_not_the_executable_inside_it(self): with Frozen("/Applications/Dikte.app/Contents/MacOS/Dikte", platform="darwin"): @@ -95,6 +98,7 @@ class WhatToStart(unittest.TestCase): class BundledTools(unittest.TestCase): """The ffmpeg the disk image carries, and how anything finds it.""" + @posix_only def test_a_mac_looks_beside_the_bundle_not_beside_the_executable(self): with Frozen("/Applications/Dikte.app/Contents/MacOS/Dikte", platform="darwin"): @@ -213,6 +217,7 @@ class Certificates(unittest.TestCase): self.assertIsNone(integrate.use_system_certificates()) +@posix_only class Linux(Home): def install(self, appimage, force=False): with Frozen("/tmp/.mount_x/usr/bin/dikte", appimage=str(appimage), @@ -348,6 +353,7 @@ class Linux(Home): self.assertEqual(integrate.ensure(), []) +@posix_only class MacOS(Home): def agent(self): return self.home / "Library/LaunchAgents/io.github.yusufipk.dikte.plist" diff --git a/tests/test_ipc.py b/tests/test_ipc.py index ff44d27..502574c 100644 --- a/tests/test_ipc.py +++ b/tests/test_ipc.py @@ -7,6 +7,8 @@ answers by saying nothing at all. import json import os +import pathlib +import shlex import sys import unittest from unittest import mock @@ -57,13 +59,17 @@ class FakeSocket: class Paths(unittest.TestCase): def test_script_path_points_at_dikte(self): - self.assertTrue(ipc.script_path().endswith("dikte/__main__.py")) + # By its parts rather than as a string: the separator is a backslash on + # Windows, and the path is what a shortcut there runs too. + path = pathlib.Path(ipc.script_path()) + self.assertEqual(path.parts[-2:], ("dikte", "__main__.py")) self.assertTrue(os.path.exists(ipc.script_path())) def test_the_shortcut_command_runs_it_with_this_interpreter(self): - command = ipc.command_for("toggle") - self.assertTrue(command.startswith(sys.executable)) - self.assertTrue(command.endswith(" toggle")) + # Read back through the same quoting it went out with: a Windows path + # is spelled with backslashes and comes out of the join quoted. + self.assertEqual(shlex.split(ipc.command_for("toggle")), + [sys.executable, ipc.script_path(), "toggle"]) def test_a_packaged_build_names_itself_and_no_interpreter(self): """There is no __main__.py on disk in one, and sys.executable is the diff --git a/tests/test_paste.py b/tests/test_paste.py index 8da2c81..360492d 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) @@ -465,5 +469,149 @@ 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 + self.held = False # another program has the clipboard open + self.out_of_memory = False + + def _keep(self, buffer): + handle = self.next_handle + self.next_handle += 1 + self.buffers[handle] = buffer + return handle + + # --- user32 + def OpenClipboard(self, owner): + return 0 if self.held else 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): + if self.out_of_memory: + return 0 + 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_a_copy_that_fails_leaves_what_was_there(self): + """EmptyClipboard is the point of no return, so nothing runs after it.""" + paste.copy("mine") + for failure in ("out_of_memory", "held"): + with self.subTest(failure=failure): + setattr(self.api, failure, True) + with self.assertRaises(paste.PasteError): + paste.copy("the dictation") + self.assertEqual(self.api.text, "mine") + setattr(self.api, failure, False) + + def test_the_handle_is_not_leaked_when_the_copy_fails(self): + self.api.held = True + with self.assertRaises(paste.PasteError): + paste.copy("the dictation") + self.assertEqual(self.api.buffers, {}) + + 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_paths.py b/tests/test_paths.py index 81517d1..8c076f4 100644 --- a/tests/test_paths.py +++ b/tests/test_paths.py @@ -15,29 +15,50 @@ from dikte import paths class Directories(unittest.TestCase): + """Spelled with forward slashes throughout. + + A backslash separates on Windows only, and every one of these runs on all + three systems: `as_posix()` is the one spelling they can all be read in. + """ + def test_linux_keeps_them_apart_and_follows_xdg(self): with mock.patch.dict(os.environ, {"XDG_CONFIG_HOME": "/c", "XDG_DATA_HOME": "/d"}): config_dir, data_dir = paths.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 = paths.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 = paths.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, _ = paths.directories("darwin") - self.assertNotIn("/c", str(config_dir)) + self.assertNotIn("/c", config_dir.as_posix()) + + def test_windows_keeps_the_models_out_of_the_roaming_profile(self): + """Settings roam with the account; several gigabytes must not.""" + with mock.patch.dict(os.environ, {"APPDATA": "C:/roam", + "LOCALAPPDATA": "C:/local"}): + config_dir, data_dir = paths.directories("win32") + self.assertEqual(config_dir.as_posix(), "C:/roam/Dikte") + self.assertEqual(data_dir.as_posix(), "C:/local/Dikte") + + def test_windows_without_the_variables_set(self): + with mock.patch.dict(os.environ, {}, clear=True): + config_dir, data_dir = paths.directories("win32") + self.assertTrue(config_dir.as_posix().endswith("/AppData/Roaming/Dikte")) + self.assertTrue(data_dir.as_posix().endswith("/AppData/Local/Dikte")) class OnePlace(unittest.TestCase): diff --git a/tests/test_ui.py b/tests/test_ui.py index 192aae7..6cd402c 100644 --- a/tests/test_ui.py +++ b/tests/test_ui.py @@ -16,9 +16,12 @@ from PyQt6.QtCore import QPoint, QPointF, Qt from PyQt6.QtGui import QWheelEvent from PyQt6.QtWidgets import QApplication, QMessageBox +from dikte import audio from dikte import cleanup from dikte import config as cfg +from dikte import ggml from dikte import hotkey +from dikte import ipc from dikte import overlay as overlay_module from dikte import paste from dikte import settings_ui @@ -281,7 +284,7 @@ class Settings(DikteTest): text = self.shortcut_tab_text(window) self.assertIn("i3 keeps no shortcut registry", text) self.assertNotIn("KWin", text) - self.assertIn("__main__.py toggle", text) + self.assertIn(ipc.command_for("toggle"), text) # Not a choice to offer where it is the only mechanism there is. self.assertTrue(window.evdev_enabled.isHidden()) self.assertFalse([button for button in @@ -458,7 +461,7 @@ class MacSettings(Settings): text = self.shortcut_tab_text(window) self.assertIn("Dikte asks macOS for these combinations", text) self.assertNotIn("KWin", text) - self.assertNotIn("__main__.py toggle", text) + self.assertNotIn(ipc.command_for("toggle"), text) def test_the_paste_keys_on_offer_are_the_ones_a_mac_uses(self): window = self.window(cfg.Config()) @@ -482,7 +485,7 @@ class KdeSettings(Settings): self.assertIn("KWin only reads shortcut settings at startup", text) self.assertIn("Install as a KDE shortcut", text) self.assertNotIn("keeps no shortcut registry", text) - self.assertNotIn("__main__.py toggle", text) + self.assertNotIn(ipc.command_for("toggle"), text) # Here it is a choice: the wait for the next login, or the key press # reaching the focused application as well. self.assertFalse(window.evdev_enabled.isHidden()) @@ -610,6 +613,36 @@ class Overlay(DikteTest): self.assertFalse(widget.muted) +class MeetingSources(DikteTest): + """What the Meeting tab says about the far side, per sound system. + + The box that picks it is empty on a system that cannot record it, and an + empty box with nothing next to it reads as a list that has not loaded yet. + """ + + def notes(self, meetings): + with mock.patch.object(audio, "sound", + return_value=audio.PULSE._replace( + meetings=meetings)), \ + only_these_tools(), \ + mock.patch.object(settings_ui.SettingsWindow, "_load_models"), \ + mock.patch.object(settings_ui.SettingsWindow, + "_load_transcribe_models"): + window = settings_ui.SettingsWindow(cfg.Config()) + self.addCleanup(window.deleteLater) + self.addCleanup(window.close) + return " ".join(label.text() + for label in window.findChildren(settings_ui.QLabel)) + + def test_a_system_that_cannot_record_the_far_side_says_so(self): + self.assertIn("nothing that records what the speakers", + self.notes(meetings=False)) + + def test_a_system_that_can_says_nothing_of_the_sort(self): + self.assertNotIn("nothing that records what the speakers", + self.notes(meetings=True)) + + if __name__ == "__main__": unittest.main() @@ -617,6 +650,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)