Add Blitztext for Linux v1.0.0 — native dictation tool

The upstream app is macOS-only (Swift/SwiftUI, CoreML/WhisperKit) and can't
run on Linux or in a container. This adds a native host tool under linux/ that
reproduces the workflow: focus any text field, press a hotkey, speak, and the
optionally-rewritten text is typed into that field.

- Engine: pynput global hotkeys → mic record → local faster-whisper →
  optional OpenAI-compatible rewrite → xdotool typing into the focused window
- Frontends: system tray (AppIndicator, default), tkinter control panel,
  and headless modes
- Config-driven workflows in ~/.config/blitztext/config.toml with per-workflow
  prompt/model/temperature overrides
- Packaging: install.sh, requirements.txt, systemd user unit
- Targets X11; local transcription runs CPU int8 on this arm64 host

See linux/CHANGELOG.md and linux/README.md for details.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
mARTin-B78 2026-06-03 22:44:28 +02:00
parent 6e9495e7e6
commit 83d4143e83
17 changed files with 1567 additions and 0 deletions

3
linux/.gitignore vendored Normal file
View File

@ -0,0 +1,3 @@
.venv/
__pycache__/
*.pyc

53
linux/CHANGELOG.md Normal file
View File

@ -0,0 +1,53 @@
# Changelog
All notable changes to **Blitztext for Linux** (the native dictation tool in
`linux/`) are documented here. The format follows
[Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project
adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
The version is defined in [`blitztext/__init__.py`](blitztext/__init__.py).
## [Unreleased]
## [1.0.0] - 2026-06-03
First release of the Linux port. The upstream project is a macOS-only menu-bar
app (Swift/SwiftUI, CoreML/WhisperKit) that cannot run on Linux or in a
container; this is a native host tool that reproduces the workflow — focus any
text field, press a hotkey, speak, and the (optionally rewritten) text is typed
into that field.
### Added
- **Native dictation engine** (`daemon.py`): global hotkeys via pynput, each
hotkey toggles record → transcribe → optional rewrite → deliver.
- **Local transcription** via faster-whisper (`transcribe.py`); `device="auto"`
tries CUDA and falls back to CPU `int8` (CPU-only on this arm64 host).
- **Microphone recording** (`recorder.py`) through pw-record / parecord /
arecord — 16 kHz mono WAV, no Python audio bindings required.
- **Optional LLM rewrite** (`rewrite.py`) against any OpenAI-compatible endpoint
(OpenAI, or a local vLLM / llama-swap), configurable per workflow.
- **Typing into the focused window** via xdotool (`paste.py`): `type` directly
or `paste` through the clipboard; re-activates the target window first.
- **Configurable workflows** (`config.py`) in `~/.config/blitztext/config.toml`,
with five defaults (Transcribe, Nicer email, Improve text, Calm down, Add
emojis), per-workflow prompt/model/temperature overrides, and a TOML writer.
- **Control-panel window** (`gui.py`, tkinter): workflow rows with monogram
avatars, hotkey badges, per-row Record buttons, a live status dot, and a
Settings window that edits config and can Save & Restart.
- **System-tray mode** (`tray.py`, AppIndicator): macOS-menu-bar-style status
icon with a menu to trigger each workflow, Show panel, Settings…, and Quit;
shares one daemon/model/hotkey set with the window. Falls back to the window
with an install hint when PyGObject is absent.
- **CLI** (`__main__.py`): `tray` (default), `gui`, `run`, `transcribe`,
`config-path`, and `--version`.
- **Packaging**: `install.sh` (venv with `--system-site-packages`),
`requirements.txt`, and a `blitztext.service` systemd user unit.
### Notes
- Targets an **X11** session (uses xdotool); Wayland would need ydotool/wtype.
- System tray requires a one-time `sudo apt install python3-gi` (the GTK /
AppIndicator typelibs and GNOME `ubuntu-appindicators` extension are already
present on the target host).
[Unreleased]: https://github.com/mARTin-B78/blitztext-app/compare/v1.0.0...HEAD
[1.0.0]: https://github.com/mARTin-B78/blitztext-app/releases/tag/v1.0.0

143
linux/README.md Normal file
View File

@ -0,0 +1,143 @@
# Blitztext for Linux (native dictation)
A native Linux port of the Blitztext workflow: **focus any text field → press a
hotkey → speak → the text is typed into that field**, optionally rewritten by an
LLM first (e.g. turn rough speech into a nicer, more detailed email).
This runs **on the host** (not in a container), so it can type into *any*
application — the Linux equivalent of the macOS app's Accessibility-based
auto-paste. (A sandboxed Docker/browser version can't do that; an earlier
experiment along those lines was moved out to
`~/Docker/correspondence/blitztext`.) Transcription is **local** via
[faster-whisper]; only the optional rewrite step calls out to an LLM.
## How it works
```
hotkey ──▶ record mic (pw-record/arecord) ──▶ faster-whisper (local)
┌── mode "transcribe" ────┤
│ └── mode "rewrite": LLM (OpenAI-compatible)
xdotool types it into the focused window
```
Each hotkey **toggles**: press to start recording, press again to stop — then it
transcribes, optionally rewrites, and types the result where your cursor is.
## Requirements
- **X11 session** (this uses `xdotool`; Wayland would need `ydotool`/`wtype`).
- Host tools: `xdotool`, `notify-send` (libnotify-bin), and a recorder
(`pw-record` from pipewire, or `arecord`/`parecord`).
```bash
sudo apt install xdotool libnotify-bin pipewire-bin
```
- Python 3.11+.
## Install
```bash
cd linux
./install.sh
```
This creates `.venv`, installs `faster-whisper` + `pynput`, and writes the
default config to `~/.config/blitztext/config.toml`.
## Run
Three front-ends, same engine (local Whisper + global hotkeys + xdotool typing):
```bash
# optional: only needed for the "rewrite" workflows
export OPENAI_API_KEY=sk-...
.venv/bin/python -m blitztext tray # system-tray menu (macOS-menu-bar-like, default)
.venv/bin/python -m blitztext gui # control-panel window
.venv/bin/python -m blitztext run # headless, hotkeys only
```
### System tray (recommended)
The tray is the closest match to the macOS menu-bar app: a status icon with a
menu listing every workflow (click to record), plus **Show panel**, **Settings…**,
and **Quit**. It needs PyGObject once (the GTK/AppIndicator typelibs and the
GNOME `ubuntu-appindicators` extension are already present here):
```bash
sudo apt install python3-gi # one-time; no build, just the bindings
.venv/bin/python -m blitztext tray
```
If PyGObject is missing, `tray` prints this hint and falls back to the window.
(The venv is created with `--system-site-packages` so it can see the
apt-installed `gi`.)
Either way, focus any text field and trigger a workflow — by tray menu, panel
button, or hotkey (defaults):
| Hotkey | Workflow | What it does |
| ----------------------- | ------------- | ----------------------------------------- |
| `Ctrl+Alt+Space` | Transcribe | Types the raw transcript |
| `Ctrl+Alt+E` | Nicer email | Rewrites speech into a polished email |
| `Ctrl+Alt+I` | Improve text | Proofreads / improves wording |
| `Ctrl+Alt+C` | Calm down | Rewrites an angry message into a calm one |
| `Ctrl+Alt+J` | Add emojis | Adds fitting emojis |
## Configuration
Everything lives in `~/.config/blitztext/config.toml` (`python -m blitztext
config-path` prints the location). You can change hotkeys, the Whisper model, and
the rewrite endpoint, and add/edit `[[workflow]]` blocks with your own prompts.
### Local Whisper
```toml
[whisper]
model = "small" # tiny|base|small|medium|large-v3, or a local model path
device = "auto" # auto tries cuda, falls back to cpu
compute_type = "auto"
```
> On this arm64 host the pip `ctranslate2` wheel is **CPU-only**, so it runs on
> the Grace CPU with `int8`. That's fast for dictation (≈2s for a 10s clip with
> `small`). `device = "auto"` attempts CUDA and falls back automatically — to get
> GPU you'd need a CUDA-enabled CTranslate2 build for aarch64/sm_121.
### Rewrite endpoint (OpenAI *or* your local LLM)
```toml
[rewrite]
base_url = "https://api.openai.com/v1" # or e.g. http://localhost:8000/v1 for vLLM/llama-swap
api_key_env = "OPENAI_API_KEY"
model = "gpt-4o-mini"
```
Point `base_url` at a local OpenAI-compatible server (vLLM, llama-swap) to keep
rewriting fully on-box too.
## Run on login
See [`blitztext.service`](blitztext.service) for a systemd **user** unit.
## Verified
On this machine (Ubuntu/GNOME, X11, GB10): recorder → valid 16 kHz WAV;
faster-whisper CPU transcription accurate; config + all hotkeys parse; and
`xdotool` typing of German text into a focused GTK field. The live global-hotkey
loop and the LLM rewrite HTTP call were not auto-tested here (the former hijacks
the keyboard during a session; the latter needs your key) — try them with the
`run` command above.
## CLI
```bash
python -m blitztext tray # tray menu (default)
python -m blitztext gui # control-panel window
python -m blitztext run # headless daemon, hotkeys only
python -m blitztext transcribe f.wav # one-shot, prints text (no hotkeys)
python -m blitztext config-path # print config location
```
[faster-whisper]: https://github.com/SYSTRAN/faster-whisper

