Руководство по программированию для Cactus Whistle: выдерживает ли речевая модель размером 16,9 МБ сравнение с Whisper Base? Частота ошибок в словах, скорость, потоковая передача, временные метки, ключевые слова и вызовы инструментов из речи на CPU

import os
import io
import re
import sys
import json
import math
import time
import random
import hashlib
import shutil
import zipfile
import platform
import traceback
import subprocess
import collections
import urllib.error
import urllib.request
 
RESULTS = {}
VERDICTS = []
 
 
def banner(title):
    print("\n" + "=" * 78)
    print(title)
    print("=" * 78)
 
 
def section(name):
    def wrap(fn):
        def run(*a, **kw):
            banner(name)
            t0 = time.time()
            try:
                out = fn(*a, **kw)
                RESULTS[name] = out if isinstance(out, str) else "ok"
                return out
            except Exception as e:
                RESULTS[name] = f"SKIPPED / FAILED -> {type(e).__name__}: {e}"
                print(f"\n[!] {name} did not complete: {type(e).__name__}: {e}")
                traceback.print_exc(limit=3)
                return None
            finally:
                print(f"\n[{time.time() - t0:.0f} s]")
                json.dump({"results": RESULTS, "verdicts": VERDICTS}, open("results.json", "w"), indent=1)
        return run
    return wrap
 
 
banner("1. Install Whistle, pin its weights and engine by Hub revision, transcribe one clip")
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "cactus-needle==3.2.0", "openai-whisper==20250625",
                "jiwer==4.0.0", "soundfile==0.14.0", "soxr==1.1.0"], check=True)
os.environ["NEEDLE_TELEMETRY"] = "0"            # the package sends anonymous usage counts unless told not to
os.environ["HF_HUB_DISABLE_PROGRESS_BARS"] = "1"
 
import numpy as np
import soundfile as sf
import soxr
from huggingface_hub import hf_hub_download
 
# cactus-needle fetches three files on first use, from Hugging Face repos whose files are replaced in place: the 16.9 MB
# whistle.cact weights, the ~1 MB engine library (version 3.3.0 in this package; the needle3 repo received new binaries
# on 2026-10-10) and the 35 MB needle3.cact used in step 12. Pinning the Hub revision of each, and printing the hashes,
# is what makes this run reproducible. The engine version is read from the package, not typed in.
WHISTLE_REV = "ca5287601bef25af26dcf2e1b2bdc0843a7c19e5"   # Cactus-Compute/whistle, 2026-10-10
NEEDLE3_REV = "93e88ede867e4311c39151285e74df513e4b835b"   # Cactus-Compute/needle3 (engine + Needle 3 weights), 2026-10-10
from needle.agent import fetch as needle_fetch
TAG = needle_fetch._platform_tag()
ENGINE_VERSION = needle_fetch.engine_version(3)
weights = hf_hub_download("Cactus-Compute/whistle", "whistle.cact", revision=WHISTLE_REV)
wheel = hf_hub_download("Cactus-Compute/needle3", needle_fetch.engine_wheel(ENGINE_VERSION, TAG, 3), revision=NEEDLE3_REV)
with zipfile.ZipFile(wheel) as z:
    member = next(n for n in z.namelist() if n.startswith("needle/libneedle"))
    os.makedirs("engine", exist_ok=True)
    engine = os.path.abspath(os.path.join("engine", os.path.basename(member)))
    open(engine, "wb").write(z.read(member))
needle3_weights = hf_hub_download("Cactus-Compute/needle3", "needle3.cact", revision=NEEDLE3_REV)
os.makedirs(needle_fetch.cache_dir(3), exist_ok=True)           # Needle(tools=...) reads its weights from this cache path
shutil.copyfile(needle3_weights, os.path.join(needle_fetch.cache_dir(3), needle_fetch.base_weights(3)))
os.environ["NEEDLE_WHISTLE_WEIGHTS"] = weights   # both variables are honoured by the package
os.environ["NEEDLE3_LIB_PATH"] = engine
 
import needle
import importlib.metadata as md
sha = lambda p: hashlib.sha256(open(p, "rb").read()).hexdigest()[:12]
CPU = platform.processor() or platform.machine()
if sys.platform == "linux" and os.path.exists("/proc/cpuinfo"):
    CPU = next((l.split(":", 1)[1].strip() for l in open("/proc/cpuinfo") if l.startswith("model name")), CPU)
