148 lines
7.0 KiB
Python
148 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Patch Hermes TTS tool to add Piper provider and remove Edge TTS fallback."""
|
|
import sys
|
|
|
|
tts_path = '/opt/hermes/.venv/lib/python3.13/site-packages/tools/tts_tool.py'
|
|
|
|
with open(tts_path) as f:
|
|
code = f.read()
|
|
|
|
# Replace the Coqui provider block with Piper
|
|
old_coqui = ' elif provider == "coqui":'
|
|
new_piper = ''' elif provider == "piper":
|
|
logger.info("Generating speech with Piper TTS (local, CPU)...")
|
|
import subprocess
|
|
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):
|
|
raise FileNotFoundError(f"Piper voice model not found: {model_path}")
|
|
cmd = [piper_binary, "--model", model_path, "--output-raw"]
|
|
proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.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]
|
|
subprocess.run(ffmpeg_cmd, input=raw_audio, capture_output=True, timeout=30)
|
|
logger.info("Piper TTS audio saved: %s", file_str)'''
|
|
|
|
if old_coqui in code:
|
|
code = code.replace(old_coqui, new_piper)
|
|
print("Coqui -> Piper replaced")
|
|
else:
|
|
# Fresh Hermes install - add Piper after the last provider (kittentts)
|
|
if 'provider == "piper"' in code:
|
|
print("Piper already present, skipping coqui replacement")
|
|
else:
|
|
print("Stock Hermes - adding Piper provider after kittentts...")
|
|
marker = ' elif provider == "kittentts":'
|
|
if marker in code:
|
|
# Find the end of the kittentts block and insert Piper before else
|
|
lines = code.split('\n')
|
|
kit_idx = None
|
|
for i, line in enumerate(lines):
|
|
if line.strip().startswith('elif provider == "kittentts":'):
|
|
kit_idx = i
|
|
break
|
|
if kit_idx is not None:
|
|
# Find the next blank line + else block
|
|
for i in range(kit_idx, len(lines)):
|
|
if lines[i].strip() == 'else:':
|
|
# Insert Piper block before this line
|
|
indent = ' '
|
|
piper_lines = new_piper.split('\n')
|
|
insert = piper_lines + ['']
|
|
lines[i:i] = insert
|
|
code = '\n'.join(lines)
|
|
print("Piper provider added after kittentts")
|
|
break
|
|
|
|
# Replace the Edge fallback with Piper fallback
|
|
old_edge = ''' 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_piper_fallback = ''' else:
|
|
# Default: Piper TTS (local, CPU, no cloud)
|
|
piper_available = False
|
|
try:
|
|
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):
|
|
piper_available = True
|
|
except Exception:
|
|
pass
|
|
|
|
if piper_available:
|
|
logger.info("Generating speech with Piper TTS (local, CPU)...")
|
|
import subprocess
|
|
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")
|
|
cmd = [piper_binary, "--model", model_path, "--output-raw"]
|
|
proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.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]
|
|
subprocess.run(ffmpeg_cmd, input=raw_audio, capture_output=True, timeout=30)
|
|
logger.info("Piper TTS audio saved: %s", file_str)
|
|
else:
|
|
return json.dumps({
|
|
"success": False,
|
|
"error": "No TTS provider available. Install Piper TTS (pip install piper-tts) "
|
|
"and download a voice model."
|
|
}, ensure_ascii=False)'''
|
|
|
|
if old_edge in code:
|
|
code = code.replace(old_edge, new_piper_fallback)
|
|
print("Edge fallback replaced with Piper")
|
|
else:
|
|
print("Edge fallback NOT found, checking if already Piper...")
|
|
if 'Default: Piper TTS' in code:
|
|
print("Piper fallback already present, skipping")
|
|
else:
|
|
print("ERROR: Could not find Edge fallback")
|
|
# Debug: show what the else block looks like
|
|
import re
|
|
match = re.search(r' else:\n # Default:', code)
|
|
if match:
|
|
print("Found else block at", match.start())
|
|
sys.exit(1)
|
|
|
|
with open(tts_path, 'w') as f:
|
|
f.write(code)
|
|
print("tts_tool.py patched successfully")
|