Add modifier-key input scheme (whisper-key style)
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 <noreply@anthropic.com>
This commit is contained in:
parent
3026f1ff90
commit
a104f2cfc4
@ -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 = "<ctrl>+<cmd>" # start recording (Ctrl+Win)
|
||||
key_stop: str = "<ctrl>" # stop -> paste
|
||||
key_send: str = "<alt>" # stop -> paste -> Enter
|
||||
key_cancel: str = "<esc>" # 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", "<ctrl>+<alt>+<space>"),
|
||||
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", "<ctrl>+<cmd>"),
|
||||
key_stop=inp.get("stop", "<ctrl>"),
|
||||
key_send=inp.get("send", "<alt>"),
|
||||
key_cancel=inp.get("cancel", "<esc>"),
|
||||
)
|
||||
|
||||
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 = "<ctrl>+<cmd>" # <cmd> = the Super/Windows key
|
||||
stop = "<ctrl>"
|
||||
send = "<alt>"
|
||||
cancel = "<esc>"
|
||||
|
||||
[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)
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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()
|
||||
|
||||
|
||||
112
linux/blitztext/inputmode.py
Normal file
112
linux/blitztext/inputmode.py
Normal file
@ -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','cmd'}; '<ctrl>' -> {'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"
|
||||
@ -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"]):
|
||||
|
||||
Loading…
Reference in New Issue
Block a user