felixm/blurt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Speed up transcription with a pluggable provider backend

Browse Source 
Transcription:
- Provider-pluggable backend: each model routes to OpenAI (multipart
  upload) or OpenRouter (JSON + base64 body). Add OpenRouter's
  whisper-large-v3-turbo (Groq-served, ~216x real-time) to the tray
  menu for far lower latency than whisper-1.
- Load API keys env -> keyring -> file for both providers via a shared
  helper.
- Encode the WAV in memory and reuse a warm requests.Session (no temp file).

Other improvements:
- blurt-toggle signals the running app directly via SIGUSR1 instead of
  paying ~0.5s to import the full stack through "uv run blurt toggle".
- Remove the redundant dead blurt shell script.
- Suppress the benign Gdk-CRITICAL warning from pystray's GtkStatusIcon.
- gitignore openrouter_key.txt.

Transcription code is platform-agnostic; the Windows/Linux platform layer
is unchanged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Felix Martin
2026-05-31 08:38:05 -04:00
parent 4d10653790
commit 777d4bef33
4 changed files with 160 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -7,5 +7,6 @@ __pycache__/
# Blurt runtime config / secrets (written by blurt.py at runtime)
api_key.txt
openrouter_key.txt
model.txt
language.txt
blurt
-86
View File
@@ -1,86 +0,0 @@
#!/bin/bash
set -euo pipefail
PID_FILE="/tmp/blurt.pid"
AUDIO_FILE="/tmp/blurt.wav"
API_URL="https://api.openai.com/v1/audio/transcriptions"
notify() {
notify-send "blurt" "$1"
}
get_api_key() {
keyring get openai-api-key felixm 2>/dev/null || {
notify "Error: Could not get API key from keyring"
exit 1
}
}
start_recording() {
notify "Recording..."
pw-record --format=s16 --rate=16000 --channels=1 "$AUDIO_FILE" &
echo $! > "$PID_FILE"
}
stop_recording() {
local pid="$1"
kill -INT "$pid" 2>/dev/null || true
wait "$pid" 2>/dev/null || true
rm -f "$PID_FILE"
}
transcribe() {
notify "Transcribing..."
local api_key
api_key=$(get_api_key)
local response
response=$(curl -s -X POST "$API_URL" \
-H "Authorization: Bearer $api_key" \
-F "file=@$AUDIO_FILE" \
-F "model=whisper-1")
local text
text=$(echo "$response" | jq -r '.text // empty')
if [[ -z "$text" ]]; then
local error
error=$(echo "$response" | jq -r '.error.message // "Unknown error"')
notify "Error: $error"
exit 1
fi
echo -n "$text" | xclip -selection clipboard
# If active window is kitty, insert text directly
local window_class
window_class=$(xdotool getactivewindow getwindowclassname 2>/dev/null || echo "")
if [[ "$window_class" == *"kitty"* ]]; then
xdotool type --clearmodifiers -- "$text"
fi
local preview="${text:0:50}"
[[ ${#text} -gt 50 ]] && preview="${preview}..."
notify "Copied: $preview"
}
main() {
if [[ -f "$PID_FILE" ]]; then
local pid
pid=$(cat "$PID_FILE")
if kill -0 "$pid" 2>/dev/null; then
stop_recording "$pid"
transcribe
else
rm -f "$PID_FILE"
start_recording
fi
else
start_recording
fi
}
main
blurt-toggle
+15 -1
View File
@@ -1,2 +1,16 @@
#!/bin/sh
cd /home/felixm/wrk/blurt && uv run blurt toggle
# Toggle Blurt recording by signaling the running tray app directly.
#
# A direct SIGUSR1 avoids the ~0.5s cost of "uv run blurt toggle", which
# imported the full numpy/scipy/sounddevice/pystray stack just to send one
# signal. The PID file is written by blurt.py's setup_toggle_trigger().
PID_FILE="${TMPDIR:-/tmp}/blurt.pid"
pid=$(cat "$PID_FILE" 2>/dev/null)
if [ -n "$pid" ] && kill -USR1 "$pid" 2>/dev/null; then
exit 0
fi
notify-send blurt "Blurt is not running" 2>/dev/null
exit 1
blurt.py
+144 -49
View File
@@ -1,3 +1,5 @@
import base64
import io
import logging
import os
import shlex
@@ -21,13 +23,20 @@ from PIL import Image, ImageDraw
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"
DEFAULT_TRANSCRIPTION_MODEL = "whisper-1"
DEFAULT_TRANSCRIPTION_LANGUAGE = "en"
# (model_id, menu_label, provider). The provider selects both the API endpoint
# and the request format: "openai" speaks multipart file upload; "openrouter"
# speaks JSON with base64-encoded audio. OpenRouter's whisper-large-v3-turbo is
# served by Groq (216x real-time, ~12% WER) and reuses your OpenRouter key.
TRANSCRIPTION_MODELS = [
("whisper-1", "OpenAI: whisper-1"),
("gpt-4o-mini-transcribe", "OpenAI: gpt-4o-mini-transcribe"),
("gpt-4o-transcribe", "OpenAI: gpt-4o-transcribe"),
("whisper-1", "OpenAI: whisper-1", "openai"),
("gpt-4o-mini-transcribe", "OpenAI: gpt-4o-mini-transcribe", "openai"),
("gpt-4o-transcribe", "OpenAI: gpt-4o-transcribe", "openai"),
("openai/whisper-large-v3-turbo", "OpenRouter: whisper-large-v3-turbo (Groq, fast)", "openrouter"),
]
PROVIDER_KEY_FILES = {"openai": "api_key.txt", "openrouter": "openrouter_key.txt"}
TRANSCRIPTION_LANGUAGES = [
("en", "English"),
("de", "German"),
@@ -44,6 +53,9 @@ KEYRING_CREDENTIALS = [
("openai-api-key", "felixm"),
("blurt", "openai-api-key"),
]
OPENROUTER_KEYRING_CREDENTIALS = [
("openrouter-api-key", "felixm"),
]
def setup_logging():
@@ -97,40 +109,57 @@ def load_config_value(env_name: str, file_name: str, default: str = "") -> str:
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
for service_name, username in KEYRING_CREDENTIALS:
try:
stored_key = keyring.get_password(service_name, username)
except Exception:
logging.exception(
"Could not read OpenAI API key from keyring entry %s/%s",
service_name,
username,
)
continue
if stored_key:
logging.info(
"Loaded OpenAI API key from keyring entry %s/%s",
service_name,
username,
)
return stored_key.strip()
key = _load_key_from_keyring(KEYRING_CREDENTIALS, "OpenAI")
if key:
return key
return load_config_value("OPENAI_API_KEY", "api_key.txt")
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
return load_config_value("OPENROUTER_API_KEY", "openrouter_key.txt")
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}
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
@@ -167,7 +196,7 @@ def save_transcription_language(language: str):
def get_model_label(model: str) -> str:
for model_id, label in TRANSCRIPTION_MODELS:
for model_id, label, _ in TRANSCRIPTION_MODELS:
if model_id == model:
return label
return model
@@ -334,7 +363,11 @@ class BlurtApp:
self.input_devices = self._get_input_devices()
self.transcription_model = load_transcription_model()
self.transcription_language = load_transcription_language()
self.api_key = load_api_key()
self.api_keys = {
"openai": load_api_key(),
"openrouter": load_openrouter_key(),
}
self.session = requests.Session()
self.pending_transcriptions = 0
logging.info(
"BlurtApp initialized with %d input devices, model=%s, language=%s",
@@ -375,14 +408,25 @@ class BlurtApp:
self.icon.title = f"{APP_NAME} - {state}"
self.icon.update_menu()
def _provider_for_model(self, model: str) -> str:
for model_id, _, provider in TRANSCRIPTION_MODELS:
if model_id == model:
return provider
return "openai"
def _current_api_key(self) -> str:
return self.api_keys.get(self._provider_for_model(self.transcription_model), "")
def _audio_callback(self, indata, frames, time_info, status):
if status:
print(f"sounddevice: {status}", file=sys.stderr)
self.audio_frames.append(indata.copy())
def _start_recording(self):
if not self.api_key:
self.platform.notify("Error", "No API key. Put your key in api_key.txt")
if not self._current_api_key():
provider = self._provider_for_model(self.transcription_model)
key_file = PROVIDER_KEY_FILES.get(provider, "api_key.txt")
self.platform.notify("Error", f"No {provider} API key. Put your key in {key_file}")
return
self.audio_frames = []
try:
@@ -430,14 +474,16 @@ class BlurtApp:
)
logging.info("Starting transcription: samples=%d", len(audio_data))
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
tmp_path = Path(tmp.name)
tmp.close()
wavfile.write(str(tmp_path), SAMPLE_RATE, audio_data)
logging.info("Wrote temporary WAV: %s (%d bytes)", tmp_path, tmp_path.stat().st_size)
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:
result = self._transcribe_openai(tmp_path)
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", "")
logging.info("Transcription text length: %d", len(text))
if "usage" in result:
@@ -479,23 +525,36 @@ class BlurtApp:
finally:
self.pending_transcriptions -= 1
self._update_icon()
try:
tmp_path.unlink(missing_ok=True)
except OSError:
pass
def _transcribe_openai(self, tmp_path: Path) -> dict:
with open(tmp_path, "rb") as f:
response = requests.post(
OPENAI_TRANSCRIPTIONS_URL,
headers={"Authorization": f"Bearer {self.api_key}"},
files={"file": ("blurt.wav", f, "audio/wav")},
data={
"model": self.transcription_model,
"language": self.transcription_language,
},
timeout=30,
)
def _transcribe_openai(self, wav_bytes: bytes) -> dict:
response = self.session.post(
OPENAI_TRANSCRIPTIONS_URL,
headers={"Authorization": f"Bearer {self.api_keys['openai']}"},
files={"file": ("blurt.wav", wav_bytes, "audio/wav")},
data={
"model": self.transcription_model,
"language": self.transcription_language,
},
timeout=30,
)
self._log_transcription_response(response)
response.raise_for_status()
return response.json()
def _transcribe_openrouter(self, wav_bytes: bytes) -> dict:
# OpenRouter's transcription endpoint takes a JSON body with base64-encoded
# audio (not OpenAI's multipart upload). The response still has a "text" field.
audio_b64 = base64.b64encode(wav_bytes).decode("ascii")
response = self.session.post(
OPENROUTER_TRANSCRIPTIONS_URL,
headers={"Authorization": f"Bearer {self.api_keys['openrouter']}"},
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()
@@ -540,7 +599,10 @@ class BlurtApp:
def _select_model(self, model: str):
def handler(icon, item):
self.transcription_model = model
self.api_key = load_api_key()
self.api_keys = {
"openai": load_api_key(),
"openrouter": load_openrouter_key(),
}
save_transcription_model(model)
self.icon.update_menu()
self.platform.notify("Model Selected", get_model_label(model))
@@ -588,7 +650,7 @@ class BlurtApp:
)
model_items = []
for model, label in TRANSCRIPTION_MODELS:
for model, label, _ in TRANSCRIPTION_MODELS:
model_items.append(
pystray.MenuItem(
label,
@@ -645,6 +707,38 @@ class BlurtApp:
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)
@@ -652,6 +746,7 @@ def main():
send_toggle()
return
_silence_benign_gtk_warnings()
platform = get_platform()
app = BlurtApp(platform)
app.run()