xavierk
e6fdb3be9b
nvidia-smi (vram_mib) and wtype (inject probe) ran without env=host_env(), so under an AppImage bundle they inherit the bundle's library paths. The AST guard now covers the measurement script as well as src/vocalinux, and the script gains the module logger the Style rules require.
562 lines
20 KiB
Python
562 lines
20 KiB
Python
"""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 logging
|
|
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
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
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/<pid>/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."""
|
|
# Lazy: the parse-only subcommands must run without the package installed.
|
|
from vocalinux.utils.host_process import host_env
|
|
|
|
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,
|
|
env=host_env(),
|
|
)
|
|
except (OSError, subprocess.SubprocessError) as error:
|
|
logger.debug("nvidia-smi unavailable: %s", error)
|
|
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
|
|
|
|
from vocalinux.utils.host_process import host_env
|
|
|
|
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,
|
|
env=host_env(),
|
|
)
|
|
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())
|