From a104f2cfc4ea424e3befcd415adf24c73f476208 Mon Sep 17 00:00:00 2001 From: mARTin-B78 Date: Thu, 4 Jun 2026 20:26:59 +0200 Subject: [PATCH] Add modifier-key input scheme (whisper-key style) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New inputmode.py: Ctrl+Win start, Ctrl stop+paste, Alt stop+paste+Enter, Esc cancel; toggle and push-to-talk variants. Low-level pynput Listener with arming so stop/send/cancel only fire while recording and only after the start chord is released — a stray Ctrl tap while idle does nothing (unit-tested). Daemon split into start/finish/cancel_dictation (+ send_enter -> press Enter), with start_input/stop_input dispatching between "modifiers" and "hotkeys" modes. New [input] config section; paste.press_enter for auto-send. Default mode is "modifiers"; voice-keyword routing still applies to what you say. Co-Authored-By: Claude Opus 4.8 --- linux/blitztext/config.py | 34 +++++++++++ linux/blitztext/daemon.py | 101 +++++++++++++++++++++++-------- linux/blitztext/gtkui.py | 4 +- linux/blitztext/inputmode.py | 112 +++++++++++++++++++++++++++++++++++ linux/blitztext/paste.py | 9 +++ 5 files changed, 234 insertions(+), 26 deletions(-) create mode 100644 linux/blitztext/inputmode.py diff --git a/linux/blitztext/config.py b/linux/blitztext/config.py index e2415c1..96e62bb 100644 --- a/linux/blitztext/config.py +++ b/linux/blitztext/config.py @@ -38,6 +38,13 @@ class Config: type_delay_ms: int = 12 notify: bool = True language: str = "de" # whisper hint; "" = autodetect + # input scheme + input_mode: str = "modifiers" # "modifiers" (Ctrl+Win/Ctrl/Alt/Esc) | "hotkeys" (combos) + push_to_talk: bool = False + key_start: str = "+" # start recording (Ctrl+Win) + key_stop: str = "" # stop -> paste + key_send: str = "" # stop -> paste -> Enter + key_cancel: str = "" # discard # whisper model: str = "small" device: str = "auto" # auto | cuda | cpu @@ -112,6 +119,7 @@ def load(path: Path = CONFIG_PATH) -> Config: w = data.get("whisper", {}) r = data.get("rewrite", {}) rt = data.get("routing", {}) + inp = data.get("input", {}) cfg = Config( recorder=g.get("recorder", "auto"), @@ -132,6 +140,12 @@ def load(path: Path = CONFIG_PATH) -> Config: routing_hotkey=rt.get("hotkey", "++"), routing_default=rt.get("default", ""), routing_threshold=float(rt.get("threshold", 0.82)), + input_mode=inp.get("mode", "modifiers"), + push_to_talk=bool(inp.get("push_to_talk", False)), + key_start=inp.get("start", "+"), + key_stop=inp.get("stop", ""), + key_send=inp.get("send", ""), + key_cancel=inp.get("cancel", ""), ) for entry in data.get("workflow", []): @@ -209,6 +223,14 @@ def save(cfg: Config, path: Path = CONFIG_PATH) -> None: "temperature": cfg.temperature, "timeout": cfg.timeout, }, + "input": { + "mode": cfg.input_mode, + "push_to_talk": cfg.push_to_talk, + "start": cfg.key_start, + "stop": cfg.key_stop, + "send": cfg.key_send, + "cancel": cfg.key_cancel, + }, "routing": { "enabled": cfg.routing_enabled, "hotkey": cfg.routing_hotkey, @@ -272,6 +294,18 @@ type_delay_ms = 12 # per-keystroke delay for xdotool type (raise if chars notify = true # desktop notifications for each phase language = "de" # Whisper language hint; "" = autodetect +[input] +# How you start/stop dictation. +# mode = "modifiers" -> Ctrl+Win start | Ctrl stop+paste | Alt stop+paste+Enter | Esc cancel +# mode = "hotkeys" -> the per-preset combos + the [routing] hotkey below +# Voice-keyword routing still applies to what you say in either mode. +mode = "modifiers" +push_to_talk = false # modifiers mode: hold Start to record, release to stop+paste +start = "+" # = the Super/Windows key +stop = "" +send = "" +cancel = "" + [whisper] model = "small" # tiny | base | small | medium | large-v3, or a local path device = "auto" # auto | cuda | cpu (auto tries cuda, falls back to cpu) diff --git a/linux/blitztext/daemon.py b/linux/blitztext/daemon.py index 7fa0e00..d6b34e4 100644 --- a/linux/blitztext/daemon.py +++ b/linux/blitztext/daemon.py @@ -81,35 +81,57 @@ class Daemon: def is_recording(self) -> bool: return self._recording is not None - # -- hotkey / button handler ---------------------------------------------- - def toggle(self, workflow: Workflow) -> None: - """Called on each trigger: start recording, or stop + process.""" + # -- recording control ---------------------------------------------------- + def start_dictation(self, workflow: Workflow | None = None) -> None: + wf = workflow or self._route_workflow with self._lock: - if not self.ready: - self._notify("Please wait", "Model still loading…", "low") - return - if self._busy: - self._notify("Busy", "Still processing the last clip…", "low") + if not self.ready or self._busy or self._recording is not None: return + self._target_window = active_window_id() + self._recording = Recording(self.recorder_name) + self._active_workflow = wf + self._emit("recording", wf.name, "Recording…") + self._notify(f"● {wf.name}", "Recording…") + def finish_dictation(self, send_enter: bool = False) -> None: + with self._lock: if self._recording is None: - self._target_window = active_window_id() - self._recording = Recording(self.recorder_name) - self._active_workflow = workflow - self._emit("recording", workflow.name, "Recording…") - self._notify(f"● {workflow.name}", "Recording… trigger again to stop.") return - rec, wf, win = self._recording, self._active_workflow, self._target_window self._recording = None self._active_workflow = None self._busy = True - audio_path = rec.stop() - threading.Thread(target=self._process, args=(audio_path, wf, win), daemon=True).start() + threading.Thread( + target=self._process, args=(audio_path, wf, win, send_enter), daemon=True + ).start() + + def cancel_dictation(self) -> None: + with self._lock: + if self._recording is None: + return + rec = self._recording + self._recording = None + self._active_workflow = None + rec.discard() + self._emit("idle", None, "Cancelled") + self._notify("Cancelled", "Recording discarded.", "low") + + def toggle(self, workflow: Workflow) -> None: + """Start recording, or stop + process (used by GUI clicks and combos).""" + if not self.ready: + self._notify("Please wait", "Model still loading…", "low") + return + if self._busy: + self._notify("Busy", "Still processing the last clip…", "low") + return + if self._recording is None: + self.start_dictation(workflow) + else: + self.finish_dictation(send_enter=False) # -- worker --------------------------------------------------------------- - def _process(self, audio_path, workflow: Workflow, window_id) -> None: + def _process(self, audio_path, workflow: Workflow, window_id, send_enter: bool = False) -> None: label = workflow.name try: self._emit("busy", label, "Transcribing…") @@ -172,6 +194,9 @@ class Daemon: window_id=window_id, type_delay_ms=self.cfg.type_delay_ms, ) + if send_enter: + from .paste import press_enter + press_enter(window_id) self._emit("done", label, text) self._notify(f"✓ {label}", text[:80] + ("…" if len(text) > 80 else "")) except Exception as exc: # noqa: BLE001 - surface any failure @@ -203,6 +228,29 @@ class Daemon: self._listener.start() return self._listener + def start_input(self): + """Start the configured input handler; returns its listener (joinable).""" + if self.cfg.input_mode == "modifiers": + from .inputmode import ModifierScheme + + self._scheme = ModifierScheme( + self, + start=self.cfg.key_start, + stop=self.cfg.key_stop, + send=self.cfg.key_send, + cancel=self.cfg.key_cancel, + push_to_talk=self.cfg.push_to_talk, + ) + return self._scheme.start_listener() + return self.start_hotkeys() + + def stop_input(self) -> None: + scheme = getattr(self, "_scheme", None) + if scheme is not None: + scheme.stop_listener() + self._scheme = None + self.stop_hotkeys() + def stop_hotkeys(self) -> None: if self._listener is not None: self._listener.stop() @@ -211,12 +259,17 @@ class Daemon: # -- headless run loop ---------------------------------------------------- def run(self) -> None: self.prepare() - lines = [f" {self.cfg.routing_hotkey} → Voice routing (speak a keyword)"] if self.cfg.routing_enabled else [] - lines += [f" {wf.hotkey} → {wf.name}" for wf in self.cfg.workflows if wf.hotkey] - print(f"[blitztext] ready. Recorder: {self.recorder_name}. Hotkeys:\n" + "\n".join(lines), file=sys.stderr) - self._notify("Blitztext ready", "Focus a text field and press a hotkey.") + if self.cfg.input_mode == "modifiers": + scheme = "Ctrl+Win start · Ctrl stop+paste · Alt stop+paste+Enter · Esc cancel" + print(f"[blitztext] ready. Recorder: {self.recorder_name}. Input: {scheme}", file=sys.stderr) + else: + lines = [f" {self.cfg.routing_hotkey} → Voice routing"] if self.cfg.routing_enabled else [] + lines += [f" {wf.hotkey} → {wf.name}" for wf in self.cfg.workflows if wf.hotkey] + print(f"[blitztext] ready. Recorder: {self.recorder_name}. Hotkeys:\n" + "\n".join(lines), file=sys.stderr) + self._notify("Blitztext ready", "Focus a text field and start dictating.") - from pynput import keyboard - - with keyboard.GlobalHotKeys(self._build_mapping()) as listener: + listener = self.start_input() + try: listener.join() + finally: + self.stop_input() diff --git a/linux/blitztext/gtkui.py b/linux/blitztext/gtkui.py index 6b5a908..1679319 100644 --- a/linux/blitztext/gtkui.py +++ b/linux/blitztext/gtkui.py @@ -189,7 +189,7 @@ class App: def _startup(self) -> None: try: self.daemon.prepare() - self.daemon.start_hotkeys() + self.daemon.start_input() except Exception as exc: # noqa: BLE001 GLib.idle_add(self._apply_status, "error", None, f"Startup failed: {exc}") @@ -274,7 +274,7 @@ class App: def quit_all(self) -> None: try: - self.daemon.stop_hotkeys() + self.daemon.stop_input() finally: Gtk.main_quit() diff --git a/linux/blitztext/inputmode.py b/linux/blitztext/inputmode.py new file mode 100644 index 0000000..cee29a6 --- /dev/null +++ b/linux/blitztext/inputmode.py @@ -0,0 +1,112 @@ +"""Modifier-key input scheme (whisper-key style), as an alternative to combos. + +Default interaction: + Ctrl+Win start recording + Ctrl stop -> transcribe -> paste + Alt stop -> transcribe -> paste -> Enter (auto-send) + Esc cancel (discard, no transcription) + +Bare modifier taps are risky (a stray Ctrl+C could misfire), so stop/send/cancel +are only armed *while recording* and only after the start modifiers are released. +A push-to-talk variant records while the start chord is held and stops on release. + +Uses a low-level pynput Listener (press/release), not GlobalHotKeys. +""" + +from __future__ import annotations + + +def _token(key) -> str | None: + """Canonical token for a pynput key: ctrl/alt/cmd/shift/esc or a char.""" + from pynput import keyboard + + if isinstance(key, keyboard.Key): + name = key.name + for mod in ("ctrl", "alt", "cmd", "shift"): + if name.startswith(mod): + return mod + return name # 'esc', 'space', 'enter', ... + if isinstance(key, keyboard.KeyCode) and key.char: + return key.char.lower() + return None + + +def parse_tokens(spec: str) -> frozenset[str]: + """'+' -> {'ctrl','cmd'}; '' -> {'ctrl'}.""" + return frozenset(p.strip().strip("<>").lower() for p in spec.split("+") if p.strip()) + + +class ModifierScheme: + def __init__(self, daemon, *, start: str, stop: str, send: str, cancel: str, push_to_talk: bool = False): + self.daemon = daemon + self.start = parse_tokens(start) + self.stop = parse_tokens(stop) + self.send = parse_tokens(send) + self.cancel = parse_tokens(cancel) + self.ptt = push_to_talk + self._pressed: set[str] = set() + self._state = "idle" # idle | arming | armed + self._listener = None + + # -- listener lifecycle --------------------------------------------------- + def start_listener(self): + from pynput import keyboard + + self._listener = keyboard.Listener(on_press=self._on_press, on_release=self._on_release) + self._listener.start() + return self._listener + + def stop_listener(self) -> None: + if self._listener is not None: + self._listener.stop() + self._listener = None + + # -- event handling ------------------------------------------------------- + def _is(self, token: str, combo: frozenset[str]) -> bool: + # A single-token trigger fires on that token; a chord needs all tokens held. + if len(combo) == 1: + return token in combo + return combo.issubset(self._pressed) + + def _on_press(self, key) -> None: + token = _token(key) + if token is None: + return + self._pressed.add(token) + + if self._state == "idle": + if self.start.issubset(self._pressed): + self._state = "arming" + self.daemon.start_dictation() + return + + # Cancel works while arming or armed. + if self._is(token, self.cancel): + self._state = "idle" + self.daemon.cancel_dictation() + return + + if self._state == "armed": + if self._is(token, self.send): + self._state = "idle" + self.daemon.finish_dictation(send_enter=True) + elif self._is(token, self.stop): + self._state = "idle" + self.daemon.finish_dictation(send_enter=False) + + def _on_release(self, key) -> None: + token = _token(key) + if token is not None: + self._pressed.discard(token) + + if self._state != "arming": + return + if self.ptt: + # Push-to-talk: releasing the start chord stops and pastes. + if not self.start.issubset(self._pressed): + self._state = "idle" + self.daemon.finish_dictation(send_enter=False) + else: + # Toggle: arm stop/send once the start modifiers are all released. + if not (self._pressed & self.start): + self._state = "armed" diff --git a/linux/blitztext/paste.py b/linux/blitztext/paste.py index d4d0b14..6c8c9a3 100644 --- a/linux/blitztext/paste.py +++ b/linux/blitztext/paste.py @@ -54,6 +54,15 @@ def deliver(text: str, *, mode: str = "type", window_id: str | None = None, type ) +def press_enter(window_id: str | None = None) -> None: + """Send Return to the focused/target window (auto-send after paste).""" + if not shutil.which("xdotool"): + return + _focus(window_id) + time.sleep(0.08) + subprocess.run(["xdotool", "key", "--clearmodifiers", "Return"], check=False) + + def _set_clipboard(text: str) -> bool: """Best-effort clipboard set; returns False if no clipboard tool is available.""" for argv in (["xclip", "-selection", "clipboard"], ["xsel", "--clipboard", "--input"], ["wl-copy"]):