elif sys.platform == "darwin":
    CPU = subprocess.run(["sysctl", "-n", "machdep.cpu.brand_string"], capture_output=True, text=True).stdout.strip() or CPU
print(f"cactus-needle {needle.__version__} (engine {ENGINE_VERSION}, wheel tag {TAG})   python {platform.python_version()}   "
      f"soundfile {md.version('soundfile')}  soxr {md.version('soxr')}  numpy {np.__version__}")
print(f"hardware: {CPU}, {os.cpu_count()} cpus visible")
print(f"whistle.cact  {os.path.getsize(weights)/1e6:.1f} MB  sha256 {sha(weights)}   @ {WHISTLE_REV[:12]}")
print(f"engine        {os.path.getsize(engine)/1e6:.1f} MB  sha256 {sha(engine)}   @ {NEEDLE3_REV[:12]}")
print(f"needle3.cact  {os.path.getsize(needle3_weights)/1e6:.1f} MB  sha256 {sha(needle3_weights)}   @ {NEEDLE3_REV[:12]}")
 
SR = 16000
W = needle.Whistle()      # one model per process, not thread-safe; every later step reuses this object
 
 
def fetch(url, tries=7):
    """GET with retries. The datasets-server rate-limits anonymous callers (HTTP 429): honour Retry-After when it is
    sent, otherwise back off 5, 10, 20, 40 ... seconds, and never give up on the first refusal."""
    for k in range(tries):
        try:
            with urllib.request.urlopen(url, timeout=60) as r:
                return r.read()
        except urllib.error.HTTPError as e:
            if k == tries - 1:
                raise
            wait = int(e.headers.get("Retry-After", 0) or 0) if e.code == 429 else 0
            time.sleep(max(wait, 5 * 2 ** k) if e.code in (429, 500, 502, 503, 504) else 2)
        except Exception:
            if k == tries - 1:
                raise
            time.sleep(3 * (k + 1))
 
 
def page(dataset, config, split, offset, length=100):
    """One page of a Hub dataset through the datasets-server rows API: metadata plus signed audio URLs."""
    url = (f"https://datasets-server.huggingface.co/rows?dataset={dataset}&config={config}"
           f"&split={split}&offset={offset}&length={length}")
    time.sleep(0.25)                                  # stay under the anonymous rate limit when paging a whole split
    return json.loads(fetch(url))
 
 
def load_audio(row):
    """Download one row's clip; return 16 kHz mono float32 samples and the sha256 of the original file."""
    data = fetch(row["audio"][0]["src"])
    audio, sr = sf.read(io.BytesIO(data), dtype="float32")
    if audio.ndim > 1:
        audio = audio.mean(axis=1)
    if sr != SR:
        audio = soxr.resample(audio, sr, SR, quality="HQ")
    return audio, hashlib.sha256(data).hexdigest()[:12]
 
 
first = page("openslr/librispeech_asr", "clean", "test", 0, 1)["rows"][0]["row"]
clip, _ = load_audio(first)
t0 = time.perf_counter()
out = W.transcribe(clip, word_timestamps=True)
wall = (time.perf_counter() - t0) * 1000
print(f"\nclip {first['id']}  {len(clip)/SR:.1f} s   reference: {first['text']}")
print(f"-> {out['text']!r}")
print(f"   language {out['language']}   engine time to first token {out['ttft_ms']:.1f} ms   decode {out['decode_tps']:.0f} tok/s"
      f"   wall {wall:.0f} ms")
print("   first words:", [(w["word"], w["start"], w["end"], w["probability"]) for w in out["words"][:3]])
RESULTS["1. Install Whistle, pin its weights and engine by Hub revision, transcribe one clip"] = \
    f"{os.path.getsize(weights)/1e6:.1f} MB weights, {len(clip)/SR:.1f} s clip in {wall:.0f} ms"

import jiwer
import whisper
import torch
from whisper.normalizers import EnglishTextNormalizer, BasicTextNormalizer
 
NORM_EN = EnglishTextNormalizer()      # the normalisers Whistle's own benchmarks use; applied to both sides, always
NORM_BASIC = BasicTextNormalizer()
SEED = 2026
N_WHISTLE = 300                        # clips per LibriSpeech split that Whistle transcribes (it is cheap)
N_PAIRED = {"other": 100, "clean": 60}  # the first N of them also go through Whisper (it is not); test-other is primary
MAX_CLIP_S = 30.0                      # Whistle's single-pass window; longer clips are declared out of scope here
WHISPER_SECONDS = collections.Counter()
 
 
def all_rows(dataset, config, split):
    out, offset = [], 0
    while True:
        rows = page(dataset, config, split, offset)["rows"]
        out += [r["row"] for r in rows]
        offset += 100
        if len(rows) < 100:
            return out
 
 
def dataset_sha(dataset):
    return json.loads(fetch(f"https://huggingface.co/api/datasets/{dataset}"))["sha"][:12]
 
 
def wer_counts(ref, hyp):
    """Errors (S+D+I) and reference words for one normalised pair; an empty hypothesis is all deletions."""
    n = len(ref.split())
    if not hyp.strip():
        return n, n
    o = jiwer.process_words(ref, hyp)
    return o.substitutions + o.deletions + o.insertions, n
 
 
def sdi(ref, hyp):
    if not hyp.strip():
        return 0, len(ref.split()), 0
    o = jiwer.process_words(ref, hyp)
    return o.substitutions, o.deletions, o.insertions
 
 
def corpus_wer(errs, words):
    return 100.0 * sum(errs) / max(sum(words), 1)
 
 
def _by_cluster(clusters, *arrays):
    tot = collections.defaultdict(lambda: [0.0] * len(arrays))
    for i, c in enumerate(clusters):
        for j, a in enumerate(arrays):
            tot[c][j] += a[i]
    return tot
 
 
def wer_ci(clusters, errs, words, B=2000, seed=0):
    """95% percentile interval of a corpus WER, resampling clusters (speakers) with replacement."""
    tot = _by_cluster(clusters, errs, words)
    keys, rng = sorted(tot), random.Random(seed)
    boots = []
    for _ in range(B):
        d = rng.choices(keys, k=len(keys))
        boots.append(100 * sum(tot[c][0] for c in d) / max(sum(tot[c][1] for c in d), 1))
    return np.percentile(boots, [2.5, 97.5])
 
 
def paired_contrast(clusters, errs_a, words, errs_b, label_a, label_b, B=2000, seed=0):
    """WER(a) - WER(b) with a paired speaker-cluster bootstrap: the same resampled speakers for both systems."""
    tot = _by_cluster(clusters, errs_a, errs_b, words)
    keys, rng = sorted(tot), random.Random(seed)
    boots = []
    for _ in range(B):
        d = rng.choices(keys, k=len(keys))
        w = max(sum(tot[c][2] for c in d), 1)
        boots.append(100 * (sum(tot[c][0] for c in d) - sum(tot[c][1] for c in d)) / w)
    boots = np.array(boots)
    lo, hi = np.percentile(boots, [2.5, 97.5])
    p = 2 * min((boots <= 0).mean(), (boots >= 0).mean())
    return {"a": label_a, "b": label_b, "diff": corpus_wer(errs_a, words) - corpus_wer(errs_b, words),
            "lo": lo, "hi": hi, "p": max(p, 1 / B), "clusters": len(keys)}
 
 
def holm(ps, alpha=0.05):
    order, keep = np.argsort(ps), [False] * len(ps)
    for rank, i in enumerate(order):
        if ps[i] <= alpha / (len(ps) - rank):
            keep[i] = True
        else:
            break
    return keep
 
 
def verdict(c, significant):
    """Verdict rule: diff = WER(other system) - WER(Whistle). CI entirely > 0 and Holm-significant -> Whistle ahead;
    entirely < 0 and significant -> Whistle behind; otherwise no significant difference at this sample size."""
    if c["lo"] > 0 and significant:
        return "Whistle ahead"
    if c["hi"] < 0 and significant:
        return "Whistle behind"
    return "no significant difference at this n"
 
 
def fmt_contrast(c):
    return f"{c['diff']:+.2f} pts (95% CI {c['lo']:+.2f}..{c['hi']:+.2f}, {c['clusters']} speakers)"
 
 
