File size: 5,400 Bytes
27caffe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | """
Audio Generation
Primary: F5-TTS with finetuned model (my_finetuned_model/model.pth)
Fallback: edge-tts (Microsoft Neural voice) — no existing audio files ever copied
"""
import os
import sys
import asyncio
import subprocess
import tempfile
from pathlib import Path
# ── paths (all relative to this file) ────────────────────────────────────────
_ROOT = Path(__file__).parent.absolute()
_INFER_CLI = _ROOT / "infer_cli.py"
_CKPT = _ROOT / "my_finetuned_model" / "model.pth"
_REF_AUDIO = _ROOT / "segment_43.wav"
_REF_TEXT = "are you feeling mentally alert?"
# ── edge-tts fallback voice ────────────────────────────────────────────────────
EDGE_TTS_VOICE = "en-US-JennyNeural"
# ─────────────────────────────────────────────────────────────────────────────
# Internal helpers
# ─────────────────────────────────────────────────────────────────────────────
async def _edge_tts_generate(text: str, output_path: str) -> bool:
try:
import edge_tts
communicate = edge_tts.Communicate(text, EDGE_TTS_VOICE)
await communicate.save(output_path)
return os.path.exists(output_path) and os.path.getsize(output_path) > 500
except Exception as e:
print(f" edge-tts error: {e}")
return False
def _f5tts_generate(text: str, output_path: str) -> bool:
"""Call the local infer_cli.py with the finetuned model checkpoint."""
try:
if not _CKPT.exists():
print(f" F5-TTS checkpoint not found: {_CKPT}")
return False
if not _REF_AUDIO.exists():
print(f" F5-TTS reference audio not found: {_REF_AUDIO}")
return False
out_path = Path(output_path)
out_path.parent.mkdir(parents=True, exist_ok=True)
# Use a temp dir so we can rename the output to the exact path
with tempfile.TemporaryDirectory() as tmp:
out_file = "tts_output.wav"
cmd = [
sys.executable, str(_INFER_CLI),
"--ckpt_file", str(_CKPT),
"--ref_audio", str(_REF_AUDIO),
"--ref_text", _REF_TEXT,
"--gen_text", text,
"--output_dir", tmp,
"--output_file", out_file,
"--remove_silence",
"--speed", "1.0",
]
result = subprocess.run(
cmd, capture_output=True, text=True, timeout=120,
cwd=str(_ROOT)
)
generated = Path(tmp) / out_file
if result.returncode == 0 and generated.exists() and generated.stat().st_size > 500:
import shutil
shutil.move(str(generated), str(out_path))
return True
else:
if result.stderr:
print(f" F5-TTS stderr: {result.stderr[-400:]}")
return False
except Exception as e:
print(f" F5-TTS exception: {e}")
return False
# ─────────────────────────────────────────────────────────────────────────────
# Public API (called by mindfull_pipeline.py)
# ─────────────────────────────────────────────────────────────────────────────
def generate_audio_simple(text: str, output_path: str) -> bool:
"""
Generate audio from text.
Tries finetuned F5-TTS first; falls back to edge-tts.
Never copies existing audio files.
"""
out = Path(output_path)
out.parent.mkdir(parents=True, exist_ok=True)
# Always start fresh
if out.exists():
out.unlink()
# ── 1. try finetuned F5-TTS ──────────────────────────────────────────────
print(f"[audio] F5-TTS generating: {text[:60]}...")
if _f5tts_generate(text, output_path) and out.exists():
print(f" ✓ F5-TTS ready: {out.name} ({out.stat().st_size:,} bytes)")
return True
# ── 2. fallback: edge-tts ────────────────────────────────────────────────
print("[audio] F5-TTS failed – falling back to edge-tts...")
success = asyncio.run(_edge_tts_generate(text, output_path))
if success:
print(f" ✓ edge-tts ready: {out.name} ({out.stat().st_size:,} bytes)")
else:
print(" ✗ Both audio methods failed.")
return success
def generate_audio_minimal(text: str, output_path: str) -> bool:
"""Alias kept for backward compatibility."""
return generate_audio_simple(text, output_path)
|