26
linux/blitztext.service Normal file
View File

@ -0,0 +1,26 @@
# Blitztext user service.
# Install:
# mkdir -p ~/.config/systemd/user
# cp blitztext.service ~/.config/systemd/user/
# # edit ExecStart path below if your checkout differs
# systemctl --user daemon-reload
# systemctl --user enable --now blitztext
#
# A global-hotkey daemon needs the graphical session, hence the env below.
[Unit]
Description=Blitztext dictation daemon
After=graphical-session.target
PartOf=graphical-session.target
[Service]
Type=simple
# Adjust this path to your checkout location:
ExecStart=%h/Docker/blitztext-app/linux/.venv/bin/python -m blitztext run
Environment=DISPLAY=:1
# If you use rewrite workflows, set your key here or in the unit's environment:
# Environment=OPENAI_API_KEY=sk-...
Restart=on-failure
RestartSec=3
[Install]
WantedBy=graphical-session.target

View File

@ -0,0 +1,9 @@
"""Blitztext for Linux — native dictation daemon.
Press a global hotkey, speak, and the transcribed (optionally LLM-rewritten)
text is typed into whatever text field currently has focus. This is the Linux
counterpart to the macOS Blitztext menu bar app: it runs natively on the host
(not in a container) so it can type into any application via xdotool.
"""
__version__ = "1.0.0"

View File

@ -0,0 +1,58 @@
"""CLI entrypoint: `python -m blitztext [run|transcribe|config-path]`."""
from __future__ import annotations
import argparse
import sys
from pathlib import Path
from . import __version__
from .config import CONFIG_PATH, ensure_default, load
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="blitztext", description="Native dictation for Linux.")
parser.add_argument("--version", action="version", version=f"blitztext {__version__}")
sub = parser.add_subparsers(dest="cmd")
sub.add_parser("tray", help="Run in the system tray with a workflow menu (default).")
sub.add_parser("gui", help="Open the control panel window.")
sub.add_parser("run", help="Start the headless hotkey daemon (no window/tray).")
sub.add_parser("config-path", help="Print the config file path and exit.")
p_tx = sub.add_parser("transcribe", help="Transcribe a WAV file and print the text (no hotkeys).")
p_tx.add_argument("audio", type=Path)
args = parser.parse_args(argv)
cmd = args.cmd or "tray"
if cmd == "config-path":
print(ensure_default(CONFIG_PATH))
return 0
if cmd in ("gui", "tray"):
ensure_default(CONFIG_PATH)
from .gui import run_gui
return run_gui(tray_mode=(cmd == "tray"))
cfg = load()
if cmd == "transcribe":
from .transcribe import Transcriber
tx = Transcriber(cfg.model, cfg.device, cfg.compute_type, cfg.beam_size)
print(tx.transcribe(args.audio, language=cfg.language))
return 0
# cmd == "run"
from .daemon import Daemon
try:
Daemon(cfg).run()
except KeyboardInterrupt:
print("\n[blitztext] stopped.", file=sys.stderr)
return 0
if __name__ == "__main__":
raise SystemExit(main())

255
linux/blitztext/config.py Normal file
View File