@section("2. Fixed, seeded samples of LibriSpeech test-clean and test-other, and the rules for the verdicts")
def sample_english():
    global EN
    print("Rules for the verdicts below, written into this notebook before its final run (exploratory probes on the")
    print("same splits came first; the prose says which):")
    print(f"  * simple random sample of {N_WHISTLE} rows per split (seed {SEED}); the first {N_PAIRED['other']} (test-other) and "
          f"{N_PAIRED['clean']} (test-clean) are the paired Whisper subsets")
    print(f"  * clips longer than {MAX_CLIP_S:.0f} s are excluded and counted (Whistle's single-pass window), as are empty references")
    print("  * the same normaliser on both sides of each comparison (Whisper's EnglishTextNormalizer for LibriSpeech, its")
    print("    BasicTextNormalizer for MLS); corpus WER = summed errors / summed reference words")
    print("  * interval: 95% speaker-cluster paired bootstrap, 2,000 draws, the same resampled speakers for every system")
    print("  * verdict per contrast: unadjusted CI of WER(other) - WER(Whistle) entirely > 0 and Holm-adjusted p < 0.05 ->")
    print("    Whistle ahead; entirely < 0 and significant -> behind; otherwise no significant difference at this n")
    print("    family A (published claims): test-other and test-clean vs base greedy and vs base beam 5, MLS pooled vs base greedy;")
    print("    family B: the keyword contrasts. Holm at alpha = 0.05 within each family. Whisper tiny rows are descriptive.\n")
    global MANIFEST
    EN, manifest = {}, {"dataset_sha": dataset_sha("openslr/librispeech_asr"), "seed": SEED, "splits": {}}
    MANIFEST = manifest
    for split in ("other", "clean"):
        t0 = time.time()
        rows = all_rows("openslr/librispeech_asr", split, "test")
        idx = random.Random(SEED).sample(range(len(rows)), N_WHISTLE)
        kept, too_long, empty = [], 0, 0
        for i in idx:
            r = rows[i]
            audio, h = load_audio(r)
            ref = NORM_EN(r["text"])
            if len(audio) / SR > MAX_CLIP_S:
                too_long += 1
                continue
            if not ref:
                empty += 1
                continue
            kept.append({"id": r["id"], "spk": str(r["speaker_id"]), "text": r["text"], "ref": ref,
                         "audio": audio, "dur": len(audio) / SR, "sha": h})
        EN[split] = kept
        manifest["splits"][split] = {"rows": len(rows), "sampled": N_WHISTLE, "too_long": too_long, "empty_ref": empty,
                                     "ids": [c["id"] for c in kept], "sha": [c["sha"] for c in kept]}
        durs = [c["dur"] for c in kept]
        pk = kept[:N_PAIRED[split]]
        print(f"test-{split}: {len(rows)} rows -> {len(kept)} clips kept ({too_long} over {MAX_CLIP_S:.0f} s, {empty} empty), "
              f"{len(set(c['spk'] for c in kept))} speakers, {sum(durs)/60:.1f} min, clip {np.mean(durs):.1f} s mean / {max(durs):.1f} s max"
              f"   [{time.time()-t0:.0f} s]")
        print(f"   paired subset: {len(pk)} clips, {len(set(c['spk'] for c in pk))} speakers, {sum(c['dur'] for c in pk):.0f} s of audio, "
              f"{sum(len(c['ref'].split()) for c in pk)} reference words")
    json.dump(manifest, open("manifest.json", "w"))
    print(f"\nmanifest.json: LibriSpeech revision {manifest['dataset_sha']}, clip ids and file hashes (the MLS and MInDS-14 clips are added to it in")
    print("steps 9 and 12), so a rerun can prove it used the same audio")
    return f"{len(EN['other'])} + {len(EN['clean'])} clips, manifest written"
 
 
sample_english()

WHISPER = {}
 
 
def whisper_model(size):
    if size not in WHISPER:
        WHISPER[size] = whisper.load_model(size, device="cpu")
    return WHISPER[size]
 
 
