diff --git a/docs/PERF_BASELINE_PROTOCOL.md b/docs/PERF_BASELINE_PROTOCOL.md new file mode 100644 index 0000000..f88a121 --- /dev/null +++ b/docs/PERF_BASELINE_PROTOCOL.md @@ -0,0 +1,80 @@ +# Installed-build performance baseline protocol (ticket #19) + +Locked measurement protocol for the Voculinux port's deferred performance +numbers. The numbers themselves are recorded as evidence on the ticket, +never here: until measured they are protocol, not results. + +## Machine class + +Void Linux (glibc, x86_64), niri Wayland session, NVIDIA RTX 2050 4 GB, +no CUDA toolkit installed (Vulkan then CPU backend order). + +## Prerequisites + +- Installed build from `install.sh` (venv or system install), default + whisper.cpp engine, debug logging available (`--debug`). +- Ladder models on disk with pinned checksums verified against + `src/vocalinux/utils/model_checksums.txt`: `large-v3-turbo-q8_0` + (default rung), `large-v3-turbo-q5_0` (first fallback), `small` and + `small.en` (floor, by language). +- Isolated XDG directories so the pass never touches a live install: + point `XDG_CONFIG_HOME`, `XDG_DATA_HOME`, `XDG_STATE_HOME` at scratch + roots and link the verified models under `/vocalinux/models/`. +- Seed `config.json` from the app's own `DEFAULT_CONFIG`: engine + `whisper_cpp`, the rung under test, keep-alive enabled with the + minimum 60 s timeout, `first_run` false, update checks off. + +## The three metrics + +All commands are `scripts/measure_perf_baseline.py` subcommands run +with the installed build's interpreter. + +1. **Idle tray memory, both states.** Start with `--debug + --start-minimized`, sample RSS with `sample-rss --pid`, and read + VRAM with `nvidia-smi`. Loaded state is right after startup; the + unloaded state follows the keep-alive timeout (watch the log for + `reason=idle_keepalive`). RSS does not reflect the unload (freed + native pages stay mapped); VRAM-used is the state signal. +2. **Wake latency, toggle to capture start.** Drive starts over D-Bus + (`busctl --user call com.vocalinux.Vocalinux + /com/vocalinux/Vocalinux com.vocalinux.Vocalinux Toggle`; the + `--toggle` client also works but spawns the full app stack). + Parse repeated runs with `parse-log --log`: each `Toggle requested + in state IDLE` pairs with the next `Opened capture stream`. Report + steady-state spread separately from the first-capture cost (model + reload after an idle unload lands on the first toggle). +3. **Transcription real-time factor, fixed sample.** Regenerate the + sample with `make-sample --out scripts/perf_baseline_sample.wav + --seed 19` (30 s speech-like babble, 16 kHz mono 16-bit, committed + next to the script so reruns are bit-identical), then `transcribe + --model-file … --backend vulkan|cpu --wav … --runs 3`. Report the + first (cold: shader/pipeline build, one-time) run separately from + warm runs. CUDA is measured only where a complete toolkit exists; + otherwise its absence is recorded, not worked around. + +## Same-pass extras + +- **Per-rung load footprint:** `probe-load --model-file … --backend …` + reports load wall time, child RSS, and VRAM used while the model is + held. Free versus total memory is distinguished via `nvidia-smi` + before/during/after. CPU loads touch no VRAM; their RSS is the + footprint. +- **Injection latency:** `inject-probe` records the RemoteDesktop + version-probe result and timing, then a session-establishment attempt + (fast failure versus hang matters more than the version answer), plus + the resolved first rung and a no-emission `wtype ""` spawn cost as a + lower bound. End-to-end key-to-screen latency is deliberately not + taken here: emitting keystrokes into a live desktop from a background + run would disturb the session. It belongs to the hardware checklist + with a dedicated target window. +- **Thresholds for the model ladder:** derive free-memory cutoffs from + the maximum observed per-rung footprint (transcribe peak, not just + load) plus explicit headroom for the compositor and other GPU + clients, and record the derivation next to the numbers. + +## Report + +`render_markdown_report` (and the JSON the subcommands print) carries +the protocol version, full host facts, and observed numbers only. +`tests/test_perf_baseline_protocol.py` covers the parsing, statistics, +threshold-derivation, and sample-determinism helpers offline. diff --git a/scripts/measure_perf_baseline.py b/scripts/measure_perf_baseline.py new file mode 100644 index 0000000..dfb4be2 --- /dev/null +++ b/scripts/measure_perf_baseline.py @@ -0,0 +1,550 @@ +"""Locked performance-baseline protocol for the Voculinux port (ticket #19). + +Runs the fixed three-metric protocol against an installed build on the +target machine class (Void, niri, RTX 2050 4 GB) and writes a JSON + +Markdown evidence report. Every number in the report is observed on the +machine the script runs on; nothing is estimated. Helpers that parse +observed bytes (``/proc`` status, ``nvidia-smi`` CSV, ``--debug`` logs) +are pure and covered by tests/test_perf_baseline_protocol.py. +""" + +from __future__ import annotations + +import argparse +import json +import math +import os +import platform +import re +import statistics +import subprocess +import sys +import time +from pathlib import Path +from typing import Any, Optional + +PROTOCOL_VERSION = "1.0.0" + +_VMRSS_RE = re.compile(r"^VmRSS:\s+(\d+)\s+kB", re.MULTILINE) + + +def parse_vmrss_kb(status_text: str) -> Optional[int]: + """Return the VmRSS value in kilobytes from /proc//status text.""" + match = _VMRSS_RE.search(status_text) + if match is None: + return None + return int(match.group(1)) + + +def read_rss_kb(pid: int) -> Optional[int]: + """Return the resident set size of ``pid`` in kilobytes, if readable.""" + try: + text = Path(f"/proc/{pid}/status").read_text(encoding="utf-8") + except OSError: + return None + return parse_vmrss_kb(text) + + +def describe(values: list[float]) -> dict[str, Any]: + """Summarise repeated runs; empty input stays explicitly empty.""" + if not values: + return {"n": 0, "mean": None, "min": None, "max": None, "stdev": None} + summary: dict[str, Any] = { + "n": len(values), + "mean": statistics.fmean(values), + "min": min(values), + "max": max(values), + "stdev": statistics.stdev(values) if len(values) > 1 else 0.0, + } + return summary + + +_LOG_TS_RE = re.compile(r"^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3})") +_TOGGLE_RE = re.compile(r"Handling action: (toggle|start)\b|Toggle requested in state IDLE") +_STOP_TOGGLE_RE = re.compile(r"Toggle requested in state (?!IDLE)(\w+)") +_CAPTURE_RE = re.compile(r"Opened capture stream:") + + +def _parse_log_ts(line: str) -> Optional[float]: + """Parse a ``%(asctime)s`` timestamp prefix to epoch seconds.""" + match = _LOG_TS_RE.match(line) + if match is None: + return None + try: + parsed = time.strptime(match.group(1), "%Y-%m-%d %H:%M:%S,%f") + except ValueError: + return None + return time.mktime(parsed) + int(match.group(1).split(",")[1]) / 1000.0 + + +def parse_wake_latencies(log_text: str) -> list[float]: + """Pair each toggle/start action with the next capture-stream opening. + + Returns wake latencies in seconds. A toggle with no later capture + start (blocked start, shutdown) is dropped rather than zero-filled. + """ + latencies: list[float] = [] + pending: Optional[float] = None + for line in log_text.splitlines(): + stamp = _parse_log_ts(line) + if stamp is None: + continue + if _TOGGLE_RE.search(line): + pending = stamp + elif _STOP_TOGGLE_RE.search(line): + pending = None + elif _CAPTURE_RE.search(line) and pending is not None: + latencies.append(stamp - pending) + pending = None + return latencies + + +_SMI_MEM_RE = re.compile(r"^(.+?),\s*([\d.]+),\s*(\d+)\s*MiB,\s*(\d+)\s*MiB\s*$", re.MULTILINE) + + +def parse_nvidia_smi_memory(csv_text: str) -> Optional[tuple[int, int]]: + """Parse ``(total_mib, free_mib)`` of the first GPU from nvidia-smi CSV. + + Expected query shape: ``name,driver_version,memory.total,memory.free``. + Returns None when no GPU row is present (CPU-only hosts). + """ + for line in csv_text.splitlines(): + match = _SMI_MEM_RE.match(line.strip()) + if match is not None: + return (int(match.group(3)), int(match.group(4))) + return None + + +def vram_mib() -> Optional[tuple[int, int]]: + """Return ``(total_mib, free_mib)`` of the first NVIDIA GPU, if any.""" + try: + proc = subprocess.run( + [ + "nvidia-smi", + "--query-gpu=name,driver_version,memory.total,memory.free", + "--format=csv,noheader", + ], + capture_output=True, + text=True, + timeout=30, + ) + except (OSError, subprocess.SubprocessError): + return None + if proc.returncode != 0: + return None + return parse_nvidia_smi_memory(proc.stdout) + + +def recommend_thresholds( + footprints_mib: dict[str, float], headroom_mib: float = 512.0 +) -> dict[str, float]: + """Derive ladder free-memory cutoffs from measured load footprints. + + Each rung is recommended only when free memory covers its measured + footprint plus ``headroom_mib`` for the desktop and other GPU clients. + Keys name the ladder of ticket #20: q8_0 default, q5_0 fallback, + small floor. + """ + q8 = footprints_mib["q8_0"] + headroom_mib + q5 = footprints_mib["q5_0"] + headroom_mib + return { + "prefer_q8_0_above_mib": q8, + "prefer_q5_0_above_mib": q5, + "floor_small_below_mib": q5, + "headroom_mib": headroom_mib, + } + + +SAMPLE_RATE_HZ = 16000 +SAMPLE_SECONDS = 30.0 + + +def make_sample_wav(path: Path, seed: int = 19) -> Path: + """Write the fixed 30 s speech-like sample every RTF run transcribes. + + Deterministic syllable babble (seeded): voiced harmonic bursts with a + pitch contour and pauses, 16 kHz mono 16-bit. Same seed means + byte-identical output, so runs stay comparable across machines and + backends. Speech-like, not speech: RTF comparability is the goal, + not a transcription accuracy figure. + """ + import random + import wave + + rng = random.Random(seed) + total = int(SAMPLE_RATE_HZ * SAMPLE_SECONDS) + samples = [0.0] * total + pos = 0 + while pos < total: + if rng.random() < 0.25: + pos += int(SAMPLE_RATE_HZ * rng.uniform(0.05, 0.25)) + continue + dur = int(SAMPLE_RATE_HZ * rng.uniform(0.12, 0.35)) + f0 = rng.uniform(90.0, 220.0) + glide = rng.uniform(0.9, 1.1) + formants = [rng.uniform(300, 800), rng.uniform(900, 2200), rng.uniform(2200, 3200)] + end = min(total, pos + dur) + n = end - pos + for i in range(n): + t = i / SAMPLE_RATE_HZ + frac = i / max(n, 1) + env = math.sin(math.pi * frac) ** 2 + pitch = f0 * (1.0 + (glide - 1.0) * frac) + value = 0.0 + for harmonic, weight in ((1, 0.6), (2, 0.25), (3, 0.15)): + value += weight * math.sin(2.0 * math.pi * pitch * harmonic * t) + for formant in formants: + value += 0.2 * math.sin(2.0 * math.pi * formant * t) + samples[pos + i] = env * value * 0.5 + pos = end + peak = max(1e-9, max(abs(s) for s in samples)) + path = Path(path) + with wave.open(str(path), "wb") as handle: + handle.setnchannels(1) + handle.setsampwidth(2) + handle.setframerate(SAMPLE_RATE_HZ) + handle.writeframes( + b"".join(int(s / peak * 32767).to_bytes(2, "little", signed=True) for s in samples) + ) + return path + + +def collect_host_facts() -> dict[str, Any]: + """Record the host facts the ticket requires next to every number.""" + facts: dict[str, Any] = { + "os": platform.freedesktop_os_release().get("PRETTY_NAME", platform.system()), + "python": platform.python_version(), + "cpu_count": os.cpu_count(), + "session_type": os.environ.get("XDG_SESSION_TYPE"), + "desktop": os.environ.get("XDG_CURRENT_DESKTOP"), + } + try: + facts["app_version"] = subprocess.run( + [sys.executable, "-m", "vocalinux.main", "--version"], + capture_output=True, + text=True, + timeout=60, + ).stdout.strip() + except (OSError, subprocess.SubprocessError): + facts["app_version"] = None + vram = vram_mib() + facts["gpu_total_mib"], facts["gpu_free_mib"] = vram if vram else (None, None) + return facts + + +def render_markdown_report(host_facts: dict[str, Any], metrics: dict[str, Any]) -> str: + """Render the evidence report; numbers come only from ``metrics``.""" + lines = [ + "# Installed-build performance baseline", + "", + f"Protocol version: {PROTOCOL_VERSION}", + "", + "## Host facts", + "", + *[f"- {key}: {value}" for key, value in host_facts.items()], + "", + "## Metrics", + "", + "```json", + json.dumps(metrics, indent=2, sort_keys=True), + "```", + "", + ] + return "\n".join(lines) + + +def realtime_factor(audio_seconds: float, transcribe_seconds: float) -> float: + """Transcription real-time factor: wall time per audio second.""" + if audio_seconds <= 0: + raise ValueError("audio duration must be positive") + return transcribe_seconds / audio_seconds + + +def _child_load_model(model_path: str, use_gpu: bool) -> dict[str, Any]: + """Load one whisper.cpp model in-process; prints RSS for the parent.""" + from pywhispercpp.model import Model + + params: dict[str, Any] = {} + if not use_gpu: + params["context_params"] = {"use_gpu": False} + started = time.monotonic() + model = Model(model_path, **params) + load_seconds = time.monotonic() - started + rss = read_rss_kb(os.getpid()) + del model + return {"load_seconds": load_seconds, "rss_kb": rss} + + +def _child_transcribe(model_path: str, use_gpu: bool, wav_path: str) -> dict[str, Any]: + """Load a model and transcribe ``wav_path`` once; report wall time.""" + import wave + + from pywhispercpp.model import Model + + with wave.open(wav_path, "rb") as handle: + frames = handle.getnframes() + rate = handle.getframerate() + audio_seconds = frames / rate + params: dict[str, Any] = {} + if not use_gpu: + params["context_params"] = {"use_gpu": False} + load_started = time.monotonic() + model = Model(model_path, **params) + load_seconds = time.monotonic() - load_started + transcribe_started = time.monotonic() + segments = model.transcribe(wav_path) + text = "".join(getattr(seg, "text", "") for seg in segments) + transcribe_seconds = time.monotonic() - transcribe_started + return { + "audio_seconds": audio_seconds, + "load_seconds": load_seconds, + "transcribe_seconds": transcribe_seconds, + "rtf": realtime_factor(audio_seconds, transcribe_seconds), + "chars": len(text), + } + + +def _sample_vram_while_child(child: "subprocess.Popen[str]", hold_seconds: float) -> Optional[int]: + """Peak VRAM-used reading while the child holds the loaded model.""" + deadline = time.monotonic() + hold_seconds + peak: Optional[int] = None + while time.monotonic() < deadline and child.poll() is None: + reading = vram_mib() + if reading is not None: + used = reading[0] - reading[1] + peak = used if peak is None else max(peak, used) + time.sleep(0.5) + return peak + + +def cmd_probe_load(args: argparse.Namespace) -> int: + """Measure one rung's load footprint: RSS, VRAM used, wall time.""" + use_gpu = args.backend == "vulkan" + before = vram_mib() + child = subprocess.Popen( + [ + sys.executable, + str(Path(__file__).resolve()), + "_child-load", + "--model-file", + args.model_file, + "--backend", + args.backend, + "--hold-seconds", + str(args.hold_seconds), + ], + stdout=subprocess.PIPE, + text=True, + ) + peak_used: Optional[int] = None + try: + peak_used = _sample_vram_while_child(child, args.hold_seconds + 10.0) + out = child.communicate(timeout=600)[0] + except subprocess.TimeoutExpired: + child.kill() + raise + try: + child_report = json.loads(out.strip().splitlines()[-1]) + except (json.JSONDecodeError, IndexError): + print(f"child produced no report: {out[-500:]}", file=sys.stderr) + return 1 + after = vram_mib() + report = { + "model_file": args.model_file, + "backend": args.backend, + "load_seconds": child_report["load_seconds"], + "rss_kb": child_report["rss_kb"], + "vram_used_before_mib": (before[0] - before[1]) if before else None, + "vram_free_before_mib": before[1] if before else None, + "vram_peak_used_mib": peak_used, + "vram_free_after_mib": after[1] if after else None, + } + print(json.dumps(report, indent=2, sort_keys=True)) + return 0 + + +def cmd_transcribe(args: argparse.Namespace) -> int: + """Transcribe the fixed sample ``--runs`` times in fresh processes.""" + runs = [] + for _ in range(args.runs): + child = subprocess.Popen( + [ + sys.executable, + str(Path(__file__).resolve()), + "_child-transcribe", + "--model-file", + args.model_file, + "--backend", + args.backend, + "--wav", + args.wav, + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + try: + peak_used = _sample_vram_while_child(child, float(args.timeout)) + out, err = child.communicate(timeout=60) + except subprocess.TimeoutExpired: + child.kill() + raise + if child.returncode != 0: + print(err[-2000:], file=sys.stderr) + return 1 + run = json.loads(out.strip().splitlines()[-1]) + run["vram_peak_used_mib"] = peak_used + runs.append(run) + rtfs = [run["rtf"] for run in runs] + print(json.dumps({"runs": runs, "rtf_stats": describe(rtfs)}, indent=2, sort_keys=True)) + return 0 + + +def cmd_sample_rss(args: argparse.Namespace) -> int: + """Sample a running tray process RSS ``--samples`` times.""" + samples = [] + for _ in range(args.samples): + rss = read_rss_kb(args.pid) + if rss is not None: + samples.append(float(rss)) + time.sleep(args.interval) + print(json.dumps({"rss_kb_stats": describe(samples)}, indent=2, sort_keys=True)) + return 0 + + +def cmd_parse_log(args: argparse.Namespace) -> int: + """Parse wake latencies from a ``--debug`` log file.""" + log_text = Path(args.log).read_text(encoding="utf-8", errors="replace") + latencies = parse_wake_latencies(log_text) + print(json.dumps({"latencies": latencies, "stats": describe(latencies)}, indent=2)) + return 0 + + +def cmd_inject_probe(args: argparse.Namespace) -> int: + """Time the portal probe and the direct path's app-side dispatch cost. + + Never emits keystrokes: end-to-end key-to-screen latency would type + into the live desktop, so it stays a hardware-checklist item (#32). + """ + from vocalinux.text_injection.text_injector import TextInjector + + probe_text = "the quick brown fox jumps over the lazy dog 0123456789" + started = time.monotonic() + injector = TextInjector() + wayland = bool(os.environ.get("WAYLAND_DISPLAY")) or getattr(injector, "wayland_mode", False) + if wayland: + backend = getattr(injector, "wayland_tool", None) or "wtype" + else: + backend = "x11" + selected_seconds = time.monotonic() - started + wtype_roundtrips: list[float] = [] + if wayland: + import shutil + + if shutil.which("wtype") is not None and backend != "x11": + for _ in range(5): + begin = time.monotonic() + subprocess.run( + ["wtype", ""], + capture_output=True, + text=True, + timeout=10, + ) + wtype_roundtrips.append(time.monotonic() - begin) + try: + from vocalinux.text_injection.remote_desktop_portal import RemoteDesktopPortal + + portal_started = time.monotonic() + portal_available = RemoteDesktopPortal().probe() + portal_seconds = time.monotonic() - portal_started + except Exception as exc: # noqa: BLE001 - probe must never fail the pass + portal_available = False + portal_seconds = None + print(f"portal probe raised: {exc}", file=sys.stderr) + print( + json.dumps( + { + "first_rung": backend, + "direct_select_seconds": selected_seconds, + "direct_probe_chars": len(probe_text), + "wtype_empty_roundtrip_stats": describe(wtype_roundtrips), + "portal_available": portal_available, + "portal_probe_seconds": portal_seconds, + }, + indent=2, + sort_keys=True, + ) + ) + return 0 + + +def main(argv: Optional[list[str]] = None) -> int: + """Entry point for the locked baseline protocol.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--version", action="store_true", help="print protocol version") + sub = parser.add_subparsers(dest="command") + + probe = sub.add_parser("probe-load", help="measure one rung load footprint") + probe.add_argument("--model-file", required=True) + probe.add_argument("--backend", choices=("vulkan", "cpu"), required=True) + probe.add_argument("--hold-seconds", type=float, default=8.0) + probe.set_defaults(func=cmd_probe_load) + + trans = sub.add_parser("transcribe", help="transcribe the fixed sample") + trans.add_argument("--model-file", required=True) + trans.add_argument("--backend", choices=("vulkan", "cpu"), required=True) + trans.add_argument("--wav", required=True) + trans.add_argument("--runs", type=int, default=3) + trans.add_argument("--timeout", type=int, default=1800) + trans.set_defaults(func=cmd_transcribe) + + sample = sub.add_parser("sample-rss", help="sample a tray process RSS") + sample.add_argument("--pid", type=int, required=True) + sample.add_argument("--samples", type=int, default=5) + sample.add_argument("--interval", type=float, default=2.0) + sample.set_defaults(func=cmd_sample_rss) + + parse = sub.add_parser("parse-log", help="wake latencies from a debug log") + parse.add_argument("--log", required=True) + parse.set_defaults(func=cmd_parse_log) + + inject = sub.add_parser("inject-probe", help="portal probe, direct dispatch cost") + inject.set_defaults(func=cmd_inject_probe) + + gen = sub.add_parser("make-sample", help="regenerate the fixed sample") + gen.add_argument("--out", required=True) + gen.add_argument("--seed", type=int, default=19) + + child_load = sub.add_parser("_child-load") + child_load.add_argument("--model-file", required=True) + child_load.add_argument("--backend", choices=("vulkan", "cpu"), required=True) + child_load.add_argument("--hold-seconds", type=float, default=8.0) + + child_trans = sub.add_parser("_child-transcribe") + child_trans.add_argument("--model-file", required=True) + child_trans.add_argument("--backend", choices=("vulkan", "cpu"), required=True) + child_trans.add_argument("--wav", required=True) + + args = parser.parse_args(argv) + if args.version: + print(PROTOCOL_VERSION) + return 0 + if args.command == "_child-load": + print(json.dumps(_child_load_model(args.model_file, args.backend == "vulkan"))) + time.sleep(args.hold_seconds) + return 0 + if args.command == "_child-transcribe": + print(json.dumps(_child_transcribe(args.model_file, args.backend == "vulkan", args.wav))) + return 0 + if args.command == "make-sample": + make_sample_wav(Path(args.out), seed=args.seed) + print(str(args.out)) + return 0 + if hasattr(args, "func"): + return int(args.func(args)) + parser.print_help() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/perf_baseline_sample.wav b/scripts/perf_baseline_sample.wav new file mode 100644 index 0000000..4fd36a5 Binary files /dev/null and b/scripts/perf_baseline_sample.wav differ diff --git a/src/vocalinux/ui/tray_indicator.py b/src/vocalinux/ui/tray_indicator.py index 71017e2..19cdd59 100644 --- a/src/vocalinux/ui/tray_indicator.py +++ b/src/vocalinux/ui/tray_indicator.py @@ -770,6 +770,7 @@ class TrayIndicator: return with self._ptt_lock: self._ptt_owner = None + logger.debug(f"Toggle requested in state {self.speech_engine.state.name}") if self.speech_engine.state == RecognitionState.IDLE: self.speech_engine.start_recognition() else: diff --git a/tests/test_perf_baseline_protocol.py b/tests/test_perf_baseline_protocol.py new file mode 100644 index 0000000..d005e9d --- /dev/null +++ b/tests/test_perf_baseline_protocol.py @@ -0,0 +1,136 @@ +"""Behaviour tests for the ticket-19 performance-baseline protocol script. + +The script (scripts/measure_perf_baseline.py) is the locked measurement +protocol for the installed-build baseline: every number it reports comes +from an observed run on the target machine class, parsed by the pure +helpers tested here. Tests run offline with fixture text only and never +touch audio hardware, the GPU, or the network. +""" + +import importlib.util +from pathlib import Path + +import pytest + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "measure_perf_baseline.py" +SPEC = importlib.util.spec_from_file_location("measure_perf_baseline", SCRIPT) +assert SPEC is not None and SPEC.loader is not None +PROTOCOL = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(PROTOCOL) + + +def test_parse_vmrss_kb_reads_kilobytes() -> None: + """VmRSS lines parse to kilobytes; anything else yields None.""" + status = "Name:\tvocalinux\nVmRSS:\t 412336 kB\nVmSwap:\t0 kB\n" + assert PROTOCOL.parse_vmrss_kb(status) == 412336 + assert PROTOCOL.parse_vmrss_kb("Name:\tvocalinux\n") is None + + +def test_describe_reports_count_mean_min_max() -> None: + """Spread statistics come from the samples, never fabricated.""" + stats = PROTOCOL.describe([1.0, 2.0, 3.0]) + assert stats["n"] == 3 + assert stats["mean"] == pytest.approx(2.0) + assert stats["min"] == pytest.approx(1.0) + assert stats["max"] == pytest.approx(3.0) + + +def test_describe_empty_is_explicit() -> None: + """No samples means n == 0, never a zero mean pretending to be data.""" + stats = PROTOCOL.describe([]) + assert stats["n"] == 0 + assert stats["mean"] is None + + +WAKE_LOG = """\ +2026-11-05 22:44:01,100 - vocalinux.ui.action_handler - DEBUG - Handling action: toggle +2026-11-05 22:44:01,350 - vocalinux.audio.capture - DEBUG - Opened capture stream: 1 channel(s) at 16000Hz +2026-11-05 22:44:05,100 - vocalinux.ui.action_handler - DEBUG - Handling action: toggle +2026-11-05 22:44:05,500 - vocalinux.audio.capture - DEBUG - Opened capture stream: 1 channel(s) at 16000Hz +""" + + +def test_parse_wake_latencies_pairs_toggle_to_capture() -> None: + """Each toggle pairs with the next capture-stream line, in seconds.""" + assert PROTOCOL.parse_wake_latencies(WAKE_LOG) == pytest.approx([0.25, 0.4]) + + +def test_parse_wake_latencies_ignores_unpaired_toggle() -> None: + """A toggle with no later capture start is dropped, not zero-filled.""" + log = ( + WAKE_LOG + + "2026-11-05 22:44:09,100 - vocalinux.ui.action_handler - DEBUG - Handling action: start\n" + ) + assert PROTOCOL.parse_wake_latencies(log) == pytest.approx([0.25, 0.4]) + + +def test_realtime_factor_is_audio_over_wall() -> None: + """RTF below one means faster than real time; zero audio is refused.""" + assert PROTOCOL.realtime_factor(30.0, 15.0) == pytest.approx(0.5) + with pytest.raises(ValueError): + PROTOCOL.realtime_factor(0.0, 1.0) + + +def test_parse_nvidia_smi_csv_reads_first_gpu() -> None: + """The locked nvidia-smi query shape parses to total/free MiB.""" + out = ( + "name, driver_version, memory.total [MiB], memory.free [MiB]\n" + "NVIDIA GeForce RTX 2050, 595.112.07, 4096 MiB, 3711 MiB\n" + ) + assert PROTOCOL.parse_nvidia_smi_memory(out) == (4096, 3711) + assert PROTOCOL.parse_nvidia_smi_memory("no gpu here\n") is None + + +def test_recommend_thresholds_come_from_footprints() -> None: + """Ladder thresholds derive from measured footprints plus headroom.""" + footprints = {"q8_0": 1000.0, "q5_0": 700.0, "small": 500.0} + advice = PROTOCOL.recommend_thresholds(footprints, headroom_mib=512.0) + assert advice["prefer_q8_0_above_mib"] == pytest.approx(1512.0) + assert advice["prefer_q5_0_above_mib"] == pytest.approx(1212.0) + assert advice["floor_small_below_mib"] == pytest.approx(1212.0) + + +def test_render_report_names_protocol_and_host() -> None: + """The evidence report carries protocol version and host facts.""" + report = PROTOCOL.render_markdown_report( + {"os": "Void", "gpu": "NVIDIA GeForce RTX 2050"}, {"idle": {}} + ) + assert PROTOCOL.PROTOCOL_VERSION in report + assert "Void" in report + assert "NVIDIA GeForce RTX 2050" in report + + +def test_sample_generator_is_deterministic(tmp_path) -> None: + """The fixed sample regenerates byte-identical from its seed.""" + first = tmp_path / "a.wav" + second = tmp_path / "b.wav" + PROTOCOL.make_sample_wav(first, seed=19) + PROTOCOL.make_sample_wav(second, seed=19) + assert first.read_bytes() == second.read_bytes() + + +def test_sample_shape_is_thirty_seconds_mono_16k(tmp_path) -> None: + """The sample is 30 s, mono, 16 kHz, 16-bit for whisper.cpp.""" + import wave + + target = tmp_path / "sample.wav" + PROTOCOL.make_sample_wav(target, seed=19) + with wave.open(str(target), "rb") as handle: + assert handle.getnchannels() == 1 + assert handle.getsampwidth() == 2 + assert handle.getframerate() == 16000 + assert handle.getnframes() / 16000 == pytest.approx(30.0) + + +DBUS_TOGGLE_LOG = """\ +2026-11-05 23:01:11,400 - vocalinux.ui.tray_indicator - DEBUG - Toggle requested in state IDLE +2026-11-05 23:01:11,546 - vocalinux.audio.capture - DEBUG - Opened capture stream: 2 channel(s) at 44100Hz +2026-11-05 23:01:16,400 - vocalinux.ui.tray_indicator - DEBUG - Toggle requested in state LISTENING +2026-11-05 23:01:22,100 - vocalinux.ui.tray_indicator - DEBUG - Toggle requested in state IDLE +2026-11-05 23:01:22,749 - vocalinux.audio.capture - DEBUG - Opened capture stream: 2 channel(s) at 44100Hz +""" + + +def test_parse_wake_latencies_pairs_dbus_toggle_starts_only() -> None: + """Only IDLE toggles open capture; a stop toggle never borrows the next run.""" + assert PROTOCOL.parse_wake_latencies(DBUS_TOGGLE_LOG) == pytest.approx([0.146, 0.649])