From 44ca0feabdb627c44f2959e4daf7b52e0a9de569 Mon Sep 17 00:00:00 2001 From: mARTin-B78 Date: Thu, 4 Jun 2026 23:13:36 +0200 Subject: [PATCH] Redesign Settings: preset/engine managers, mic + level meter, autostart New GTK settings with a dropdown+editor pattern across four tabs: - Presets: select/add/delete prompt presets; edit name, keywords, hotkey, mode, per-preset model/temperature, and a clearly-bordered prompt textfield. - Engines: STT and LLM engine managers with green/red online-offline status dots, add/edit/delete, active selector, and an STT record-and-benchmark Test. - Input: input scheme + keys + quality-gate settings. - General: microphone picker with a live sounddevice level meter, output, language, notifications, launch-on-login toggle, and local-Whisper settings. Adds audio.py (mic enumeration via pactl + LevelMeter) and autostart.py (~/.config/autostart entry). Recorder + daemon honour the selected mic. Requirements gain sounddevice + tomli-w. Co-Authored-By: Claude Opus 4.8 --- linux/blitztext/audio.py | 94 ++++++ linux/blitztext/autostart.py | 44 +++ linux/blitztext/config.py | 3 + linux/blitztext/daemon.py | 2 +- linux/blitztext/gtksettings.py | 549 +++++++++++++++++++++++++-------- linux/blitztext/gtkui.py | 2 +- linux/blitztext/recorder.py | 15 +- linux/requirements.txt | 2 + 8 files changed, 582 insertions(+), 129 deletions(-) create mode 100644 linux/blitztext/audio.py create mode 100644 linux/blitztext/autostart.py diff --git a/linux/blitztext/audio.py b/linux/blitztext/audio.py new file mode 100644 index 0000000..2a19879 --- /dev/null +++ b/linux/blitztext/audio.py @@ -0,0 +1,94 @@ +"""Audio helpers: enumerate input devices and a live input-level meter. + +Mic enumeration uses pactl (PipeWire/PulseAudio source names, which pw-record +and parecord accept via --target/-d). The level meter uses sounddevice to read +the chosen input and report a 0..1 level to a callback. +""" + +from __future__ import annotations + +import shutil +import subprocess +import threading + + +def list_mics() -> list[tuple[str, str]]: + """Return [(source_name, friendly_label)] for real input sources. + + The first entry is always the system default ("", "Default device"). + Monitor sources (loopback of outputs) are excluded. + """ + mics: list[tuple[str, str]] = [("", "Default device")] + if not shutil.which("pactl"): + return mics + try: + out = subprocess.run( + ["pactl", "list", "short", "sources"], capture_output=True, text=True, check=True + ).stdout + except (OSError, subprocess.CalledProcessError): + return mics + for line in out.splitlines(): + parts = line.split("\t") + if len(parts) < 2: + continue + name = parts[1] + if name.endswith(".monitor"): + continue + label = name.replace("alsa_input.", "").replace("-", " ") + mics.append((name, label)) + return mics + + +class LevelMeter: + """Open the given input device and call `on_level(0..1)` periodically.""" + + def __init__(self, device: str = "", on_level=None): + self.device = device or None + self.on_level = on_level + self._stream = None + self._lock = threading.Lock() + + def start(self) -> bool: + import numpy as np + import sounddevice as sd + + def _cb(indata, _frames, _time, _status): + level = float(np.sqrt(np.mean(np.square(indata)))) if indata.size else 0.0 + if self.on_level: + # Scale RMS (typically small) into a usable 0..1 range. + self.on_level(min(1.0, level * 12.0)) + + try: + self._stream = sd.InputStream( + samplerate=16000, channels=1, dtype="float32", + blocksize=1600, device=self._resolve_device(), callback=_cb, + ) + self._stream.start() + return True + except Exception: # noqa: BLE001 - device may be busy/unavailable + self._stream = None + return False + + def _resolve_device(self): + # sounddevice wants an index/name it knows; pactl names rarely match, so + # fall back to the default input when the name isn't resolvable. + if not self.device: + return None + try: + import sounddevice as sd + + for i, d in enumerate(sd.query_devices()): + if d["max_input_channels"] > 0 and self.device in d["name"]: + return i + except Exception: # noqa: BLE001 + pass + return None + + def stop(self) -> None: + with self._lock: + if self._stream is not None: + try: + self._stream.stop() + self._stream.close() + finally: + self._stream = None diff --git a/linux/blitztext/autostart.py b/linux/blitztext/autostart.py new file mode 100644 index 0000000..b4de76d --- /dev/null +++ b/linux/blitztext/autostart.py @@ -0,0 +1,44 @@ +"""Launch-on-login via a freedesktop autostart entry. + +Writes ~/.config/autostart/blitztext.desktop so the tray starts with the GNOME +session. No root needed. +""" + +from __future__ import annotations + +import os +import shutil +import sys +from pathlib import Path + +_AUTOSTART_DIR = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / "autostart" +_ENTRY = _AUTOSTART_DIR / "blitztext.desktop" + + +def _exec_command() -> str: + """Prefer an installed `blitztext` launcher; else run this venv's module.""" + launcher = shutil.which("blitztext") + if launcher: + return f"{launcher} tray" + return f"{sys.executable} -m blitztext tray" + + +def is_enabled() -> bool: + return _ENTRY.exists() + + +def set_enabled(enabled: bool) -> None: + if not enabled: + _ENTRY.unlink(missing_ok=True) + return + _AUTOSTART_DIR.mkdir(parents=True, exist_ok=True) + _ENTRY.write_text( + "[Desktop Entry]\n" + "Type=Application\n" + "Name=Blitztext\n" + f"Exec={_exec_command()}\n" + "Icon=blitztext\n" + "Terminal=false\n" + "X-GNOME-Autostart-enabled=true\n", + encoding="utf-8", + ) diff --git a/linux/blitztext/config.py b/linux/blitztext/config.py index c749a26..efc441f 100644 --- a/linux/blitztext/config.py +++ b/linux/blitztext/config.py @@ -34,6 +34,7 @@ class Workflow: class Config: # general recorder: str = "auto" + mic: str = "" # pactl/pipewire source name; "" = default device output: str = "type" # type | paste type_delay_ms: int = 12 notify: bool = True @@ -129,6 +130,7 @@ def load(path: Path = CONFIG_PATH) -> Config: cfg = Config( recorder=g.get("recorder", "auto"), + mic=g.get("mic", ""), output=g.get("output", "type"), type_delay_ms=int(g.get("type_delay_ms", 4)), notify=bool(g.get("notify", True)), @@ -215,6 +217,7 @@ def save(cfg: Config, path: Path = CONFIG_PATH) -> None: data: dict = { "general": { "recorder": cfg.recorder, + "mic": cfg.mic, "output": cfg.output, "type_delay_ms": cfg.type_delay_ms, "notify": cfg.notify, diff --git a/linux/blitztext/daemon.py b/linux/blitztext/daemon.py index cd5ea1b..ead1119 100644 --- a/linux/blitztext/daemon.py +++ b/linux/blitztext/daemon.py @@ -88,7 +88,7 @@ class Daemon: 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._recording = Recording(self.recorder_name, self.cfg.mic) self._active_workflow = wf self._emit("recording", wf.name, "Recording…") self._notify(f"● {wf.name}", "Recording…") diff --git a/linux/blitztext/gtksettings.py b/linux/blitztext/gtksettings.py index 7ffcf1e..712fc8d 100644 --- a/linux/blitztext/gtksettings.py +++ b/linux/blitztext/gtksettings.py @@ -1,194 +1,493 @@ -"""GTK settings dialog for Blitztext. +"""GTK settings: dropdown + editor managers for prompt presets and engines. -A functional editor for the engine, rewrite endpoint, and per-workflow prompts. -The richer presets manager + remote-engine controls build on this next. +Tabs: + Presets - prompt presets (select from a dropdown, edit, add, delete) + Engines - STT and LLM engine presets with online/offline status + STT test + Input - input scheme, keys, quality gate + General - mic (with live level meter), output, language, notifications, autostart """ from __future__ import annotations import os import sys +import threading +import time +from pathlib import Path import gi gi.require_version("Gtk", "3.0") -from gi.repository import Gtk # noqa: E402 +from gi.repository import GLib, Gtk # noqa: E402 +from . import audio, autostart, llm, stt # noqa: E402 from .config import Config, save # noqa: E402 +from .llm import LLMEngine # noqa: E402 +from .stt import STTEngine # noqa: E402 +from .config import Workflow # noqa: E402 RESP_SAVE = 1 RESP_SAVE_RESTART = 2 +GREEN, RED, GREY = "#34c759", "#ff3b30", "#b8b8be" -def _row(parent: Gtk.Box, label: str) -> Gtk.Box: - box = Gtk.Box(spacing=10) - box.set_margin_top(4) - box.set_margin_bottom(4) - lbl = Gtk.Label(label=label, xalign=0.0) - lbl.set_size_request(150, -1) - box.pack_start(lbl, False, False, 0) - parent.pack_start(box, False, False, 0) - return box +def _dot(color: str) -> str: + return f'●' -def _entry(parent: Gtk.Box, label: str, value) -> Gtk.Entry: - box = _row(parent, label) - e = Gtk.Entry() - e.set_text(str(value)) - e.set_hexpand(True) - box.pack_start(e, True, True, 0) - return e +def _labeled(parent: Gtk.Box, label: str, widget: Gtk.Widget, width: int = 130) -> Gtk.Widget: + row = Gtk.Box(spacing=10) + row.set_margin_top(3); row.set_margin_bottom(3) + lbl = Gtk.Label(label=label, xalign=0.0); lbl.set_size_request(width, -1) + row.pack_start(lbl, False, False, 0) + row.pack_start(widget, True, True, 0) + parent.pack_start(row, False, False, 0) + return widget -def _combo(parent: Gtk.Box, label: str, value: str, options: list[str]) -> Gtk.ComboBoxText: - box = _row(parent, label) +def _entry(text="") -> Gtk.Entry: + e = Gtk.Entry(); e.set_text(str(text)); e.set_hexpand(True); return e + + +def _combo(options, active=None) -> Gtk.ComboBoxText: c = Gtk.ComboBoxText() - for i, o in enumerate(options): + for o in options: c.append_text(o) - if o == value: - c.set_active(i) - if c.get_active() < 0: + if active in options: + c.set_active(options.index(active)) + elif options: c.set_active(0) - box.pack_start(c, False, False, 0) return c def _page(nb: Gtk.Notebook, title: str) -> Gtk.Box: - page = Gtk.Box(orientation=Gtk.Orientation.VERTICAL) - page.set_margin_top(12) - page.set_margin_bottom(12) - page.set_margin_start(14) - page.set_margin_end(14) - nb.append_page(page, Gtk.Label(label=title)) - return page + box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL) + for m in ("top", "bottom", "start", "end"): + getattr(box, f"set_margin_{m}")(12) + nb.append_page(box, Gtk.Label(label=title)) + return box class SettingsDialog: - def __init__(self, parent: Gtk.Window, cfg: Config): + def __init__(self, parent: Gtk.Window, cfg: Config, daemon=None): self.cfg = cfg + self.daemon = daemon + self._meter = None + self._wf_idx = self._stt_idx = self._llm_idx = 0 + self.dlg = Gtk.Dialog(title="Blitztext — Settings", transient_for=parent, modal=True) - self.dlg.set_default_size(580, 580) + self.dlg.set_default_size(620, 600) self.dlg.add_button("Close", Gtk.ResponseType.CLOSE) self.dlg.add_button("Save", RESP_SAVE) self.dlg.add_button("Save & Restart", RESP_SAVE_RESTART) nb = Gtk.Notebook() self.dlg.get_content_area().pack_start(nb, True, True, 0) - - # Engine - p = _page(nb, "Engine") - self.e_model = _entry(p, "Whisper model", cfg.model) - self.e_device = _combo(p, "Device", cfg.device, ["auto", "cpu", "cuda"]) - self.e_compute = _combo(p, "Compute type", cfg.compute_type, ["auto", "int8", "float16", "int8_float16"]) - self.e_lang = _entry(p, "Language hint", cfg.language) - self.e_output = _combo(p, "Output", cfg.output, ["type", "paste"]) - self.e_delay = _entry(p, "Type delay (ms)", cfg.type_delay_ms) - nbox = _row(p, "Notifications") - self.e_notify = Gtk.Switch() - self.e_notify.set_active(cfg.notify) - self.e_notify.set_halign(Gtk.Align.START) - nbox.pack_start(self.e_notify, False, False, 0) - - # Rewrite - p = _page(nb, "Rewrite LLM") - self.r_url = _entry(p, "Base URL", cfg.base_url) - self.r_keyenv = _entry(p, "API key env var", cfg.api_key_env) - self.r_model = _entry(p, "Model", cfg.rewrite_model) - self.r_temp = _entry(p, "Temperature", cfg.temperature) - self.r_timeout = _entry(p, "Timeout (s)", cfg.timeout) - hint = Gtk.Label(xalign=0.0, wrap=True) - hint.set_markup( - f'OpenAI-compatible endpoint (OpenAI, vLLM, llama-swap…). ' - f'Env {cfg.api_key_env}: ' - f'{"set ✓" if cfg.api_key else "NOT set"}.' - ) - p.pack_start(hint, False, False, 8) - - # Per-workflow prompts - self.wf_widgets: list[dict] = [] - for wf in cfg.workflows: - p = _page(nb, wf.name[:14]) - w = { - "name": _entry(p, "Name", wf.name), - "description": _entry(p, "Description", wf.description), - "hotkey": _entry(p, "Hotkey", wf.hotkey), - "model": _entry(p, "Model (opt.)", wf.model or ""), - "temperature": _entry(p, "Temp (opt.)", "" if wf.temperature is None else wf.temperature), - "mode": _combo(p, "Mode", wf.mode, ["transcribe", "rewrite"]), - } - p.pack_start(Gtk.Label(label="Rewrite prompt (system):", xalign=0.0), False, False, 6) - sw = Gtk.ScrolledWindow() - sw.set_min_content_height(150) - sw.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC) - tv = Gtk.TextView() - tv.set_wrap_mode(Gtk.WrapMode.WORD) - tv.get_buffer().set_text(wf.prompt) - sw.add(tv) - p.pack_start(sw, True, True, 0) - w["prompt"] = tv - self.wf_widgets.append(w) + self._build_presets(_page(nb, "Presets")) + self._build_engines(_page(nb, "Engines")) + self._build_input(_page(nb, "Input")) + self._build_general(_page(nb, "General")) self.dlg.connect("response", self._on_response) + self.dlg.connect("destroy", lambda *_: self._stop_meter()) - # -- helpers -------------------------------------------------------------- - @staticmethod - def _tv_text(tv: Gtk.TextView) -> str: - b = tv.get_buffer() - return b.get_text(b.get_start_iter(), b.get_end_iter(), True).strip() + # ===== Presets ========================================================== + def _build_presets(self, page: Gtk.Box) -> None: + self._wf_idx = 0 + bar = Gtk.Box(spacing=8) + self.wf_combo = Gtk.ComboBoxText() + for wf in self.cfg.workflows: + self.wf_combo.append_text(wf.name) + self.wf_combo.set_active(0) + self.wf_combo.connect("changed", self._wf_changed) + bar.pack_start(self.wf_combo, True, True, 0) + add = Gtk.Button(label="+ Add"); add.connect("clicked", self._wf_add) + rm = Gtk.Button(label="Delete"); rm.connect("clicked", self._wf_delete) + bar.pack_start(add, False, False, 0); bar.pack_start(rm, False, False, 0) + page.pack_start(bar, False, False, 4) + form = Gtk.Box(orientation=Gtk.Orientation.VERTICAL) + page.pack_start(form, True, True, 6) + self.wf_name = _labeled(form, "Name", _entry()) + self.wf_desc = _labeled(form, "Description", _entry()) + self.wf_keywords = _labeled(form, "Keywords (comma)", _entry()) + self.wf_hotkey = _labeled(form, "Hotkey (optional)", _entry()) + self.wf_mode = _labeled(form, "Mode", _combo(["transcribe", "rewrite"])) + self.wf_model = _labeled(form, "LLM model (opt.)", _entry()) + self.wf_temp = _labeled(form, "Temperature (opt.)", _entry()) + + form.pack_start(Gtk.Label(label="Prompt sent to the LLM (rewrite mode):", xalign=0.0), False, False, 6) + frame = Gtk.Frame(); frame.set_shadow_type(Gtk.ShadowType.IN) + sw = Gtk.ScrolledWindow(); sw.set_min_content_height(150) + self.wf_prompt = Gtk.TextView(); self.wf_prompt.set_wrap_mode(Gtk.WrapMode.WORD) + self.wf_prompt.set_left_margin(6); self.wf_prompt.set_top_margin(6) + sw.add(self.wf_prompt); frame.add(sw) + form.pack_start(frame, True, True, 0) + self._wf_load(0) + + def _wf_load(self, idx: int) -> None: + if not (0 <= idx < len(self.cfg.workflows)): + return + wf = self.cfg.workflows[idx] + self.wf_name.set_text(wf.name) + self.wf_desc.set_text(wf.description) + self.wf_keywords.set_text(", ".join(wf.keywords)) + self.wf_hotkey.set_text(wf.hotkey) + self.wf_mode.set_active(["transcribe", "rewrite"].index(wf.mode) if wf.mode in ("transcribe", "rewrite") else 0) + self.wf_model.set_text(wf.model or "") + self.wf_temp.set_text("" if wf.temperature is None else str(wf.temperature)) + self.wf_prompt.get_buffer().set_text(wf.prompt) + self._wf_idx = idx + + def _wf_commit(self) -> None: + idx = self._wf_idx + if not (0 <= idx < len(self.cfg.workflows)): + return + wf = self.cfg.workflows[idx] + wf.name = self.wf_name.get_text().strip() or wf.name + wf.description = self.wf_desc.get_text().strip() + wf.keywords = [k.strip() for k in self.wf_keywords.get_text().split(",") if k.strip()] + wf.hotkey = self.wf_hotkey.get_text().strip() + wf.mode = self.wf_mode.get_active_text() or "transcribe" + wf.model = self.wf_model.get_text().strip() or None + t = self.wf_temp.get_text().strip() + wf.temperature = float(t) if _isfloat(t) else None + b = self.wf_prompt.get_buffer() + wf.prompt = b.get_text(b.get_start_iter(), b.get_end_iter(), True).strip() + # reflect a possibly-changed name in the dropdown + self.wf_combo.remove(idx); self.wf_combo.insert_text(idx, wf.name) + self.wf_combo.set_active(idx) + + def _wf_changed(self, combo) -> None: + new = combo.get_active() + if new < 0 or new == self._wf_idx: + return + self._wf_commit() + self._wf_load(new) + + def _wf_add(self, _b) -> None: + self._wf_commit() + wf = Workflow(name="New preset", hotkey="", mode="rewrite", prompt="", description="") + self.cfg.workflows.append(wf) + self.wf_combo.append_text(wf.name) + self.wf_combo.set_active(len(self.cfg.workflows) - 1) + self._wf_load(len(self.cfg.workflows) - 1) + + def _wf_delete(self, _b) -> None: + if len(self.cfg.workflows) <= 1: + return + idx = self._wf_idx + del self.cfg.workflows[idx] + self.wf_combo.remove(idx) + self._wf_idx = -1 + self.wf_combo.set_active(0) + self._wf_load(0) + + # ===== Engines ========================================================== + def _build_engines(self, page: Gtk.Box) -> None: + page.pack_start(self._stt_section(), False, False, 4) + page.pack_start(Gtk.Separator(), False, False, 8) + page.pack_start(self._llm_section(), False, False, 4) + self._refresh_status() + + def _stt_section(self) -> Gtk.Box: + box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL) + box.pack_start(Gtk.Label(label="Speech-to-text engine", xalign=0.0), False, False, 2) + self._stt_idx = 0 + bar = Gtk.Box(spacing=8) + self.stt_combo = Gtk.ComboBoxText() + for e in self.cfg.stt_engines: + self.stt_combo.append_text(e.name) + self.stt_combo.set_active(self._index_of(self.cfg.stt_engines, self.cfg.stt_active)) + self.stt_combo.connect("changed", self._stt_changed) + self.stt_dot = Gtk.Label(); self.stt_dot.set_markup(_dot(GREY)) + bar.pack_start(self.stt_dot, False, False, 0) + bar.pack_start(self.stt_combo, True, True, 0) + for label, cb in (("+ Add", self._stt_add), ("Delete", self._stt_delete), + ("Test", self._stt_test), ("Refresh", lambda _b: self._refresh_status())): + b = Gtk.Button(label=label); b.connect("clicked", cb); bar.pack_start(b, False, False, 0) + box.pack_start(bar, False, False, 2) + + form = Gtk.Box(orientation=Gtk.Orientation.VERTICAL); box.pack_start(form, False, False, 2) + self.stt_type = _labeled(form, "Type", _combo(["local", "openai"])) + self.stt_url = _labeled(form, "URL", _entry()) + self.stt_model = _labeled(form, "Model", _entry()) + self.stt_key = _labeled(form, "API key env", _entry()) + self.stt_result = Gtk.Label(xalign=0.0); self.stt_result.set_line_wrap(True) + box.pack_start(self.stt_result, False, False, 2) + self._stt_load(self.stt_combo.get_active()) + return box + + def _llm_section(self) -> Gtk.Box: + box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL) + box.pack_start(Gtk.Label(label="Language model (rewrite)", xalign=0.0), False, False, 2) + self._llm_idx = 0 + bar = Gtk.Box(spacing=8) + self.llm_combo = Gtk.ComboBoxText() + for e in self.cfg.llm_engines: + self.llm_combo.append_text(e.name) + self.llm_combo.set_active(self._index_of(self.cfg.llm_engines, self.cfg.llm_active)) + self.llm_combo.connect("changed", self._llm_changed) + self.llm_dot = Gtk.Label(); self.llm_dot.set_markup(_dot(GREY)) + bar.pack_start(self.llm_dot, False, False, 0) + bar.pack_start(self.llm_combo, True, True, 0) + for label, cb in (("+ Add", self._llm_add), ("Delete", self._llm_delete), + ("Refresh", lambda _b: self._refresh_status())): + b = Gtk.Button(label=label); b.connect("clicked", cb); bar.pack_start(b, False, False, 0) + box.pack_start(bar, False, False, 2) + + form = Gtk.Box(orientation=Gtk.Orientation.VERTICAL); box.pack_start(form, False, False, 2) + self.llm_url = _labeled(form, "Base URL", _entry()) + self.llm_model = _labeled(form, "Model", _entry()) + self.llm_key = _labeled(form, "API key env", _entry()) + self.llm_temp = _labeled(form, "Temperature", _entry()) + self._llm_load(self.llm_combo.get_active()) + return box + + # -- STT engine editor --- + def _stt_load(self, idx: int) -> None: + if not (0 <= idx < len(self.cfg.stt_engines)): + return + e = self.cfg.stt_engines[idx] + self.stt_type.set_active(["local", "openai"].index(e.type) if e.type in ("local", "openai") else 0) + self.stt_url.set_text(e.url); self.stt_model.set_text(e.model); self.stt_key.set_text(e.api_key_env) + self._stt_idx = idx + + def _stt_commit(self) -> None: + idx = self._stt_idx + if not (0 <= idx < len(self.cfg.stt_engines)): + return + e = self.cfg.stt_engines[idx] + e.type = self.stt_type.get_active_text() or "local" + e.url = self.stt_url.get_text().strip().rstrip("/") + e.model = self.stt_model.get_text().strip() + e.api_key_env = self.stt_key.get_text().strip() + + def _stt_changed(self, combo): + new = combo.get_active() + if new < 0 or new == self._stt_idx: + return + self._stt_commit(); self._stt_load(new); self._refresh_status() + + def _stt_add(self, _b): + self._stt_commit() + e = STTEngine("New STT", "openai", "http://localhost:8010/v1", "whisper-1") + self.cfg.stt_engines.append(e); self.stt_combo.append_text(e.name) + self.stt_combo.set_active(len(self.cfg.stt_engines) - 1) + self._stt_load(len(self.cfg.stt_engines) - 1) + + def _stt_delete(self, _b): + if len(self.cfg.stt_engines) <= 1: + return + del self.cfg.stt_engines[self._stt_idx] + self.stt_combo.remove(self._stt_idx); self._stt_idx = -1 + self.stt_combo.set_active(0); self._stt_load(0) + + def _stt_test(self, _b): + self._stt_commit() + e = self.cfg.stt_engines[self._stt_idx] + self.stt_result.set_markup("Recording 4s — speak now…") + threading.Thread(target=self._run_stt_test, args=(e,), daemon=True).start() + + def _run_stt_test(self, engine): + from .recorder import Recording, detect_recorder + try: + rec = Recording(detect_recorder(self.cfg.recorder), self.cfg.mic) + time.sleep(4.0) + wav = rec.stop() + tr = self.daemon.transcriber if (self.daemon and engine.is_local) else None + res = stt.benchmark(engine, wav, language=self.cfg.language, local_transcriber=tr) + wav.unlink(missing_ok=True) + if res.ok: + msg = f"{res.seconds:.2f}s · {GLib.markup_escape_text(res.text or '(empty)')}" + else: + msg = f'{GLib.markup_escape_text(res.error)}' + except Exception as exc: # noqa: BLE001 + msg = f'{GLib.markup_escape_text(str(exc))}' + GLib.idle_add(self.stt_result.set_markup, msg) + + # -- LLM engine editor --- + def _llm_load(self, idx: int) -> None: + if not (0 <= idx < len(self.cfg.llm_engines)): + return + e = self.cfg.llm_engines[idx] + self.llm_url.set_text(e.url); self.llm_model.set_text(e.model) + self.llm_key.set_text(e.api_key_env); self.llm_temp.set_text(str(e.temperature)) + self._llm_idx = idx + + def _llm_commit(self) -> None: + idx = self._llm_idx + if not (0 <= idx < len(self.cfg.llm_engines)): + return + e = self.cfg.llm_engines[idx] + e.url = self.llm_url.get_text().strip().rstrip("/") + e.model = self.llm_model.get_text().strip() + e.api_key_env = self.llm_key.get_text().strip() + if _isfloat(self.llm_temp.get_text().strip()): + e.temperature = float(self.llm_temp.get_text().strip()) + + def _llm_changed(self, combo): + new = combo.get_active() + if new < 0 or new == self._llm_idx: + return + self._llm_commit(); self._llm_load(new); self._refresh_status() + + def _llm_add(self, _b): + self._llm_commit() + e = LLMEngine("New LLM", "http://localhost:28080/v1", "model", "") + self.cfg.llm_engines.append(e); self.llm_combo.append_text(e.name) + self.llm_combo.set_active(len(self.cfg.llm_engines) - 1) + self._llm_load(len(self.cfg.llm_engines) - 1) + + def _llm_delete(self, _b): + if len(self.cfg.llm_engines) <= 1: + return + del self.cfg.llm_engines[self._llm_idx] + self.llm_combo.remove(self._llm_idx); self._llm_idx = -1 + self.llm_combo.set_active(0); self._llm_load(0) + + # -- status dots (threaded) --- + def _refresh_status(self) -> None: + s = self.cfg.stt_engines[self._stt_idx] if 0 <= self._stt_idx < len(self.cfg.stt_engines) else None + self._stt_commit() + l = self.cfg.llm_engines[self._llm_idx] if 0 <= self._llm_idx < len(self.cfg.llm_engines) else None + self._llm_commit() + + def check(): + sc = GREEN if (s and stt.status(s)) else RED if s else GREY + lc = GREEN if (l and llm.status(l)) else RED if l else GREY + GLib.idle_add(self.stt_dot.set_markup, _dot(sc)) + GLib.idle_add(self.llm_dot.set_markup, _dot(lc)) + threading.Thread(target=check, daemon=True).start() + + # ===== Input ============================================================ + def _build_input(self, page: Gtk.Box) -> None: + self.in_mode = _labeled(page, "Input mode", _combo(["modifiers", "hotkeys"], self.cfg.input_mode)) + self.in_ptt = Gtk.Switch(); self.in_ptt.set_active(self.cfg.push_to_talk); self.in_ptt.set_halign(Gtk.Align.START) + _labeled(page, "Push-to-talk", self.in_ptt) + self.in_start = _labeled(page, "Start", _entry(self.cfg.key_start)) + self.in_stop = _labeled(page, "Stop + paste", _entry(self.cfg.key_stop)) + self.in_send = _labeled(page, "Stop + paste + Enter", _entry(self.cfg.key_send)) + self.in_cancel = _labeled(page, "Cancel", _entry(self.cfg.key_cancel)) + page.pack_start(Gtk.Separator(), False, False, 8) + page.pack_start(Gtk.Label(label="Quality gate", xalign=0.0), False, False, 2) + self.q_min = _labeled(page, "Min seconds", _entry(self.cfg.min_speech_seconds)) + self.q_rms = _labeled(page, "Silence RMS", _entry(self.cfg.silence_rms)) + self.q_halluc = Gtk.Switch(); self.q_halluc.set_active(self.cfg.reject_hallucinations); self.q_halluc.set_halign(Gtk.Align.START) + _labeled(page, "Reject hallucinations", self.q_halluc) + self.q_strip = Gtk.Switch(); self.q_strip.set_active(self.cfg.strip_trailing_punctuation); self.q_strip.set_halign(Gtk.Align.START) + _labeled(page, "Strip trailing punctuation", self.q_strip) + + # ===== General ========================================================== + def _build_general(self, page: Gtk.Box) -> None: + self._mics = audio.list_mics() + names = [label for _, label in self._mics] + cur = next((lbl for nm, lbl in self._mics if nm == self.cfg.mic), names[0]) + self.gen_mic = _labeled(page, "Microphone", _combo(names, cur)) + + self.mic_level = Gtk.LevelBar(); self.mic_level.set_min_value(0); self.mic_level.set_max_value(1) + _labeled(page, "Input level", self.mic_level) + self.gen_mic.connect("changed", lambda _c: self._restart_meter()) + + self.gen_output = _labeled(page, "Output", _combo(["type", "paste"], self.cfg.output)) + self.gen_lang = _labeled(page, "Language hint", _entry(self.cfg.language)) + self.gen_notify = Gtk.Switch(); self.gen_notify.set_active(self.cfg.notify); self.gen_notify.set_halign(Gtk.Align.START) + _labeled(page, "Notifications", self.gen_notify) + self.gen_boot = Gtk.Switch(); self.gen_boot.set_active(autostart.is_enabled()); self.gen_boot.set_halign(Gtk.Align.START) + _labeled(page, "Launch on login", self.gen_boot) + + page.pack_start(Gtk.Separator(), False, False, 8) + page.pack_start(Gtk.Label(label="Local Whisper (for the local STT engine)", xalign=0.0), False, False, 2) + self.gen_model = _labeled(page, "Model", _entry(self.cfg.model)) + self.gen_device = _labeled(page, "Device", _combo(["auto", "cpu", "cuda"], self.cfg.device)) + self.gen_compute = _labeled(page, "Compute type", _combo(["auto", "int8", "float16", "int8_float16"], self.cfg.compute_type)) + self._start_meter() + + def _selected_mic_name(self) -> str: + i = self.gen_mic.get_active() + return self._mics[i][0] if 0 <= i < len(self._mics) else "" + + def _start_meter(self) -> None: + self._stop_meter() + self._meter = audio.LevelMeter(self._selected_mic_name(), on_level=lambda v: GLib.idle_add(self.mic_level.set_value, v)) + self._meter.start() + + def _restart_meter(self) -> None: + self._start_meter() + + def _stop_meter(self) -> None: + if self._meter is not None: + self._meter.stop(); self._meter = None + + # ===== save / collect ==================================================== def _collect(self) -> bool: try: + self._wf_commit(); self._stt_commit(); self._llm_commit() c = self.cfg - c.model = self.e_model.get_text().strip() - c.device = self.e_device.get_active_text() - c.compute_type = self.e_compute.get_active_text() - c.language = self.e_lang.get_text().strip() - c.output = self.e_output.get_active_text() - c.type_delay_ms = int(self.e_delay.get_text()) - c.notify = self.e_notify.get_active() - c.base_url = self.r_url.get_text().strip().rstrip("/") - c.api_key_env = self.r_keyenv.get_text().strip() - c.rewrite_model = self.r_model.get_text().strip() - c.temperature = float(self.r_temp.get_text()) - c.timeout = int(self.r_timeout.get_text()) - for wf, w in zip(c.workflows, self.wf_widgets): - wf.name = w["name"].get_text().strip() or wf.name - wf.description = w["description"].get_text().strip() - wf.hotkey = w["hotkey"].get_text().strip() - wf.model = w["model"].get_text().strip() or None - t = w["temperature"].get_text().strip() - wf.temperature = float(t) if t else None - wf.mode = w["mode"].get_active_text() - wf.prompt = self._tv_text(w["prompt"]) + c.stt_active = self.stt_combo.get_active_text() or c.stt_active + c.llm_active = self.llm_combo.get_active_text() or c.llm_active + c.input_mode = self.in_mode.get_active_text() or "modifiers" + c.push_to_talk = self.in_ptt.get_active() + c.key_start = self.in_start.get_text().strip() + c.key_stop = self.in_stop.get_text().strip() + c.key_send = self.in_send.get_text().strip() + c.key_cancel = self.in_cancel.get_text().strip() + c.min_speech_seconds = float(self.q_min.get_text()) + c.silence_rms = float(self.q_rms.get_text()) + c.reject_hallucinations = self.q_halluc.get_active() + c.strip_trailing_punctuation = self.q_strip.get_active() + c.mic = self._selected_mic_name() + c.output = self.gen_output.get_active_text() or "type" + c.language = self.gen_lang.get_text().strip() + c.notify = self.gen_notify.get_active() + c.model = self.gen_model.get_text().strip() + c.device = self.gen_device.get_active_text() or "auto" + c.compute_type = self.gen_compute.get_active_text() or "auto" + autostart.set_enabled(self.gen_boot.get_active()) except ValueError as exc: self._error(f"Check numeric fields: {exc}") return False return True def _error(self, msg: str) -> None: - d = Gtk.MessageDialog(transient_for=self.dlg, modal=True, - message_type=Gtk.MessageType.ERROR, buttons=Gtk.ButtonsType.OK, text=msg) - d.run() - d.destroy() + d = Gtk.MessageDialog(transient_for=self.dlg, modal=True, message_type=Gtk.MessageType.ERROR, + buttons=Gtk.ButtonsType.OK, text=msg) + d.run(); d.destroy() def _info(self, msg: str) -> None: - d = Gtk.MessageDialog(transient_for=self.dlg, modal=True, - message_type=Gtk.MessageType.INFO, buttons=Gtk.ButtonsType.OK, text=msg) - d.run() - d.destroy() + d = Gtk.MessageDialog(transient_for=self.dlg, modal=True, message_type=Gtk.MessageType.INFO, + buttons=Gtk.ButtonsType.OK, text=msg) + d.run(); d.destroy() - def _on_response(self, dlg: Gtk.Dialog, resp: int) -> None: + def _on_response(self, dlg, resp): if resp == RESP_SAVE: if self._collect(): save(self.cfg) - self._info("Saved. Restart Blitztext to apply hotkey/model changes.") + self._info("Saved. Restart Blitztext to apply engine/hotkey changes.") return if resp == RESP_SAVE_RESTART: if self._collect(): save(self.cfg) + self._stop_meter() os.execv(sys.executable, [sys.executable, "-m", "blitztext", "tray"]) return + self._stop_meter() dlg.destroy() + @staticmethod + def _index_of(engines, name): + for i, e in enumerate(engines): + if e.name == name: + return i + return 0 + def run_dialog(self) -> None: self.dlg.show_all() + + +def _isfloat(s: str) -> bool: + try: + float(s) + return True + except (TypeError, ValueError): + return False diff --git a/linux/blitztext/gtkui.py b/linux/blitztext/gtkui.py index 1679319..74a6924 100644 --- a/linux/blitztext/gtkui.py +++ b/linux/blitztext/gtkui.py @@ -256,7 +256,7 @@ class App: def open_settings(self) -> None: from .gtksettings import SettingsDialog - SettingsDialog(self.win, self.cfg).run_dialog() + SettingsDialog(self.win, self.cfg, daemon=self.daemon).run_dialog() def show_panel(self) -> None: self.win.show_all() diff --git a/linux/blitztext/recorder.py b/linux/blitztext/recorder.py index ae2a3f0..969d2b5 100644 --- a/linux/blitztext/recorder.py +++ b/linux/blitztext/recorder.py @@ -19,6 +19,13 @@ _RECORDERS: dict[str, list[str]] = { "arecord": ["arecord", "-q", "-f", "S16_LE", "-r", "16000", "-c", "1", "-t", "wav"], } +# How to point each recorder at a specific pactl/pipewire source. +_DEVICE_FLAG: dict[str, list[str]] = { + "pw-record": ["--target"], + "parecord": ["-d"], + "arecord": [], # ALSA names differ from pactl; fall back to default device +} + def detect_recorder(preference: str = "auto") -> str: if preference != "auto": @@ -34,9 +41,13 @@ def detect_recorder(preference: str = "auto") -> str: class Recording: """A single in-progress recording. Start on construction, call stop().""" - def __init__(self, recorder: str): + def __init__(self, recorder: str, device: str = ""): self._tmp = Path(tempfile.mkstemp(prefix="blitztext-", suffix=".wav")[1]) - argv = list(_RECORDERS[recorder]) + [str(self._tmp)] + argv = list(_RECORDERS[recorder]) + flag = _DEVICE_FLAG.get(recorder, []) + if device and flag: + argv += flag + [device] + argv += [str(self._tmp)] # arecord writes to the file given as a positional arg; the others too. self._proc = subprocess.Popen( argv, diff --git a/linux/requirements.txt b/linux/requirements.txt index 48737cf..576a2d1 100644 --- a/linux/requirements.txt +++ b/linux/requirements.txt @@ -1,2 +1,4 @@ faster-whisper>=1.0 pynput>=1.7 +tomli-w>=1.0 +sounddevice>=0.4