benchmark: remote GPU detection, URL column, test-all-models (v2.0.6)

- stt.detect_remote_device(): probes /info (faster-whisper-server) then
  /metadata (NVIDIA NIM) to detect CUDA vs CPU; cached per unique URL
- BenchRow gains url field; Device column now shows "CUDA" for GPU remotes
  instead of the generic "remote"
- Benchmark table gains URL column (scheme stripped, max 180px wide)
- "Test all models per engine" checkbox: fetches list_models() for each
  remote engine and expands to one row per model when checked
- benchmark.run() gains expand_models parameter

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
mARTin-B78 2026-06-09 18:19:42 +02:00
parent bc6e2a6e54
commit 1c16cfeb0f
4 changed files with 86 additions and 17 deletions

View File

@ -6,4 +6,4 @@ 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. (not in a container) so it can type into any application via xdotool.
""" """
__version__ = "2.0.5" __version__ = "2.0.6"

View File

@ -7,6 +7,7 @@ Benchmark tab to find the fastest and most accurate engine/model.
from __future__ import annotations from __future__ import annotations
import dataclasses
import re import re
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
@ -18,6 +19,7 @@ from .routing import normalize
@dataclass @dataclass
class BenchRow: class BenchRow:
engine: str engine: str
url: str # base URL of the engine (empty for local)
model: str model: str
device: str # "CPU" | "CUDA" | "remote" device: str # "CPU" | "CUDA" | "remote"
best_for: str # "Short clips" | "Short / medium" | "Long / batch" | "Streaming" best_for: str # "Short clips" | "Short / medium" | "Long / batch" | "Streaming"
@ -59,10 +61,13 @@ def wer(reference: str, hypothesis: str, *, case_sensitive: bool = False) -> flo
return _edit_distance(ref, hyp) / len(ref) return _edit_distance(ref, hyp) / len(ref)
def _engine_device(engine, transcriber) -> str: def _engine_device(engine, transcriber, _cache: dict) -> str:
if engine.is_local: if engine.is_local:
return "CUDA" if getattr(transcriber, "device", "cpu") == "cuda" else "CPU" return "CUDA" if getattr(transcriber, "device", "cpu") == "cuda" else "CPU"
return "remote" url = engine.url
if url not in _cache:
_cache[url] = stt.detect_remote_device(url)
return _cache[url]
def _engine_best_for(engine) -> str: def _engine_best_for(engine) -> str:
@ -85,20 +90,37 @@ def _engine_best_for(engine) -> str:
def run(engines, wav_path: Path, reference: str, *, language: str = "", def run(engines, wav_path: Path, reference: str, *, language: str = "",
case_sensitive: bool = True, get_local_transcriber=None, progress=None) -> list[BenchRow]: case_sensitive: bool = True, get_local_transcriber=None, progress=None,
expand_models: bool = False) -> list[BenchRow]:
"""Benchmark each engine; calls progress(row) as each finishes. """Benchmark each engine; calls progress(row) as each finishes.
expand_models=True fetches available models for each remote engine and
runs one benchmark row per model instead of just the configured one.
Accuracy is case-sensitive by default so wrong capitalisation counts. Accuracy is case-sensitive by default so wrong capitalisation counts.
""" """
rows: list[BenchRow] = [] # Build the run list, optionally expanding remote engines by their models
run_list: list = []
for e in engines: for e in engines:
if expand_models and not e.is_local and not e.is_streaming:
models = stt.list_models(e.url, e.api_key_env)
if len(models) > 1:
for m in models:
run_list.append(dataclasses.replace(e, model=m,
name=f"{e.name} [{m}]"))
continue
run_list.append(e)
device_cache: dict = {}
rows: list[BenchRow] = []
for e in run_list:
tr = get_local_transcriber(e) if (e.is_local and get_local_transcriber) else None tr = get_local_transcriber(e) if (e.is_local and get_local_transcriber) else None
res = stt.benchmark(e, wav_path, language=language, local_transcriber=tr) res = stt.benchmark(e, wav_path, language=language, local_transcriber=tr)
w = wer(reference, res.text, case_sensitive=case_sensitive) if res.ok else 1.0 w = wer(reference, res.text, case_sensitive=case_sensitive) if res.ok else 1.0
row = BenchRow( row = BenchRow(
engine=e.name, engine=e.name,
url=e.url,
model=e.model or ("local" if e.is_local else "(default)"), model=e.model or ("local" if e.is_local else "(default)"),
device=_engine_device(e, tr), device=_engine_device(e, tr, device_cache),
best_for=_engine_best_for(e), best_for=_engine_best_for(e),
ok=res.ok, ok=res.ok,
seconds=res.seconds, seconds=res.seconds,

View File

@ -1853,23 +1853,33 @@ notebook.bt-nb tab:checked label {
break break
wavf.connect("file-set", _on_wav_set) wavf.connect("file-set", _on_wav_set)
run_row = Gtk.Box(spacing=16)
run_row.set_margin_bottom(2)
run = Gtk.Button(label="Run benchmark"); run.connect("clicked", self._run_bench) run = Gtk.Button(label="Run benchmark"); run.connect("clicked", self._run_bench)
run.set_halign(Gtk.Align.START) run.set_halign(Gtk.Align.START)
page.pack_start(run, False, False, 6) run_row.pack_start(run, False, False, 0)
self.bench_expand = Gtk.CheckButton(label="Test all models per engine")
self.bench_expand.set_tooltip_text(
"Fetch every available model from each remote engine and run a separate "
"benchmark row per model — useful to compare models on the same server.")
run_row.pack_start(self.bench_expand, False, False, 0)
page.pack_start(run_row, False, False, 6)
# cols: engine, model, device, best_for, time, accuracy, output_friendly, tooltip_full # cols: engine, url, model, device, best_for, time, accuracy, output_friendly, tooltip_full
self.bench_store = Gtk.ListStore(str, str, str, str, str, str, str, str) self.bench_store = Gtk.ListStore(str, str, str, str, str, str, str, str, str)
tree = Gtk.TreeView(model=self.bench_store) tree = Gtk.TreeView(model=self.bench_store)
tree.set_has_tooltip(True) tree.set_has_tooltip(True)
tree.set_tooltip_column(7) # hover any row → full error text tree.set_tooltip_column(8) # hover any row → full error / output text
for title, i, expand in [("Engine", 0, False), ("Model", 1, False), ("Device", 2, False), for title, i, expand in [("Engine", 0, False), ("URL", 1, False),
("Best for", 3, False), ("Time (s)", 4, False), ("Model", 2, False), ("Device", 3, False),
("Accuracy", 5, False), ("Output", 6, True)]: ("Best for", 4, False), ("Time (s)", 5, False),
("Accuracy", 6, False), ("Output", 7, True)]:
r = Gtk.CellRendererText() r = Gtk.CellRendererText()
if i == 6: r.set_property("ellipsize", Pango.EllipsizeMode.END)
r.set_property("ellipsize", Pango.EllipsizeMode.END)
col = Gtk.TreeViewColumn(title, r, text=i); col.set_resizable(True) col = Gtk.TreeViewColumn(title, r, text=i); col.set_resizable(True)
col.set_expand(expand) col.set_expand(expand)
if i == 1:
col.set_max_width(180) # keep URL column from dominating
tree.append_column(col) tree.append_column(col)
sw = Gtk.ScrolledWindow(); sw.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC) sw = Gtk.ScrolledWindow(); sw.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
sw.add(tree); page.pack_start(sw, True, True, 4) sw.add(tree); page.pack_start(sw, True, True, 4)
@ -1959,11 +1969,14 @@ notebook.bt-nb tab:checked label {
else: else:
self.bench_summary.set_markup("<i>Running… (local models load on first use)</i>") self.bench_summary.set_markup("<i>Running… (local models load on first use)</i>")
expand = self.bench_expand.get_active()
def work(): def work():
def prog(row): def prog(row):
GLib.idle_add(self._bench_add_row, row) GLib.idle_add(self._bench_add_row, row)
rows = benchmark.run(engines, Path(wav), reference, language=self.cfg.language, rows = benchmark.run(engines, Path(wav), reference, language=self.cfg.language,
get_local_transcriber=self._transcriber_for, progress=prog) get_local_transcriber=self._transcriber_for, progress=prog,
expand_models=expand)
GLib.idle_add(self._bench_done, rows) GLib.idle_add(self._bench_done, rows)
threading.Thread(target=work, daemon=True).start() threading.Thread(target=work, daemon=True).start()
@ -1995,7 +2008,9 @@ notebook.bt-nb tab:checked label {
else: else:
out_friendly = f"{self._bench_friendly_error(row.error)}" out_friendly = f"{self._bench_friendly_error(row.error)}"
tooltip = row.error tooltip = row.error
self.bench_store.append([row.engine, row.model, row.device, row.best_for, # Strip scheme from URL for display brevity (http://192.168.1.1:8080 → 192.168.1.1:8080)
url_display = row.url.removeprefix("https://").removeprefix("http://").rstrip("/")
self.bench_store.append([row.engine, url_display, row.model, row.device, row.best_for,
f"{row.seconds:.2f}", acc, out_friendly, tooltip]) f"{row.seconds:.2f}", acc, out_friendly, tooltip])
return False return False

View File

@ -105,6 +105,38 @@ def list_models(base_url: str, api_key_env: str = "", timeout: float = 5.0) -> l
return [] return []
def detect_remote_device(base_url: str, timeout: float = 3.0) -> str:
"""Best-effort GPU/CPU detection for a remote STT server.
Tries faster-whisper-server's /info endpoint (returns {"device":"cuda",...}),
then NVIDIA NIM /metadata (GPU-only service). Falls back to "remote".
"""
if not base_url:
return "remote"
base = base_url.rstrip("/")
try:
req = urllib.request.Request(base + "/info")
with urllib.request.urlopen(req, timeout=timeout) as resp:
data = json.loads(resp.read().decode("utf-8"))
if isinstance(data, dict):
dev = str(data.get("device") or data.get("compute_type") or "")
if "cuda" in dev.lower():
return "CUDA"
if dev:
return dev.upper()[:16]
except Exception:
pass
try:
req = urllib.request.Request(base + "/metadata")
with urllib.request.urlopen(req, timeout=timeout) as resp:
data = json.loads(resp.read().decode("utf-8"))
if isinstance(data, dict) and data.get("modelInfo"):
return "CUDA" # NVIDIA NIM is always GPU
except Exception:
pass
return "remote"
def _host_port(url: str) -> tuple[str | None, int]: def _host_port(url: str) -> tuple[str | None, int]:
try: try:
u = urlparse(url if "://" in url else "http://" + url) u = urlparse(url if "://" in url else "http://" + url)