def run_whisper(size, audio, language="en", **kw):
    """openai-whisper with its defaults (greedy with temperature fallback) unless kw says otherwise. Seeded per call,
    because the fallback samples. Returns text, whether any segment needed the fallback, and the wall time."""
    torch.manual_seed(0)
    WHISPER_SECONDS[size] += len(audio) / SR
    t0 = time.perf_counter()
    r = whisper_model(size).transcribe(audio, language=language, fp16=False, **kw)
    fell_back = any(s.get("temperature", 0) > 0 for s in r["segments"])
    return r["text"], fell_back, time.perf_counter() - t0
 
 
SYSTEMS = {"other": [("whisper tiny, greedy", "tiny", {}), ("whisper base, greedy", "base", {}), ("whisper base, beam 5", "base", {"beam_size": 5})],
           "clean": [("whisper tiny, greedy", "tiny", {}), ("whisper base, greedy", "base", {}), ("whisper base, beam 5", "base", {"beam_size": 5})]}
FAMILY_A = {"test-clean: Whistle vs whisper base, greedy", "test-clean: Whistle vs whisper base, beam 5",
            "test-other: Whistle vs whisper base, greedy", "test-other: Whistle vs whisper base, beam 5",
            "MLS pooled: Whistle vs whisper base, greedy"}
PUBLISHED = {"other": "Whistle 10.49 / Whisper base 11.0 (paper table 9, beam) / 12.4 (table 8, greedy)",
             "clean": "Whistle 4.31 / Whisper base 4.9 (paper table 9, beam) / 5.0 (table 8, greedy)"}
CONTRASTS = []
 
 
@section("3. Word error rate against Whisper tiny and base, paired by clip, with a verdict per contrast")
def english_wer():
    global HYP
    HYP = {}
    for split in ("other", "clean"):
        clips = EN[split]
        t0 = time.perf_counter()
        hw = [W.transcribe(c["audio"], language="en")["text"] for c in clips]
        tw = time.perf_counter() - t0
        HYP[(split, "Whistle")] = hw
        e_w, n_w = zip(*[wer_counts(c["ref"], NORM_EN(h)) for c, h in zip(clips, hw)])
        lo, hi = wer_ci([c["spk"] for c in clips], e_w, n_w)
        print(f"test-{split}: Whistle on all {len(clips)} clips: WER {corpus_wer(e_w, n_w):.2f}% (95% speaker-cluster CI {lo:.2f}..{hi:.2f}) "
              f"in {tw:.1f} s ({1000*tw/len(clips):.0f} ms per clip)")
        print(f"   published full-split figures, different sample: {PUBLISHED[split]}")
        pk = clips[:N_PAIRED[split]]
        spk = [c["spk"] for c in pk]
        e_wp, n_p = list(e_w[:len(pk)]), list(n_w[:len(pk)])
        s = np.sum([sdi(c["ref"], NORM_EN(h)) for c, h in zip(pk, hw[:len(pk)])], axis=0)
        print(f"   paired subset, {len(pk)} clips / {len(set(spk))} speakers / {sum(n_p)} words:")
        print(f"   {'Whistle, 5-beam':22s} {corpus_wer(e_wp, n_p):6.2f}%  S/D/I {s[0]:3d}/{s[1]:3d}/{s[2]:3d}")
        for label, size, kw in SYSTEMS[split]:
            res = [run_whisper(size, c["audio"], **kw) for c in pk]
            hyps = [r[0] for r in res]
            HYP[(split, label)] = hyps
            e_o = [wer_counts(c["ref"], NORM_EN(h))[0] for c, h in zip(pk, hyps)]
            s = np.sum([sdi(c["ref"], NORM_EN(h)) for c, h in zip(pk, hyps)], axis=0)
            con = paired_contrast(spk, e_o, n_p, e_wp, label, "Whistle")
            con["name"] = f"test-{split}: Whistle vs {label}"
            CONTRASTS.append(con)
            better, worse = sum(a < b for a, b in zip(e_wp, e_o)), sum(a > b for a, b in zip(e_wp, e_o))
            print(f"   {label:22s} {corpus_wer(e_o, n_p):6.2f}%  S/D/I {s[0]:3d}/{s[1]:3d}/{s[2]:3d}  fallback on {sum(r[1] for r in res):2d} clips  "
                  f"{np.median([r[2] for r in res])*1000:5.0f} ms/clip  | {fmt_contrast(con)}  clips Whistle better/worse {better}/{worse}")
    fam = [c for c in CONTRASTS if c["name"] in FAMILY_A]
    print(f"\nContrasts so far (family A: {len(fam)} of 5; the Holm correction and the verdicts come once the MLS contrast is in):")
    for c in CONTRASTS:
        tag = "family A" if c["name"] in FAMILY_A else "descriptive"
        print(f"   {c['name']:48s} {fmt_contrast(c)}  p~{c['p']:.3f}  [{tag}]")
    print("\n   Whistle decodes with a 5-beam length-normalised search. openai-whisper's default is greedy with temperature")
    print("   fallback; the Whisper paper's LibriSpeech numbers come from beam search, so the beam-5 rows are the like-for-like ones.")
    print("   Every gap here is a system gap (model plus decoder), not a model-only one.")
    return " | ".join(f"test-{s}: Whistle {corpus_wer(*zip(*[wer_counts(c['ref'], NORM_EN(h)) for c, h in zip(EN[s][:N_PAIRED[s]], HYP[(s, 'Whistle')][:N_PAIRED[s]])])):.2f}% "
                      f"vs base greedy {corpus_wer(*zip(*[wer_counts(c['ref'], NORM_EN(h)) for c, h in zip(EN[s][:N_PAIRED[s]], HYP[(s, 'whisper base, greedy')])])):.2f}%" for s in ("other", "clean")) 
 
 
