import base64 import io import logging import os import random import re import shlex import signal import subprocess import sys import tempfile import threading from dataclasses import dataclass from logging.handlers import RotatingFileHandler from math import gcd from pathlib import Path from typing import Callable import numpy as np import keyring import pyperclip import pystray import requests import sounddevice as sd from PIL import Image, ImageDraw from scipy import signal as dsp from scipy.io import wavfile OPENAI_TRANSCRIPTIONS_URL = "https://api.openai.com/v1/audio/transcriptions" OPENROUTER_TRANSCRIPTIONS_URL = "https://openrouter.ai/api/v1/audio/transcriptions" OPENROUTER_APP_URL = "https://git.felixm.de/felixm/blurt" OPENROUTER_APP_TITLE = "Blurt" DEFAULT_TRANSCRIPTION_MODEL = "whisper-1" DEFAULT_TRANSCRIPTION_LANGUAGE = "en" TRANSCRIPTION_MODELS = [ ("whisper-1", "OpenAI: whisper-1", "openai"), ("gpt-4o-mini-transcribe", "OpenAI: gpt-4o-mini-transcribe", "openai"), ( "gpt-4o-mini-transcribe-2025-12-15", "OpenAI: gpt-4o-mini-transcribe (2025-12-15)", "openai", ), ("gpt-4o-transcribe", "OpenAI: gpt-4o-transcribe", "openai"), ("gpt-4o-transcribe-diarize", "OpenAI: gpt-4o-transcribe-diarize", "openai"), ( "openai/whisper-large-v3-turbo", "OpenRouter: whisper-large-v3-turbo (fast)", "openrouter", ), ( "openai/whisper-large-v3", "OpenRouter: whisper-large-v3 (accurate)", "openrouter", ), ( "openai/gpt-4o-mini-transcribe", "OpenRouter: gpt-4o-mini-transcribe", "openrouter", ), ( "openai/gpt-4o-transcribe", "OpenRouter: gpt-4o-transcribe", "openrouter", ), ] PROVIDER_KEY_HINTS = { "openai": ("set OPENAI_API_KEY or configure keyring/Credential Manager entry blurt/openai-api-key"), "openrouter": ( "set OPENROUTER_API_KEY or configure Windows Credential Manager entry " "blurt/openrouter-api-key" ), } TRANSCRIPTION_LANGUAGES = [ ("en", "English"), ("de", "German"), ] SAMPLE_RATE = 16000 CHANNELS = 1 DTYPE = "float32" # capture with headroom; the WAV sent to the API is int16 WARMUP_SECONDS = 0.15 # drop the click many mics emit when the stream opens HIGHPASS_HZ = 70.0 # strip DC bias and rumble (raw digital mics carry a lot) TARGET_RMS_DBFS = -20.0 PEAK_CEILING_DBFS = -1.0 MAX_GAIN_DB = 30.0 SILENT_RMS_DBFS = -65.0 # Playback endpoints that JACK re-exposes as inputs, and raw ALSA `hw:` devices # that a sound server already owns — offering either in the menu only invites a # hang or a silent recording. OUTPUT_ENDPOINT_HINTS = ("HDMI", "DisplayPort", "Headphones") RAW_ALSA_DEVICE = re.compile(r"\(hw:\d+,\d+\)") SOUND_SERVER_DEVICES = ("pipewire", "pulse", "default") APP_NAME = "Blurt" APP_DIR = Path(__file__).resolve().parent AGENT_PREFIX = "to my agent" PID_FILE = Path(tempfile.gettempdir()) / "blurt.pid" LOG_FILE = Path(tempfile.gettempdir()) / "blurt.log" KEYRING_CREDENTIALS = [ ("openai-api-key", "felixm"), ("blurt", "openai-api-key"), ] OPENROUTER_KEYRING_CREDENTIALS = [ ("openrouter-api-key", "felixm"), ("blurt", "openrouter-api-key"), ] def setup_logging(): logger = logging.getLogger() if logger.handlers: return logger.setLevel(logging.INFO) handler = RotatingFileHandler( LOG_FILE, maxBytes=512 * 1024, backupCount=3, encoding="utf-8", ) handler.setFormatter( logging.Formatter("%(asctime)s %(levelname)s [%(threadName)s] %(message)s") ) logger.addHandler(handler) def log_thread_exception(args): logging.critical( "Uncaught thread exception in %s", args.thread.name if args.thread else "", exc_info=(args.exc_type, args.exc_value, args.exc_traceback), ) threading.excepthook = log_thread_exception def load_config_value(env_name: str, file_name: str, default: str = "") -> str: env_value = os.environ.get(env_name, "").strip() if env_value: logging.info("Loaded %s from environment", env_name) return env_value config_file = APP_DIR / file_name try: value = config_file.read_text().strip() logging.info( "Loaded %s from %s: %s", env_name, config_file, "present" if value else "empty", ) return value except FileNotFoundError: if default: logging.info("Config file not found, using default for %s: %s", env_name, default) else: logging.warning("Config file not found for %s: %s", env_name, config_file) return default def _load_key_from_keyring(credentials: list[tuple[str, str]], label: str) -> str: for service_name, username in credentials: try: stored_key = keyring.get_password(service_name, username) except Exception: logging.exception( "Could not read %s key from keyring entry %s/%s", label, service_name, username, ) continue if stored_key: logging.info("Loaded %s key from keyring entry %s/%s", label, service_name, username) return stored_key.strip() return "" def load_api_key() -> str: env_key = os.environ.get("OPENAI_API_KEY", "").strip() if env_key: logging.info("Loaded OPENAI_API_KEY from environment") return env_key key = _load_key_from_keyring(KEYRING_CREDENTIALS, "OpenAI") if key: return key logging.warning("No OpenAI API key found in OPENAI_API_KEY or configured keyring entries") return "" def load_openrouter_key() -> str: env_key = os.environ.get("OPENROUTER_API_KEY", "").strip() if env_key: logging.info("Loaded OPENROUTER_API_KEY from environment") return env_key key = _load_key_from_keyring(OPENROUTER_KEYRING_CREDENTIALS, "OpenRouter") if key: return key logging.warning( "No OpenRouter API key found in OPENROUTER_API_KEY or configured keyring entries" ) return "" def load_transcription_model() -> str: model = load_config_value( "BLURT_TRANSCRIPTION_MODEL", "model.txt", DEFAULT_TRANSCRIPTION_MODEL, ) valid_models = {model_id for model_id, _, _ in TRANSCRIPTION_MODELS} if model not in valid_models: logging.warning("Unknown transcription model %s, using %s", model, DEFAULT_TRANSCRIPTION_MODEL) return DEFAULT_TRANSCRIPTION_MODEL return model def save_transcription_model(model: str): model_file = APP_DIR / "model.txt" model_file.write_text(f"{model}\n") logging.info("Saved transcription model to %s: %s", model_file, model) def load_transcription_language() -> str: language = load_config_value( "BLURT_TRANSCRIPTION_LANGUAGE", "language.txt", DEFAULT_TRANSCRIPTION_LANGUAGE, ) valid_languages = {language_id for language_id, _ in TRANSCRIPTION_LANGUAGES} if language not in valid_languages: logging.warning( "Unknown transcription language %s, using %s", language, DEFAULT_TRANSCRIPTION_LANGUAGE, ) return DEFAULT_TRANSCRIPTION_LANGUAGE return language def save_transcription_language(language: str): language_file = APP_DIR / "language.txt" language_file.write_text(f"{language}\n") logging.info("Saved transcription language to %s: %s", language_file, language) def load_input_device_name() -> str: return load_config_value("BLURT_INPUT_DEVICE", "device.txt") def save_input_device_name(name: str): device_file = APP_DIR / "device.txt" device_file.write_text(f"{name}\n") logging.info("Saved input device to %s: %s", device_file, name or "") def audio_stats(samples: np.ndarray) -> dict: """Level report for a float32 mono clip, in the units that explain bad audio.""" if len(samples) == 0: return {"rms_dbfs": -120.0, "peak_dbfs": -120.0, "dc": 0.0, "clipped_pct": 0.0} dc = float(samples.mean()) centered = samples - dc rms = float(np.sqrt(np.mean(np.square(centered)))) peak = float(np.abs(samples).max()) return { "rms_dbfs": 20 * np.log10(max(rms, 1e-6)), "peak_dbfs": 20 * np.log10(max(peak, 1e-6)), "dc": dc, "clipped_pct": float((np.abs(samples) >= 0.999).mean() * 100), } def input_stream_settings(device) -> tuple[int, int]: """Rate and channel count the device will actually open with. Only the resampling backends (`pipewire`, `default`) accept 16 kHz mono: raw ALSA `hw:` devices and PipeWire's JACK endpoints run at their own rate, so we capture natively and resample ourselves.""" info = sd.query_devices(device, "input") native_rate = int(info["default_samplerate"]) max_channels = max(int(info["max_input_channels"]), CHANNELS) for rate in dict.fromkeys((SAMPLE_RATE, native_rate)): for channels in dict.fromkeys((CHANNELS, max_channels)): try: sd.check_input_settings( device=device, samplerate=rate, channels=channels, dtype=DTYPE ) except Exception: continue return rate, channels logging.warning( "No probed capture format worked for %s; trying native %d Hz / %d ch", info["name"], native_rate, max_channels, ) return native_rate, max_channels def resample_to_target(samples: np.ndarray, rate: int) -> np.ndarray: if rate == SAMPLE_RATE: return samples divisor = gcd(rate, SAMPLE_RATE) resampled = dsp.resample_poly(samples, SAMPLE_RATE // divisor, rate // divisor) logging.info("Resampled %d Hz -> %d Hz", rate, SAMPLE_RATE) return resampled.astype(np.float32) def without_warmup(samples: np.ndarray) -> np.ndarray: """The clip minus the stream-open click, which otherwise skews every stat.""" warmup = int(WARMUP_SECONDS * SAMPLE_RATE) return samples[warmup:] if len(samples) > warmup * 2 else samples def speech_level_dbfs(samples: np.ndarray) -> float: """Level of the loudest tenth of the clip: the part that should be speech. Robust to the click at stream start and to long pauses.""" frame = SAMPLE_RATE // 50 if len(samples) < frame * 2: return -120.0 centered = samples - samples.mean() usable = centered[: len(centered) // frame * frame].reshape(-1, frame) energy = np.sqrt(np.mean(np.square(usable), axis=1)) return 20 * np.log10(max(float(np.percentile(energy, 90)), 1e-6)) def condition_audio(samples: np.ndarray) -> np.ndarray: """Make raw capture look like what a Windows mic driver would have handed us: no start-up click, no DC bias, speech at a predictable level.""" trimmed = len(samples) samples = without_warmup(samples) trimmed -= len(samples) sos = dsp.butter(2, HIGHPASS_HZ, btype="highpass", fs=SAMPLE_RATE, output="sos") # Seed the filter with the first sample, or the DC step at t=0 rings for # ~100 ms and lands right on top of the first word. zi = dsp.sosfilt_zi(sos) * (samples[0] if len(samples) else 0.0) filtered = dsp.sosfilt(sos, samples, zi=zi)[0].astype(np.float32) rms = float(np.sqrt(np.mean(np.square(filtered)))) if len(filtered) else 0.0 # 99.9th percentile, not the max: one keyboard clack must not hold the # whole clip 20 dB down. Stray peaks above the ceiling get clipped below. peak = float(np.percentile(np.abs(filtered), 99.9)) if len(filtered) else 0.0 if rms > 0 and peak > 0: gain = min( 10 ** ((TARGET_RMS_DBFS - 20 * np.log10(rms)) / 20), 10 ** ((PEAK_CEILING_DBFS - 20 * np.log10(peak)) / 20), 10 ** (MAX_GAIN_DB / 20), ) else: gain = 1.0 logging.info( "Conditioned audio: trimmed=%d samples highpass=%gHz gain=%+.1f dB", trimmed, HIGHPASS_HZ, 20 * np.log10(gain), ) return np.clip(filtered * gain, -1.0, 1.0) def get_model_label(model: str) -> str: for model_id, label, _ in TRANSCRIPTION_MODELS: if model_id == model: return label return model def get_language_label(language: str) -> str: for language_id, label in TRANSCRIPTION_LANGUAGES: if language_id == language: return label return language @dataclass class Platform: notify: Callable[[str, str], None] dispatch: Callable[[str], None] open_logs: Callable[[], None] setup_toggle_trigger: Callable[["BlurtApp"], None] cleanup_toggle_trigger: Callable[[], None] def _make_windows_platform() -> Platform: from pynput import keyboard from winotify import Notification listener = None def notify(title: str, message: str): try: toast = Notification(app_id=APP_NAME, title=title, msg=message) toast.show() except Exception: logging.exception("Windows notification failed") def dispatch(task: str): wezterm = r"C:\Program Files\WezTerm\wezterm-gui.exe" dispatch_sh = "/mnt/c/Users/felix.martin/AppData/Roaming/FlowLauncher/Plugins/Dispatch-1.0.0/dispatch.sh" safe_query = shlex.quote(task) cmd = [ wezterm, "-e", "wsl.exe", "-d", "Ubuntu", "-e", "zsh", "-c", f"{dispatch_sh} {safe_query}", ] logging.info("Dispatching agent task with %d characters", len(task)) subprocess.Popen(cmd, creationflags=subprocess.DETACHED_PROCESS) def open_logs(): subprocess.Popen(["notepad.exe", str(LOG_FILE)]) def setup_toggle_trigger(app: "BlurtApp"): nonlocal listener def on_press(key): if key == keyboard.Key.f20: app.toggle_recording() listener = keyboard.Listener(on_press=on_press) listener.start() logging.info("Started Windows F20 hotkey listener") def cleanup_toggle_trigger(): nonlocal listener if listener: listener.stop() return Platform( notify=notify, dispatch=dispatch, open_logs=open_logs, setup_toggle_trigger=setup_toggle_trigger, cleanup_toggle_trigger=cleanup_toggle_trigger, ) def _make_linux_platform() -> Platform: prev_handler = None def notify(title: str, message: str): try: subprocess.Popen( ["notify-send", APP_NAME, f"{title}: {message}"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) except Exception: logging.exception("Linux notification failed") def dispatch(task: str): dispatch_sh = Path.home() / ".local" / "bin" / "dispatch.sh" safe_query = shlex.quote(task) cmd = ["bash", str(dispatch_sh), safe_query] logging.info("Dispatching agent task with %d characters", len(task)) subprocess.Popen( cmd, start_new_session=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) def open_logs(): subprocess.Popen( ["xdg-open", str(LOG_FILE)], start_new_session=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) def setup_toggle_trigger(app: "BlurtApp"): nonlocal prev_handler PID_FILE.write_text(str(os.getpid())) logging.info("Wrote toggle PID file: %s", PID_FILE) def on_sigusr1(signum, frame): threading.Thread(target=app.toggle_recording, daemon=True).start() prev_handler = signal.signal(signal.SIGUSR1, on_sigusr1) def cleanup_toggle_trigger(): nonlocal prev_handler if prev_handler is not None: signal.signal(signal.SIGUSR1, prev_handler) PID_FILE.unlink(missing_ok=True) return Platform( notify=notify, dispatch=dispatch, open_logs=open_logs, setup_toggle_trigger=setup_toggle_trigger, cleanup_toggle_trigger=cleanup_toggle_trigger, ) def get_platform() -> Platform: if sys.platform == "win32": return _make_windows_platform() return _make_linux_platform() class AwakeKeeper: """Periodically clicks the Windows Start button area to prevent PC lock.""" def __init__(self): from pynput.mouse import Controller, Button self._mouse = Controller() self._button = Button.left self._stop_event = threading.Event() self._thread: threading.Thread | None = None @property def is_active(self) -> bool: return self._thread is not None and self._thread.is_alive() def start(self): if self.is_active: return self._stop_event.clear() self._thread = threading.Thread( target=self._run, daemon=True, name="AwakeKeeper" ) self._thread.start() logging.info("AwakeKeeper started") def stop(self): self._stop_event.set() self._thread = None logging.info("AwakeKeeper stopped") def _run(self): while not self._stop_event.wait(timeout=random.uniform(60, 120)): x = random.randint(5, 20) y = random.randint(5, 20) self._mouse.position = (x, y) self._mouse.click(self._button) logging.debug("AwakeKeeper clicked at (%d, %d)", x, y) def send_toggle(): """Send toggle signal to running blurt instance (Linux only).""" if sys.platform == "win32": print("'blurt toggle' is not supported on Windows.", file=sys.stderr) sys.exit(1) try: pid = int(PID_FILE.read_text().strip()) logging.info("Sending toggle signal to PID %s", pid) os.kill(pid, signal.SIGUSR1) except (FileNotFoundError, ValueError, ProcessLookupError): print("No running blurt instance found.", file=sys.stderr) sys.exit(1) class BlurtApp: def __init__(self, platform: Platform): self.platform = platform self.recording = False self.audio_frames: list[np.ndarray] = [] self.stream: sd.InputStream | None = None self.capture_rate = SAMPLE_RATE self.capture_channels = CHANNELS self.lock = threading.Lock() self.icon: pystray.Icon | None = None self.input_devices = self._get_input_devices() self.device_name = load_input_device_name() self.device_index = self._resolve_device(self.device_name) self.api_keys = { "openai": load_api_key(), "openrouter": load_openrouter_key(), } self.transcription_model = self._available_model_or_fallback( load_transcription_model() ) self.transcription_language = load_transcription_language() self.session = requests.Session() self._awake_keeper: AwakeKeeper | None = ( AwakeKeeper() if sys.platform == "win32" else None ) self.pending_transcriptions = 0 logging.info( "BlurtApp initialized with %d input devices, model=%s, language=%s, awake_available=%s", len(self.input_devices), self.transcription_model, self.transcription_language, self._awake_keeper is not None, ) def _get_input_devices(self) -> list[dict]: devices = sd.query_devices() self.all_input_devices = [ {"index": i, "name": d["name"]} for i, d in enumerate(devices) if d["max_input_channels"] > 0 ] input_devices = [] server_present = any( dev["name"] in SOUND_SERVER_DEVICES for dev in self.all_input_devices ) for dev in self.all_input_devices: if any(hint in dev["name"] for hint in OUTPUT_ENDPOINT_HINTS): logging.info("Skipping output endpoint %s: %s", dev["index"], dev["name"]) continue if server_present and RAW_ALSA_DEVICE.search(dev["name"]): logging.info("Skipping raw ALSA device %s: %s", dev["index"], dev["name"]) continue logging.info("Input device %s: %s", dev["index"], dev["name"]) input_devices.append(dev) return input_devices def _resolve_device(self, name: str) -> int | None: """Devices are addressed by name, not index: PortAudio indices shift whenever a USB mic or dock comes and goes.""" if not name: logging.info("No input device configured; recording from the system default") return None for dev in self.all_input_devices: if dev["name"] == name: logging.info("Using configured input device %s: %s", dev["index"], name) return dev["index"] logging.warning("Configured input device %r is not connected; using system default", name) return None def _create_icon_image(self, color: tuple[int, int, int]) -> Image.Image: img = Image.new("RGBA", (64, 64), (0, 0, 0, 0)) draw = ImageDraw.Draw(img) draw.ellipse([8, 8, 56, 56], fill=color) return img def _update_icon(self): if self.icon: if self.recording: color = (220, 40, 40) # red state = "Recording..." elif self.pending_transcriptions > 0: color = (40, 100, 220) # blue state = "Transcribing..." else: color = (120, 120, 120) # gray state = "Ready" self.icon.icon = self._create_icon_image(color) self.icon.title = f"{APP_NAME} - {state}" self.icon.update_menu() def _audio_callback(self, indata, frames, time_info, status): if status: logging.warning("sounddevice input status: %s", status) self.audio_frames.append(indata.copy()) def _provider_for_model(self, model: str) -> str: for model_id, _, provider in TRANSCRIPTION_MODELS: if model_id == model: return provider return "openai" def _available_model_or_fallback(self, model: str) -> str: provider = self._provider_for_model(model) if self.api_keys.get(provider): return model for fallback_model, _, fallback_provider in TRANSCRIPTION_MODELS: if self.api_keys.get(fallback_provider): logging.warning( "Model %s requires a %s API key, which is unavailable; " "using available model %s for this session", model, provider, fallback_model, ) return fallback_model logging.error( "Model %s requires a %s API key, but no API keys are available for any provider", model, provider, ) return model def _current_api_key(self) -> str: return self.api_keys.get(self._provider_for_model(self.transcription_model), "") def _api_key_label(self) -> str: provider = self._provider_for_model(self.transcription_model) provider_label = "OpenRouter" if provider == "openrouter" else "OpenAI" key = self._current_api_key() if not key: return f"API key: {provider_label} (missing)" return f"API key: {provider_label} (…{key[-4:]})" def _start_recording(self): if not self._current_api_key(): provider = self._provider_for_model(self.transcription_model) key_hint = PROVIDER_KEY_HINTS.get(provider, f"set the {provider} API key") logging.error( "Recording blocked: model=%s provider=%s has no API key; " "%s or select a model with an available key", self.transcription_model, provider, key_hint, ) self.platform.notify("Error", f"No {provider} API key; {key_hint}.") return self.audio_frames = [] try: self.capture_rate, self.capture_channels = input_stream_settings(self.device_index) self.stream = sd.InputStream( device=self.device_index, samplerate=self.capture_rate, channels=self.capture_channels, dtype=DTYPE, callback=self._audio_callback, ) self.stream.start() logging.info( "Started recording: device_index=%s name=%s rate=%d channels=%d", self.device_index, self.device_name or "", self.capture_rate, self.capture_channels, ) except Exception as e: logging.exception( "Could not start recording on device_index=%s name=%s", self.device_index, self.device_name or "", ) self.platform.notify("Error", f"Could not start recording: {e}") return self.recording = True self._update_icon() self.platform.notify("Recording", "Press hotkey or click tray to stop.") def _stop_recording(self) -> np.ndarray | None: if self.stream: self.stream.stop() self.stream.close() self.stream = None self.recording = False self._update_icon() if not self.audio_frames: logging.warning("Stopped recording with no audio frames") return None frames = np.concatenate(self.audio_frames, axis=0) self.audio_frames = [] audio_data = frames.mean(axis=1) if frames.shape[1] > 1 else frames[:, 0] audio_data = resample_to_target(audio_data, self.capture_rate) raw = audio_stats(without_warmup(audio_data)) speech_level = speech_level_dbfs(without_warmup(audio_data)) logging.info( "Stopped recording: samples=%d (%.1fs) speech=%.1f dBFS rms=%.1f dBFS " "peak=%.1f dBFS dc=%+.4f clipped=%.3f%% device=%s", len(audio_data), len(audio_data) / SAMPLE_RATE, speech_level, raw["rms_dbfs"], raw["peak_dbfs"], raw["dc"], raw["clipped_pct"], self.device_name or "", ) if speech_level < SILENT_RMS_DBFS: logging.error( "Capture is essentially silent (speech level %.1f dBFS) on device %s; " "not sending it for transcription", speech_level, self.device_name or "", ) return None if raw["clipped_pct"] > 1.0: logging.warning( "Capture clips on %.2f%% of samples; lower the input gain for %s", raw["clipped_pct"], self.device_name or "", ) conditioned = condition_audio(audio_data) clean = audio_stats(conditioned) logging.info( "Conditioned capture: rms=%.1f dBFS peak=%.1f dBFS dc=%+.4f", clean["rms_dbfs"], clean["peak_dbfs"], clean["dc"], ) return (conditioned * 32767).astype(np.int16) def _transcribe(self, audio_data: np.ndarray): self.platform.notify( "Transcribing", f"Using {get_model_label(self.transcription_model)} ({self.transcription_language})", ) logging.info("Starting transcription: samples=%d", len(audio_data)) buffer = io.BytesIO() wavfile.write(buffer, SAMPLE_RATE, audio_data) wav_bytes = buffer.getvalue() logging.info("Encoded WAV in memory: %d bytes", len(wav_bytes)) try: if self._provider_for_model(self.transcription_model) == "openrouter": result = self._transcribe_openrouter(wav_bytes) else: result = self._transcribe_openai(wav_bytes) text = result.get("text", "").lstrip() logging.info("Transcription text length: %d", len(text)) if "usage" in result: logging.info("Transcription usage: %s", result["usage"]) if not text: error_msg = result.get("error", {}).get("message", "Unknown error") logging.error("Transcription response did not include text: %s", result) self.platform.notify("Error", f"Transcription failed: {error_msg}") return if text.lower().startswith(AGENT_PREFIX): task = text[len(AGENT_PREFIX):].lstrip(" ,.:;-") if task: pyperclip.copy(task) try: self.platform.dispatch(task) except Exception as e: self.platform.notify("Dispatch Error", str(e)) preview = task[:50] + ("..." if len(task) > 50 else "") self.platform.notify("Dispatched", preview) logging.info("Dispatched transcription task with %d characters", len(task)) else: pyperclip.copy(text) self.platform.notify("Error", f"No task after '{AGENT_PREFIX}'") logging.warning("Agent prefix detected with no task") else: pyperclip.copy(text) preview = text[:50] + ("..." if len(text) > 50 else "") self.platform.notify("Copied", preview) logging.info("Copied transcription to clipboard") except requests.RequestException as e: logging.exception("API request failed") self.platform.notify("Error", f"API request failed: {e}") except Exception as e: logging.exception("Unexpected transcription failure") self.platform.notify("Error", f"Unexpected transcription failure: {e}") finally: self.pending_transcriptions -= 1 self._update_icon() def _transcribe_openai(self, wav_bytes: bytes) -> dict: data = { "model": self.transcription_model, "language": self.transcription_language, } if self.transcription_model == "gpt-4o-transcribe-diarize": data.update(response_format="diarized_json", chunking_strategy="auto") response = self.session.post( OPENAI_TRANSCRIPTIONS_URL, headers={"Authorization": f"Bearer {self.api_keys['openai']}"}, files={"file": ("blurt.wav", wav_bytes, "audio/wav")}, data=data, timeout=30, ) self._log_transcription_response(response) response.raise_for_status() return response.json() def _transcribe_openrouter(self, wav_bytes: bytes) -> dict: audio_b64 = base64.b64encode(wav_bytes).decode("ascii") response = self.session.post( OPENROUTER_TRANSCRIPTIONS_URL, headers={ "Authorization": f"Bearer {self.api_keys['openrouter']}", "HTTP-Referer": OPENROUTER_APP_URL, "X-OpenRouter-Title": OPENROUTER_APP_TITLE, }, json={ "model": self.transcription_model, "input_audio": {"data": audio_b64, "format": "wav"}, "language": self.transcription_language, }, timeout=30, ) self._log_transcription_response(response) response.raise_for_status() return response.json() def _log_transcription_response(self, response: requests.Response): logging.info( "Transcription response: status=%s content_type=%s bytes=%d", response.status_code, response.headers.get("content-type", ""), len(response.content), ) if not response.ok: logging.error("Transcription error response: %s", response.text[:1000]) def toggle_recording(self): with self.lock: if not self.recording: self._start_recording() else: audio_data = self._stop_recording() if audio_data is not None: self.pending_transcriptions += 1 self._update_icon() t = threading.Thread( target=self._transcribe, args=(audio_data,), daemon=True ) t.start() else: self.platform.notify("Error", "No audio recorded.") def _on_click(self, icon, item): self.toggle_recording() def _show_logs(self, icon, item): try: logging.info("Opening log file: %s", LOG_FILE) self.platform.open_logs() except Exception as e: logging.exception("Could not open log file") self.platform.notify("Error", f"Could not open log file: {e}") def _toggle_awake(self, icon, item): if self._awake_keeper is None: return if self._awake_keeper.is_active: self._awake_keeper.stop() else: self._awake_keeper.start() self.icon.update_menu() def _select_model(self, model: str): def handler(icon, item): self.api_keys = { "openai": load_api_key(), "openrouter": load_openrouter_key(), } provider = self._provider_for_model(model) if not self.api_keys.get(provider): logging.error( "Model selection blocked: model=%s provider=%s has no API key", model, provider, ) self.platform.notify( "Model Unavailable", f"No {provider} API key is configured.", ) self.icon.update_menu() return self.transcription_model = model save_transcription_model(model) self.icon.update_menu() self.platform.notify("Model Selected", get_model_label(model)) return handler def _select_language(self, language: str): def handler(icon, item): self.transcription_language = language save_transcription_language(language) self.icon.update_menu() self.platform.notify("Language Selected", get_language_label(language)) return handler def _is_model_selected(self, model: str): def check(item): return self.transcription_model == model return check def _is_provider_available(self, provider: str): def check(item): return bool(self.api_keys.get(provider)) return check def _is_language_selected(self, language: str): def check(item): return self.transcription_language == language return check def _select_device(self, name: str): def handler(icon, item): self.device_name = name self.device_index = self._resolve_device(name) save_input_device_name(name) self.icon.update_menu() return handler def _is_device_selected(self, name: str): def check(item): return self.device_name == name return check def _build_menu(self): device_items = [ pystray.MenuItem( "System default", self._select_device(""), checked=self._is_device_selected(""), radio=True, ) ] for dev in self.input_devices: device_items.append( pystray.MenuItem( dev["name"], self._select_device(dev["name"]), checked=self._is_device_selected(dev["name"]), radio=True, ) ) model_items = [] for model, label, provider in TRANSCRIPTION_MODELS: model_items.append( pystray.MenuItem( label, self._select_model(model), checked=self._is_model_selected(model), enabled=self._is_provider_available(provider), radio=True, ) ) language_items = [] for language, label in TRANSCRIPTION_LANGUAGES: language_items.append( pystray.MenuItem( label, self._select_language(language), checked=self._is_language_selected(language), radio=True, ) ) awake_items = [] if self._awake_keeper is not None: awake_items.append( pystray.MenuItem( "Awake", self._toggle_awake, checked=lambda item: self._awake_keeper.is_active, ) ) return pystray.Menu( pystray.MenuItem( lambda item: "Stop Recording" if self.recording else "Start Recording", self._on_click, default=True, ), pystray.Menu.SEPARATOR, pystray.MenuItem("Input Device", pystray.Menu(*device_items)), pystray.MenuItem("Model", pystray.Menu(*model_items)), pystray.MenuItem(lambda item: self._api_key_label(), None, enabled=False), pystray.MenuItem("Language", pystray.Menu(*language_items)), pystray.MenuItem("Show Logs", self._show_logs), *awake_items, pystray.Menu.SEPARATOR, pystray.MenuItem("Quit", self._on_quit), ) def _on_quit(self, icon, item): if self.recording: self._stop_recording() if self._awake_keeper and self._awake_keeper.is_active: self._awake_keeper.stop() self.platform.cleanup_toggle_trigger() icon.stop() def _setup(self, icon): icon.visible = True def run(self): self.platform.setup_toggle_trigger(self) logging.info("Starting tray icon") self.icon = pystray.Icon( APP_NAME, icon=self._create_icon_image((120, 120, 120)), title=f"{APP_NAME} - Ready", menu=self._build_menu(), ) self.icon.run(setup=self._setup) def _silence_benign_gtk_warnings(): """Drop a harmless, repeating Gdk-CRITICAL emitted by Gtk.StatusIcon. pystray's GTK backend (the one qtile's XEmbed Systray widget needs) uses the deprecated Gtk.StatusIcon, which logs this assertion on icon/menu updates: gdk_window_thaw_toplevel_updates: assertion 'window->update_and_descendants_freeze_count > 0' failed It is cosmetic. Suppress only that message and re-emit every other Gdk log so genuine problems still surface. """ try: from gi.repository import GLib except Exception: return benign = "gdk_window_thaw_toplevel_updates" def handler(domain, level, message, _user_data): if benign not in message: print(f"{domain}: {message}", file=sys.stderr) GLib.log_set_handler( "Gdk", GLib.LogLevelFlags.LEVEL_CRITICAL | GLib.LogLevelFlags.LEVEL_WARNING | GLib.LogLevelFlags.FLAG_FATAL | GLib.LogLevelFlags.FLAG_RECURSION, handler, None, ) def main(): setup_logging() logging.info("Starting %s argv=%s platform=%s log_file=%s", APP_NAME, sys.argv, sys.platform, LOG_FILE) if len(sys.argv) > 1 and sys.argv[1] == "toggle": send_toggle() return _silence_benign_gtk_warnings() platform = get_platform() app = BlurtApp(platform) app.run() if __name__ == "__main__": main()