diff --git a/linux/.gitignore b/linux/.gitignore new file mode 100644 index 0000000..77ac754 --- /dev/null +++ b/linux/.gitignore @@ -0,0 +1,3 @@ +.venv/ +__pycache__/ +*.pyc diff --git a/linux/CHANGELOG.md b/linux/CHANGELOG.md new file mode 100644 index 0000000..f2b548b --- /dev/null +++ b/linux/CHANGELOG.md @@ -0,0 +1,63 @@ +# 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.1] - 2026-06-03 + +### Changed +- Redesigned the control-panel window: minimal flat layout, Ubuntu font + throughout, clickable workflow rows with hover (click to record / stop), + subtle dividers, and text-style Settings/Quit actions. Dropped the monogram + avatars and per-row buttons in favour of a cleaner, simpler look. The Settings + window picks up the same font and styling. + +## [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.1...HEAD +[1.0.1]: https://github.com/mARTin-B78/blitztext-app/compare/v1.0.0...v1.0.1 +[1.0.0]: https://github.com/mARTin-B78/blitztext-app/releases/tag/v1.0.0 diff --git a/linux/README.md b/linux/README.md new file mode 100644 index 0000000..76a9f93 --- /dev/null +++ b/linux/README.md @@ -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 diff --git a/linux/blitztext.service b/linux/blitztext.service new file mode 100644 index 0000000..3564531 --- /dev/null +++ b/linux/blitztext.service @@ -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 diff --git a/linux/blitztext/__init__.py b/linux/blitztext/__init__.py new file mode 100644 index 0000000..1689a27 --- /dev/null +++ b/linux/blitztext/__init__.py @@ -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.1" diff --git a/linux/blitztext/__main__.py b/linux/blitztext/__main__.py new file mode 100644 index 0000000..c928f65 --- /dev/null +++ b/linux/blitztext/__main__.py @@ -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()) diff --git a/linux/blitztext/config.py b/linux/blitztext/config.py new file mode 100644 index 0000000..bc11962 --- /dev/null +++ b/linux/blitztext/config.py @@ -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. "++space". +DEFAULT_CONFIG = """\ +# Blitztext for Linux — configuration +# Hotkey format follows pynput: + 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 = "++" +mode = "transcribe" + +[[workflow]] +name = "Nicer email" +icon = "✉" +description = "Rough notes → polished email." +hotkey = "++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 = "++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 = "++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 = "++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.''' +""" diff --git a/linux/blitztext/daemon.py b/linux/blitztext/daemon.py new file mode 100644 index 0000000..ecb1797 --- /dev/null +++ b/linux/blitztext/daemon.py @@ -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() diff --git a/linux/blitztext/gui.py b/linux/blitztext/gui.py new file mode 100644 index 0000000..242af9e --- /dev/null +++ b/linux/blitztext/gui.py @@ -0,0 +1,472 @@ +"""tkinter control panel for Blitztext — a Linux analogue of the macOS menu bar. + +A minimal, flat design: clickable workflow rows with hover, a single status dot, +and the Ubuntu font throughout. 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 +import tkinter.font as tkfont +from tkinter import messagebox, ttk + +from . import __version__ +from .config import Config, load, save +from .daemon import Daemon + +# --- palette (light, minimal) ------------------------------------------------ +WIN = "#ffffff" +TEXT = "#1a1a1c" +SUBTLE = "#8e8e93" +FAINT = "#b8b8be" +HOVER = "#f5f5f7" +LINE = "#ececee" +ACCENT = "#0a84ff" +GREEN = "#34c759" +RED = "#ff3b30" +AMBER = "#ff9f0a" + +DOT = {"loading": AMBER, "idle": GREEN, "recording": RED, "busy": AMBER, "done": GREEN, "error": RED} +DOT_LABEL = {"loading": "Loading…", "idle": "Ready", "recording": "Recording", + "busy": "Working…", "done": "Ready", "error": "Error"} + +_FONT = "TkDefaultFont" +_MONO = "TkFixedFont" + + +def _pick_fonts(root: tk.Tk) -> None: + """Apply the nicest available UI/mono fonts to the default named fonts.""" + global _FONT, _MONO + fams = set(tkfont.families(root)) + for f in ("Ubuntu", "Cantarell", "Noto Sans", "DejaVu Sans"): + if f in fams: + _FONT = f + break + for f in ("Ubuntu Mono", "DejaVu Sans Mono", "Noto Sans Mono"): + if f in fams: + _MONO = f + break + if _FONT != "TkDefaultFont": + for name in ("TkDefaultFont", "TkTextFont", "TkMenuFont", "TkHeadingFont"): + try: + tkfont.nametofont(name).configure(family=_FONT, size=11) + except tk.TclError: + pass + + +def pretty_hotkey(hotkey: str) -> str: + names = {"": "Ctrl", "": "Alt", "": "Shift", "": "Super", "": "Space"} + parts = [] + for raw in hotkey.split("+"): + inner = raw.strip("<>") + parts.append(names.get(raw, inner.upper() if len(inner) == 1 else inner.title())) + return " ".join(parts) + + +class Row: + """One clickable workflow row.""" + + def __init__(self, app: "BlitztextGUI", parent: tk.Widget, wf): + self.app = app + self.wf = wf + self.dimmed = False + + self.frame = tk.Frame(parent, bg=WIN, cursor="hand2") + self.frame.pack(fill="x") + self.strip = tk.Frame(self.frame, bg=WIN, width=3) + self.strip.pack(side="left", fill="y") + + body = tk.Frame(self.frame, bg=WIN) + body.pack(side="left", fill="x", expand=True, padx=(18, 16), pady=11) + + top = tk.Frame(body, bg=WIN) + top.pack(fill="x") + self.name = tk.Label(top, text=wf.name, font=(_FONT, 13), bg=WIN, fg=TEXT, anchor="w") + self.name.pack(side="left") + self.hint = tk.Label(top, text=pretty_hotkey(wf.hotkey), font=(_MONO, 9), bg=WIN, fg=FAINT, anchor="e") + self.hint.pack(side="right") + + self.desc = None + if wf.description: + self.desc = tk.Label(body, text=wf.description, font=(_FONT, 9), bg=WIN, fg=SUBTLE, anchor="w") + self.desc.pack(fill="x") + + self._widgets = [self.frame, body, top, self.name, self.hint] + ([self.desc] if self.desc else []) + for w in self._widgets: + w.bind("", lambda _e: self.app.on_row_click(self)) + w.bind("", lambda _e: self._hover(True)) + w.bind("", lambda _e: self._hover(False)) + + def _paint(self, bg: str) -> None: + for w in self._widgets: + w.configure(bg=bg) + + def _hover(self, on: bool) -> None: + if self.dimmed or not self.app.daemon.ready: + return + self._paint(HOVER if on else WIN) + self.strip.configure(bg=HOVER if on else WIN) + + def set_idle(self) -> None: + self.dimmed = False + self._paint(WIN) + self.strip.configure(bg=WIN) + self.name.configure(fg=TEXT) + if self.desc: + self.desc.configure(fg=SUBTLE) + self.hint.configure(text=pretty_hotkey(self.wf.hotkey), fg=FAINT) + + def set_recording(self) -> None: + self.dimmed = False + self._paint(WIN) + self.strip.configure(bg=RED) + self.name.configure(fg=TEXT) + if self.desc: + self.desc.configure(fg=SUBTLE) + self.hint.configure(text="● Stop", fg=RED) + + def set_busy(self) -> None: + self.hint.configure(text="Working…", fg=AMBER) + + def set_dimmed(self) -> None: + self.dimmed = True + self._paint(WIN) + self.strip.configure(bg=WIN) + self.name.configure(fg=FAINT) + if self.desc: + self.desc.configure(fg=FAINT) + self.hint.configure(fg=FAINT) + + +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._rows: list[Row] = [] + self._active: str | None = None + + self.root = tk.Tk() + self.root.title("Blitztext") + self.root.configure(bg=WIN) + _pick_fonts(self.root) + self.root.minsize(380, 320) + + self._build_header() + tk.Frame(self.root, bg=LINE, height=1).pack(fill="x", padx=22) + self._build_rows() + tk.Frame(self.root, bg=LINE, height=1).pack(fill="x", padx=22) + self._build_footer() + + self.root.after(80, self._drain_events) + 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}")) + + # -- layout --------------------------------------------------------------- + def _build_header(self) -> None: + head = tk.Frame(self.root, bg=WIN) + head.pack(fill="x", padx=22, pady=(20, 14)) + tk.Label(head, text="Blitztext", font=(_FONT, 17), bg=WIN, fg=TEXT).pack(side="left") + wrap = tk.Frame(head, bg=WIN) + wrap.pack(side="right") + self.dot = tk.Label(wrap, text="●", font=(_FONT, 10), bg=WIN, fg=AMBER) + self.dot.pack(side="left", padx=(0, 5)) + self.status_label = tk.Label(wrap, text="Starting…", font=(_FONT, 11), bg=WIN, fg=SUBTLE) + self.status_label.pack(side="left") + + def _build_rows(self) -> None: + body = tk.Frame(self.root, bg=WIN) + body.pack(fill="both", expand=True, padx=8, pady=6) + for wf in self.cfg.workflows: + self._rows.append(Row(self, body, wf)) + + def _build_footer(self) -> None: + foot = tk.Frame(self.root, bg=WIN) + foot.pack(fill="x", padx=22, pady=(12, 16)) + tk.Label(foot, text=f"v{__version__}", font=(_FONT, 9), bg=WIN, fg=FAINT).pack(side="left") + self._text_button(foot, "Quit", self.quit_all).pack(side="right") + self._text_button(foot, "Settings", self.open_settings).pack(side="right", padx=(0, 18)) + + def _text_button(self, parent, label, cmd) -> tk.Label: + b = tk.Label(parent, text=label, font=(_FONT, 11), bg=WIN, fg=ACCENT, cursor="hand2") + b.bind("", lambda _e: cmd()) + b.bind("", lambda _e: b.configure(fg=TEXT)) + b.bind("", lambda _e: b.configure(fg=ACCENT)) + return b + + # -- interaction ---------------------------------------------------------- + def on_row_click(self, row: Row) -> None: + if not self.daemon.ready or self.daemon._busy: + return + if self._active and row.wf.name != self._active: + return # another workflow is recording + self.trigger_workflow(row.wf) + + def trigger_workflow(self, wf) -> None: + if not self.daemon.ready: + return + threading.Thread(target=lambda: self.daemon.toggle(wf), daemon=True).start() + + # -- 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: + self._apply_status(*self._events.get_nowait()) + 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, SUBTLE)) + self.status_label.configure(text=label) + if self.tray is not None: + self.tray.update_status(state, label) + + if state == "recording": + self._active = workflow + for r in self._rows: + r.set_recording() if r.wf.name == workflow else r.set_dimmed() + elif state == "busy": + for r in self._rows: + if r.wf.name == workflow: + r.set_busy() + else: + r.set_dimmed() + elif state == "loading": + for r in self._rows: + r.set_dimmed() + else: # idle / done / error + self._active = None + for r in self._rows: + r.set_idle() + if state == "error" and message: + self.status_label.configure(text=message[:42], fg=RED) + + # -- 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: + 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() + 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=WIN) + self.win.minsize(560, 500) + + style = ttk.Style(self.win) + try: + style.configure("TNotebook", background=WIN, borderwidth=0) + style.configure("TNotebook.Tab", padding=(14, 7), font=(_FONT, 10)) + except tk.TclError: + pass + + nb = ttk.Notebook(self.win) + nb.pack(fill="both", expand=True, padx=14, pady=14) + + 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=WIN) + bar.pack(fill="x", padx=14, pady=(0, 14)) + save_btn = tk.Label(bar, text="Save & Restart", font=(_FONT, 11), bg=ACCENT, fg="white", + padx=14, pady=7, cursor="hand2") + save_btn.bind("", lambda _e: self._save_restart()) + save_btn.pack(side="right") + save2 = tk.Label(bar, text="Save", font=(_FONT, 11), bg=HOVER, fg=TEXT, padx=14, pady=7, cursor="hand2") + save2.bind("", lambda _e: self._save()) + save2.pack(side="right", padx=8) + tk.Label(bar, text="Hotkey & model changes apply after restart.", + font=(_FONT, 9), bg=WIN, fg=SUBTLE).pack(side="left") + + def _field(self, parent, label, value, key): + frame = tk.Frame(parent, bg=WIN) + frame.pack(fill="x", padx=16, pady=6) + tk.Label(frame, text=label, font=(_FONT, 10), bg=WIN, fg=TEXT, width=16, anchor="w").pack(side="left") + var = tk.StringVar(value=str(value)) + self.vars[key] = var + tk.Entry(frame, textvariable=var, font=(_FONT, 10), relief="solid", bd=1, + highlightthickness=0).pack(side="left", fill="x", expand=True, ipady=3) + return var + + def _combo(self, parent, label, value, options, key): + frame = tk.Frame(parent, bg=WIN) + frame.pack(fill="x", padx=16, pady=6) + tk.Label(frame, text=label, font=(_FONT, 10), bg=WIN, 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 _build_general(self, nb): + tab = tk.Frame(nb, bg=WIN) + 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=WIN) + f.pack(fill="x", padx=16, pady=6) + tk.Checkbutton(f, text="Desktop notifications", variable=nv, bg=WIN, fg=TEXT, + font=(_FONT, 10), activebackground=WIN, selectcolor=WIN).pack(side="left") + + def _build_rewrite(self, nb): + tab = tk.Frame(nb, bg=WIN) + 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=(_FONT, 9), bg=WIN, fg=SUBTLE, wraplength=500, justify="left").pack(fill="x", padx=16, pady=(10, 0)) + + def _build_workflow_tab(self, nb, idx, wf): + tab = tk.Frame(nb, bg=WIN) + 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=WIN) + frame.pack(fill="x", padx=16, pady=5) + tk.Label(frame, text=label, font=(_FONT, 10), bg=WIN, fg=TEXT, width=14, anchor="w").pack(side="left") + sv = tk.StringVar(value=str(val)) + v[key] = sv + tk.Entry(frame, textvariable=sv, font=(_FONT, 10), relief="solid", bd=1, + highlightthickness=0).pack(side="left", fill="x", expand=True, ipady=3) + + mframe = tk.Frame(tab, bg=WIN) + mframe.pack(fill="x", padx=16, pady=5) + tk.Label(mframe, text="Mode", font=(_FONT, 10), bg=WIN, 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=(_FONT, 10), bg=WIN, fg=TEXT, anchor="w").pack(fill="x", padx=16, pady=(10, 2)) + txt = tk.Text(tab, height=9, wrap="word", font=(_FONT, 10), relief="solid", bd=1, highlightthickness=0) + txt.insert("1.0", wf.prompt) + txt.pack(fill="both", expand=True, padx=16, pady=(0, 12)) + self.prompt_texts[idx] = txt + + 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", "tray"]) + + +def run_gui(tray_mode: bool = False) -> int: + cfg = load() + BlitztextGUI(cfg, tray_mode=tray_mode).run() + return 0 diff --git a/linux/blitztext/notify.py b/linux/blitztext/notify.py new file mode 100644 index 0000000..8e8c4d6 --- /dev/null +++ b/linux/blitztext/notify.py @@ -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) diff --git a/linux/blitztext/paste.py b/linux/blitztext/paste.py new file mode 100644 index 0000000..d4d0b14 --- /dev/null +++ b/linux/blitztext/paste.py @@ -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 diff --git a/linux/blitztext/recorder.py b/linux/blitztext/recorder.py new file mode 100644 index 0000000..ae2a3f0 --- /dev/null +++ b/linux/blitztext/recorder.py @@ -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) diff --git a/linux/blitztext/rewrite.py b/linux/blitztext/rewrite.py new file mode 100644 index 0000000..36e453e --- /dev/null +++ b/linux/blitztext/rewrite.py @@ -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 diff --git a/linux/blitztext/transcribe.py b/linux/blitztext/transcribe.py new file mode 100644 index 0000000..770b424 --- /dev/null +++ b/linux/blitztext/transcribe.py @@ -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() diff --git a/linux/blitztext/tray.py b/linux/blitztext/tray.py new file mode 100644 index 0000000..37c02bc --- /dev/null +++ b/linux/blitztext/tray.py @@ -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) diff --git a/linux/install.sh b/linux/install.sh new file mode 100755 index 0000000..4577cca --- /dev/null +++ b/linux/install.sh @@ -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 <=1.0 +pynput>=1.7