@ -0,0 +1,255 @@
"""Configuration loading and the default config template."""
from __future__ import annotations
import os
import tomllib
from dataclasses import dataclass, field
from pathlib import Path
CONFIG_DIR = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / "blitztext"
CONFIG_PATH = CONFIG_DIR / "config.toml"
@dataclass
class Workflow:
name: str
hotkey: str
mode: str # "transcribe" | "rewrite"
prompt: str = ""
# Optional per-workflow overrides of the [rewrite] defaults.
model: str | None = None
temperature: float | None = None
# Cosmetic, used by the GUI.
description: str = ""
icon: str = ""
@dataclass
class Config:
# general
recorder: str = "auto"
output: str = "type" # type | paste
type_delay_ms: int = 12
notify: bool = True
language: str = "de" # whisper hint; "" = autodetect
# whisper
model: str = "small"
device: str = "auto" # auto | cuda | cpu
compute_type: str = "auto" # auto | int8 | float16 | int8_float16
beam_size: int = 5
# rewrite (OpenAI-compatible)
base_url: str = "https://api.openai.com/v1"
api_key_env: str = "OPENAI_API_KEY"
rewrite_model: str = "gpt-4o-mini"
temperature: float = 0.3
timeout: int = 45
# workflows
workflows: list[Workflow] = field(default_factory=list)
@property
def api_key(self) -> str | None:
return os.environ.get(self.api_key_env) or None
def load(path: Path = CONFIG_PATH) -> Config:
"""Load config from TOML, creating a default file on first run."""
if not path.exists():
ensure_default(path)
with path.open("rb") as fh:
data = tomllib.load(fh)
g = data.get("general", {})
w = data.get("whisper", {})
r = data.get("rewrite", {})
cfg = Config(
recorder=g.get("recorder", "auto"),
output=g.get("output", "type"),
type_delay_ms=int(g.get("type_delay_ms", 4)),
notify=bool(g.get("notify", True)),
language=g.get("language", "de"),
model=w.get("model", "small"),
device=w.get("device", "auto"),
compute_type=w.get("compute_type", "auto"),
beam_size=int(w.get("beam_size", 5)),
base_url=r.get("base_url", "https://api.openai.com/v1").rstrip("/"),
api_key_env=r.get("api_key_env", "OPENAI_API_KEY"),
rewrite_model=r.get("model", "gpt-4o-mini"),
temperature=float(r.get("temperature", 0.3)),
timeout=int(r.get("timeout", 45)),
)
for entry in data.get("workflow", []):
cfg.workflows.append(
Workflow(
name=entry["name"],
hotkey=entry["hotkey"],
mode=entry.get("mode", "transcribe"),
prompt=entry.get("prompt", ""),
model=entry.get("model"),
temperature=entry.get("temperature"),
description=entry.get("description", ""),
icon=entry.get("icon", ""),
)
)
if not cfg.workflows:
raise ValueError(f"No [[workflow]] entries defined in {path}")
return cfg
def save(cfg: Config, path: Path = CONFIG_PATH) -> None:
"""Write the config back to TOML (used by the settings UI).
Note: inline comments from the template are not preserved on save.
"""
import tomli_w
data: dict = {
"general": {
"recorder": cfg.recorder,
"output": cfg.output,
"type_delay_ms": cfg.type_delay_ms,
"notify": cfg.notify,
"language": cfg.language,
},
"whisper": {
"model": cfg.model,
"device": cfg.device,
"compute_type": cfg.compute_type,
"beam_size": cfg.beam_size,
},
"rewrite": {
"base_url": cfg.base_url,
"api_key_env": cfg.api_key_env,
"model": cfg.rewrite_model,
"temperature": cfg.temperature,
"timeout": cfg.timeout,
},
"workflow": [],
}
for wf in cfg.workflows:
entry: dict = {"name": wf.name, "hotkey": wf.hotkey, "mode": wf.mode}
if wf.prompt:
entry["prompt"] = wf.prompt
if wf.model:
entry["model"] = wf.model
if wf.temperature is not None:
entry["temperature"] = wf.temperature
if wf.description:
entry["description"] = wf.description
if wf.icon and wf.icon != "":
entry["icon"] = wf.icon
data["workflow"].append(entry)
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("wb") as fh:
tomli_w.dump(data, fh)
def ensure_default(path: Path = CONFIG_PATH) -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
if not path.exists():
path.write_text(DEFAULT_CONFIG, encoding="utf-8")
return path
# Hotkey syntax is pynput's GlobalHotKeys format, e.g. "<ctrl>+<alt>+space".
DEFAULT_CONFIG = """\
# Blitztext for Linux — configuration
# Hotkey format follows pynput: <ctrl> <alt> <shift> <cmd> + a letter/keyname.
# Each hotkey TOGGLES recording: press to start speaking, press again to finish.
[general]
recorder = "auto" # auto | pw-record | parecord | arecord
output = "type" # "type" = xdotool types it; "paste" = clipboard + Ctrl+V
type_delay_ms = 12 # per-keystroke delay for xdotool type (raise if chars drop)
notify = true # desktop notifications for each phase
language = "de" # Whisper language hint; "" = autodetect
[whisper]
model = "small" # tiny | base | small | medium | large-v3, or a local path
device = "auto" # auto | cuda | cpu (auto tries cuda, falls back to cpu)
compute_type = "auto" # auto | int8 | float16 | int8_float16
beam_size = 5
[rewrite]
# OpenAI-compatible chat endpoint. Define your own provider/API/model here.
# Point base_url at OpenAI, OR any local server that speaks the OpenAI chat API
# (vLLM, llama-swap, Ollama's /v1, LM Studio, ...). Examples:
# base_url = "https://api.openai.com/v1" (OpenAI)
# base_url = "http://localhost:8000/v1" (local vLLM / llama-swap)
# api_key_env names the ENV VAR holding the key (local servers often ignore it).
base_url = "https://api.openai.com/v1"
api_key_env = "OPENAI_API_KEY"
model = "gpt-4o-mini" # default model for rewrite workflows
temperature = 0.3
timeout = 45
# ----------------------------------------------------------------------------
# Workflows. mode = "transcribe" types the raw transcript. mode = "rewrite"
# sends the transcript through the LLM with `prompt` as the system prompt.
# Any workflow may override the [rewrite] defaults with its own:
# model = "gpt-4o"
# temperature = 0.4
# ----------------------------------------------------------------------------
[[workflow]]
name = "Transcribe"
icon = ""
description = "Speak, get plain text."
hotkey = "<ctrl>+<alt>+<space>"
mode = "transcribe"
[[workflow]]
name = "Nicer email"
icon = ""
description = "Rough notes → polished email."
hotkey = "<ctrl>+<alt>+e"
mode = "rewrite"
prompt = '''Du bist ein Schreibassistent fuer E-Mails. Du erhaeltst ein gesprochenes Transkript.
Schreibe daraus eine freundliche, gut formulierte und etwas ausfuehrlichere E-Mail:
- Korrigiere Rechtschreibung und Grammatik
- Formuliere hoeflich, klar und professionell
- Ergaenze sinnvolle Hoeflichkeitsfloskeln (Anrede/Gruss), wenn passend
- Behalte die urspruengliche Aussage und Absicht bei, erfinde keine Fakten
- Antworte in der Sprache des Transkripts
- Gib NUR den E-Mail-Text zurueck, keine Erklaerungen'''
[[workflow]]
name = "Improve text"
icon = ""
description = "Speak → cleaner writing."
hotkey = "<ctrl>+<alt>+i"
mode = "rewrite"
prompt = '''Du bist ein Lektor und Schreibassistent. Verbessere den folgenden gesprochenen Text:
- Korrigiere Rechtschreibung und Grammatik
- Verbessere Formulierung und Lesefluss, behalte die Bedeutung bei
- Antworte in der Sprache des Transkripts
- Gib NUR den verbesserten Text zurueck, keine Erklaerungen'''
[[workflow]]
name = "Calm down"
icon = ""
description = "Frustrated in → calm out."
hotkey = "<ctrl>+<alt>+c"
mode = "rewrite"
prompt = '''Du erhaeltst ein gesprochenes, frustriertes oder veraergertes Transkript.
Formuliere es in eine ruhige, sachliche und hoefliche Nachricht um, die dasselbe
Anliegen professionell vermittelt. Antworte in der Sprache des Transkripts.
Gib NUR die umformulierte Nachricht zurueck, keine Erklaerungen.'''
[[workflow]]
name = "Add emojis"
icon = ""
description = "Text in → emojis out."
hotkey = "<ctrl>+<alt>+j"
mode = "rewrite"
prompt = '''Du erhaeltst ein gesprochenes Transkript. Gib den Text moeglichst originalgetreu
zurueck, fuege aber regelmaessig passende Emojis ein (etwa alle 1-2 Saetze).
Korrigiere offensichtliche Fehler, behalte Stil und Bedeutung bei.
Gib NUR den Text mit Emojis zurueck, keine Erklaerungen.'''
"""

171
linux/blitztext/daemon.py Normal file
View File

@ -0,0 +1,171 @@
"""Hotkey-driven engine: toggle recording per workflow, then transcribe + deliver.
Used by both the headless CLI (`run`) and the tkinter GUI. A status callback
lets the GUI reflect each phase; desktop notifications fire regardless.
"""
from __future__ import annotations
import sys
import threading
from typing import Callable
from .config import Config, Workflow
from .notify import notify
from .paste import active_window_id, deliver
from .recorder import Recording, detect_recorder
from .rewrite import RewriteError, rewrite
from .transcribe import Transcriber
# status_cb(state, workflow_name, message)
# state in {"loading", "idle", "recording", "busy", "done", "error"}
StatusCallback = Callable[[str, str | None, str], None]
class Daemon:
def __init__(self, cfg: Config, status_cb: StatusCallback | None = None):
self.cfg = cfg
self.status_cb = status_cb
self._lock = threading.Lock()
self._recording: Recording | None = None
self._active_workflow: Workflow | None = None
self._target_window: str | None = None
self._busy = False
self._listener = None
self.recorder_name = detect_recorder(cfg.recorder)
self.transcriber: Transcriber | None = None
# -- feedback -------------------------------------------------------------
def _notify(self, title: str, body: str = "", urgency: str = "normal") -> None:
notify(title, body, urgency=urgency, enabled=self.cfg.notify)
def _emit(self, state: str, workflow: str | None = None, message: str = "") -> None:
if self.status_cb:
try:
self.status_cb(state, workflow, message)
except Exception: # noqa: BLE001 - never let UI errors break the engine
pass
# -- model load (slow; call off the UI thread) ----------------------------
def prepare(self) -> None:
self._emit("loading", None, f"Loading Whisper '{self.cfg.model}'")
self._notify("Loading model…", f"Whisper '{self.cfg.model}' ({self.cfg.device})")
self.transcriber = Transcriber(
model=self.cfg.model,
device=self.cfg.device,
compute_type=self.cfg.compute_type,
beam_size=self.cfg.beam_size,
)
self._emit("idle", None, "Ready")
@property
def ready(self) -> bool:
return self.transcriber is not None
@property
def is_recording(self) -> bool:
return self._recording is not None
# -- hotkey / button handler ----------------------------------------------
def toggle(self, workflow: Workflow) -> None:
"""Called on each trigger: start recording, or stop + process."""
with self._lock:
if not self.ready:
self._notify("Please wait", "Model still loading…", "low")
return
if self._busy:
self._notify("Busy", "Still processing the last clip…", "low")
return
if self._recording is None:
self._target_window = active_window_id()
self._recording = Recording(self.recorder_name)
self._active_workflow = workflow
self._emit("recording", workflow.name, "Recording…")
self._notify(f"{workflow.name}", "Recording… trigger again to stop.")
return
rec, wf, win = self._recording, self._active_workflow, self._target_window
self._recording = None
self._active_workflow = None
self._busy = True
audio_path = rec.stop()
threading.Thread(target=self._process, args=(audio_path, wf, win), daemon=True).start()
# -- worker ---------------------------------------------------------------
def _process(self, audio_path, workflow: Workflow, window_id) -> None:
try:
self._emit("busy", workflow.name, "Transcribing…")
self._notify(f"{workflow.name}", "Transcribing…")
text = self.transcriber.transcribe(audio_path, language=self.cfg.language)
if not text:
self._emit("idle", workflow.name, "No speech detected")
self._notify("Nothing heard", "No speech detected.", "low")
return
if workflow.mode == "rewrite" and workflow.prompt:
self._emit("busy", workflow.name, "Rewriting…")
self._notify(f"{workflow.name}", "Rewriting…")
try:
text = rewrite(
text,
workflow.prompt,
base_url=self.cfg.base_url,
api_key=self.cfg.api_key,
model=workflow.model or self.cfg.rewrite_model,
temperature=workflow.temperature if workflow.temperature is not None else self.cfg.temperature,
timeout=self.cfg.timeout,
)
except RewriteError as exc:
self._emit("error", workflow.name, str(exc))
self._notify("Rewrite failed", str(exc), "critical")
return
deliver(
text,
mode=self.cfg.output,
window_id=window_id,
type_delay_ms=self.cfg.type_delay_ms,
)
self._emit("done", workflow.name, text)
self._notify(f"{workflow.name}", text[:80] + ("" if len(text) > 80 else ""))
except Exception as exc: # noqa: BLE001 - surface any failure
self._emit("error", workflow.name, str(exc))
self._notify("Error", str(exc), "critical")
print(f"[blitztext] error: {exc}", file=sys.stderr)
finally:
audio_path.unlink(missing_ok=True)
with self._lock:
self._busy = False
self._emit("idle", None, "Ready")
# -- hotkeys --------------------------------------------------------------
def start_hotkeys(self):
"""Register global hotkeys non-blocking; returns the pynput listener."""
from pynput import keyboard
mapping = {wf.hotkey: (lambda wf=wf: self.toggle(wf)) for wf in self.cfg.workflows}
self._listener = keyboard.GlobalHotKeys(mapping)
self._listener.start()
return self._listener
def stop_hotkeys(self) -> None:
if self._listener is not None:
self._listener.stop()
self._listener = None
# -- headless run loop ----------------------------------------------------
def run(self) -> None:
self.prepare()
lines = "\n".join(f" {wf.hotkey}{wf.name}" for wf in self.cfg.workflows)
print(f"[blitztext] ready. Recorder: {self.recorder_name}. Hotkeys:\n{lines}", file=sys.stderr)
self._notify("Blitztext ready", "Focus a text field and press a hotkey.")
from pynput import keyboard
mapping = {wf.hotkey: (lambda wf=wf: self.toggle(wf)) for wf in self.cfg.workflows}
with keyboard.GlobalHotKeys(mapping) as listener:
listener.join()

389
linux/blitztext/gui.py Normal file
View File

@ -0,0 +1,389 @@
"""tkinter control panel for Blitztext — a Linux analogue of the macOS menu bar.
Shows the workflow list with per-row Record buttons and a live status dot, plus
a Settings window to edit hotkeys, the Whisper engine, the rewrite endpoint, and
each workflow's prompt. Global hotkeys keep working in the background, so the
panel is optional once it's running.
"""
from __future__ import annotations
import os
import queue
import sys
import threading
import tkinter as tk
from tkinter import messagebox, ttk
from . import __version__
from .config import Config, load, save
from .daemon import Daemon
# Light palette, loosely matching the macOS panel.
BG = "#ececf1"
CARD = "#ffffff"
TEXT = "#1c1c1e"
MUTED = "#8a8a8e"
BADGE = "#e3e3e8"
DIVIDER = "#d8d8de"
AVATAR_COLORS = ["#0a84ff", "#34c759", "#ff9f0a", "#ff375f", "#bf5af2", "#5ac8fa"]
DOT = {
"loading": "#ff9f0a",
"idle": "#34c759",
"recording": "#ff3b30",
"busy": "#ff9f0a",
"done": "#34c759",
"error": "#ff3b30",
}
DOT_LABEL = {
"loading": "Loading model…",
"idle": "Ready",
"recording": "Recording…",
"busy": "Working…",
"done": "Ready",
"error": "Error",
}
def pretty_hotkey(hotkey: str) -> str:
names = {"<ctrl>": "Ctrl", "<alt>": "Alt", "<shift>": "Shift", "<cmd>": "Super", "<space>": "Space"}
parts = []
for raw in hotkey.split("+"):
parts.append(names.get(raw, raw.strip("<>").upper() if len(raw.strip("<>")) == 1 else raw.strip("<>").title()))
return "+".join(parts)
class BlitztextGUI:
def __init__(self, cfg: Config, tray_mode: bool = False):
self.cfg = cfg
self.tray_mode = tray_mode
self.tray = None
self.daemon = Daemon(cfg, status_cb=self._status_cb)
self._events: queue.Queue = queue.Queue()
self._active_wf: str | None = None
self._row_buttons: dict[str, tk.Button] = {}
self.root = tk.Tk()
self.root.title("Blitztext")
self.root.configure(bg=BG)
self.root.minsize(420, 360)
self._build_header()
self._build_rows()
self._build_footer()
# Disable controls until the model is loaded.
self._set_buttons_enabled(False)
self.root.after(80, self._drain_events)
# Load model + start hotkeys off the UI thread.
threading.Thread(target=self._startup, daemon=True).start()
self.root.protocol("WM_DELETE_WINDOW", self._on_close)
# -- startup --------------------------------------------------------------
def _startup(self) -> None:
try:
self.daemon.prepare()
self.daemon.start_hotkeys()
except Exception as exc: # noqa: BLE001
self._events.put(("error", None, f"Startup failed: {exc}"))
# -- header ---------------------------------------------------------------
def _build_header(self) -> None:
head = tk.Frame(self.root, bg=BG)
head.pack(fill="x", padx=16, pady=(14, 8))
tk.Label(head, text="Blitztext", font=("", 15, "bold"), bg=BG, fg=TEXT).pack(side="left")
status = tk.Frame(head, bg=BG)
status.pack(side="right")
self.dot = tk.Label(status, text="", font=("", 12), bg=BG, fg=DOT["loading"])
self.dot.pack(side="left")
self.status_label = tk.Label(status, text="Starting…", font=("", 11), bg=BG, fg=MUTED)
self.status_label.pack(side="left", padx=(4, 0))
# -- workflow rows --------------------------------------------------------
def _build_rows(self) -> None:
body = tk.Frame(self.root, bg=CARD, highlightthickness=1, highlightbackground=DIVIDER)
body.pack(fill="both", expand=True, padx=12, pady=4)
for i, wf in enumerate(self.cfg.workflows):
if i:
tk.Frame(body, bg=DIVIDER, height=1).pack(fill="x", padx=12)
row = tk.Frame(body, bg=CARD)
row.pack(fill="x", padx=12, pady=8)
color = AVATAR_COLORS[i % len(AVATAR_COLORS)]
avatar = tk.Canvas(row, width=32, height=32, bg=CARD, highlightthickness=0)
avatar.create_oval(3, 3, 29, 29, fill=color, outline="")
avatar.create_text(16, 16, text=(wf.name[:1] or "?").upper(), fill="white", font=("", 13, "bold"))
avatar.pack(side="left", padx=(0, 10))
mid = tk.Frame(row, bg=CARD)
mid.pack(side="left", fill="x", expand=True)
tk.Label(mid, text=wf.name, font=("", 12, "bold"), bg=CARD, fg=TEXT, anchor="w").pack(fill="x")
sub = wf.description or ("Transcribe only" if wf.mode == "transcribe" else "Transcribe → rewrite")
tk.Label(mid, text=sub, font=("", 10), bg=CARD, fg=MUTED, anchor="w").pack(fill="x")
badge = tk.Label(row, text=pretty_hotkey(wf.hotkey), font=("", 9), bg=BADGE, fg=MUTED, padx=6, pady=2)
badge.pack(side="left", padx=8)
btn = tk.Button(
row, text="● Rec", font=("", 10, "bold"), width=7,
relief="flat", bg="#e8453c", fg="white", activebackground="#c93b33", activeforeground="white",
command=lambda w=wf: self._on_record(w),
)
btn.pack(side="right")
self._row_buttons[wf.name] = btn
# -- footer ---------------------------------------------------------------
def _build_footer(self) -> None:
foot = tk.Frame(self.root, bg=BG)
foot.pack(fill="x", padx=16, pady=(6, 12))
tk.Label(foot, text=f"recorder: {self.daemon.recorder_name} · v{__version__}",
font=("", 9), bg=BG, fg=MUTED).pack(side="left")
tk.Button(foot, text="Quit", font=("", 10), relief="flat", bg=BADGE, fg=TEXT,
command=self.quit_all).pack(side="right")
tk.Button(foot, text="Settings", font=("", 10), relief="flat", bg=BADGE, fg=TEXT,
command=self.open_settings).pack(side="right", padx=6)
# -- record button --------------------------------------------------------
def _on_record(self, wf) -> None:
self.trigger_workflow(wf)
def trigger_workflow(self, wf) -> None:
"""Start/stop a workflow (from a row button or the tray menu)."""
if not self.daemon.ready:
return
threading.Thread(target=lambda: self.daemon.toggle(wf), daemon=True).start()
def _set_buttons_enabled(self, enabled: bool) -> None:
state = "normal" if enabled else "disabled"
for btn in self._row_buttons.values():
btn.configure(state=state)
# -- status plumbing (thread-safe via queue) ------------------------------
def _status_cb(self, state: str, workflow: str | None, message: str) -> None:
self._events.put((state, workflow, message))
def _drain_events(self) -> None:
try:
while True:
state, workflow, message = self._events.get_nowait()
self._apply_status(state, workflow, message)
except queue.Empty:
pass
self.root.after(80, self._drain_events)
def _apply_status(self, state: str, workflow: str | None, message: str) -> None:
label = DOT_LABEL.get(state, message) or message
self.dot.configure(fg=DOT.get(state, MUTED))
self.status_label.configure(text=label)
if self.tray is not None:
self.tray.update_status(state, label)
if state == "recording":
self._active_wf = workflow
for name, btn in self._row_buttons.items():
if name == workflow:
btn.configure(text="■ Stop", state="normal")
else:
btn.configure(state="disabled")
elif state in ("busy", "loading"):
self._set_buttons_enabled(False)
elif state in ("idle", "done", "error"):
self._active_wf = None
for btn in self._row_buttons.values():
btn.configure(text="● Rec", state="normal")
if state == "error" and message:
self.status_label.configure(text=message[:48])
# -- settings / panel -----------------------------------------------------
def open_settings(self) -> None:
SettingsWindow(self.root, self.cfg)
def show_panel(self) -> None:
self.root.deiconify()
self.root.lift()
self.root.focus_force()
def hide_panel(self) -> None:
self.root.withdraw()
# -- lifecycle ------------------------------------------------------------
def _on_close(self) -> None:
# In tray mode the window close button just hides to the tray.
if self.tray is not None:
self.hide_panel()
else:
self.quit_all()
def quit_all(self) -> None:
try:
self.daemon.stop_hotkeys()
finally:
self.root.destroy()
def _pump_gtk(self) -> None:
if self.tray is not None:
self.tray.pump()
self.root.after(50, self._pump_gtk)
def run(self) -> None:
if self.tray_mode:
from . import tray as tray_mod
if tray_mod.gi_available():
self.tray = tray_mod.Tray(self)
self.root.withdraw() # live in the tray; panel opens on demand
self._pump_gtk()
else:
print(tray_mod.INSTALL_HINT, file=sys.stderr)
self.tray_mode = False
self.root.mainloop()
class SettingsWindow:
"""Edit config and save. Hotkey/model changes apply after a restart."""
def __init__(self, parent: tk.Misc, cfg: Config):
self.cfg = cfg
self.win = tk.Toplevel(parent)
self.win.title("Blitztext — Settings")
self.win.configure(bg=BG)
self.win.minsize(560, 480)
nb = ttk.Notebook(self.win)
nb.pack(fill="both", expand=True, padx=10, pady=10)
self.vars: dict[str, tk.Variable] = {}
self._build_general(nb)
self._build_rewrite(nb)
self.prompt_texts: dict[int, tk.Text] = {}
self.wf_vars: dict[int, dict[str, tk.Variable]] = {}
for idx, wf in enumerate(cfg.workflows):
self._build_workflow_tab(nb, idx, wf)
bar = tk.Frame(self.win, bg=BG)
bar.pack(fill="x", padx=10, pady=(0, 10))
tk.Button(bar, text="Save & Restart", font=("", 10, "bold"), relief="flat",
bg="#0a84ff", fg="white", command=self._save_restart).pack(side="right")
tk.Button(bar, text="Save", font=("", 10), relief="flat", bg=BADGE, fg=TEXT,
command=self._save).pack(side="right", padx=6)
tk.Label(bar, text="Hotkey & model changes apply after restart.",
font=("", 9), bg=BG, fg=MUTED).pack(side="left")
def _field(self, parent, label, value, key, width=42):
frame = tk.Frame(parent, bg=BG)
frame.pack(fill="x", padx=14, pady=5)
tk.Label(frame, text=label, font=("", 10), bg=BG, fg=TEXT, width=16, anchor="w").pack(side="left")
var = tk.StringVar(value=str(value))
self.vars[key] = var
tk.Entry(frame, textvariable=var, width=width).pack(side="left", fill="x", expand=True)
return var
def _build_general(self, nb):
tab = tk.Frame(nb, bg=BG)
nb.add(tab, text="Engine")
self._field(tab, "Whisper model", self.cfg.model, "model")
self._combo(tab, "Device", self.cfg.device, ["auto", "cpu", "cuda"], "device")
self._combo(tab, "Compute type", self.cfg.compute_type, ["auto", "int8", "float16", "int8_float16"], "compute_type")
self._field(tab, "Language hint", self.cfg.language, "language")
self._combo(tab, "Output", self.cfg.output, ["type", "paste"], "output")
self._field(tab, "Type delay (ms)", self.cfg.type_delay_ms, "type_delay_ms")
nv = tk.BooleanVar(value=self.cfg.notify)
self.vars["notify"] = nv
f = tk.Frame(tab, bg=BG); f.pack(fill="x", padx=14, pady=5)
tk.Checkbutton(f, text="Desktop notifications", variable=nv, bg=BG, fg=TEXT,
activebackground=BG, selectcolor=CARD).pack(side="left")
def _build_rewrite(self, nb):
tab = tk.Frame(nb, bg=BG)
nb.add(tab, text="Rewrite LLM")
self._field(tab, "Base URL", self.cfg.base_url, "base_url")
self._field(tab, "API key env var", self.cfg.api_key_env, "api_key_env")
self._field(tab, "Model", self.cfg.rewrite_model, "rewrite_model")
self._field(tab, "Temperature", self.cfg.temperature, "temperature")
self._field(tab, "Timeout (s)", self.cfg.timeout, "timeout")
present = "set ✓" if self.cfg.api_key else "NOT set"
tk.Label(tab, text=f"OpenAI-compatible endpoint (OpenAI, vLLM, llama-swap…). "
f"Env {self.cfg.api_key_env}: {present}.",
font=("", 9), bg=BG, fg=MUTED, wraplength=500, justify="left").pack(fill="x", padx=14, pady=(8, 0))
def _build_workflow_tab(self, nb, idx, wf):
tab = tk.Frame(nb, bg=BG)
nb.add(tab, text=wf.name[:14])
v: dict[str, tk.Variable] = {}
self.wf_vars[idx] = v
for label, key, val in [("Name", "name", wf.name),
("Description", "description", wf.description),
("Hotkey", "hotkey", wf.hotkey),
("Model (opt.)", "model", wf.model or ""),
("Temp (opt.)", "temperature", "" if wf.temperature is None else wf.temperature)]:
frame = tk.Frame(tab, bg=BG); frame.pack(fill="x", padx=14, pady=4)
tk.Label(frame, text=label, font=("", 10), bg=BG, fg=TEXT, width=14, anchor="w").pack(side="left")
sv = tk.StringVar(value=str(val))
v[key] = sv
tk.Entry(frame, textvariable=sv).pack(side="left", fill="x", expand=True)
mframe = tk.Frame(tab, bg=BG); mframe.pack(fill="x", padx=14, pady=4)
tk.Label(mframe, text="Mode", font=("", 10), bg=BG, fg=TEXT, width=14, anchor="w").pack(side="left")
mv = tk.StringVar(value=wf.mode); v["mode"] = mv
ttk.Combobox(mframe, textvariable=mv, values=["transcribe", "rewrite"], state="readonly", width=14).pack(side="left")
tk.Label(tab, text="Rewrite prompt (system):", font=("", 10), bg=BG, fg=TEXT, anchor="w").pack(fill="x", padx=14, pady=(8, 2))
txt = tk.Text(tab, height=10, wrap="word")
txt.insert("1.0", wf.prompt)
txt.pack(fill="both", expand=True, padx=14, pady=(0, 10))
self.prompt_texts[idx] = txt
def _combo(self, parent, label, value, options, key):
frame = tk.Frame(parent, bg=BG); frame.pack(fill="x", padx=14, pady=5)
tk.Label(frame, text=label, font=("", 10), bg=BG, fg=TEXT, width=16, anchor="w").pack(side="left")
var = tk.StringVar(value=value); self.vars[key] = var
ttk.Combobox(frame, textvariable=var, values=options, state="readonly", width=18).pack(side="left")
return var
def _collect(self) -> bool:
try:
self.cfg.model = self.vars["model"].get().strip()
self.cfg.device = self.vars["device"].get()
self.cfg.compute_type = self.vars["compute_type"].get()
self.cfg.language = self.vars["language"].get().strip()
self.cfg.output = self.vars["output"].get()
self.cfg.type_delay_ms = int(self.vars["type_delay_ms"].get())
self.cfg.notify = bool(self.vars["notify"].get())
self.cfg.base_url = self.vars["base_url"].get().strip().rstrip("/")
self.cfg.api_key_env = self.vars["api_key_env"].get().strip()
self.cfg.rewrite_model = self.vars["rewrite_model"].get().strip()
self.cfg.temperature = float(self.vars["temperature"].get())
self.cfg.timeout = int(self.vars["timeout"].get())
for idx, wf in enumerate(self.cfg.workflows):
v = self.wf_vars[idx]
wf.name = v["name"].get().strip() or wf.name
wf.description = v["description"].get().strip()
wf.hotkey = v["hotkey"].get().strip()
wf.mode = v["mode"].get()
wf.model = v["model"].get().strip() or None
temp = v["temperature"].get().strip()
wf.temperature = float(temp) if temp else None
wf.prompt = self.prompt_texts[idx].get("1.0", "end").strip()
except ValueError as exc:
messagebox.showerror("Invalid value", f"Check numeric fields: {exc}", parent=self.win)
return False
return True
def _save(self) -> None:
if not self._collect():
return
save(self.cfg)
messagebox.showinfo("Saved", "Settings saved. Restart Blitztext to apply hotkey/model changes.", parent=self.win)
def _save_restart(self) -> None:
if not self._collect():
return
save(self.cfg)
os.execv(sys.executable, [sys.executable, "-m", "blitztext", "gui"])
def run_gui(tray_mode: bool = False) -> int:
cfg = load()
BlitztextGUI(cfg, tray_mode=tray_mode).run()
return 0

34
linux/blitztext/notify.py Normal file
View File

@ -0,0 +1,34 @@
"""Lightweight desktop feedback via notify-send, with stdout fallback."""
from __future__ import annotations
import shutil
import subprocess
import sys
_HAVE_NOTIFY = shutil.which("notify-send") is not None
_APP = "Blitztext"
# Reuse one notification bubble instead of stacking them.
_REPLACE_ID = "99317"
def notify(title: str, body: str = "", *, urgency: str = "normal", enabled: bool = True) -> None:
if enabled and _HAVE_NOTIFY:
try:
subprocess.run(
[
"notify-send",
"--app-name", _APP,
"--urgency", urgency,
"--hint", f"string:x-canonical-private-synchronous:{_REPLACE_ID}",
title,
body,
],
check=False,
)
return
except OSError:
pass
# Always echo to the console too — useful when running in a terminal.
line = f"[{_APP}] {title}" + (f"{body}" if body else "")
print(line, file=sys.stderr, flush=True)

66
linux/blitztext/paste.py Normal file
View File

@ -0,0 +1,66 @@
"""Deliver text into the focused window via xdotool (X11).
Two strategies:
- "type" : xdotool types the text directly (no clipboard side effects).
- "paste" : put text on the clipboard, then send Ctrl+V.
The target window is captured when recording starts and re-activated before
delivery, so a brief focus change during processing doesn't misfire.
"""
from __future__ import annotations
import shutil
import subprocess
import time
def active_window_id() -> str | None:
if not shutil.which("xdotool"):
return None
try:
out = subprocess.run(
["xdotool", "getactivewindow"],
capture_output=True, text=True, check=True,
)
return out.stdout.strip() or None
except (OSError, subprocess.CalledProcessError):
return None
def _focus(window_id: str | None) -> None:
if window_id:
subprocess.run(["xdotool", "windowactivate", "--sync", window_id], check=False)
time.sleep(0.05)
def deliver(text: str, *, mode: str = "type", window_id: str | None = None, type_delay_ms: int = 4) -> None:
if not text:
return
if not shutil.which("xdotool"):
raise RuntimeError("xdotool not found; cannot type into the focused window.")
_focus(window_id)
# Give the user time to release the hotkey modifiers before we synthesize input.
time.sleep(0.12)
if mode == "paste" and _set_clipboard(text):
subprocess.run(["xdotool", "key", "--clearmodifiers", "ctrl+v"], check=False)
return
subprocess.run(
["xdotool", "type", "--clearmodifiers", "--delay", str(type_delay_ms), "--", text],
check=False,
)
def _set_clipboard(text: str) -> bool:
"""Best-effort clipboard set; returns False if no clipboard tool is available."""
for argv in (["xclip", "-selection", "clipboard"], ["xsel", "--clipboard", "--input"], ["wl-copy"]):
if shutil.which(argv[0]):
try:
subprocess.run(argv, input=text.encode("utf-8"), check=True)
return True
except (OSError, subprocess.CalledProcessError):
continue
return False

View File

@ -0,0 +1,63 @@
"""Microphone recording by shelling out to a system recorder.
Records 16 kHz mono WAV (what Whisper wants) using whichever recorder is
available no Python audio bindings or device configuration required.
"""
from __future__ import annotations
import shutil
import signal
import subprocess
import tempfile
from pathlib import Path
# Ordered by preference. Each maps a recorder name to an argv builder.
_RECORDERS: dict[str, list[str]] = {
"pw-record": ["pw-record", "--rate", "16000", "--channels", "1", "--format", "s16"],
"parecord": ["parecord", "--rate=16000", "--channels=1", "--format=s16le", "--file-format=wav"],
"arecord": ["arecord", "-q", "-f", "S16_LE", "-r", "16000", "-c", "1", "-t", "wav"],
}
def detect_recorder(preference: str = "auto") -> str:
if preference != "auto":
if shutil.which(preference):
return preference
raise RuntimeError(f"Configured recorder '{preference}' not found on PATH.")
for name in _RECORDERS:
if shutil.which(name):
return name
raise RuntimeError("No recorder found (need one of: pw-record, parecord, arecord).")
class Recording:
"""A single in-progress recording. Start on construction, call stop()."""
def __init__(self, recorder: str):
self._tmp = Path(tempfile.mkstemp(prefix="blitztext-", suffix=".wav")[1])
argv = list(_RECORDERS[recorder]) + [str(self._tmp)]
# arecord writes to the file given as a positional arg; the others too.
self._proc = subprocess.Popen(
argv,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
def stop(self) -> Path:
"""Stop recording, finalize the WAV, and return its path."""
if self._proc.poll() is None:
# SIGINT lets pw-record/arecord flush the WAV header cleanly.
self._proc.send_signal(signal.SIGINT)
try:
self._proc.wait(timeout=5)
except subprocess.TimeoutExpired:
self._proc.terminate()
self._proc.wait(timeout=5)
return self._tmp
def discard(self) -> None:
if self._proc.poll() is None:
self._proc.terminate()
self._proc.wait(timeout=5)
self._tmp.unlink(missing_ok=True)

View File

@ -0,0 +1,73 @@
"""Optional LLM rewrite via an OpenAI-compatible chat endpoint.
Uses only the standard library (urllib) so the daemon needs no extra deps for
the rewrite step. Works against OpenAI or any local server that implements the
/chat/completions API (e.g. vLLM, llama-swap).
"""
from __future__ import annotations
import json
import urllib.error
import urllib.request
class RewriteError(RuntimeError):
pass
def rewrite(
text: str,
system_prompt: str,
*,
base_url: str,
api_key: str | None,
model: str,
temperature: float = 0.3,
timeout: int = 45,
) -> str:
if not api_key:
raise RewriteError(
"No API key for rewrite. Set the env var named by rewrite.api_key_env, "
"or use a local endpoint that ignores auth."
)
payload = json.dumps(
{
"model": model,
"temperature": temperature,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": text},
],
}
).encode("utf-8")
req = urllib.request.Request(
f"{base_url}/chat/completions",
data=payload,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
body = json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", "replace")
raise RewriteError(f"HTTP {exc.code}: {detail[:300]}") from exc
except urllib.error.URLError as exc:
raise RewriteError(f"Connection failed: {exc.reason}") from exc
try:
content = body["choices"][0]["message"]["content"]
except (KeyError, IndexError, TypeError) as exc:
raise RewriteError(f"Unexpected response: {str(body)[:300]}") from exc
content = (content or "").strip()
if not content:
raise RewriteError("Empty response from model.")
return content

View File

@ -0,0 +1,59 @@
"""Local speech-to-text via faster-whisper (CTranslate2).
The model is loaded once and reused. On this arm64 host the CTranslate2 wheel is
CPU-only, so device="auto" attempts CUDA and falls back to CPU automatically.
"""
from __future__ import annotations
import sys
from pathlib import Path
class Transcriber:
def __init__(
self,
model: str = "small",
device: str = "auto",
compute_type: str = "auto",
beam_size: int = 5,
):
self.beam_size = beam_size
self._model = self._load(model, device, compute_type)
@staticmethod
def _resolve_compute_type(device: str, requested: str) -> str:
if requested != "auto":
return requested
return "float16" if device == "cuda" else "int8"
def _load(self, model: str, device: str, compute_type: str):
from faster_whisper import WhisperModel
attempts: list[str]
if device == "auto":
attempts = ["cuda", "cpu"]
else:
attempts = [device]
last_err: Exception | None = None
for dev in attempts:
try:
ct = self._resolve_compute_type(dev, compute_type)
m = WhisperModel(model, device=dev, compute_type=ct)
print(f"[blitztext] Whisper '{model}' loaded on {dev} ({ct})", file=sys.stderr)
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)
raise RuntimeError(f"Failed to load Whisper model '{model}': {last_err}")
def transcribe(self, audio_path: Path, language: str = "") -> str:
segments, _info = self._model.transcribe(
str(audio_path),
language=language or None,
beam_size=self.beam_size,
vad_filter=True,
)
return " ".join(seg.text.strip() for seg in segments).strip()

114
linux/blitztext/tray.py Normal file
View File

@ -0,0 +1,114 @@
"""System-tray (AppIndicator) integration — the macOS-menu-bar-style entry.
Requires PyGObject (apt: python3-gi); the AyatanaAppIndicator3 + GTK-3 typelibs
are already present on this system. The tray coexists with the tkinter panel by
pumping the GLib main context from tkinter's event loop, so both share one
daemon, one Whisper model, and one set of global hotkeys.
"""
from __future__ import annotations
INSTALL_HINT = (
"The system tray needs PyGObject, which isn't installed.\n"
" Install it (no build required): sudo apt install python3-gi\n"
"The GTK/AppIndicator typelibs are already present. Then run:\n"
" python -m blitztext tray\n"
"Meanwhile, `python -m blitztext gui` (window) and `run` (headless) work now."
)
# Symbolic theme icons used for the panel indicator per state.
ICONS = {
"loading": "content-loading-symbolic",
"idle": "audio-input-microphone-symbolic",
"recording": "media-record-symbolic",
"busy": "content-loading-symbolic",
"done": "audio-input-microphone-symbolic",
"error": "dialog-error-symbolic",
}
def gi_available() -> bool:
"""True if PyGObject + an AppIndicator + GTK-3 are importable."""
try:
import gi
gi.require_version("Gtk", "3.0")
try:
gi.require_version("AyatanaAppIndicator3", "0.1")
except ValueError:
gi.require_version("AppIndicator3", "0.1")
return True
except (ImportError, ValueError):
return False
class Tray:
"""An AppIndicator with a workflow menu, driven by a BlitztextGUI app."""
def __init__(self, app):
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk
try:
gi.require_version("AyatanaAppIndicator3", "0.1")
from gi.repository import AyatanaAppIndicator3 as AppIndicator3
except ValueError:
gi.require_version("AppIndicator3", "0.1")
from gi.repository import AppIndicator3
self._Gtk = Gtk
self.app = app
self.indicator = AppIndicator3.Indicator.new(
"blitztext",
ICONS["idle"],
AppIndicator3.IndicatorCategory.APPLICATION_STATUS,
)
self.indicator.set_status(AppIndicator3.IndicatorStatus.ACTIVE)
self.indicator.set_title("Blitztext")
self._build_menu()
def _build_menu(self) -> None:
from .gui import pretty_hotkey
Gtk = self._Gtk
menu = Gtk.Menu()
self.status_item = Gtk.MenuItem(label="● Ready")
self.status_item.set_sensitive(False)
menu.append(self.status_item)
menu.append(Gtk.SeparatorMenuItem())
for wf in self.app.cfg.workflows:
item = Gtk.MenuItem(label=f"{wf.name}({pretty_hotkey(wf.hotkey)})")
item.connect("activate", lambda _i, w=wf: self.app.trigger_workflow(w))
menu.append(item)
menu.append(Gtk.SeparatorMenuItem())
for label, cb in (
("Show panel", self.app.show_panel),
("Settings…", self.app.open_settings),
):
it = Gtk.MenuItem(label=label)
it.connect("activate", lambda _i, c=cb: c())
menu.append(it)
menu.append(Gtk.SeparatorMenuItem())
quit_item = Gtk.MenuItem(label="Quit Blitztext")
quit_item.connect("activate", lambda _i: self.app.quit_all())
menu.append(quit_item)
menu.show_all()
self.menu = menu
self.indicator.set_menu(menu)
def update_status(self, state: str, message: str) -> None:
self.indicator.set_icon_full(ICONS.get(state, ICONS["idle"]), state)
self.status_item.set_label(f"{message or state.title()}")
def pump(self) -> None:
"""Service pending GLib/GTK events; called from tkinter's loop."""
Gtk = self._Gtk
while Gtk.events_pending():
Gtk.main_iteration_do(False)

49
linux/install.sh Executable file
View File

@ -0,0 +1,49 @@
#!/usr/bin/env bash
# Set up Blitztext for Linux: create a venv, install deps, check host tools,
# and write the default config. Run from the linux/ directory.
set -euo pipefail
cd "$(dirname "$0")"
VENV=".venv"
echo "==> Checking host tools"
need_pkg=()
command -v xdotool >/dev/null || need_pkg+=("xdotool")
command -v notify-send >/dev/null || need_pkg+=("libnotify-bin")
if ! command -v pw-record >/dev/null && ! command -v arecord >/dev/null && ! command -v parecord >/dev/null; then
need_pkg+=("pipewire-bin (or alsa-utils)")
fi
if ((${#need_pkg[@]})); then
echo " Missing host tools: ${need_pkg[*]}"
echo " On Ubuntu/Debian: sudo apt install xdotool libnotify-bin pipewire-bin"
echo " (continuing — install them before running the daemon)"
fi
# --system-site-packages lets the venv see an apt-installed PyGObject (python3-gi)
# for the optional system-tray mode; harmless if it's not installed.
echo "==> Creating venv at $VENV"
python3 -m venv --system-site-packages "$VENV"
"$VENV/bin/pip" install --upgrade pip -q
"$VENV/bin/pip" install -r requirements.txt
echo "==> Writing default config (if absent)"
"$VENV/bin/python" -m blitztext config-path
cat <<EOF
Done. To start it:
$VENV/bin/python -m blitztext tray # system tray (needs: sudo apt install python3-gi)
$VENV/bin/python -m blitztext gui # control-panel window
$VENV/bin/python -m blitztext run # headless, hotkeys only
Edit your config (hotkeys, Whisper model, rewrite endpoint) at:
\$($VENV/bin/python -m blitztext config-path)
For the rewrite workflows, export your key first, e.g.:
export OPENAI_API_KEY=sk-...
To run it in the background on login, see blitztext.service in this folder.
EOF

2
linux/requirements.txt Normal file
View File

@ -0,0 +1,2 @@
faster-whisper>=1.0
pynput>=1.7