Epictetus
772ef4f6fd
- Autonomous agent runtime (JSONL + fcntl locking, no token caps) - Moltbook-style prompts: agents share what they built, not scripted turns - Episodes 001-008 transcripts, conversation logs, system prompts - Producer brief for Hermes v0.18
113 lines
3.9 KiB
Python
113 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
AI Talk Show — TTS Pipeline
|
|
Renders a transcript to audio using edge-tts with two distinct voices.
|
|
Sequential rendering with retry for reliability.
|
|
"""
|
|
|
|
import json
|
|
import asyncio
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import edge_tts
|
|
from pathlib import Path
|
|
|
|
# ── Config ──────────────────────────────────────────────────────────
|
|
VOICES = {
|
|
"Leonard": "en-US-GuyNeural", # Warm, mid-range, conversational
|
|
"Charlie": "en-US-EricNeural", # Deeper, more measured/rational
|
|
}
|
|
RATE_ADJUST = "-8%"
|
|
PROJECT_DIR = Path(__file__).parent.parent
|
|
OUTPUT_DIR = PROJECT_DIR / "outputs"
|
|
SEGMENTS_DIR = OUTPUT_DIR / "segments"
|
|
SEGMENTS_DIR.mkdir(exist_ok=True)
|
|
|
|
MAX_RETRIES = 3
|
|
RETRY_DELAY = 2
|
|
|
|
def load_transcript(path: Path) -> dict:
|
|
return json.loads(path.read_text())
|
|
|
|
async def speak(text: str, agent: str, turn_num: int, output_path: Path) -> None:
|
|
"""Generate TTS for one turn with retry."""
|
|
voice = VOICES[agent]
|
|
for attempt in range(MAX_RETRIES):
|
|
try:
|
|
communicate = edge_tts.Communicate(text, voice, rate=RATE_ADJUST)
|
|
await communicate.save(str(output_path))
|
|
# Verify output is non-empty
|
|
if output_path.stat().st_size > 0:
|
|
print(f" ✓ {agent} turn {turn_num}: {output_path.name} ({output_path.stat().st_size} bytes)")
|
|
return
|
|
else:
|
|
print(f" ⚠ Empty output, retry {attempt+1}/{MAX_RETRIES}")
|
|
time.sleep(RETRY_DELAY)
|
|
except Exception as e:
|
|
print(f" ✗ {agent} turn {turn_num}: {e} (attempt {attempt+1}/{MAX_RETRIES})")
|
|
time.sleep(RETRY_DELAY * (attempt + 1))
|
|
raise RuntimeError(f"Failed TTS for {agent} turn {turn_num} after {MAX_RETRIES} retries")
|
|
|
|
def stitch_segments(episode_id: str, topic: str) -> Path:
|
|
segment_files = sorted(SEGMENTS_DIR.glob(f"ep{episode_id}_{topic}_seg_*.mp3"))
|
|
if not segment_files:
|
|
raise FileNotFoundError(f"No segment files found for ep{episode_id}_{topic}")
|
|
|
|
output_path = OUTPUT_DIR / f"ep{episode_id}_{topic}_audio.mp3"
|
|
concat_list = SEGMENTS_DIR / "concat.txt"
|
|
with open(concat_list, "w") as f:
|
|
for seg in segment_files:
|
|
f.write(f"file '{seg.absolute()}'\n")
|
|
|
|
subprocess.run([
|
|
"ffmpeg", "-y",
|
|
"-f", "concat", "-safe", "0",
|
|
"-i", str(concat_list),
|
|
"-c", "copy",
|
|
str(output_path)
|
|
], check=True, capture_output=True)
|
|
|
|
print(f"✅ Stitched audio: {output_path} ({output_path.stat().st_size:,} bytes)")
|
|
return output_path
|
|
|
|
def main():
|
|
if len(sys.argv) < 3:
|
|
print("Usage: python tts_pipeline.py <episode_id> <topic>")
|
|
sys.exit(1)
|
|
|
|
episode_id = sys.argv[1]
|
|
topic = sys.argv[2]
|
|
transcript_path = OUTPUT_DIR / f"ep{episode_id}_{topic}_transcript.json"
|
|
|
|
if not transcript_path.exists():
|
|
print(f"❌ Transcript not found: {transcript_path}")
|
|
sys.exit(1)
|
|
|
|
data = load_transcript(transcript_path)
|
|
conversation = data["conversation"]
|
|
|
|
print(f"=== TTS: Episode {episode_id} — {topic} ===\n", flush=True)
|
|
|
|
# Sequential rendering (reliable)
|
|
async def generate_all():
|
|
for turn in conversation:
|
|
agent = turn["agent"]
|
|
num = turn["turn"]
|
|
seg_path = SEGMENTS_DIR / f"ep{episode_id}_{topic}_seg_{num:02d}.mp3"
|
|
await speak(turn["content"], agent, num, seg_path)
|
|
|
|
asyncio.run(generate_all())
|
|
|
|
# Stitch
|
|
audio_path = stitch_segments(episode_id, topic)
|
|
|
|
# Summary
|
|
total_words = sum(t.get("word_count", len(t["content"].split())) for t in conversation)
|
|
estimated_minutes = total_words / 150
|
|
print(f"\n📊 {total_words} words, ~{estimated_minutes:.1f} min estimated")
|
|
print(f"🎧 Audio: {audio_path}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|