hermes/blitztext-app-linux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Log tab + in-memory log buffer

Browse Source 
New logbuffer.py captures app messages (log()) and library logs (faster-whisper,
huggingface_hub) into a ring buffer, mirrored to stderr. Settings gains a Log
tab: monospace view, 1s refresh, auto-scroll, Copy, Clear — so model
download/load progress is visible instead of an opaque "Loading…". transcribe
and daemon now log via the buffer.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
mARTin-B78 2026-06-05 09:42:48 +02:00
parent a508ecc69c
commit 78350bdb99
5 changed files with 128 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
linux/blitztext/blitztext.py
View File

@ -29,6 +29,9 @@ def main(argv: list[str] | None = None) -> int:
args = parser.parse_args(argv)
cmd = args.cmd or "tray"
from .logbuffer import install_logging
install_logging()
if cmd == "config-path":
print(ensure_default(CONFIG_PATH))
return 0

15
linux/blitztext/daemon.py
View File

@ -14,6 +14,7 @@ from typing import Callable
from . import llm, quality, stt
from .config import Config, Workflow
from .llm import LLMError
from .logbuffer import log
from .notify import notify
from .paste import active_window_id, deliver
from .recorder import Recording, detect_recorder
@ -56,6 +57,7 @@ class Daemon:
# -- model load (slow; call off the UI thread) ----------------------------
def prepare(self) -> None:
engine = self.cfg.active_stt
log(f"STT engine: {engine.name} ({'local' if engine.is_local else engine.url})")
if engine.is_local:
model = engine.model or self.cfg.model
self._emit("loading", None, f"Loading Whisper '{model}'…")
@ -70,7 +72,9 @@ class Daemon:
# Remote STT engine — no local model to load.
self.transcriber = None
self._emit("loading", None, f"Using {engine.name}")
log(f"Using remote STT '{engine.name}' — no local model to load")
self._prepared = True
log("Ready.")
self._emit("idle", None, "Ready")
@property
@ -208,11 +212,12 @@ class Daemon:
from .paste import press_enter
press_enter(window_id)
self._emit("done", label, text)
log(f"✓ {label}: {text[:120]}")
self._notify(f"✓ {label}", text[:80] + ("…" if len(text) > 80 else ""))
except Exception as exc: # noqa: BLE001 - surface any failure
self._emit("error", label, str(exc))
self._notify("Error", str(exc), "critical")
print(f"[blitztext] error: {exc}", file=sys.stderr)
log(f"ERROR ({label}): {exc}")
finally:
audio_path.unlink(missing_ok=True)
with self._lock:
@ -270,12 +275,10 @@ class Daemon:
def run(self) -> None:
self.prepare()
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)
log(f"Recorder: {self.recorder_name}. Input: Ctrl+Win start · Ctrl stop+paste · Alt stop+paste+Enter · Esc cancel")
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)
keys = ", ".join([f"{self.cfg.routing_hotkey}→Voice"] if self.cfg.routing_enabled else [])
log(f"Recorder: {self.recorder_name}. Hotkeys: {keys}")
self._notify("Blitztext ready", "Focus a text field and start dictating.")
listener = self.start_input()

55
linux/blitztext/gtksettings.py
View File

@ -21,7 +21,7 @@ gi.require_version("Gtk", "3.0")
gi.require_version("Gdk", "3.0")
from gi.repository import Gdk, GLib, Gtk # noqa: E402
from . import audio, autostart, llm, stt # noqa: E402
from . import audio, autostart, llm, logbuffer, stt # noqa: E402
from .config import Config, save # noqa: E402
from .llm import LLMEngine # noqa: E402
from .stt import STTEngine # noqa: E402
@ -209,13 +209,14 @@ class SettingsDialog:
self._build_engines(_page(nb, "Engines"))
self._build_input(_page(nb, "Input"))
self._build_general(_page(nb, "General"))
self._build_log(_page(nb, "Log"))
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._stop_meter())
self.dlg.connect("destroy", lambda *_: self._cleanup())
# ===== Presets ==========================================================
def _build_presets(self, page: Gtk.Box) -> None:
@ -632,6 +633,52 @@ class SettingsDialog:
if self._meter is not None:
self._meter.stop(); self._meter = None
# ===== Log ==============================================================
def _build_log(self, page: Gtk.Box) -> None:
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:
@ -681,10 +728,10 @@ class SettingsDialog:
if resp == RESP_SAVE_RESTART:
if self._collect():
save(self.cfg)
self._stop_meter()
self._cleanup()
os.execv(sys.executable, [sys.executable, "-m", "blitztext", "tray"])
return
self._stop_meter()
self._cleanup()
dlg.destroy()
@staticmethod

60
linux/blitztext/logbuffer.py Normal file
View File

@ -0,0 +1,60 @@
"""In-memory log buffer shown in Settings → Log, mirrored to stderr.
Our own messages go through log(); library logs (faster-whisper, huggingface_hub)
are captured via a logging handler so model download/load progress is visible
instead of an opaque "Loading…".
"""
from __future__ import annotations
import logging
import sys
import threading
import time
from collections import deque
_LINES: deque[str] = deque(maxlen=2000)
_LOCK = threading.Lock()
def log(msg: str, *, echo: bool = True) -> None:
line = f"{time.strftime('%H:%M:%S')} {msg}"
with _LOCK:
_LINES.append(line)
if echo:
print(line, file=sys.stderr, flush=True)
def lines() -> list[str]:
with _LOCK:
return list(_LINES)
def clear() -> None:
with _LOCK:
_LINES.clear()
class _BufferHandler(logging.Handler):
def emit(self, record: logging.LogRecord) -> None:
try:
log(self.format(record), echo=False)
except Exception: # noqa: BLE001 - never let logging break the app
pass
_installed = False
def install_logging() -> None:
"""Capture library log records into the buffer (once)."""
global _installed
if _installed:
return
_installed = True
handler = _BufferHandler()
handler.setFormatter(logging.Formatter("%(name)s: %(message)s"))
for name in ("faster_whisper", "huggingface_hub", "blitztext"):
lg = logging.getLogger(name)
lg.addHandler(handler)
lg.setLevel(logging.INFO)

7
linux/blitztext/transcribe.py
View File

@ -30,6 +30,8 @@ class Transcriber:
def _load(self, model: str, device: str, compute_type: str):
from faster_whisper import WhisperModel
from .logbuffer import log
attempts: list[str]
if device == "auto":
attempts = ["cuda", "cpu"]
@ -40,13 +42,14 @@ class Transcriber:
for dev in attempts:
try:
ct = self._resolve_compute_type(dev, compute_type)
log(f"Loading Whisper '{model}' on {dev} ({ct})… (first run may download the model)")
m = WhisperModel(model, device=dev, compute_type=ct)
print(f"[blitztext] Whisper '{model}' loaded on {dev} ({ct})", file=sys.stderr)
log(f"Whisper '{model}' ready on {dev} ({ct})")
return m
except Exception as exc: # noqa: BLE001 - CUDA libs may be absent
last_err = exc
if dev != attempts[-1]:
print(f"[blitztext] {dev} unavailable ({exc}); trying next device", file=sys.stderr)
log(f"{dev} unavailable ({exc}); trying next device")
raise RuntimeError(f"Failed to load Whisper model '{model}': {last_err}")
def transcribe(self, audio_path: Path, language: str = "", hotwords: str = "") -> str: