Hitonabi
09a1c98514
- 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>
58 lines
2.6 KiB
Python
58 lines
2.6 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Warum 'manchmal zu laut' + 'Stimme verändert sich am Anfang'? Stream-Kopf-Logik N× laufen lassen
|
||
und Gain, Crop-Punkt, Kopf-Segmente, Whisper-Start, rms/peak protokollieren."""
|
||
import os, numpy as np, librosa, soundfile as sf
|
||
from pocket_tts import TTSModel
|
||
from faster_whisper import WhisperModel
|
||
|
||
BASE = r"F:\Coding Stuff\mission-control-2\client\lucy-tts"
|
||
src,_ = librosa.load(os.path.join(BASE,"ref.mp3"), sr=24000, mono=True)
|
||
srt,_ = librosa.effects.trim(src, top_db=30); ref=os.path.join(BASE,"ref.wav")
|
||
sf.write(ref, srt[:int(18*24000)], 24000)
|
||
TARGET_RMS=0.09; LEAD="Tja. "
|
||
m = TTSModel.load_model(language="german_24l", lsd_decode_steps=6, temp=0.9)
|
||
vs = m.get_state_for_audio_prompt(ref); sr=m.sample_rate
|
||
w = WhisperModel("small", device="cpu", compute_type="int8")
|
||
TEXTS = ["Hallo Commander, ich höre dich.",
|
||
"Natürlich, Commander. Das Backup ist sauber durchgelaufen."]
|
||
|
||
def run_once(text):
|
||
chunks=[]
|
||
for c in m.generate_audio_stream(vs, LEAD+text, frames_after_eos=4):
|
||
c=c.numpy() if hasattr(c,"numpy") else np.asarray(c)
|
||
chunks.append(np.asarray(c,dtype=np.float32).reshape(-1))
|
||
full=np.concatenate(chunks)
|
||
# Kopf adaptiv (wie Server): bis >=2 Segmente oder 2.5s
|
||
head=[]; hl=0; head_arr=None
|
||
for c in chunks:
|
||
head.append(c); hl+=c.size
|
||
if hl < int(0.5*sr): continue
|
||
a=np.concatenate(head)
|
||
if len(librosa.effects.split(a, top_db=35))>=2 or hl>=int(2.5*sr):
|
||
head_arr=a; break
|
||
if head_arr is None: head_arr=np.concatenate(head)
|
||
iv=librosa.effects.split(head_arr, top_db=35)
|
||
segs=[(round(s/sr,2),round(e/sr,2)) for s,e in iv]
|
||
# Gain aus voiced (aktuelle Server-Logik)
|
||
voiced=np.concatenate([head_arr[s:e] for s,e in iv]) if len(iv) else head_arr
|
||
rms=float(np.sqrt(np.mean(voiced**2))) or 1e-9
|
||
gain=TARGET_RMS/rms
|
||
# Crop (aktuell): kurz vor echtem Wort
|
||
if len(iv)>=2:
|
||
cut=max(iv[0][1], iv[1][0]-int(0.06*sr)); cropped=full[cut:]
|
||
else:
|
||
cropped=full
|
||
out=np.clip(cropped*gain,-0.95,0.95)
|
||
sf.write(os.path.join(BASE,"out_diag_s.wav"), out, sr)
|
||
seg,_=w.transcribe(os.path.join(BASE,"out_diag_s.wav"), language="de", beam_size=5)
|
||
start=(" ".join(x.text for x in seg)).strip()[:30]
|
||
return dict(segs=segs, voiced_rms=round(rms,3), gain=round(gain,2),
|
||
out_rms=round(float(np.sqrt(np.mean(out**2))),3), out_peak=round(float(np.abs(out).max()),3),
|
||
start=start)
|
||
|
||
for text in TEXTS:
|
||
print(f"=== {text[:30]!r} ===")
|
||
for i in range(4):
|
||
print(f" run{i}:", run_once(text))
|
||
print("DIAG2_DONE")
|