english_wer()

def audio_seconds(clips, seconds):
    """Exactly `seconds` of speech: clips concatenated in sample order, then trimmed."""
    out, total = [], 0
    for c in clips:
        out.append(c["audio"])
        total += c["dur"]
        if total >= seconds:
            break
    return np.concatenate(out)[:int(seconds * SR)]
 
 
PROBE_SCRIPT = r"""
import sys, json, time, resource, numpy as np
kind, paths = sys.argv[1], sys.argv[2:]
scale = 1 << 20 if sys.platform == "darwin" else 1 << 10        # ru_maxrss is bytes on macOS, kilobytes on Linux
rss = lambda: resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / scale
out = {"kind": kind}
if kind.startswith("whistle"):
    import needle
    out["after_import"] = rss()
    W = needle.Whistle()
    run = lambda a: W.transcribe(a, language="en")
else:
    import whisper, torch
    out["after_import"] = rss()
    size, kw = kind.split("-")[1], ({"beam_size": 5} if kind.endswith("beam5") else {})
    m = whisper.load_model(size, device="cpu")
    def run(a):
        torch.manual_seed(0)
        return m.transcribe(a, language="en", fp16=False, **kw)
clips = {p: np.load(p) for p in paths}
run(clips[paths[0]])                                               # warm-up
reps = 5 if kind.startswith("whistle") else 3
for p, a in clips.items():
    ts = []
    for _ in range(reps):
        t0 = time.perf_counter(); r = run(a); ts.append(time.perf_counter() - t0)
    out[p] = {"wall_ms": 1000 * float(np.median(ts)), "spread_ms": 1000 * float(np.ptp(ts))}
    if kind.startswith("whistle"):
        out[p].update(ttft_ms=r["ttft_ms"], tps=r["decode_tps"])
out["peak"] = rss()
print(json.dumps(out))
"""
 
 
@section("4. Speed and footprint: identical arrays, one fresh process per engine, total wall time and peak memory")
def speed():
    pool = EN["other"]
    print(f"hardware: {CPU}, {os.cpu_count()} cpus visible. Each engine runs in its own fresh process so torch's threads cannot slow")
    print("Whistle down (or the reverse); Whistle's engine picks its own thread count. Primary measure: total wall time on the same")
    print("float32 array, median of 5 (Whistle) or 3 (Whisper) after a warm-up. ttft_ms is the engine's own clock, for reference.\n")
    for secs in (5, 10, 30):
        np.save(f"clip{secs}.npy", audio_seconds(pool, secs))
    open("probe.py", "w").write(PROBE_SCRIPT)
    res = {}
    for kind in ("whistle", "whisper-tiny", "whisper-base", "whisper-base-beam5"):
        out = subprocess.run([sys.executable, "probe.py", kind, "clip5.npy", "clip10.npy", "clip30.npy"], capture_output=True, text=True, env=os.environ)
        if out.returncode != 0:
            raise RuntimeError(f"{kind} probe failed: {out.stderr[-300:]}")
        res[kind] = json.loads(out.stdout.strip().splitlines()[-1])
    print(f"{'clip':>6s} | {'Whistle wall':>12s} {'spread':>7s} {'ttft':>6s} {'tok/s':>6s} | {'tiny greedy':>11s} | {'base greedy':>11s} | {'base beam 5':>11s}")
    for secs in (5, 10, 30):
        p = f"clip{secs}.npy"
        w = res["whistle"][p]
        print(f"{secs:5d} s | {w['wall_ms']:9.0f} ms {w['spread_ms']:5.0f} ms {w['ttft_ms']:6.1f} {w['tps']:6.0f} | "
              f"{res['whisper-tiny'][p]['wall_ms']:8.0f} ms | {res['whisper-base'][p]['wall_ms']:8.0f} ms | {res['whisper-base-beam5'][p]['wall_ms']:8.0f} ms")
    w5, w10, w30 = (res["whistle"][p]["wall_ms"] for p in ("clip5.npy", "clip10.npy", "clip30.npy"))
    b5, b10, b30 = (res["whisper-base"][p]["wall_ms"] for p in ("clip5.npy", "clip10.npy", "clip30.npy"))
    bb5, bb10, bb30 = (res["whisper-base-beam5"][p]["wall_ms"] for p in ("clip5.npy", "clip10.npy", "clip30.npy"))
    print(f"\nreal-time factor on the 10 s clip: Whistle {w10/10000:.4f}, base greedy {b10/10000:.3f}, base beam 5 {bb10/10000:.3f}")
    print(f"base greedy / Whistle wall time: {b5/w5:.1f}x at 5 s, {b10/w10:.1f}x at 10 s, {b30/w30:.1f}x at 30 s; beam 5 / Whistle: "
          f"{bb5/w5:.1f}x, {bb10/w10:.1f}x, {bb30/w30:.1f}x")
    print(f"from 5 to 10 s the wall time grows {w10/w5:.2f}x for Whistle, {b10/b5:.2f}x for base greedy, {bb10/bb5:.2f}x for base beam 5;")
    print(f"from 10 to 30 s: {w30/w10:.2f}x, {b30/b10:.2f}x, {bb30/bb10:.2f}x. Whisper's encoder always sees a padded 30 s window; its decoder")
    print("still grows with the words it emits, so the padding fixes only part of its cost. The encoder is not timed separately here.")
    print("\npeak RSS of each process (model loaded, three clips transcribed) and the RSS right after its imports:")
    for kind in ("whistle", "whisper-tiny", "whisper-base"):
        r = res[kind]
        print(f"   {kind:13s} peak {r['peak']:6.0f} MB   after imports {r['after_import']:4.0f} MB   -> {r['peak']-r['after_import']:4.0f} MB for the model and the run")
    sizes = {"whistle.cact + engine": (os.path.getsize(weights) + os.path.getsize(engine)) / 1e6}
    for size in ("tiny", "base"):
        p = os.path.expanduser(f"~/.cache/whisper/{size}.pt")
        sizes[f"whisper {size}.pt"] = os.path.getsize(p) / 1e6 if os.path.exists(p) else float("nan")
    print("files on disk: " + ", ".join(f"{k} {v:.1f} MB" for k, v in sizes.items()) + "  (file sizes, not deployed footprint)")
    off = subprocess.run([sys.executable, "-c", "import needle, numpy as np; W = needle.Whistle(); print(W.transcribe(np.zeros(16000, np.float32))['text'] == '')"],
                         capture_output=True, text=True, env={**os.environ, "HF_HUB_OFFLINE": "1"})
    print(f"fresh process with HF_HUB_OFFLINE=1 and the pinned paths: {'loads and runs' if 'True' in off.stdout else 'FAILED: ' + off.stderr[-200:]}")
    return f"10 s clip: Whistle {w10:.0f} ms vs base greedy {b10:.0f} ms ({b10/w10:.1f}x, {b5/w5:.1f}x at 5 s, {b30/w30:.1f}x at 30 s); peak RSS {res['whistle']['peak']:.0f} vs {res['whisper-base']['peak']:.0f} MB"
 
 
speed()