Add Windows support

Windows joins the three systems as its own entry in each table: DirectShow
through ffmpeg for capture, the Win32 clipboard and SendInput for the paste,
RegisterHotKey for the global shortcut, and the whisper.cpp and llama.cpp
Windows zips (the OpenBLAS whisper build, which transcribes about twice as
fast on a plain CPU). Settings go to APPDATA, data to LOCALAPPDATA, and
install.ps1 adds the Start Menu entry, the dikte command and an optional
autostart. Meetings are not supported yet: Windows offers nothing to record
the far side from.

Porting surfaced three fixes that were not Windows specific:

- A stopped or overlong download tried to delete its .part file while still
  holding it open, which Windows refuses. The unlinks now wait for the handle.
- The CLI transcribed files without handing the local servers their settings
  first, so a local provider failed with "no model downloaded" wherever the
  GUI had not run in the same process.
- The audio content types are pinned instead of asked of the registry, which
  answers differently machine to machine.

One fix is Windows specific but sits in shared code: shutdown() does not end
a blocked recv there, so stopping a request also closes the socket handle.

Co-Authored-By: Claude Fable 5 <[email protected]>
This commit is contained in:
huseyin-emre-tigci
2026-08-14 16:53:21 +03:00
co-authored by Claude Fable 5
parent 77b26e76be
commit 3436e6b426
24 changed files with 1317 additions and 48 deletions
+9 -3
View File
@@ -5,9 +5,9 @@ machine by default, a model cleans it up (dropping the *uh*s, the restarts, the
missing punctuation), and the result lands in your clipboard and is pasted into missing punctuation), and the result lands in your clipboard and is pasted into
whatever window you were typing in. whatever window you were typing in.
Built for KDE Plasma 6 on Wayland, and runs on GNOME X11 and macOS too. No Built for KDE Plasma 6 on Wayland, and runs on GNOME X11, macOS and
dependencies beyond system packages: just the Python standard library, 3.11 or [Windows](README.windows.md) too. No dependencies beyond system packages: just
newer, and PyQt6. the Python standard library, 3.11 or newer, and PyQt6.
*[Türkçe README](README.tr.md)* *[Türkçe README](README.tr.md)*
@@ -61,6 +61,12 @@ that catches the keys is the mechanism there and there is nothing to run:
needs BlackHole or Loopback, because nothing else offers what the speakers are needs BlackHole or Loopback, because nothing else offers what the speakers are
playing. playing.
Windows works the same way, holding the keys through the system's own hotkey
service while Dikte runs: `winget install Gyan.FFmpeg`, `pip install PyQt6`,
then `python dikte.py`, with an optional `install.ps1` for the Start Menu entry
and the `dikte` command. Meetings are not supported there yet; the details are
in the [Windows README](README.windows.md).
`install.sh` adds the `dikte` command, a menu entry, an autostart entry and the `install.sh` adds the `dikte` command, a menu entry, an autostart entry and the
two global shortcuts, whose keys are its two arguments. `./update.sh` pulls and two global shortcuts, whose keys are its two arguments. `./update.sh` pulls and
puts all of that back, keeping the keys you chose; `./uninstall.sh` takes it away puts all of that back, keeping the keys you chose; `./uninstall.sh` takes it away
+72
View File
@@ -0,0 +1,72 @@
# Dikte on Windows
Press `Ctrl+Space`, talk, press again: what you said is transcribed, cleaned
up and pasted where your cursor is.
## Requirements
- **Windows 10/11**
- **Python 3.11+** with **PyQt6** (`pip install PyQt6`; install.ps1 installs
it when it is missing)
- **ffmpeg** for microphone capture: `winget install Gyan.FFmpeg`
## Installing
```powershell
powershell -ExecutionPolicy Bypass -File install.ps1
```
This adds a **Dikte** entry to the Start Menu and a **`dikte`** command to the
terminal. Add `-Autostart` to also start it at sign-in; `-Uninstall` removes
all of it and leaves the repository and your settings alone.
To try it without installing anything:
```sh
python dikte.py
```
## First run
1. The tray icon appears and the Settings window opens.
2. Under **API and models**, download a local whisper model (the whisper.cpp
Windows build is fetched automatically) or enter an OpenAI, Groq or
OpenRouter key.
3. The shortcut defaults to `Ctrl+Space` and is changed under Shortcuts.
While Dikte runs, Windows' own hotkey service (RegisterHotKey) listens for
it: nothing to install and no permission to grant.
## What is different from Linux and macOS
- **Meeting recording (microphone + speakers) is not supported yet.** Windows
does not offer what the speakers are playing as a capture device, so there
is nothing to record the far side from. Everything else works, including
transcribing audio and video files.
- **The shortcut is swallowed**: while Dikte holds `Ctrl+Space`, the focused
application does not see it. This is how macOS behaves too, and unlike the
Linux listener, which shares the key.
- No external tools for the clipboard or the key press: both go straight
through the Windows API (the clipboard, SendInput).
- Settings live under `%APPDATA%\Dikte`, models and recordings under
`%LOCALAPPDATA%\Dikte`.
## Performance
- The local install fetches whisper.cpp's **OpenBLAS build**, which
transcribes about twice as fast as the stock one on a plain CPU. There is
no GPU build to fetch for machines without an NVIDIA card.
- Setting Settings → API and models → **Threads** near your physical core
count helps noticeably; the server's own default is 4.
- If speed matters more than accuracy, `ggml-small` and `ggml-base` are much
faster; `ggml-large-v3-turbo-q5_0` transcribes best.
## Troubleshooting
- **Recording does not start:** does `ffmpeg -version` run? Does
`dikte devices` list your microphone?
- **Nothing is pasted:** a normal-privilege process cannot type into an
elevated (administrator) window; run Dikte elevated too, or paste by hand.
The text lands on the clipboard either way.
- **The shortcut does nothing:** another application already holds the
combination. Dikte says so in a tray notification when it asks for the key;
pick a different one under Settings → Shortcuts.
+13 -1
View File
@@ -18,6 +18,7 @@ import mimetypes
import os import os
import secrets import secrets
import socket import socket
import sys
import threading import threading
import urllib.error import urllib.error
import urllib.request import urllib.request
@@ -148,6 +149,12 @@ def _stop_using(conn):
if sock is not None: if sock is not None:
with contextlib.suppress(OSError): with contextlib.suppress(OSError):
sock.shutdown(socket.SHUT_RDWR) 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): with contextlib.suppress(OSError):
conn.close() conn.close()
@@ -258,7 +265,12 @@ def _multipart(fields, file_field, file_path):
out += str(value).encode("utf-8") + b"\r\n" out += str(value).encode("utf-8") + b"\r\n"
filename = os.path.basename(file_path) 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: with open(file_path, "rb") as fh:
payload = fh.read() payload = fh.read()
out += f"--{boundary}\r\n".encode() out += f"--{boundary}\r\n".encode()
+1
View File
@@ -378,6 +378,7 @@ def _stream(cmd, conf, on_event, should_stop):
cmd, cwd=working_dir(conf), stdin=subprocess.DEVNULL, cmd, cwd=working_dir(conf), stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, encoding="utf-8", errors="replace", bufsize=1, text=True, encoding="utf-8", errors="replace", bufsize=1,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
) )
except OSError as exc: except OSError as exc:
raise AssistantError(t("Could not run {binary}: {error}", raise AssistantError(t("Could not run {binary}: {error}",
+104 -5
View File
@@ -30,6 +30,10 @@ from PyQt6.QtCore import QObject, pyqtSignal
from i18n import t 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 RATE = 16000
CHANNELS = 1 CHANNELS = 1
SAMPLE_WIDTH = 2 # s16 SAMPLE_WIDTH = 2 # s16
@@ -39,6 +43,19 @@ CHUNK_LATENCY_MS = round(CHUNK_FRAMES / RATE * 1000)
MIN_FRAMES = int(RATE * 0.25) MIN_FRAMES = int(RATE * 0.25)
def _interrupt(proc):
"""Ask a recorder process to end.
SIGINT is the polite way everywhere it exists; Windows has no equivalent a
child can be sent, so the process is terminated outright. The captured
audio is not lost either way: it has already been read from the pipe.
"""
if sys.platform == "win32":
proc.terminate()
else:
proc.send_signal(signal.SIGINT)
class Recorder(QObject): class Recorder(QObject):
"""Runs the available sound-server recorder and reads raw PCM from stdout.""" """Runs the available sound-server recorder and reads raw PCM from stdout."""
@@ -70,7 +87,8 @@ class Recorder(QObject):
try: try:
self._proc = subprocess.Popen( self._proc = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0 cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0,
creationflags=NO_WINDOW,
) )
except OSError as exc: except OSError as exc:
self.failed.emit(t("Could not start recording: {error}", error=exc)) self.failed.emit(t("Could not start recording: {error}", error=exc))
@@ -125,7 +143,7 @@ class Recorder(QObject):
proc = self._proc proc = self._proc
if proc and proc.poll() is None: if proc and proc.poll() is None:
try: try:
proc.send_signal(signal.SIGINT) _interrupt(proc)
proc.wait(timeout=1.5) proc.wait(timeout=1.5)
except (subprocess.TimeoutExpired, OSError): except (subprocess.TimeoutExpired, OSError):
try: try:
@@ -246,7 +264,8 @@ class MeetingRecorder(QObject):
# nobody drains would eventually block it, so it writes to a file. # nobody drains would eventually block it, so it writes to a file.
self._log = tempfile.TemporaryFile() self._log = tempfile.TemporaryFile()
self._proc = subprocess.Popen( self._proc = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=self._log, bufsize=0 cmd, stdout=subprocess.PIPE, stderr=self._log, bufsize=0,
creationflags=NO_WINDOW,
) )
except (OSError, wave.Error) as exc: except (OSError, wave.Error) as exc:
self._close_file() self._close_file()
@@ -298,7 +317,7 @@ class MeetingRecorder(QObject):
proc = self._proc proc = self._proc
if proc and proc.poll() is None: if proc and proc.poll() is None:
try: try:
proc.send_signal(signal.SIGINT) _interrupt(proc)
proc.wait(timeout=2) proc.wait(timeout=2)
except (subprocess.TimeoutExpired, OSError): except (subprocess.TimeoutExpired, OSError):
try: try:
@@ -603,6 +622,71 @@ def _avfoundation_default_output():
return "" return ""
# Windows records through DirectShow, the one capture API ffmpeg's Windows
# builds all ship with. What the speakers are playing is not offered as a
# device at all, so a meeting has nothing to record the far side from yet.
def _dshow_devices():
"""[(name, name)] for every DirectShow audio capture device.
The list comes out on stderr of a command that then fails, the same
documented trick AVFoundation uses above. Names are the only stable handle
dshow offers a user; they are what the recorder is given back.
"""
if not shutil.which("ffmpeg"):
return []
try:
result = subprocess.run(
["ffmpeg", "-hide_banner", "-list_devices", "true",
"-f", "dshow", "-i", "dummy"],
capture_output=True, timeout=8, check=False, creationflags=NO_WINDOW,
)
except (subprocess.SubprocessError, OSError):
return []
devices = []
for line in result.stderr.decode("utf-8", "replace").splitlines():
if "(audio)" not in line:
continue
match = re.search(r'"([^"]+)"\s*\([^)]*audio[^)]*\)', line)
if match:
devices.append((match.group(1), match.group(1)))
return devices
def _dshow_record(target):
if not shutil.which("ffmpeg"):
return []
# dshow has no "default" device: an unset target means the first one listed.
device = target
if not device:
inputs = _dshow_devices()
if not inputs:
return []
device = inputs[0][0]
return [
"ffmpeg", "-hide_banner", "-nostdin", "-loglevel", "error",
# dshow holds half a second of audio before handing anything over;
# asked for the chunk the level meter is measured in instead.
"-f", "dshow", "-audio_buffer_size", str(CHUNK_LATENCY_MS),
"-i", f"audio={device}",
"-ac", str(CHANNELS), "-ar", str(RATE), "-f", "s16le", "-",
]
def _dshow_meeting(mic_target, system_target):
return [] # no monitor devices to record the far side from
def _dshow_no_outputs():
return []
def _dshow_no_default_output():
return ""
Sound = collections.namedtuple( Sound = collections.namedtuple(
"Sound", "Sound",
# How to capture one source and two at once, the two device lists, which # How to capture one source and two at once, the two device lists, which
@@ -633,9 +717,24 @@ COREAUDIO = Sound(
) )
DSHOW = Sound(
record=_dshow_record,
meeting=_dshow_meeting,
inputs=_dshow_devices,
outputs=_dshow_no_outputs,
default_output=_dshow_no_default_output,
missing="ffmpeg or a microphone was not found. Install ffmpeg with: "
"winget install Gyan.FFmpeg",
)
def sound(): def sound():
"""The programs this machine records through.""" """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(): def list_sources():
+1
View File
@@ -199,6 +199,7 @@ def _output(cmd, timeout, service):
cmd, cwd=os.path.expanduser("~"), stdin=subprocess.DEVNULL, cmd, cwd=os.path.expanduser("~"), stdin=subprocess.DEVNULL,
capture_output=True, text=True, encoding="utf-8", errors="replace", capture_output=True, text=True, encoding="utf-8", errors="replace",
timeout=timeout, timeout=timeout,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
) )
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
raise CleanupError(t("{service} did not finish within {seconds} seconds.", raise CleanupError(t("{service} did not finish within {seconds} seconds.",
+14
View File
@@ -17,6 +17,7 @@ import json
import os import os
import shutil import shutil
import signal import signal
import subprocess
import sys import sys
import time import time
@@ -134,6 +135,16 @@ def launch_gui(verb=""):
if verb: if verb:
args.append(verb) args.append(verb)
args.append("--gui") args.append("--gui")
if sys.platform == "win32":
# execv on Windows mangles arguments with spaces and would leave the
# application tied to this console; start it detached instead.
subprocess.Popen(
args,
creationflags=(subprocess.DETACHED_PROCESS
| subprocess.CREATE_NEW_PROCESS_GROUP),
close_fds=True,
)
sys.exit(0)
os.execv(sys.executable, args) os.execv(sys.executable, args)
@@ -297,6 +308,9 @@ def cmd_transcribe(opts):
return fail(opts, f"no such file: {path}") return fail(opts, f"no such file: {path}")
conf = cfg.Config() 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"]) timestamps = opts.srt or _pick(opts.timestamps, conf["file_timestamps"])
worker = filetranscribe.FileTranscriber(conf) worker = filetranscribe.FileTranscriber(conf)
+6
View File
@@ -27,6 +27,12 @@ def _directories(platform=None):
if (platform or sys.platform) == "darwin": if (platform or sys.platform) == "darwin":
support = pathlib.Path.home() / "Library/Application Support/Dikte" support = pathlib.Path.home() / "Library/Application Support/Dikte"
return support, support return support, support
if (platform or sys.platform) == "win32":
roaming = pathlib.Path(
os.environ.get("APPDATA") or pathlib.Path.home() / "AppData/Roaming")
local = pathlib.Path(
os.environ.get("LOCALAPPDATA") or pathlib.Path.home() / "AppData/Local")
return roaming / "Dikte", local / "Dikte"
return (_xdg("XDG_CONFIG_HOME", "~/.config") / "dikte", return (_xdg("XDG_CONFIG_HOME", "~/.config") / "dikte",
_xdg("XDG_DATA_HOME", "~/.local/share") / "dikte") _xdg("XDG_DATA_HOME", "~/.local/share") / "dikte")
+17 -1
View File
@@ -12,6 +12,7 @@ import json
import os import os
import signal import signal
import socket import socket
import subprocess
import sys import sys
import threading import threading
@@ -881,6 +882,17 @@ class Dikte:
self.settings_window.close() self.settings_window.close()
self.shutdown() self.shutdown()
QLocalServer.removeServer(SERVER_NAME) QLocalServer.removeServer(SERVER_NAME)
if sys.platform == "win32":
# execv on Windows mangles arguments with spaces and leaves the two
# processes sharing a console; a detached start does neither.
subprocess.Popen(
[sys.executable, ipc.script_path(), "--gui"],
creationflags=(subprocess.DETACHED_PROCESS
| subprocess.CREATE_NEW_PROCESS_GROUP),
close_fds=True,
)
QApplication.instance().quit()
return
os.execv(sys.executable, [sys.executable, ipc.script_path(), "--gui"]) os.execv(sys.executable, [sys.executable, ipc.script_path(), "--gui"])
def shutdown(self): def shutdown(self):
@@ -951,7 +963,11 @@ def install_signal_handlers(app):
app.quit() # aboutToQuit runs shutdown() app.quit() # aboutToQuit runs shutdown()
notifier.activated.connect(woken) 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 # A handler that does nothing, so that the default action, stopping the
# process where it stands, is replaced by the wakeup above. # process where it stands, is replaced by the wakeup above.
signal.signal(sig, lambda *_: None) signal.signal(sig, lambda *_: None)
+1
View File
@@ -283,6 +283,7 @@ def _ffmpeg(args, out, aborter=None):
["ffmpeg", "-nostdin", "-y", *args], ["ffmpeg", "-nostdin", "-y", *args],
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, 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 # 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. # doing nothing unless the abort reaches the process itself.
+77 -13
View File
@@ -43,6 +43,7 @@ import threading
import time import time
import urllib.error import urllib.error
import urllib.request import urllib.request
import zipfile
import hub import hub
from i18n import t from i18n import t
@@ -51,8 +52,15 @@ HOST = "127.0.0.1"
# The path api.py asks for, so its URL and the server's line up. # The path api.py asks for, so its URL and the server's line up.
INFERENCE_PATH = "/v1/audio/transcriptions" INFERENCE_PATH = "/v1/audio/transcriptions"
DATA_DIR = (pathlib.Path(os.environ.get("XDG_DATA_HOME") def _data_dir():
or os.path.expanduser("~/.local/share")) / "dikte") if sys.platform == "win32":
return pathlib.Path(os.environ.get("LOCALAPPDATA")
or os.path.expanduser("~/AppData/Local")) / "Dikte"
return pathlib.Path(os.environ.get("XDG_DATA_HOME")
or os.path.expanduser("~/.local/share")) / "dikte"
DATA_DIR = _data_dir()
BIN_DIR = DATA_DIR / "bin" BIN_DIR = DATA_DIR / "bin"
MODELS_DIR = DATA_DIR / "models" MODELS_DIR = DATA_DIR / "models"
@@ -151,11 +159,14 @@ def download(item, target, on_progress=None, should_stop=None, require_hash=True
try: try:
with urllib.request.urlopen(request, timeout=60) as response: with urllib.request.urlopen(request, timeout=60) as response:
total = int(response.headers.get("Content-Length") or item.size or 0) 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: with open(part, "wb") as out:
while True: while True:
if should_stop is not None and should_stop(): if should_stop is not None and should_stop():
part.unlink(missing_ok=True) stopped = True
return False break
block = response.read(DOWNLOAD_CHUNK) block = response.read(DOWNLOAD_CHUNK)
if not block: if not block:
break break
@@ -165,11 +176,17 @@ def download(item, target, on_progress=None, should_stop=None, require_hash=True
# More than was announced: a body that does not end is the # More than was announced: a body that does not end is the
# one way this loop could run until the disk is full. # one way this loop could run until the disk is full.
if total and done > total: if total and done > total:
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) part.unlink(missing_ok=True)
raise LocalError(t("{name} is longer than it said it " raise LocalError(t("{name} is longer than it said it "
"would be.", name=item.name)) "would be.", name=item.name))
if on_progress is not None:
on_progress(done, total)
# A proxy notice or an error page that came back as 200 would otherwise # 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. # be renamed into place and only fail when something tries to read it.
if total and done != total: if total and done != total:
@@ -215,9 +232,11 @@ def _has_vulkan():
llama.cpp publishes no CUDA build for Linux, so Vulkan is what a graphics 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 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 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): def _wanted_assets(program):
@@ -230,6 +249,16 @@ def _wanted_assets(program):
arch = _arch() arch = _arch()
if sys.platform == "darwin": if sys.platform == "darwin":
return () if program is WHISPER else (f"bin-macos-{arch}.tar.gz",) return () if program is WHISPER else (f"bin-macos-{arch}.tar.gz",)
if sys.platform == "win32":
if program is WHISPER:
# The BLAS build first: on a plain CPU it transcribes about twice
# as fast as the stock one, and it carries everything it needs.
# Full names, because "bin-x64.zip" alone would also match the
# CUDA archives, whichever the release happened to list first.
return ("whisper-blas-bin-x64.zip", "whisper-bin-x64.zip")
if _has_vulkan() and arch == "x64":
return ("bin-win-vulkan-x64.zip", f"bin-win-cpu-{arch}.zip")
return (f"bin-win-cpu-{arch}.zip",)
if program is LLAMA and _has_vulkan(): 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-vulkan-{arch}.tar.gz", f"bin-ubuntu-{arch}.tar.gz")
return (f"bin-ubuntu-{arch}.tar.gz",) return (f"bin-ubuntu-{arch}.tar.gz",)
@@ -275,6 +304,11 @@ def system_program(program):
return bool(shutil.which(program.binary)) 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): def _find_binary(root, name):
for path in sorted(pathlib.Path(root).rglob(name)): for path in sorted(pathlib.Path(root).rglob(name)):
if path.is_file(): if path.is_file():
@@ -283,19 +317,24 @@ def _find_binary(root, name):
def _extract(archive, into): 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 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 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: try:
if str(archive).endswith(".zip"):
with zipfile.ZipFile(archive) as bundle:
bundle.extractall(into)
return
with tarfile.open(archive, "r:gz") as tar: with tarfile.open(archive, "r:gz") as tar:
try: try:
tar.extractall(into, filter="data") tar.extractall(into, filter="data")
except TypeError: # Python without the extraction filters except TypeError: # Python without the extraction filters
tar.extractall(into) 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}", raise LocalError(t("Could not unpack {name}: {error}",
name=os.path.basename(str(archive)), error=exc)) from exc name=os.path.basename(str(archive)), error=exc)) from exc
@@ -334,7 +373,7 @@ def install_program(program, tag="", on_progress=None, should_stop=None,
if not download(item, archive, on_progress, should_stop): if not download(item, archive, on_progress, should_stop):
return "" return ""
_extract(archive, into) _extract(archive, into)
binary = _find_binary(into, program.binary) binary = _find_binary(into, _binary_file(program))
if binary is None: if binary is None:
raise LocalError(t("{name} was not in the download.", raise LocalError(t("{name} was not in the download.",
name=program.binary)) name=program.binary))
@@ -491,6 +530,26 @@ def _tail(path, lines=3):
return " | ".join(found[-lines:]) 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: class Server:
"""One process, started when something needs it and stopped when nothing does. """One process, started when something needs it and stopped when nothing does.
@@ -592,6 +651,8 @@ class Server:
args + ["--host", HOST, "--port", str(port)], args + ["--host", HOST, "--port", str(port)],
stdout=sink, stderr=subprocess.STDOUT, stdout=sink, stderr=subprocess.STDOUT,
stdin=subprocess.DEVNULL, stdin=subprocess.DEVNULL,
# No console window of its own on Windows.
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
) )
except OSError as exc: except OSError as exc:
raise LocalError(t("Could not start {name}: {error}", raise LocalError(t("Could not start {name}: {error}",
@@ -689,8 +750,11 @@ class Server:
number could belong to something else entirely, and killing it would be 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 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 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: try:
blob = pathlib.Path(f"/proc/{pid}/cmdline").read_bytes() blob = pathlib.Path(f"/proc/{pid}/cmdline").read_bytes()
except OSError: except OSError:
+186 -14
View File
@@ -418,12 +418,175 @@ def _carbon():
return 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 desktop's own shortcut -------------------------------------------
def _macos(): def _macos():
return sys.platform == "darwin" return sys.platform == "darwin"
def _windows():
return sys.platform == "win32"
def _gnome(): def _gnome():
desktop = os.environ.get("XDG_CURRENT_DESKTOP", "").lower() desktop = os.environ.get("XDG_CURRENT_DESKTOP", "").lower()
return "gnome" in desktop and shutil.which("gsettings") is not None return "gnome" in desktop and shutil.which("gsettings") is not None
@@ -548,37 +711,44 @@ def gnome_shortcut_status(desktop_id=DESKTOP_ID):
def listener(parent=None): def listener(parent=None):
"""The thing that hears the key, for whichever system this is.""" """The thing that hears the key, for whichever system this is."""
return CarbonHotkey(parent) if _macos() else EvdevHotkey(parent) if _macos():
return CarbonHotkey(parent)
if _windows():
return WinHotkey(parent)
return EvdevHotkey(parent)
def valid_shortcut(text): def valid_shortcut(text):
"""Whether this machine can bind the combination as it was typed.""" """Whether this machine can bind the combination as it was typed."""
parse = parse_macos_shortcut if _macos() else parse_shortcut if _macos():
return parse(text)[1] is not None 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(): def installs_shortcuts():
"""Whether this system keeps a shortcut registry to write into. """Whether this system keeps a shortcut registry to write into.
KDE and GNOME do, and something outside Dikte reads it, so the combination KDE and GNOME do, and something outside Dikte reads it, so the combination
survives Dikte being closed. macOS does not: there is nothing to install, survives Dikte being closed. macOS and Windows do not: there is nothing to
nothing to remove, and Settings should not offer either. install, nothing to remove, and Settings should not offer either.
""" """
return not _macos() return not _macos() and not _windows()
def shortcut_needs_restart(): def shortcut_needs_restart():
"""Whether an installed shortcut waits for the next login before it works. """Whether an installed shortcut waits for the next login before it works.
KWin reads kglobalshortcutsrc once, when it starts. GNOME picks a binding KWin reads kglobalshortcutsrc once, when it starts. GNOME picks a binding
up as it is written, and macOS never had one to write. up as it is written, and macOS and Windows never had one to write.
""" """
return not _macos() and not _gnome() return not _macos() and not _windows() and not _gnome()
def install_shortcut(shortcut, exec_command, name="Dikte: start/stop recording", def install_shortcut(shortcut, exec_command, name="Dikte: start/stop recording",
desktop_id=DESKTOP_ID): desktop_id=DESKTOP_ID):
if _macos(): if _macos() or _windows():
_REGISTERED[desktop_id] = shortcut _REGISTERED[desktop_id] = shortcut
return True, t( return True, t(
"Shortcut saved: {shortcut}\nDikte holds this one itself while it " "Shortcut saved: {shortcut}\nDikte holds this one itself while it "
@@ -591,7 +761,7 @@ def install_shortcut(shortcut, exec_command, name="Dikte: start/stop recording",
def remove_shortcut(desktop_id=DESKTOP_ID): def remove_shortcut(desktop_id=DESKTOP_ID):
if _macos(): if _macos() or _windows():
_REGISTERED.pop(desktop_id, None) _REGISTERED.pop(desktop_id, None)
elif _gnome(): elif _gnome():
remove_gnome_shortcut(desktop_id) remove_gnome_shortcut(desktop_id)
@@ -600,7 +770,7 @@ def remove_shortcut(desktop_id=DESKTOP_ID):
def shortcut_status(desktop_id=DESKTOP_ID): def shortcut_status(desktop_id=DESKTOP_ID):
if _macos(): if _macos() or _windows():
return _REGISTERED.get(desktop_id) return _REGISTERED.get(desktop_id)
return (gnome_shortcut_status(desktop_id) if _gnome() return (gnome_shortcut_status(desktop_id) if _gnome()
else kde_shortcut_status(desktop_id)) else kde_shortcut_status(desktop_id))
@@ -609,6 +779,8 @@ def shortcut_status(desktop_id=DESKTOP_ID):
def desktop_name(): def desktop_name():
if _macos(): if _macos():
return "macOS" return "macOS"
if _windows():
return "Windows"
return "GNOME" if _gnome() else "KDE" return "GNOME" if _gnome() else "KDE"
@@ -693,9 +865,9 @@ def kde_shortcut_status(desktop_id=DESKTOP_ID):
def conflicting_shortcuts(shortcut, desktop_id=DESKTOP_ID): def conflicting_shortcuts(shortcut, desktop_id=DESKTOP_ID):
"""Names of other KDE entries bound to the same combination.""" """Names of other KDE entries bound to the same combination."""
if _macos(): if _macos() or _windows():
# There is no list to read: macOS answers the question by refusing the # There is no list to read: both answer the question by refusing the
# registration, which CarbonHotkey reports when it asks for the key. # registration, which their listeners report when they ask for the key.
return [] return []
try: try:
text = SHORTCUTS_FILE.read_text(encoding="utf-8") text = SHORTCUTS_FILE.read_text(encoding="utf-8")
+10
View File
@@ -106,6 +106,10 @@ TR = {
"Ses kayıt aracı bulunamadı. pulseaudio-utils ya da pipewire-audio kur.", "Ses kayıt aracı bulunamadı. pulseaudio-utils ya da pipewire-audio kur.",
"ffmpeg not found. Install it with: brew install ffmpeg": "ffmpeg not found. Install it with: brew install ffmpeg":
"ffmpeg bulunamadı. Şununla kur: 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}": "Audio recorder stopped before receiving sound: {error}":
"Ses kayıt aracı veri alamadan kapandı: {error}", "Ses kayıt aracı veri alamadan kapandı: {error}",
"Could not copy to clipboard: {error}": "Panoya kopyalanamadı: {error}", "Could not copy to clipboard: {error}": "Panoya kopyalanamadı: {error}",
@@ -355,6 +359,12 @@ TR = {
"macOS would not give Dikte {shortcut}; another application already holds it.": "macOS would not give Dikte {shortcut}; another application already holds it.":
"macOS {shortcut} kombinasyonunu Dikte'ye vermedi; başka bir uygulama " "macOS {shortcut} kombinasyonunu Dikte'ye vermedi; başka bir uygulama "
"onu şimdiden tutuyor.", "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" "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)": " sudo usermod -aG input $USER (then log out and back in)":
"/dev/input okunamıyor. Kullanıcının 'input' grubunda olması gerekir:\n" "/dev/input okunamıyor. Kullanıcının 'input' grubunda olması gerekir:\n"
+74
View File
@@ -0,0 +1,74 @@
# Dikte'yi bu Windows kullanicisi icin kurar: Baslat Menusu kisayolu, istege
# bagli otomatik baslangic ve her yerden calisan bir `dikte` komutu.
#
# powershell -ExecutionPolicy Bypass -File install.ps1 # kur
# powershell -ExecutionPolicy Bypass -File install.ps1 -Autostart # + oturum acilisinda baslat
# powershell -ExecutionPolicy Bypass -File install.ps1 -Uninstall # kaldir
param(
[switch]$Autostart,
[switch]$Uninstall
)
$ErrorActionPreference = "Stop"
$repo = $PSScriptRoot
$startMenu = [Environment]::GetFolderPath("Programs")
$startup = [Environment]::GetFolderPath("Startup")
$shortcut = Join-Path $startMenu "Dikte.lnk"
$autostartLink = Join-Path $startup "Dikte.lnk"
# WindowsApps kullanici PATH'inde hazir durur; oraya birakilan dikte.cmd her
# terminalden calisir.
$cmdShim = Join-Path $env:LOCALAPPDATA "Microsoft\WindowsApps\dikte.cmd"
if ($Uninstall) {
foreach ($path in @($shortcut, $autostartLink, $cmdShim)) {
if (Test-Path $path) { Remove-Item $path -Force; Write-Host "silindi: $path" }
}
Write-Host "Dikte kisayollari kaldirildi. Depo klasoru ve ayarlar duruyor."
exit 0
}
# --- gereksinimler ----------------------------------------------------------
$python = Get-Command python -ErrorAction SilentlyContinue
if (-not $python) {
Write-Error "Python bulunamadi. Kurun: winget install Python.Python.3.12"
}
$version = & python -c "import sys; print('%d.%d' % sys.version_info[:2])"
if ([version]$version -lt [version]"3.11") {
Write-Error "Python 3.11+ gerekli, bulunan: $version"
}
& python -c "import PyQt6.QtWidgets" 2>$null
if ($LASTEXITCODE -ne 0) {
Write-Host "PyQt6 kuruluyor..."
& python -m pip install PyQt6
if ($LASTEXITCODE -ne 0) { Write-Error "PyQt6 kurulamadi." }
}
if (-not (Get-Command ffmpeg -ErrorAction SilentlyContinue)) {
Write-Warning "ffmpeg bulunamadi. Ses kaydi icin gerekli: winget install Gyan.FFmpeg"
}
# pythonw.exe konsol penceresi acmadan calistirir.
$pythonw = Join-Path (Split-Path $python.Source) "pythonw.exe"
if (-not (Test-Path $pythonw)) { $pythonw = $python.Source }
# --- Baslat Menusu kisayolu -------------------------------------------------
$shell = New-Object -ComObject WScript.Shell
foreach ($path in @($shortcut) + $(if ($Autostart) { @($autostartLink) } else { @() })) {
$link = $shell.CreateShortcut($path)
$link.TargetPath = $pythonw
$link.Arguments = "`"$repo\dikte.py`" --gui"
$link.WorkingDirectory = $repo
$link.Description = "Dikte: sesli dikte"
$link.Save()
Write-Host "kisayol: $path"
}
# --- dikte komutu -----------------------------------------------------------
$shimDir = Split-Path $cmdShim
if (Test-Path $shimDir) {
"@echo off`r`npython `"$repo\dikte.py`" %*" | Out-File $cmdShim -Encoding ascii
Write-Host "komut: dikte ($cmdShim)"
}
Write-Host ""
Write-Host "Kurulum tamam. Baslat Menusu'nden 'Dikte' ile ya da terminalden 'dikte' yazarak baslatin."
Write-Host "Ilk acilista Ayarlar penceresi acilir: oradan model indirin ve kisayolu secin (varsayilan Ctrl+Space)."
+3 -1
View File
@@ -14,7 +14,9 @@ import sys
from PyQt6.QtNetwork import QLocalSocket 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 # 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. # "nothing is running" is not a noticeable pause in front of a key press.
+3 -2
View File
@@ -61,9 +61,10 @@ class Overlay(QWidget):
| Qt.WindowType.Tool | Qt.WindowType.Tool
| Qt.WindowType.WindowDoesNotAcceptFocus | 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 # 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 flags |= Qt.WindowType.X11BypassWindowManagerHint
# One that can be clicked away has to receive the click, which means it # 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 # also swallows one aimed at whatever is underneath it. The rest stay
+181 -1
View File
@@ -278,6 +278,159 @@ def _macos_press(shortcut, delay):
core.CFRelease(up) core.CFRelease(up)
def _win_keys(shortcut):
"""'Ctrl+V' -> [0x11, 0x56]: Windows virtual-key codes, modifiers first."""
codes = []
for key in _keys(shortcut):
if key not in WIN_KEYCODES:
raise PasteError(t("Unknown key: {key}", key=key))
codes.append(WIN_KEYCODES[key])
return codes
# Windows virtual-key codes (winuser.h). Like Apple's, they say where the key
# sits rather than what a layout prints on it.
WIN_KEYCODES = {
"ctrl": 0x11, "control": 0x11, "shift": 0x10, "alt": 0x12,
"super": 0x5B, "meta": 0x5B,
"v": 0x56, "insert": 0x2D, "enter": 0x0D, "return": 0x0D,
}
_WIN_KEYUP = 0x0002 # KEYEVENTF_KEYUP
_WIN_CF_UNICODETEXT = 13 # what the clipboard calls UTF-16 text
_WIN_GMEM_MOVEABLE = 0x0002
@functools.lru_cache(maxsize=1)
def _win_api():
"""user32 and kernel32 with their prototypes spelled out.
The default return type is a 32-bit int, which silently truncates the
64-bit handles and pointers every one of these calls trades in.
"""
user32 = ctypes.WinDLL("user32", use_last_error=True)
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
user32.OpenClipboard.argtypes = [ctypes.c_void_p]
user32.GetClipboardData.restype = ctypes.c_void_p
user32.GetClipboardData.argtypes = [ctypes.c_uint]
user32.SetClipboardData.restype = ctypes.c_void_p
user32.SetClipboardData.argtypes = [ctypes.c_uint, ctypes.c_void_p]
kernel32.GlobalAlloc.restype = ctypes.c_void_p
kernel32.GlobalAlloc.argtypes = [ctypes.c_uint, ctypes.c_size_t]
kernel32.GlobalLock.restype = ctypes.c_void_p
kernel32.GlobalLock.argtypes = [ctypes.c_void_p]
kernel32.GlobalUnlock.argtypes = [ctypes.c_void_p]
kernel32.GlobalFree.argtypes = [ctypes.c_void_p]
return user32, kernel32
def _win_error():
"""GetLastError where it exists, so the failure paths run under any test."""
return getattr(ctypes, "get_last_error", lambda: 0)()
def _win_open_clipboard(user32):
"""The clipboard is a lock another program may hold for a moment."""
for _ in range(10):
if user32.OpenClipboard(None):
return True
time.sleep(0.01)
return False
def _win_read_text():
"""The clipboard's text, '' when it holds none, None when it cannot be read."""
user32, kernel32 = _win_api()
if not _win_open_clipboard(user32):
return None
try:
handle = user32.GetClipboardData(_WIN_CF_UNICODETEXT)
if not handle:
return ""
pointer = kernel32.GlobalLock(handle)
if not pointer:
return None
try:
return ctypes.wstring_at(pointer)
finally:
kernel32.GlobalUnlock(handle)
finally:
user32.CloseClipboard()
def _win_write_text(text):
user32, kernel32 = _win_api()
payload = str(text).encode("utf-16-le") + b"\x00\x00"
if not _win_open_clipboard(user32):
raise PasteError(t("Could not copy to clipboard: {error}",
error="the clipboard is held by another program"))
handle = None
try:
user32.EmptyClipboard()
handle = kernel32.GlobalAlloc(_WIN_GMEM_MOVEABLE, len(payload))
pointer = kernel32.GlobalLock(handle) if handle else None
if not pointer:
raise PasteError(t("Could not copy to clipboard: {error}",
error="out of memory"))
ctypes.memmove(pointer, payload, len(payload))
kernel32.GlobalUnlock(handle)
if not user32.SetClipboardData(_WIN_CF_UNICODETEXT, handle):
raise PasteError(t("Could not copy to clipboard: {error}",
error=f"error {_win_error()}"))
handle = None # the clipboard owns it now
finally:
if handle:
kernel32.GlobalFree(handle)
user32.CloseClipboard()
class _WinKeybdInput(ctypes.Structure):
_fields_ = [("wVk", ctypes.c_ushort), ("wScan", ctypes.c_ushort),
("dwFlags", ctypes.c_ulong), ("time", ctypes.c_ulong),
("dwExtraInfo", ctypes.c_size_t)]
class _WinMouseInput(ctypes.Structure):
_fields_ = [("dx", ctypes.c_long), ("dy", ctypes.c_long),
("mouseData", ctypes.c_ulong), ("dwFlags", ctypes.c_ulong),
("time", ctypes.c_ulong), ("dwExtraInfo", ctypes.c_size_t)]
class _WinInputUnion(ctypes.Union):
_fields_ = [("mi", _WinMouseInput), ("ki", _WinKeybdInput)]
class _WinInput(ctypes.Structure):
# The union carries the mouse shape too: SendInput sizes its argument by
# the biggest member whether or not it is the one being sent.
_fields_ = [("type", ctypes.c_ulong), ("union", _WinInputUnion)]
def _win_press(shortcut, delay):
"""Post the presses and releases straight into the input queue.
No permission stands in front of SendInput the way Accessibility does on
macOS: whatever window has focus receives the combination.
"""
codes = _win_keys(shortcut)
user32, _ = _win_api()
time.sleep(delay) # let the selection settle and focus come back
events = ([(code, 0) for code in codes]
+ [(code, _WIN_KEYUP) for code in reversed(codes)])
inputs = (_WinInput * len(events))()
for entry, (code, flags) in zip(inputs, events):
entry.type = 1 # INPUT_KEYBOARD
entry.union.ki = _WinKeybdInput(code, 0, flags, 0, 0)
sent = user32.SendInput(len(inputs), inputs, ctypes.sizeof(_WinInput))
if sent != len(inputs):
raise PasteError(t("Could not run {tool}: {error}", tool="SendInput",
error=f"error {_win_error()}"))
def _win_ready():
return True
# --- which of them is here ------------------------------------------------- # --- which of them is here -------------------------------------------------
Desktop = collections.namedtuple( Desktop = collections.namedtuple(
@@ -310,6 +463,17 @@ X11 = Desktop(
**_program_keyboard("xdotool", _xdotool_command), **_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( MACOS = Desktop(
clipboard="pbcopy", clipboard="pbcopy",
packages="", # both are part of macOS; there is nothing to install packages="", # both are part of macOS; there is nothing to install
@@ -331,6 +495,8 @@ def desktop():
""" """
if sys.platform == "darwin": if sys.platform == "darwin":
return MACOS return MACOS
if sys.platform == "win32":
return WINDOWS
if os.environ.get("XDG_SESSION_TYPE") == "x11": if os.environ.get("XDG_SESSION_TYPE") == "x11":
return X11 return X11
if os.environ.get("DISPLAY") and not os.environ.get("WAYLAND_DISPLAY"): if os.environ.get("DISPLAY") and not os.environ.get("WAYLAND_DISPLAY"):
@@ -375,6 +541,9 @@ def _macos_restore(snapshot):
def read_clipboard(): def read_clipboard():
here = desktop() 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"): if here is MACOS and shutil.which("osascript"):
snapshot = _macos_snapshot() snapshot = _macos_snapshot()
if snapshot is not None: if snapshot is not None:
@@ -402,6 +571,9 @@ def _run_copy(payload):
def copy(text): def copy(text):
here = desktop() here = desktop()
if here is WINDOWS:
_win_write_text(text)
return
if not shutil.which(here.clipboard): if not shutil.which(here.clipboard):
raise PasteError( raise PasteError(
t("{tool} not found. Install {packages}.", t("{tool} not found. Install {packages}.",
@@ -421,7 +593,15 @@ def copy_bytes(data):
if isinstance(data, _MAC_SNAPSHOT): if isinstance(data, _MAC_SNAPSHOT):
_macos_restore(data) _macos_restore(data)
return 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 return
try: try:
_run_copy(data) _run_copy(data)
+78
View File
@@ -606,5 +606,83 @@ class MacRecordingCommand(OnMacOS, DikteTest):
self.assertFalse(recorder.active) self.assertFalse(recorder.active)
class OnWindows:
"""A test that runs as if the machine ran Windows."""
def setUp(self):
super().setUp()
self.enterContext(mock.patch.object(sys, "platform", "win32"))
class WindowsDevices(OnWindows, DikteTest):
"""The one ffmpeg listing the device questions are answered from.
dshow names devices rather than numbering them, and the names carry
whatever alphabet the machine speaks, so the listing here does too.
"""
LISTING = (
'[dshow @ 0000020c] "Integrated Camera" (video)\n'
'[dshow @ 0000020c] Alternative name "@device_pnp_\\...."\n'
'[dshow @ 0000020c] "Mikrofon Dizisi (Intel Smart Sound)" (audio)\n'
'[dshow @ 0000020c] Alternative name "@device_cm_{33D9A762}...."\n'
'[dshow @ 0000020c] "Kulaklık (Soundcore Life Q30)" (audio)\n'
"dummy: Immediate exit requested\n"
).encode("utf-8")
@contextlib.contextmanager
def listing(self, stderr=None, tools=("ffmpeg",)):
completed = FakeCompleted(
returncode=1, stderr=self.LISTING if stderr is None else stderr)
with only_these_tools(*tools), \
mock.patch.object(subprocess, "run", return_value=completed):
yield
def test_windows_records_through_dshow(self):
self.assertIs(audio.sound(), audio.DSHOW)
def test_the_audio_lines_are_the_only_ones_read(self):
with self.listing():
self.assertEqual(audio.list_sources(), [
("Mikrofon Dizisi (Intel Smart Sound)",
"Mikrofon Dizisi (Intel Smart Sound)"),
("Kulaklık (Soundcore Life Q30)",
"Kulaklık (Soundcore Life Q30)"),
])
def test_no_ffmpeg_installed(self):
with only_these_tools():
self.assertEqual(audio.list_sources(), [])
self.assertEqual(audio.recording_command(), [])
def test_the_name_is_what_the_recorder_is_given_back(self):
with self.listing():
cmd = audio.recording_command("Kulaklık (Soundcore Life Q30)")
self.assertEqual(cmd[cmd.index("-f") + 1], "dshow")
self.assertIn("audio=Kulaklık (Soundcore Life Q30)", cmd)
def test_no_microphone_named_means_the_first_one_listed(self):
"""dshow has no default device for an empty target to mean."""
with self.listing():
self.assertIn("audio=Mikrofon Dizisi (Intel Smart Sound)",
audio.recording_command())
def test_a_machine_with_no_microphone_at_all(self):
with self.listing(stderr=b'[dshow @ 0] "Integrated Camera" (video)\n'):
self.assertEqual(audio.recording_command(), [])
def test_an_ffmpeg_that_will_not_run(self):
with only_these_tools("ffmpeg"), \
mock.patch.object(subprocess, "run", side_effect=OSError("nope")):
self.assertEqual(audio.list_sources(), [])
def test_nothing_offers_the_far_side_of_a_meeting(self):
"""What the speakers play is not a capture device Windows hands out."""
with self.listing():
self.assertEqual(audio.list_monitors(), [])
self.assertEqual(audio.default_monitor(), "")
self.assertEqual(audio.meeting_command("mic", "sys"), [])
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+23
View File
@@ -14,6 +14,7 @@ from unittest import mock
import cli import cli
import config as cfg import config as cfg
import ggml
import hotkey import hotkey
import ipc import ipc
from tests.support import DikteTest, fake_urlopen from tests.support import DikteTest, fake_urlopen
@@ -564,5 +565,27 @@ class Replies(DikteTest):
self.assertEqual(cli.run(["cancel"]), 0) self.assertEqual(cli.run(["cancel"]), 0)
class TranscribeRunsHere(DikteTest):
"""`dikte transcribe` runs in this process, not in the instance."""
def test_the_local_servers_are_handed_the_settings_first(self):
# The GUI does this at startup; a CLI run has no GUI to have done it,
# and without it the whisper server holds an empty model name.
wav = self.path("clip.wav")
wav.write_bytes(b"RIFF not really audio")
self.write_config({"local_model": "ggml-base.bin"})
self.addCleanup(ggml.whisper.configure,
model="", threads=0, gpu=True, binary="")
opts = cli.build_parser().parse_args(["transcribe", str(wav)])
with mock.patch.object(cli.filetranscribe, "FileTranscriber"), \
mock.patch.object(cli, "_headless",
return_value={"error": "stopped"}), \
captured():
cli.cmd_transcribe(opts)
self.assertEqual(ggml.whisper.settings()["model"], "ggml-base.bin")
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+20 -6
View File
@@ -8,6 +8,7 @@ config and now shadows the default.
import json import json
import os import os
import sys
import unittest import unittest
from unittest import mock from unittest import mock
@@ -89,6 +90,8 @@ class Saving(DikteTest):
cfg.Config().save() cfg.Config().save()
self.assertTrue(cfg.CONFIG_FILE.exists()) 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): def test_the_file_is_readable_by_nobody_else(self):
"""It holds two API keys.""" """It holds two API keys."""
cfg.Config().save() cfg.Config().save()
@@ -467,25 +470,33 @@ class Directories(unittest.TestCase):
with mock.patch.dict(os.environ, {"XDG_CONFIG_HOME": "/c", with mock.patch.dict(os.environ, {"XDG_CONFIG_HOME": "/c",
"XDG_DATA_HOME": "/d"}): "XDG_DATA_HOME": "/d"}):
config_dir, data_dir = cfg._directories("linux") config_dir, data_dir = cfg._directories("linux")
self.assertEqual(str(config_dir), "/c/dikte") self.assertEqual(config_dir.as_posix(), "/c/dikte")
self.assertEqual(str(data_dir), "/d/dikte") self.assertEqual(data_dir.as_posix(), "/d/dikte")
def test_linux_without_the_variables_set(self): def test_linux_without_the_variables_set(self):
with mock.patch.dict(os.environ, {}, clear=True): with mock.patch.dict(os.environ, {}, clear=True):
config_dir, data_dir = cfg._directories("linux") config_dir, data_dir = cfg._directories("linux")
self.assertTrue(str(config_dir).endswith("/.config/dikte")) self.assertTrue(config_dir.as_posix().endswith("/.config/dikte"))
self.assertTrue(str(data_dir).endswith("/.local/share/dikte")) self.assertTrue(data_dir.as_posix().endswith("/.local/share/dikte"))
def test_a_mac_keeps_both_in_application_support(self): def test_a_mac_keeps_both_in_application_support(self):
config_dir, data_dir = cfg._directories("darwin") config_dir, data_dir = cfg._directories("darwin")
self.assertEqual(config_dir, data_dir) 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): 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.""" """A Mac with them set from some other tool still stores in one place."""
with mock.patch.dict(os.environ, {"XDG_CONFIG_HOME": "/c"}): with mock.patch.dict(os.environ, {"XDG_CONFIG_HOME": "/c"}):
config_dir, _ = cfg._directories("darwin") config_dir, _ = cfg._directories("darwin")
self.assertNotIn("/c", str(config_dir)) self.assertNotIn("/c", config_dir.as_posix())
def test_windows_keeps_settings_and_data_apart(self):
with mock.patch.dict(os.environ, {"APPDATA": r"C:\roam",
"LOCALAPPDATA": r"C:\local"}):
config_dir, data_dir = cfg._directories("win32")
self.assertEqual(config_dir.as_posix(), "C:/roam/Dikte")
self.assertEqual(data_dir.as_posix(), "C:/local/Dikte")
if __name__ == "__main__": if __name__ == "__main__":
@@ -519,6 +530,9 @@ class ReadyToRun(DikteTest):
def setUp(self): def setUp(self):
super().setUp() super().setUp()
self.patch_attr(ggml, "MODELS_DIR", self.path("models")) 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): def install(self, name):
path = ggml.whisper_model_path(name) path = ggml.whisper_model_path(name)
+85
View File
@@ -15,6 +15,7 @@ import tarfile
import textwrap import textwrap
import threading import threading
import time import time
import zipfile
from unittest import mock from unittest import mock
import ggml import ggml
@@ -77,6 +78,15 @@ def tarball(entries):
return buf.getvalue() 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): class Local(DikteTest):
"""A test with its own bin, models and cache directories.""" """A test with its own bin, models and cache directories."""
@@ -681,3 +691,78 @@ class Sizes(DikteTest):
self.assertEqual(ggml.human_size(512), "512 B") self.assertEqual(ggml.human_size(512), "512 B")
self.assertEqual(ggml.human_size(574041195), "547.4 MB") self.assertEqual(ggml.human_size(574041195), "547.4 MB")
self.assertEqual(ggml.human_size(3_095_033_483), "2.9 GB") self.assertEqual(ggml.human_size(3_095_033_483), "2.9 GB")
# --- Windows ----------------------------------------------------------------
class WindowsAssets(Local):
"""Which archive a Windows machine is handed."""
def setUp(self):
super().setUp()
self.patch_attr(sys, "platform", "win32")
self.patch_attr(ggml, "_arch", lambda: "x64")
def test_whisper_prefers_the_blas_build(self):
# On a plain CPU it transcribes about twice as fast as the stock one.
self.assertEqual(ggml._wanted_assets(ggml.WHISPER),
("whisper-blas-bin-x64.zip", "whisper-bin-x64.zip"))
def test_llama_takes_the_vulkan_build_when_there_is_a_loader(self):
self.patch_attr(ggml, "_has_vulkan", lambda: True)
self.assertEqual(ggml._wanted_assets(ggml.LLAMA),
("bin-win-vulkan-x64.zip", "bin-win-cpu-x64.zip"))
def test_llama_falls_back_to_the_cpu_build_without_one(self):
self.patch_attr(ggml, "_has_vulkan", lambda: False)
self.assertEqual(ggml._wanted_assets(ggml.LLAMA),
("bin-win-cpu-x64.zip",))
def test_an_arm_machine_is_not_handed_the_x64_build(self):
self.patch_attr(ggml, "_arch", lambda: "arm64")
self.patch_attr(ggml, "_has_vulkan", lambda: True)
self.assertEqual(ggml._wanted_assets(ggml.LLAMA),
("bin-win-cpu-arm64.zip",))
class InstallOnWindows(Local):
"""The Windows releases are zips, and the binary carries .exe."""
def setUp(self):
super().setUp()
self.patch_attr(sys, "platform", "win32")
self.patch_attr(ggml, "_arch", lambda: "x64")
self.archive = zipball({
"Release/whisper-server.exe": b"MZ not really a program",
"Release/whisper.dll": b"not really a library",
})
def release(self, *names):
digest = hashlib.sha256(self.archive)
return {"tag_name": "v1.9.1", "assets": [
{"name": name, "browser_download_url": f"https://example.invalid/{name}",
"size": 10, "digest": "sha256:" + digest.hexdigest()}
for name in names]}
def test_the_zip_lands_and_the_exe_inside_it_is_found(self):
with serving(self.release("whisper-blas-bin-x64.zip"), self.archive):
path = ggml.install_program(ggml.WHISPER)
self.assertTrue(path.endswith("whisper-server.exe"))
self.assertTrue(os.path.isfile(path))
self.assertTrue(os.path.isfile(os.path.join(os.path.dirname(path),
"whisper.dll")))
self.assertEqual(ggml.installed_program(ggml.WHISPER), path)
def test_the_blas_build_is_the_one_fetched_when_both_are_offered(self):
listing = self.release("whisper-bin-x64.zip", "whisper-blas-bin-x64.zip")
with serving(listing, self.archive) as calls:
ggml.install_program(ggml.WHISPER)
urls = [call.args[0].full_url for call in calls.call_args_list]
self.assertTrue(urls[1].endswith("whisper-blas-bin-x64.zip"))
def test_a_release_with_nothing_for_windows_says_so(self):
with fake_urlopen(json_body(self.release("whisper-bin-ubuntu-x64.tar.gz"))):
with self.assertRaises(ggml.LocalError) as caught:
ggml.install_program(ggml.WHISPER)
self.assertIn("this machine", str(caught.exception))
+203
View File
@@ -2,10 +2,14 @@
import contextlib import contextlib
import os import os
import queue
import subprocess import subprocess
import time
import unittest import unittest
from unittest import mock from unittest import mock
from PyQt6.QtCore import Qt
import config as cfg import config as cfg
import hotkey import hotkey
from tests.support import DikteTest, FakeCompleted, linux_only from tests.support import DikteTest, FakeCompleted, linux_only
@@ -681,5 +685,204 @@ class MacChooser(DikteTest):
self.assertFalse(hotkey.valid_shortcut("Cmd+Space")) 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__": if __name__ == "__main__":
unittest.main() unittest.main()
+127
View File
@@ -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. cannot quietly break the platform nobody is sitting at.
""" """
import ctypes
import os import os
import pathlib import pathlib
import subprocess import subprocess
@@ -56,6 +57,9 @@ class Chooser(DikteTest):
def test_a_mac(self): def test_a_mac(self):
self.assertIs(self.under("darwin"), paste.MACOS) 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): 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.""" """XQuartz sets DISPLAY, and none of X's programs are what pastes here."""
self.assertIs(self.under("darwin", DISPLAY=":0"), paste.MACOS) self.assertIs(self.under("darwin", DISPLAY=":0"), paste.MACOS)
@@ -429,5 +433,128 @@ class MacClipboardSnapshot(DikteTest):
self.assertFalse(os.path.exists(directory)) self.assertFalse(os.path.exists(directory))
class FakeWin32:
"""user32 and kernel32, as much of both as paste.py calls.
The clipboard is a string held here. A read materialises it as this
machine's own wide characters, which is what wstring_at reads wherever the
test runs; a write arrives as the UTF-16 the real clipboard is handed, so
what the code sent is exactly what is checked.
"""
def __init__(self):
self.text = None
self.buffers = {}
self.next_handle = 1
self.pressed = [] # (virtual key, flags), in the order sent
self.send_result = None # None: report every event as delivered
def _keep(self, buffer):
handle = self.next_handle
self.next_handle += 1
self.buffers[handle] = buffer
return handle
# --- user32
def OpenClipboard(self, owner):
return 1
def CloseClipboard(self):
return 1
def EmptyClipboard(self):
self.text = None
return 1
def GetClipboardData(self, fmt):
if self.text is None:
return 0
return self._keep(ctypes.create_unicode_buffer(self.text))
def SetClipboardData(self, fmt, handle):
raw = self.buffers[handle].raw
self.text = raw.decode("utf-16-le").split("\x00", 1)[0]
return handle
def SendInput(self, count, inputs, size):
self.pressed.extend((entry.union.ki.wVk, entry.union.ki.dwFlags)
for entry in inputs)
return count if self.send_result is None else self.send_result
# --- kernel32
def GlobalAlloc(self, flags, size):
return self._keep(ctypes.create_string_buffer(size))
def GlobalLock(self, handle):
buffer = self.buffers.get(handle)
return ctypes.addressof(buffer) if buffer else 0
def GlobalUnlock(self, handle):
return 1
def GlobalFree(self, handle):
self.buffers.pop(handle, None)
return 1
class Windows(Standing, DikteTest):
"""Windows shells out to nothing: both halves are calls into the system."""
platform = "win32"
here = paste.WINDOWS
def setUp(self):
super().setUp()
self.api = FakeWin32()
self.patch_attr(paste, "_win_api", lambda: (self.api, self.api))
self.patch_attr(paste.time, "sleep", lambda seconds: None)
def test_what_is_copied_is_what_reads_back(self):
paste.copy("ığüşöç İ")
self.assertEqual(paste.read_clipboard(), "ığüşöç İ".encode("utf-8"))
def test_an_empty_clipboard_reads_as_empty_text(self):
self.assertEqual(paste.read_clipboard(), b"")
def test_what_was_saved_goes_back_after_the_paste(self):
paste.copy("mine")
saved = paste.read_clipboard()
paste.copy("the dictation")
paste.copy_bytes(saved)
self.assertEqual(self.api.text, "mine")
def test_readiness_asks_for_no_program_and_no_permission(self):
with only_these_tools():
self.assertTrue(paste.paste_ready())
def test_the_keys_go_down_in_order_and_up_in_reverse(self):
paste.press("ctrl+v")
keyup = 0x0002
self.assertEqual(self.api.pressed,
[(0x11, 0), (0x56, 0), (0x56, keyup), (0x11, keyup)])
def test_three_keys(self):
paste.press("ctrl+shift+v")
self.assertEqual([code for code, _ in self.api.pressed],
[0x11, 0x10, 0x56, 0x56, 0x10, 0x11])
def test_the_other_spellings_land_on_the_same_keys(self):
paste.press("super+enter")
first, self.api.pressed = self.api.pressed, []
paste.press("meta+return")
self.assertEqual(self.api.pressed, first)
def test_a_key_nobody_mapped_is_refused_before_anything_is_sent(self):
with self.assertRaises(paste.PasteError):
paste.press("ctrl+f13")
self.assertEqual(self.api.pressed, [])
def test_a_press_the_system_did_not_take_says_so(self):
self.api.send_result = 0
with self.assertRaises(paste.PasteError) as caught:
paste.press("ctrl+v")
self.assertIn("SendInput", str(caught.exception))
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+8
View File
@@ -15,6 +15,7 @@ from PyQt6.QtWidgets import QApplication, QMessageBox
import cleanup import cleanup
import config as cfg import config as cfg
import ggml
import hotkey import hotkey
import overlay as overlay_module import overlay as overlay_module
import paste import paste
@@ -449,6 +450,13 @@ if __name__ == "__main__":
class LocalModels(DikteTest): class LocalModels(DikteTest):
"""The download boxes, without a network and without either program.""" """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): def window(self, conf):
window = settings_ui.SettingsWindow(conf) window = settings_ui.SettingsWindow(conf)
self.addCleanup(window.deleteLater) self.addCleanup(window.deleteLater)