simple_audio_gen.py

"""
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)