compose/ai/hermes/patch_tts_tool.py

#!/usr/bin/env python3
"""Patch Hermes TTS tool: add Piper TTS provider, remove Edge TTS as default.

Patches ALL copies of tts_tool.py found (venv site-packages + /opt/hermes/tools/).

Searches multiple paths for tts_tool.py so it works both at build time
(in the image venv) and at runtime (on the mounted data volume).

Idempotent: if already patched, does nothing.
"""

import sys
import os

# ---------------------------------------------------------------------------
# Search for all copies of tts_tool.py
# ---------------------------------------------------------------------------
CANDIDATE_PATHS = [
    "/opt/hermes/.venv/lib/python3.13/site-packages/tools/tts_tool.py",
    "/opt/hermes/tools/tts_tool.py",
]

found_paths = []

for p in CANDIDATE_PATHS:
    if os.path.exists(p):
        found_paths.append(p)
        print(f"Found tts_tool.py at: {p}")

# Also try to find via Python import
import subprocess
try:
    result = subprocess.run(
        [sys.executable, "-c", "import tools.tts_tool; print(tools.tts_tool.__file__)"],
        capture_output=True, text=True, timeout=5
    )
    if result.returncode == 0:
        p = result.stdout.strip()
        if os.path.exists(p) and p not in found_paths:
            found_paths.append(p)
            print(f"Found tts_tool.py via import at: {p}")
except Exception:
    pass

if not found_paths:
    print("WARNING: tts_tool.py not found anywhere. Patching deferred to runtime.")
    print(f"Searched: {CANDIDATE_PATHS}")
    sys.exit(0)

# ---------------------------------------------------------------------------
# Old else block: the Edge TTS default fallback to replace
# ---------------------------------------------------------------------------
old_else = '''        else:
            # Default: Edge TTS (free), with NeuTTS as local fallback
            edge_available = True
            try:
                _import_edge_tts()
            except ImportError:
                edge_available = False

            if edge_available:
                logger.info("Generating speech with Edge TTS...")
                try:
                    import concurrent.futures
                    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
                        pool.submit(
                            lambda: asyncio.run(_generate_edge_tts(text, file_str, tts_config))
                        ).result(timeout=60)
                except RuntimeError:
                    asyncio.run(_generate_edge_tts(text, file_str, tts_config))
            elif _check_neutts_available():
                logger.info("Edge TTS not available, falling back to NeuTTS (local)...")
                provider = "neutts"
                _generate_neutts(text, file_str, tts_config)
            else:
                return json.dumps({
                    "success": False,
                    "error": "No TTS provider available. Install edge-tts (pip install edge-tts) "
                             "or set up NeuTTS for local synthesis."
                }, ensure_ascii=False)'''

# ---------------------------------------------------------------------------
# New block: elif provider == "piper" + else: fallback with Piper only
# ---------------------------------------------------------------------------
new_block = '''        elif provider == "piper":
            # Piper TTS (local, CPU, no cloud, no Microsoft)
            piper_binary = "/opt/hermes/.venv/bin/piper"
            piper_config = tts_config.get("piper", {})
            voice = piper_config.get("voice", "en_US-lessac-medium")
            model_dir = piper_config.get("model_dir", "/opt/hermes/.venv/share/piper/voices")
            model_path = os.path.join(model_dir, f"{voice}.onnx")
            if not os.path.exists(model_path):
                return json.dumps({
                    "success": False,
                    "error": "Piper TTS voice model not found. "
                             "Install Piper TTS and download a voice model."
                }, ensure_ascii=False)
            logger.info("Generating speech with Piper TTS (local, CPU)...")
            import subprocess as _sp
            cmd = [piper_binary, "--model", model_path, "--output-raw"]
            try:
                proc = _sp.Popen(cmd, stdin=_sp.PIPE, stdout=_sp.PIPE, stderr=_sp.PIPE)
                raw_audio, stderr = proc.communicate(input=text.encode(), timeout=60)
                if proc.returncode != 0:
                    raise RuntimeError(f"Piper TTS failed: {stderr.decode()[:200]}")
                ffmpeg_cmd = ["ffmpeg", "-f", "s16le", "-ar", "22050", "-ac", "1", "-i", "-", "-y", file_str]
                _sp.run(ffmpeg_cmd, input=raw_audio, capture_output=True, timeout=30)
            except Exception as e:
                return json.dumps({
                    "success": False,
                    "error": f"Piper TTS failed: {e}"
                }, ensure_ascii=False)

        else:
            # Default: Piper TTS (local, CPU, no cloud, no Microsoft)
            piper_binary = "/opt/hermes/.venv/bin/piper"
            piper_config = tts_config.get("piper", {})
            voice = piper_config.get("voice", "en_US-lessac-medium")
            model_dir = piper_config.get("model_dir", "/opt/hermes/.venv/share/piper/voices")
            model_path = os.path.join(model_dir, f"{voice}.onnx")
            if os.path.exists(model_path) and os.path.exists(piper_binary):
                logger.info("Generating speech with Piper TTS (local, CPU)...")
                import subprocess as _sp
                cmd = [piper_binary, "--model", model_path, "--output-raw"]
                try:
                    proc = _sp.Popen(cmd, stdin=_sp.PIPE, stdout=_sp.PIPE, stderr=_sp.PIPE)
                    raw_audio, stderr = proc.communicate(input=text.encode(), timeout=60)
                    if proc.returncode != 0:
                        raise RuntimeError(stderr.decode()[:200])
                    ffmpeg_cmd = ["ffmpeg", "-f", "s16le", "-ar", "22050", "-ac", "1", "-i", "-", "-y", file_str]
                    _sp.run(ffmpeg_cmd, input=raw_audio, capture_output=True, timeout=30)
                except Exception:
                    pass
            else:
                return json.dumps({
                    "success": False,
                    "error": "Piper TTS not available. Install piper-tts and download a voice model."
                }, ensure_ascii=False)'''

# ---------------------------------------------------------------------------
# Apply the patch to all copies found
# ---------------------------------------------------------------------------
patched_any = False

for tts_path in found_paths:
    with open(tts_path) as f:
        code = f.read()

    if 'provider == "piper"' in code:
        print(f"ALREADY PATCHED: {tts_path}")
        continue

    if old_else in code:
        code = code.replace(old_else, new_block, 1)
        with open(tts_path, 'w') as f:
            f.write(code)
        print(f"PATCHED: {tts_path}")
        patched_any = True
    else:
        print(f"SKIP {tts_path}: Edge fallback pattern not found")
        import re
        for m in re.finditer(r'        else:\n            # Default:', code):
            start = max(0, m.start() - 100)
            end = min(len(code), m.end() + 300)
            print(f"  Found 'else:/# Default:' at position {m.start()}:")
            print(f"  {code[start:end]}")
            print("  ---")
        # Don't exit with error — if one copy isn't patchable, try the others

if not patched_any:
    all_patched = all(
        'provider == "piper"' in open(p).read()
        for p in found_paths
    )
    if all_patched:
        print("All copies already patched.")
        sys.exit(0)
    print("WARNING: Could not patch any copy of tts_tool.py")
    sys.exit(1)

print("tts_tool.py patched successfully across all copies.")