mARTin-B78
2d777382f5
Add a plain-language info box to the top of every Settings tab (Presets, Engines, Input, General, Benchmark, Log, About), exposed to screen readers, and tooltips (wired to ATK descriptions) on the preset, general, and input-mode controls — for non-technical and blind users (barrierefrei). Complete the unreleased 1.2.0 changelog with the engine manager, Benchmark tab, custom audio cues, Log tab, click-to-bind hotkeys, the GTK rebuild, voice routing, the modifier hotkey scheme, and the quality gate. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1188 lines
55 KiB
Python
1188 lines
55 KiB
Python
"""GTK settings: dropdown + editor managers for prompt presets and engines.
|
||
|
||
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
|
||
Benchmark- compare STT engines against a reference clip
|
||
Log - runtime log output
|
||
About - version, source, changelog, and license
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import gi
|
||
|
||
gi.require_version("Gtk", "3.0")
|
||
gi.require_version("Gdk", "3.0")
|
||
from gi.repository import Gdk, GLib, Gtk, Pango # noqa: E402
|
||
|
||
from . import __version__, audio, autostart, benchmark, llm, logbuffer, 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 _dot(color: str) -> str:
|
||
return f'<span foreground="{color}">●</span>'
|
||
|
||
|
||
# --- key capture (for the "Set" bind buttons) -------------------------------
|
||
_MOD_ORDER = ["ctrl", "alt", "shift", "cmd"]
|
||
|
||
|
||
def _keyval_token(keyval: int) -> str | None:
|
||
name = Gdk.keyval_name(keyval) or ""
|
||
table = {
|
||
"Control_L": "ctrl", "Control_R": "ctrl",
|
||
"Alt_L": "alt", "Alt_R": "alt", "ISO_Level3_Shift": "alt",
|
||
"Super_L": "cmd", "Super_R": "cmd", "Meta_L": "cmd", "Meta_R": "cmd",
|
||
"Shift_L": "shift", "Shift_R": "shift",
|
||
"Escape": "esc", "space": "space", "Return": "enter", "Tab": "tab",
|
||
"BackSpace": "backspace", "Delete": "delete",
|
||
}
|
||
if name in table:
|
||
return table[name]
|
||
if len(name) == 1 and name.isprintable():
|
||
return name.lower()
|
||
if len(name) >= 2 and name[0] in "Ff" and name[1:].isdigit():
|
||
return name.lower() # F1..F12
|
||
return None
|
||
|
||
|
||
def _format_combo(tokens: list[str]) -> str:
|
||
mods = [t for t in _MOD_ORDER if t in tokens]
|
||
rest = [t for t in tokens if t not in _MOD_ORDER]
|
||
parts = [(t if len(t) == 1 else f"<{t}>") for t in mods + rest]
|
||
return "+".join(parts)
|
||
|
||
|
||
def _labeled(parent: Gtk.Box, label: str, widget: Gtk.Widget, width: int = 130, tooltip: str = "") -> 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)
|
||
|
||
# ATK Accessibility linking
|
||
if hasattr(lbl, "set_mnemonic_widget"):
|
||
lbl.set_mnemonic_widget(widget)
|
||
atk = widget.get_accessible()
|
||
if atk and label:
|
||
atk.set_name(label.replace("_", ""))
|
||
if tooltip:
|
||
widget.set_tooltip_text(tooltip)
|
||
lbl.set_tooltip_text(tooltip)
|
||
if atk:
|
||
atk.set_description(tooltip)
|
||
|
||
row.pack_start(lbl, False, False, 0)
|
||
row.pack_start(widget, True, True, 0)
|
||
parent.pack_start(row, False, False, 0)
|
||
return widget
|
||
|
||
|
||
def _infobox(parent: Gtk.Box, text: str) -> Gtk.Box:
|
||
"""A plain-language help box at the top of a tab (read by screen readers)."""
|
||
box = Gtk.Box(spacing=8)
|
||
box.set_margin_top(2); box.set_margin_bottom(10)
|
||
icon = Gtk.Image.new_from_icon_name("dialog-information-symbolic", Gtk.IconSize.BUTTON)
|
||
icon.set_valign(Gtk.Align.START)
|
||
box.pack_start(icon, False, False, 0)
|
||
lbl = Gtk.Label(label=text, xalign=0.0)
|
||
lbl.set_line_wrap(True); lbl.set_xalign(0.0); lbl.set_max_width_chars(72)
|
||
box.pack_start(lbl, True, True, 0)
|
||
acc = box.get_accessible()
|
||
if acc:
|
||
acc.set_name("Information")
|
||
acc.set_description(text)
|
||
parent.pack_start(box, False, False, 0)
|
||
return box
|
||
|
||
|
||
def _entry(text="", placeholder="") -> Gtk.Entry:
|
||
e = Gtk.Entry(); e.set_text(str(text)); e.set_hexpand(True)
|
||
if placeholder:
|
||
e.set_placeholder_text(placeholder)
|
||
return e
|
||
|
||
|
||
def _combo(options, active=None) -> Gtk.ComboBoxText:
|
||
c = Gtk.ComboBoxText()
|
||
for o in options:
|
||
c.append_text(o)
|
||
if active in options:
|
||
c.set_active(options.index(active))
|
||
elif options:
|
||
c.set_active(0)
|
||
return c
|
||
|
||
|
||
class ModelPicker(Gtk.Box):
|
||
"""An editable model field + a ▾ button opening a popover with a search bar
|
||
on top and a scrollable, filtered list of models."""
|
||
|
||
def __init__(self, placeholder: str = ""):
|
||
super().__init__(spacing=4)
|
||
self.entry = Gtk.Entry(); self.entry.set_hexpand(True)
|
||
if placeholder:
|
||
self.entry.set_placeholder_text(placeholder)
|
||
self.pack_start(self.entry, True, True, 0)
|
||
self.btn = Gtk.Button.new_from_icon_name("pan-down-symbolic", Gtk.IconSize.BUTTON)
|
||
self.btn.set_tooltip_text("Browse models")
|
||
self.pack_start(self.btn, False, False, 0)
|
||
|
||
self.pop = Gtk.Popover(); self.pop.set_relative_to(self.btn)
|
||
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6)
|
||
for m in ("top", "bottom", "start", "end"):
|
||
getattr(box, f"set_margin_{m}")(6)
|
||
self.search = Gtk.SearchEntry(); self.search.set_placeholder_text("Search models…")
|
||
box.pack_start(self.search, False, False, 0)
|
||
sw = Gtk.ScrolledWindow()
|
||
sw.set_min_content_height(300); sw.set_min_content_width(380)
|
||
sw.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC)
|
||
self.listbox = Gtk.ListBox()
|
||
self.listbox.set_filter_func(self._filter)
|
||
sw.add(self.listbox); box.pack_start(sw, True, True, 0)
|
||
self.pop.add(box)
|
||
|
||
self.btn.connect("clicked", self._open)
|
||
self.search.connect("search-changed", lambda _s: self.listbox.invalidate_filter())
|
||
self.listbox.connect("row-activated", self._activated)
|
||
|
||
def _filter(self, row) -> bool:
|
||
return self.search.get_text().lower() in row.get_child().get_text().lower()
|
||
|
||
def _open(self, _b) -> None:
|
||
self.pop.show_all(); self.search.set_text(""); self.search.grab_focus()
|
||
|
||
def _activated(self, _lb, row) -> None:
|
||
self.entry.set_text(row.get_child().get_text())
|
||
self.pop.popdown()
|
||
|
||
def set_models(self, models) -> None:
|
||
for child in self.listbox.get_children():
|
||
self.listbox.remove(child)
|
||
for m in models:
|
||
row = Gtk.ListBoxRow()
|
||
lbl = Gtk.Label(label=m, xalign=0.0)
|
||
lbl.set_margin_top(4); lbl.set_margin_bottom(4); lbl.set_margin_start(8); lbl.set_margin_end(8)
|
||
row.add(lbl); self.listbox.add(row)
|
||
self.listbox.show_all()
|
||
|
||
def get_text(self) -> str:
|
||
return self.entry.get_text().strip()
|
||
|
||
def set_text(self, value: str) -> None:
|
||
self.entry.set_text(value or "")
|
||
|
||
|
||
def _model_combo(placeholder="") -> ModelPicker:
|
||
return ModelPicker(placeholder)
|
||
|
||
|
||
def _combo_text(c: ModelPicker) -> str:
|
||
return c.get_text()
|
||
|
||
|
||
def _fill_combo(combo: ModelPicker, options, current: str) -> None:
|
||
combo.set_models(list(options))
|
||
combo.set_text(current or "")
|
||
|
||
|
||
def _url_field(parent: Gtk.Box, label: str, placeholder: str, on_reload) -> Gtk.Entry:
|
||
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(130, -1)
|
||
row.pack_start(lbl, False, False, 0)
|
||
e = Gtk.Entry(); e.set_hexpand(True)
|
||
if placeholder:
|
||
e.set_placeholder_text(placeholder)
|
||
|
||
lbl.set_mnemonic_widget(e)
|
||
atk = e.get_accessible()
|
||
if atk and label:
|
||
atk.set_name(label)
|
||
|
||
row.pack_start(e, True, True, 0)
|
||
btn = Gtk.Button.new_from_icon_name("view-refresh-symbolic", Gtk.IconSize.BUTTON)
|
||
btn.set_tooltip_text("Load models from this URL")
|
||
if btn.get_accessible():
|
||
btn.get_accessible().set_name("Refresh models")
|
||
btn.connect("clicked", lambda _b: on_reload())
|
||
row.pack_start(btn, False, False, 0)
|
||
parent.pack_start(row, False, False, 0)
|
||
return e
|
||
|
||
|
||
def _page(nb: Gtk.Notebook, title: str) -> Gtk.Box:
|
||
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
|
||
|
||
|
||
def _read_first(paths: list[Path], fallback: str = "Not available in this install.") -> str:
|
||
for path in paths:
|
||
try:
|
||
if path.exists():
|
||
return path.read_text(encoding="utf-8", errors="replace").strip()
|
||
except OSError:
|
||
continue
|
||
return fallback
|
||
|
||
|
||
def _app_paths() -> dict[str, list[Path]]:
|
||
pkg_dir = Path(__file__).resolve().parent
|
||
linux_dir = pkg_dir.parent
|
||
repo_dir = linux_dir.parent
|
||
return {
|
||
"changelog": [linux_dir / "CHANGELOG.md", Path("/opt/blitztext/CHANGELOG.md")],
|
||
"license": [repo_dir / "LICENSE", Path("/usr/share/doc/blitztext/copyright")],
|
||
}
|
||
|
||
|
||
def _text_panel(text: str, *, monospace: bool = True, height: int = 180) -> Gtk.ScrolledWindow:
|
||
sw = Gtk.ScrolledWindow()
|
||
sw.set_min_content_height(height)
|
||
sw.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
|
||
view = Gtk.TextView()
|
||
view.set_editable(False)
|
||
view.set_cursor_visible(False)
|
||
view.set_wrap_mode(Gtk.WrapMode.WORD_CHAR)
|
||
view.set_left_margin(8)
|
||
view.set_right_margin(8)
|
||
view.set_top_margin(8)
|
||
view.set_bottom_margin(8)
|
||
view.set_monospace(monospace)
|
||
view.get_buffer().set_text(text)
|
||
sw.add(view)
|
||
return sw
|
||
|
||
|
||
class SettingsDialog:
|
||
def __init__(self, parent: Gtk.Window, cfg: Config, daemon=None):
|
||
self.cfg = cfg
|
||
self.daemon = daemon
|
||
self._meter = None
|
||
self._tr_cache: dict = {}
|
||
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(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)
|
||
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._build_benchmark(_page(nb, "Benchmark"))
|
||
self._build_log(_page(nb, "Log"))
|
||
self._build_about(_page(nb, "About"))
|
||
|
||
self._bind_entry = None
|
||
self._bind_pressed: list[str] = []
|
||
self.dlg.connect("key-press-event", self._on_bind_press)
|
||
self.dlg.connect("key-release-event", self._on_bind_release)
|
||
self.dlg.connect("response", self._on_response)
|
||
self.dlg.connect("destroy", lambda *_: self._cleanup())
|
||
|
||
# ===== Presets ==========================================================
|
||
def _build_presets(self, page: Gtk.Box) -> None:
|
||
_infobox(page, "Presets are your dictation actions. Pick one to edit it, or add your "
|
||
"own. A preset can simply type what you say, or rewrite it first "
|
||
"(for example into a polished email). Trigger it by speaking its "
|
||
"keyword, or with an optional keyboard shortcut.")
|
||
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(placeholder="Preset name"),
|
||
tooltip="A short name for this action, shown in the main panel.")
|
||
self.wf_desc = _labeled(form, "Description", _entry(placeholder="Short description shown in the panel"),
|
||
tooltip="One line explaining what this preset does.")
|
||
self.wf_keywords = _labeled(form, "Keywords (comma)", _entry(placeholder="nicer email, bessere email"),
|
||
tooltip="Spoken trigger words. Say one at the start or end of your speech to use this preset.")
|
||
self.wf_hotkey = self._key_field(form, "Hotkey (optional)", "", placeholder="click Set, or e.g. <ctrl>+<alt>+e", width=130)
|
||
self.wf_mode = _labeled(form, "Mode", _combo(["transcribe", "rewrite", "stream"]),
|
||
tooltip="‘transcribe’ types your words as-is. ‘rewrite’ sends them to the language model first. ‘stream’ shows live text from a realtime engine.")
|
||
self.wf_model = _labeled(form, "LLM model (opt.)", _entry(placeholder="blank = use the active LLM engine's model"),
|
||
tooltip="Override the language model just for this preset. Leave blank to use the active LLM engine.")
|
||
self.wf_temp = _labeled(form, "Temperature (opt.)", _entry(placeholder="blank = engine default (e.g. 0.3)"),
|
||
tooltip="Creativity of the rewrite, 0–1. Lower is more predictable. Blank uses the engine default.")
|
||
|
||
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", "stream"].index(wf.mode) if wf.mode in ("transcribe", "rewrite", "stream") 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:
|
||
_infobox(page, "Engines do the work. The speech-to-text engine turns your voice into "
|
||
"text; the language model rewrites it. Each can run locally on this "
|
||
"computer or on a server you enter. A green dot means it is reachable, "
|
||
"red means offline. Use Test to try the speech engine.")
|
||
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), ("+ Stream", self._stt_add_stream),
|
||
("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_name = _labeled(form, "Name", _entry(placeholder="e.g. faster-whisper GPU"))
|
||
self.stt_type = _labeled(form, "Type", _combo(["local", "openai", "riva_realtime"]))
|
||
self.stt_url = _url_field(form, "URL", "http://localhost:8010/v1 · realtime: http://localhost:8006/v1",
|
||
lambda: self._populate_models(self.stt_model, self.stt_url.get_text().strip(), self.stt_key.get_text().strip()))
|
||
self.stt_model = _labeled(form, "Model", _model_combo("blank = server default · tiny/base/small… for local"))
|
||
self.stt_key = _labeled(form, "API key env", _entry(placeholder="env var name, e.g. GROQ_API_KEY (optional)"))
|
||
self.stt_url.connect("changed", lambda _e: self._schedule_models("stt"))
|
||
self.stt_key.connect("changed", lambda _e: self._schedule_models("stt"))
|
||
self.stt_type.connect("changed", self._stt_type_changed)
|
||
|
||
sep = Gtk.Separator(); sep.set_margin_top(4); form.pack_start(sep, False, False, 4)
|
||
form.pack_start(Gtk.Label(label="Local engine (faster-whisper) — device & precision", xalign=0.0), False, False, 0)
|
||
self.stt_device = _labeled(form, "Device", _combo(["auto", "cpu", "cuda"], self.cfg.device))
|
||
self.stt_compute = _labeled(form, "Compute type", _combo(["auto", "int8", "float16", "int8_float16"], self.cfg.compute_type))
|
||
|
||
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_name = _labeled(form, "Name", _entry(placeholder="e.g. Local Qwen"))
|
||
self.llm_type = _labeled(form, "Type", _combo(["local", "cloud"]))
|
||
self.llm_url = _url_field(form, "Base URL", "http://localhost:28080/v1 · https://api.openai.com/v1",
|
||
lambda: self._populate_models(self.llm_model, self.llm_url.get_text().strip(), self.llm_key.get_text().strip()))
|
||
self.llm_model = _labeled(form, "Model", _model_combo("pick after entering URL"))
|
||
self.llm_key = _labeled(form, "API key env", _entry(placeholder="env var name, e.g. OPENAI_API_KEY (blank for local)"))
|
||
self.llm_temp = _labeled(form, "Temperature", _entry(placeholder="0.3"))
|
||
self.llm_url.connect("changed", lambda _e: self._schedule_models("llm"))
|
||
self.llm_key.connect("changed", lambda _e: self._schedule_models("llm"))
|
||
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_name.set_text(e.name)
|
||
self.stt_type.set_active(["local", "openai", "riva_realtime"].index(e.type) if e.type in ("local", "openai", "riva_realtime") else 0)
|
||
self.stt_url.set_text(e.url); self.stt_key.set_text(e.api_key_env)
|
||
if e.type == "local":
|
||
_fill_combo(self.stt_model, ["tiny", "base", "small", "medium", "large-v3"], e.model or self.cfg.model)
|
||
elif e.type == "openai":
|
||
_fill_combo(self.stt_model, [], e.model)
|
||
self._populate_models(self.stt_model, e.url, e.api_key_env)
|
||
else:
|
||
_fill_combo(self.stt_model, [], e.model)
|
||
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]
|
||
new_name = self.stt_name.get_text().strip() or e.name
|
||
e.name = new_name
|
||
e.type = self.stt_type.get_active_text() or "local"
|
||
e.url = self.stt_url.get_text().strip().rstrip("/")
|
||
e.model = _combo_text(self.stt_model)
|
||
e.api_key_env = self.stt_key.get_text().strip()
|
||
if self.stt_combo.get_active_text() != new_name:
|
||
self.stt_combo.remove(idx); self.stt_combo.insert_text(idx, new_name); self.stt_combo.set_active(idx)
|
||
|
||
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_add_stream(self, _b):
|
||
self._stt_commit()
|
||
e = STTEngine("Nemotron ASR Streaming", "riva_realtime", "http://127.0.0.1:8006/v1", "")
|
||
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]
|
||
if e.is_streaming:
|
||
self.stt_result.set_markup("<i>Streaming engines are live-only. Use a preset with mode = stream.</i>")
|
||
return
|
||
self.stt_result.set_markup("<i>Recording 4s — speak now…</i>")
|
||
threading.Thread(target=self._run_stt_test, args=(e,), daemon=True).start()
|
||
|
||
def _transcriber_for(self, engine):
|
||
"""A local Transcriber for the engine, loading on demand (cached)."""
|
||
if not engine.is_local:
|
||
return None
|
||
model = engine.model or self.cfg.model
|
||
active = self.cfg.active_stt
|
||
if (self.daemon and self.daemon.transcriber and active.is_local
|
||
and (active.model or self.cfg.model) == model):
|
||
return self.daemon.transcriber
|
||
key = (model, self.cfg.device, self.cfg.compute_type)
|
||
if key not in self._tr_cache:
|
||
from .transcribe import Transcriber
|
||
logbuffer.log(f"Loading local model '{model}' ({self.cfg.device}) for test/benchmark…")
|
||
self._tr_cache[key] = Transcriber(model, self.cfg.device, self.cfg.compute_type, self.cfg.beam_size)
|
||
return self._tr_cache[key]
|
||
|
||
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()
|
||
GLib.idle_add(self.stt_result.set_markup, "<i>Transcribing…</i>")
|
||
tr = self._transcriber_for(engine)
|
||
res = stt.benchmark(engine, wav, language=self.cfg.language, local_transcriber=tr)
|
||
wav.unlink(missing_ok=True)
|
||
if res.ok:
|
||
msg = f"<b>{res.seconds:.2f}s</b> · {GLib.markup_escape_text(res.text or '(empty)')}"
|
||
else:
|
||
msg = f'<span foreground="{RED}">{GLib.markup_escape_text(res.error)}</span>'
|
||
except Exception as exc: # noqa: BLE001
|
||
msg = f'<span foreground="{RED}">{GLib.markup_escape_text(str(exc))}</span>'
|
||
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_name.set_text(e.name)
|
||
self.llm_type.set_active(["local", "cloud"].index(e.type) if e.type in ("local", "cloud") else 1)
|
||
self.llm_url.set_text(e.url)
|
||
self.llm_key.set_text(e.api_key_env); self.llm_temp.set_text(str(e.temperature))
|
||
_fill_combo(self.llm_model, [], e.model)
|
||
self._populate_models(self.llm_model, e.url, e.api_key_env)
|
||
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]
|
||
new_name = self.llm_name.get_text().strip() or e.name
|
||
e.name = new_name
|
||
e.type = self.llm_type.get_active_text() or "cloud"
|
||
e.url = self.llm_url.get_text().strip().rstrip("/")
|
||
e.model = _combo_text(self.llm_model)
|
||
e.api_key_env = self.llm_key.get_text().strip()
|
||
if self.llm_combo.get_active_text() != new_name:
|
||
self.llm_combo.remove(idx); self.llm_combo.insert_text(idx, new_name); self.llm_combo.set_active(idx)
|
||
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)
|
||
|
||
def _stt_type_changed(self, _c) -> None:
|
||
typ = self.stt_type.get_active_text()
|
||
if typ == "local":
|
||
_fill_combo(self.stt_model, ["tiny", "base", "small", "medium", "large-v3"], _combo_text(self.stt_model))
|
||
elif typ == "openai":
|
||
self._schedule_models("stt")
|
||
else:
|
||
_fill_combo(self.stt_model, [], _combo_text(self.stt_model))
|
||
|
||
# -- model dropdowns (fetched from {url}/models) ---
|
||
def _populate_models(self, combo, url: str, key_env: str) -> None:
|
||
def work():
|
||
models = stt.list_models(url, key_env) if url else []
|
||
|
||
def apply():
|
||
cur = _combo_text(combo)
|
||
_fill_combo(combo, models, cur)
|
||
return False
|
||
GLib.idle_add(apply)
|
||
threading.Thread(target=work, daemon=True).start()
|
||
|
||
def _schedule_models(self, which: str) -> None:
|
||
attr = f"_mt_{which}"
|
||
old = getattr(self, attr, 0)
|
||
if old:
|
||
GLib.source_remove(old)
|
||
|
||
def fire():
|
||
setattr(self, attr, 0)
|
||
if which == "stt" and self.stt_type.get_active_text() == "openai":
|
||
self._populate_models(self.stt_model, self.stt_url.get_text().strip(),
|
||
self.stt_key.get_text().strip())
|
||
elif which == "llm":
|
||
self._populate_models(self.llm_model, self.llm_url.get_text().strip(),
|
||
self.llm_key.get_text().strip())
|
||
return False
|
||
setattr(self, attr, GLib.timeout_add(700, fire))
|
||
|
||
# -- 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()
|
||
|
||
# -- key binding ("Set" button captures the next keypress) ---
|
||
def _key_field(self, page: Gtk.Box, label: str, value: str, placeholder: str = "", width: int = 150) -> Gtk.Entry:
|
||
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)
|
||
entry = Gtk.Entry(); entry.set_text(value); entry.set_hexpand(True)
|
||
if placeholder:
|
||
entry.set_placeholder_text(placeholder)
|
||
row.pack_start(entry, True, True, 0)
|
||
btn = Gtk.Button(label="Set")
|
||
btn.connect("clicked", lambda _b, e=entry: self._bind_key(e))
|
||
row.pack_start(btn, False, False, 0)
|
||
page.pack_start(row, False, False, 0)
|
||
return entry
|
||
|
||
def _bind_key(self, entry: Gtk.Entry) -> None:
|
||
self._bind_entry = entry
|
||
self._bind_pressed = []
|
||
entry.set_text("")
|
||
entry.set_placeholder_text("press the key to bind…")
|
||
|
||
def _on_bind_press(self, _w, event) -> bool:
|
||
if self._bind_entry is None:
|
||
return False
|
||
tok = _keyval_token(event.keyval)
|
||
if tok and tok not in self._bind_pressed:
|
||
self._bind_pressed.append(tok)
|
||
return True # swallow while binding
|
||
|
||
def _on_bind_release(self, _w, event) -> bool:
|
||
if self._bind_entry is None:
|
||
return False
|
||
combo = _format_combo(self._bind_pressed)
|
||
if combo:
|
||
self._bind_entry.set_text(combo)
|
||
self._bind_entry = None
|
||
return True
|
||
|
||
# ===== Input ============================================================
|
||
def _build_input(self, page: Gtk.Box) -> None:
|
||
_infobox(page, "Choose how you start and stop dictating. With the default "
|
||
"‘modifiers’ mode: hold Ctrl and the Windows key to talk, then press "
|
||
"Ctrl to stop and paste. Below you can tune the noise filter, set up "
|
||
"hands-free wakeword, and pick sounds that confirm start and stop.")
|
||
self.in_mode = _labeled(page, "Input mode", _combo(["modifiers", "hotkeys"], self.cfg.input_mode),
|
||
tooltip="‘modifiers’: hold/press the keys below. ‘hotkeys’: each preset has its own shortcut combo.")
|
||
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 = self._key_field(page, "Start", self.cfg.key_start)
|
||
self.in_stop = self._key_field(page, "Stop + paste", self.cfg.key_stop)
|
||
self.in_send = self._key_field(page, "Stop + paste + Enter", self.cfg.key_send)
|
||
self.in_cancel = self._key_field(page, "Cancel", 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), tooltip="Minimum audio length. Shorter clips are ignored.")
|
||
self.q_rms = _labeled(page, "Silence RMS", _entry(self.cfg.silence_rms), tooltip="Microphone volume threshold to ignore background noise.")
|
||
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, tooltip="Automatically drop STT ghost outputs like 'Thank you.' or 'Bye.'")
|
||
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, tooltip="Remove ending periods from pasted text, useful for code inserts.")
|
||
page.pack_start(Gtk.Separator(), False, False, 8)
|
||
page.pack_start(Gtk.Label(label="Hands-free (Wakeword)", xalign=0.0), False, False, 2)
|
||
self.ww_enabled = Gtk.Switch(); self.ww_enabled.set_active(self.cfg.wakeword_enabled); self.ww_enabled.set_halign(Gtk.Align.START)
|
||
_labeled(page, "Enable wakeword", self.ww_enabled)
|
||
self.ww_uri = _url_field(page, "Wyoming URI", "tcp://127.0.0.1:10400", self._ww_load)
|
||
self.ww_uri.set_text(self.cfg.wakeword_uri)
|
||
self.ww_model = _labeled(page, "Model name", _model_combo("Search models…"))
|
||
_fill_combo(self.ww_model, [], self.cfg.wakeword_model)
|
||
|
||
self.ww_mic_level = Gtk.LevelBar(); self.ww_mic_level.set_min_value(0); self.ww_mic_level.set_max_value(1)
|
||
_labeled(page, "Input level", self.ww_mic_level)
|
||
|
||
self.ww_test_btn = Gtk.Button(label="Test Wakeword"); self.ww_test_btn.set_halign(Gtk.Align.START)
|
||
self.ww_test_btn.connect("clicked", self._ww_test)
|
||
self.ww_test_lbl = Gtk.Label(label=""); self.ww_test_lbl.set_xalign(0.0)
|
||
box = Gtk.Box(spacing=10); box.pack_start(self.ww_test_btn, False, False, 0); box.pack_start(self.ww_test_lbl, False, False, 0)
|
||
_labeled(page, "", box)
|
||
|
||
page.pack_start(Gtk.Separator(), False, False, 8)
|
||
page.pack_start(Gtk.Label(label="Audio cues", xalign=0.0), False, False, 2)
|
||
self.snd_before = self._sound_field(
|
||
page, "Play before", self.cfg.sound_before,
|
||
"Sound played when recording starts — your confirmation that Blitztext is listening.")
|
||
self.snd_after = self._sound_field(
|
||
page, "Play after", self.cfg.sound_after,
|
||
"Sound played when recording stops (on paste, paste+Enter, or auto-stop on silence).")
|
||
|
||
def _sound_field(self, page: Gtk.Box, label: str, value: str, tooltip: str = "") -> Gtk.FileChooserButton:
|
||
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(150, -1)
|
||
if tooltip:
|
||
lbl.set_tooltip_text(tooltip)
|
||
row.pack_start(lbl, False, False, 0)
|
||
chooser = Gtk.FileChooserButton(title=label, action=Gtk.FileChooserAction.OPEN)
|
||
af = Gtk.FileFilter(); af.set_name("Audio")
|
||
for pat in ("*.wav", "*.oga", "*.ogg", "*.flac"):
|
||
af.add_pattern(pat)
|
||
chooser.add_filter(af)
|
||
if value:
|
||
chooser.set_filename(value)
|
||
chooser.set_hexpand(True)
|
||
if tooltip:
|
||
chooser.set_tooltip_text(tooltip + " Leave empty to use the built-in system sound.")
|
||
row.pack_start(chooser, True, True, 0)
|
||
play = Gtk.Button.new_from_icon_name("media-playback-start-symbolic", Gtk.IconSize.BUTTON)
|
||
play.set_tooltip_text("Play this sound now")
|
||
play.connect("clicked", lambda _b, c=chooser: self._play_sound_file(c.get_filename()))
|
||
row.pack_start(play, False, False, 0)
|
||
clr = Gtk.Button.new_from_icon_name("edit-clear-symbolic", Gtk.IconSize.BUTTON)
|
||
clr.set_tooltip_text("Clear — use the built-in system sound")
|
||
clr.connect("clicked", lambda _b, c=chooser: c.unselect_all())
|
||
row.pack_start(clr, False, False, 0)
|
||
page.pack_start(row, False, False, 0)
|
||
return chooser
|
||
|
||
def _play_sound_file(self, path) -> None:
|
||
from . import sound
|
||
if path:
|
||
sound.play(path)
|
||
|
||
# ===== General ==========================================================
|
||
def _build_general(self, page: Gtk.Box) -> None:
|
||
_infobox(page, "General options: choose your microphone and watch its level bar move "
|
||
"when you speak, set how the text is delivered, the spoken language, "
|
||
"desktop notifications, and whether Blitztext starts automatically "
|
||
"when you log in.")
|
||
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),
|
||
tooltip="Which microphone Blitztext records from.")
|
||
|
||
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,
|
||
tooltip="Live microphone level. The bar should move when you speak.")
|
||
self.gen_mic.connect("changed", lambda _c: self._restart_meter())
|
||
|
||
self.gen_output = _labeled(page, "Output", _combo(["type", "paste"], self.cfg.output),
|
||
tooltip="‘type’ types the text key by key. ‘paste’ copies it and presses Ctrl+V (faster for long text).")
|
||
self.gen_lang = _labeled(page, "Language hint", _entry(self.cfg.language, placeholder="de, en, … (blank = autodetect)"),
|
||
tooltip="Spoken language code (de, en, …). Leave blank to auto-detect.")
|
||
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)
|
||
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()
|
||
def on_level(v):
|
||
GLib.idle_add(self.mic_level.set_value, v)
|
||
if hasattr(self, "ww_mic_level"):
|
||
GLib.idle_add(self.ww_mic_level.set_value, v)
|
||
self._meter = audio.LevelMeter(self._selected_mic_name(), on_level=on_level)
|
||
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
|
||
|
||
def _ww_load(self) -> None:
|
||
def work():
|
||
import socket, json
|
||
from urllib.parse import urlparse
|
||
uri = self.ww_uri.get_text().strip()
|
||
parsed = urlparse(uri)
|
||
host = parsed.hostname or "127.0.0.1"
|
||
port = parsed.port or 10400
|
||
try:
|
||
with socket.create_connection((host, port), timeout=2.0) as s:
|
||
s.sendall(json.dumps({"type": "describe"}).encode() + b"\n")
|
||
line = b""
|
||
while not line.endswith(b"\n"):
|
||
b = s.recv(1)
|
||
if not b: break
|
||
line += b
|
||
info = json.loads(line.decode("utf-8"))
|
||
payload_len = info.get("data_length", info.get("payload_length", 0))
|
||
payload = b""
|
||
while len(payload) < payload_len:
|
||
chunk = s.recv(payload_len - len(payload))
|
||
if not chunk: break
|
||
payload += chunk
|
||
data = json.loads(payload.decode("utf-8"))
|
||
models = []
|
||
for w in data.get("wake", []):
|
||
for m in w.get("models", []):
|
||
models.append(m.get("name"))
|
||
def apply():
|
||
_fill_combo(self.ww_model, models, models[0] if models else "")
|
||
self._info("Loaded models successfully.")
|
||
GLib.idle_add(apply)
|
||
except Exception as e:
|
||
GLib.idle_add(lambda: self._error(f"Failed to load models:\n{e}"))
|
||
threading.Thread(target=work, daemon=True).start()
|
||
|
||
def _ww_test(self, _b) -> None:
|
||
self.ww_test_btn.set_sensitive(False)
|
||
self.ww_test_lbl.set_text("Listening for 10s...")
|
||
|
||
def work():
|
||
from .wakeword import WakewordListener
|
||
import time
|
||
detected = False
|
||
def on_detect():
|
||
nonlocal detected
|
||
detected = True
|
||
|
||
listener = WakewordListener(
|
||
uri=self.ww_uri.get_text().strip(),
|
||
model=_combo_text(self.ww_model),
|
||
mic=self._selected_mic_name(),
|
||
on_detect=on_detect
|
||
)
|
||
listener.start()
|
||
|
||
start_t = time.time()
|
||
while time.time() - start_t < 10 and not detected:
|
||
time.sleep(0.1)
|
||
|
||
listener.stop()
|
||
|
||
def finish():
|
||
self.ww_test_btn.set_sensitive(True)
|
||
if detected:
|
||
self.ww_test_lbl.set_markup("<span foreground='green'><b>Detected!</b></span>")
|
||
else:
|
||
self.ww_test_lbl.set_markup("<span foreground='red'>Timed out.</span>")
|
||
GLib.idle_add(finish)
|
||
|
||
threading.Thread(target=work, daemon=True).start()
|
||
|
||
# ===== Benchmark ========================================================
|
||
def _build_benchmark(self, page: Gtk.Box) -> None:
|
||
_infobox(page, "Compare your speech-to-text engines. Pick a recording (.wav) and a "
|
||
"text file (.txt) containing exactly what is said, then press Run "
|
||
"benchmark. The table shows how fast each engine is and how accurate, "
|
||
"and names the fastest and most accurate. Add an engine preset for "
|
||
"each model you want in the comparison.")
|
||
|
||
wavf = Gtk.FileChooserButton(title="WAV file", action=Gtk.FileChooserAction.OPEN)
|
||
fa = Gtk.FileFilter(); fa.set_name("Audio (.wav)"); fa.add_pattern("*.wav"); wavf.add_filter(fa)
|
||
self.bench_wav = _labeled(page, "Audio (.wav)", wavf)
|
||
reff = Gtk.FileChooserButton(title="Reference transcript", action=Gtk.FileChooserAction.OPEN)
|
||
ft = Gtk.FileFilter(); ft.set_name("Text (.txt)"); ft.add_pattern("*.txt"); reff.add_filter(ft)
|
||
self.bench_ref = _labeled(page, "Reference (.txt)", reff)
|
||
|
||
def _on_wav_set(_b):
|
||
fn = wavf.get_filename()
|
||
if not fn: return
|
||
p = Path(fn)
|
||
for ext in (".reference.txt", ".txt"):
|
||
cand = p.with_name(p.stem + ext)
|
||
if cand.exists():
|
||
reff.set_filename(str(cand))
|
||
break
|
||
wavf.connect("file-set", _on_wav_set)
|
||
|
||
run = Gtk.Button(label="Run benchmark"); run.connect("clicked", self._run_bench)
|
||
run.set_halign(Gtk.Align.START)
|
||
page.pack_start(run, False, False, 6)
|
||
|
||
self.bench_store = Gtk.ListStore(str, str, str, str, str, str)
|
||
tree = Gtk.TreeView(model=self.bench_store)
|
||
for title, i, expand in [("Engine", 0, False), ("Model", 1, False), ("Device", 2, False),
|
||
("Time (s)", 3, False), ("Accuracy", 4, False), ("Output", 5, True)]:
|
||
r = Gtk.CellRendererText()
|
||
if i == 5:
|
||
r.set_property("ellipsize", Pango.EllipsizeMode.END)
|
||
col = Gtk.TreeViewColumn(title, r, text=i); col.set_resizable(True)
|
||
col.set_expand(expand)
|
||
tree.append_column(col)
|
||
sw = Gtk.ScrolledWindow(); sw.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
|
||
sw.add(tree); page.pack_start(sw, True, True, 4)
|
||
|
||
self.bench_summary = Gtk.Label(xalign=0.0); self.bench_summary.set_line_wrap(True)
|
||
page.pack_start(self.bench_summary, False, False, 4)
|
||
|
||
def _run_bench(self, _b) -> None:
|
||
wav = self.bench_wav.get_filename()
|
||
refp = self.bench_ref.get_filename()
|
||
if not wav or not refp:
|
||
self._error("Pick a .wav file and a matching reference .txt file.")
|
||
return
|
||
reference = Path(refp).read_text(errors="replace")
|
||
self.bench_store.clear()
|
||
self.bench_summary.set_markup("<i>Running… (local models load on first use)</i>")
|
||
self._stt_commit()
|
||
engines = list(self.cfg.stt_engines)
|
||
|
||
def work():
|
||
def prog(row):
|
||
GLib.idle_add(self._bench_add_row, row)
|
||
rows = benchmark.run(engines, Path(wav), reference, language=self.cfg.language,
|
||
get_local_transcriber=self._transcriber_for, progress=prog)
|
||
GLib.idle_add(self._bench_done, rows)
|
||
threading.Thread(target=work, daemon=True).start()
|
||
|
||
def _bench_add_row(self, row) -> bool:
|
||
acc = f"{row.accuracy:.1f}%" if row.ok else "—"
|
||
out = row.text if row.ok else f"⚠ {row.error}"
|
||
self.bench_store.append([row.engine, row.model, row.device, f"{row.seconds:.2f}", acc, out])
|
||
return False
|
||
|
||
def _bench_done(self, rows) -> bool:
|
||
fastest, acc = benchmark.best(rows)
|
||
if not fastest:
|
||
self.bench_summary.set_markup('<span foreground="#ff3b30">All engines failed — check the Log tab.</span>')
|
||
return False
|
||
self.bench_summary.set_markup(
|
||
f"<b>Fastest:</b> {GLib.markup_escape_text(fastest.engine)} ({fastest.seconds:.2f}s)"
|
||
f" · <b>Most accurate:</b> {GLib.markup_escape_text(acc.engine)} ({acc.accuracy:.1f}%)")
|
||
return False
|
||
|
||
# ===== About ============================================================
|
||
def _build_about(self, page: Gtk.Box) -> None:
|
||
_infobox(page, "About Blitztext: version, source code, recent changes, and licence.")
|
||
paths = _app_paths()
|
||
changelog = _read_first(paths["changelog"])
|
||
license_text = _read_first(paths["license"])
|
||
|
||
title = Gtk.Label(label="Blitztext App Linux", xalign=0.0)
|
||
title.set_markup("<b>Blitztext App Linux</b>")
|
||
page.pack_start(title, False, False, 0)
|
||
|
||
version = Gtk.Label(label=f"Version {__version__}", xalign=0.0)
|
||
version.set_selectable(True)
|
||
page.pack_start(version, False, False, 2)
|
||
|
||
source = Gtk.LinkButton.new_with_label(
|
||
"https://github.com/mARTin-B78/blitztext-app-linux",
|
||
"Source: github.com/mARTin-B78/blitztext-app-linux",
|
||
)
|
||
source.set_halign(Gtk.Align.START)
|
||
page.pack_start(source, False, False, 4)
|
||
|
||
license_label = Gtk.Label(label="License: MIT", xalign=0.0)
|
||
license_label.set_selectable(True)
|
||
page.pack_start(license_label, False, False, 2)
|
||
|
||
nb = Gtk.Notebook()
|
||
nb.set_margin_top(8)
|
||
page.pack_start(nb, True, True, 0)
|
||
|
||
changelog_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
|
||
changelog_box.pack_start(_text_panel(changelog, height=300), True, True, 0)
|
||
nb.append_page(changelog_box, Gtk.Label(label="Changelog"))
|
||
|
||
license_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
|
||
license_box.pack_start(_text_panel(license_text, height=300), True, True, 0)
|
||
nb.append_page(license_box, Gtk.Label(label="License"))
|
||
|
||
|
||
# ===== Log ==============================================================
|
||
def _build_log(self, page: Gtk.Box) -> None:
|
||
_infobox(page, "A live activity log — useful to see what Blitztext is doing, for "
|
||
"example while a speech model loads or downloads. Press Copy to put it "
|
||
"on the clipboard when you want to report a problem.")
|
||
sw = Gtk.ScrolledWindow(); sw.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
|
||
self.log_view = Gtk.TextView()
|
||
self.log_view.set_editable(False); self.log_view.set_cursor_visible(False)
|
||
self.log_view.set_monospace(True); self.log_view.set_left_margin(6); self.log_view.set_top_margin(6)
|
||
sw.add(self.log_view)
|
||
page.pack_start(sw, True, True, 0)
|
||
|
||
bar = Gtk.Box(spacing=8); bar.set_margin_top(6)
|
||
self.log_autoscroll = Gtk.CheckButton(label="Auto-scroll"); self.log_autoscroll.set_active(True)
|
||
bar.pack_start(self.log_autoscroll, False, False, 0)
|
||
clear = Gtk.Button(label="Clear"); clear.connect("clicked", lambda _b: (logbuffer.clear(), self._log_refresh()))
|
||
copy = Gtk.Button(label="Copy"); copy.connect("clicked", lambda _b: self._log_copy())
|
||
bar.pack_end(clear, False, False, 0); bar.pack_end(copy, False, False, 0)
|
||
page.pack_start(bar, False, False, 0)
|
||
|
||
self._log_last = None
|
||
self._log_refresh()
|
||
self._log_timer = GLib.timeout_add(1000, self._log_tick)
|
||
|
||
def _log_tick(self) -> bool:
|
||
self._log_refresh()
|
||
return True
|
||
|
||
def _log_refresh(self) -> None:
|
||
text = "\n".join(logbuffer.lines())
|
||
if text == getattr(self, "_log_last", None):
|
||
return
|
||
self._log_last = text
|
||
buf = self.log_view.get_buffer()
|
||
buf.set_text(text)
|
||
if self.log_autoscroll.get_active():
|
||
self.log_view.scroll_to_iter(buf.get_end_iter(), 0.0, False, 0, 0)
|
||
|
||
def _log_copy(self) -> None:
|
||
cb = Gtk.Clipboard.get(Gdk.SELECTION_CLIPBOARD)
|
||
cb.set_text("\n".join(logbuffer.lines()), -1)
|
||
|
||
def _cleanup(self) -> None:
|
||
self._stop_meter()
|
||
t = getattr(self, "_log_timer", 0)
|
||
if t:
|
||
GLib.source_remove(t)
|
||
self._log_timer = 0
|
||
|
||
# ===== save / collect ====================================================
|
||
def _collect(self) -> bool:
|
||
try:
|
||
self._wf_commit(); self._stt_commit(); self._llm_commit()
|
||
c = self.cfg
|
||
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.sound_before = self.snd_before.get_filename() or ""
|
||
c.sound_after = self.snd_after.get_filename() or ""
|
||
c.wakeword_enabled = self.ww_enabled.get_active()
|
||
c.wakeword_uri = self.ww_uri.get_text().strip()
|
||
c.wakeword_model = _combo_text(self.ww_model)
|
||
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.device = self.stt_device.get_active_text() or "auto"
|
||
c.compute_type = self.stt_compute.get_active_text() or "auto"
|
||
autostart.set_enabled(self.gen_boot.get_active())
|
||
|
||
if c.wakeword_enabled:
|
||
import socket
|
||
from urllib.parse import urlparse
|
||
parsed = urlparse(c.wakeword_uri)
|
||
host = parsed.hostname or "127.0.0.1"
|
||
port = parsed.port or 10400
|
||
try:
|
||
with socket.create_connection((host, port), timeout=1.5):
|
||
pass
|
||
except OSError as e:
|
||
self._error(f"Cannot connect to Wakeword server at {c.wakeword_uri}:\n{e}\n\nPlease check your server or disable wakeword.")
|
||
return False
|
||
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()
|
||
|
||
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()
|
||
|
||
def _on_response(self, dlg, resp):
|
||
if resp == RESP_SAVE:
|
||
if self._collect():
|
||
save(self.cfg)
|
||
self._info("Saved. Restart Blitztext to apply engine/hotkey changes.")
|
||
return
|
||
if resp == RESP_SAVE_RESTART:
|
||
if self._collect():
|
||
save(self.cfg)
|
||
self._cleanup()
|
||
os.execv(sys.executable, [sys.executable, "-m", "blitztext", "tray"])
|
||
return
|
||
self._cleanup()
|
||
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
|