Lucy-TTS/F5: Skripte + Batches versionieren, schwere Assets ignoriert

- pocket_server.py (Produktions-TTS mit Stimmen-Waechter), text_norm, Bench-/Diag-Skripte - lucy-f5: f5_server/f5_test/bench_dml (DirectML-Experiment, Phase C/D offen) - .gitignore: venvs/Modelle/Audio/Logs der beiden Ordner + box_recon/gemma_swap-Scratch Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-02 10:29:33 +02:00
parent aff0105700
commit 09a1c98514
49 changed files with 8231 additions and 0 deletions
@@ -0,0 +1,109 @@
+# -*- coding: utf-8 -*-
+"""bench_lucy.py — A3/B1: misst Lucys Stimme (pocket-tts) über temp/lsd.
+
+Metriken je Konfig: RTF, gen-Zeit und ROHE Kollaps-Rate (EINE Generierung, OHNE best-of-N-Gate)
+— also genau das, was der Stimm-Wächter aktuell wegbügeln muss (= Latenz-Treiber). Niedrigere
+temp sollte die rohe Kollaps-Rate senken -> weniger Regenerationen -> niedrigere effektive Latenz.
+
+Nutzung (im ptts-venv der lucy-tts-Maschine):
+  python bench_lucy.py                          # Default-Sweep
+  python bench_lucy.py --reps 5
+  python bench_lucy.py --temps 0.7,0.8,0.9 --lsds 8,10,12
+  python bench_lucy.py --whisper                # + Verständlichkeits-Check (faster-whisper)
+
+Reuse: Kollaps-Logik (F0 + MFCC-Fingerabdruck) und cleanup() kommen 1:1 aus pocket_server.py,
+damit der Benchmark exakt die Produktions-Kriterien misst.
+"""
+import os, time, argparse, statistics as st
+import numpy as np, soundfile as sf, librosa
+from pocket_tts import TTSModel
+import pocket_server as ps
+
+SENT = {
+    "kurz":   "Hallo Commander, ich höre dich.",
+    "mittel": "Guten Morgen, Commander. Das Backup ist sauber durchgelaufen und es gab keine Fehler.",
+    "lang":   "Natürlich kümmere ich mich darum, Commander. Ich starte den Dienst neu, prüfe die "
+              "Protokolle und melde mich, sobald alles wieder läuft.",
+}
+
+
+def _voice_state(m):
+    """Voice-State laden: bevorzugt das gecachte safetensors (A1), sonst aus der Referenz klonen."""
+    if os.path.exists(ps.VOICE_ST):
+        try:
+            return m.get_state_for_audio_prompt(ps.VOICE_ST)
+        except Exception as e:
+            print(f"  (safetensors-Load fehlgeschlagen, klone aus Referenz: {e})")
+    return m.get_state_for_audio_prompt(ps._prep_ref())
+
+
+def _raw_gen(m, vs, ref_fp, text, sr):
+    """EINE rohe Generierung (kein Gate). -> (gen_s, audio_s, male, sim, audio)."""
+    t0 = time.time()
+    audio = m.generate_audio(vs, ps.LEAD + text, frames_after_eos=ps.FEOS)
+    gen = time.time() - t0
+    a = audio.numpy() if hasattr(audio, "numpy") else np.asarray(audio)
+    a = np.asarray(a, dtype=np.float32).reshape(-1)
+    f0 = ps._voiced_f0(a, sr)
+    male = (f0 == f0) and f0 < ps.F0_FLOOR              # NaN-sicher
+    fp = ps._fingerprint(ps._crop_lead(a, sr), sr)
+    sim = float(np.dot(ref_fp, fp)) if (ref_fp is not None and fp is not None) else 1.0
+    return gen, a.size / sr, bool(male), sim, a
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--temps", default="0.7,0.8,0.9")
+    ap.add_argument("--lsds", default="8,10")
+    ap.add_argument("--reps", type=int, default=4)
+    ap.add_argument("--whisper", action="store_true")
+    args = ap.parse_args()
+    temps = [float(x) for x in args.temps.split(",")]
+    lsds = [int(x) for x in args.lsds.split(",")]
+
+    OUT = os.path.join(ps.BASE, "out_bench"); os.makedirs(OUT, exist_ok=True)
+    ref = ps._prep_ref()
+    print(f"LANG={ps.LANG} quantize={ps.QUANTIZE} noise_clamp={ps.NOISE_CLAMP} "
+          f"F0_FLOOR={ps.F0_FLOOR} FP_FLOOR={ps.FP_FLOOR} reps={args.reps}")
+    print(f"{'temp':>5} {'lsd':>4} | {'RTF':>5} {'gen/s':>6} | "
+          f"{'collapse%':>9} {'sim_min':>7} {'sim_avg':>7}")
+    rows = []
+    for lsd in lsds:
+        for temp in temps:
+            m = TTSModel.load_model(language=ps.LANG, lsd_decode_steps=lsd, temp=temp,
+                                    noise_clamp=ps.NOISE_CLAMP, quantize=ps.QUANTIZE)
+            sr = m.sample_rate
+            ref_audio, _ = librosa.load(ref, sr=sr, mono=True)
+            ref_fp = ps._fingerprint(ref_audio, sr)
+            vs = _voice_state(m)
+            rtfs, gens, sims, collapses, n = [], [], [], 0, 0
+            for name, text in SENT.items():
+                for r in range(args.reps):
+                    gen, asec, male, sim, a = _raw_gen(m, vs, ref_fp, text, sr)
+                    rtfs.append(gen / max(asec, 0.01)); gens.append(gen); sims.append(sim)
+                    collapses += int(male or sim < ps.FP_FLOOR); n += 1
+                    if r == 0:  # ein Sample je Satz zum Reinhören
+                        sf.write(os.path.join(OUT, f"t{temp}_l{lsd}_{name}.wav"), ps.cleanup(a, sr), sr)
+            cr = 100.0 * collapses / max(n, 1)
+            print(f"{temp:>5} {lsd:>4} | {st.mean(rtfs):>5.2f} {st.mean(gens):>6.2f} | "
+                  f"{cr:>8.1f}% {min(sims):>7.3f} {st.mean(sims):>7.3f}")
+            rows.append((temp, lsd, st.mean(rtfs), cr, min(sims)))
+            del m
+
+    best = sorted(rows, key=lambda x: (x[3], x[2]))[0]  # min Kollaps, dann beste RTF
+    print(f"\n>> Vorschlag: temp={best[0]} lsd={best[1]} "
+          f"(collapse={best[3]:.1f}%, RTF={best[2]:.2f}) -> in pocket_server via "
+          f"LUCY_TEMP / LUCY_LSD setzen.")
+
+    if args.whisper:
+        print("\n=== Whisper-Verständlichkeit (faster-whisper small/int8) ===")
+        from faster_whisper import WhisperModel
+        import glob
+        wm = WhisperModel("small", device="cpu", compute_type="int8")
+        for p in sorted(glob.glob(os.path.join(OUT, "*.wav"))):
+            segs, _ = wm.transcribe(p, language="de", beam_size=5)
+            print(os.path.basename(p), "::", " ".join(s.text.strip() for s in segs))
+
+
+if __name__ == "__main__":
+    main()