path: root/generate_jingles.py

#!/usr/bin/env python3
"""
Jingle Factory for Radio Susan

Phrases use {sfx_name} markers to insert sound effects at breakpoints.
Each phrase gets TTS'd with a random voice, then stitched with SFX.
Background ambience (static/noise) is layered underneath.

Usage: python3 generate_jingles.py [--list-sfx] [--preview PHRASE]

SFX are stored in /var/lib/radio/sfx/
Jingles output to /var/lib/radio/jingles/
"""

import asyncio
import json
import os
import random
import re
import subprocess
import sys
import tempfile
from pathlib import Path

RADIO_DIR = Path("/var/lib/radio")
SFX_DIR = RADIO_DIR / "sfx"
JINGLES_DIR = RADIO_DIR / "jingles"
VENV_BIN = RADIO_DIR / "venv" / "bin"
EDGE_TTS = VENV_BIN / "edge-tts"

# Voices to randomly pick from — mix of dramatic, cheesy, authoritative
VOICES = [
    "en-US-GuyNeural",          # Passion — classic radio voice
    "en-US-ChristopherNeural",  # Authority — news anchor
    "en-US-EricNeural",         # Rational — deadpan
    "en-US-BrianNeural",        # Casual bro
    "en-GB-RyanNeural",         # British bloke
    "en-GB-ThomasNeural",       # Posh British
    "en-US-AriaNeural",         # Confident woman
    "en-US-JennyNeural",        # Friendly woman
    "en-GB-SoniaNeural",        # British woman
]

# Phrase definitions — {sfx_name} marks where sound effects go
# Add as many as you want!
PHRASES = [
    "You're listening to {crash} ninety nine point oh {airhorn} Susan Radio!",
    "This is {static} Susan Radio {crash} ninety nine point oh {reverb_hit} all day, all night",
    "{static} You're tuned in to {crash} Susan Radio! {airhorn}",
    "Susan Radio {reverb_hit} ninety nine point oh {crash} on your dial",
    "{static} Don't touch that dial! {crash} Susan Radio {airhorn} ninety nine point oh!",
    "Coming at you live {crash} this is {reverb_hit} Susan Radio! {airhorn}",
    "{static} Lock it in {crash} ninety nine point oh {reverb_hit} Susan Radio!",
]


def generate_sfx():
    """Generate built-in sound effects using ffmpeg."""
    SFX_DIR.mkdir(parents=True, exist_ok=True)
    
    sfx_specs = {
        "crash": {
            # White noise burst with sharp attack, fast decay
            "filter": "anoisesrc=d=0.8:c=white:a=0.6,afade=t=in:d=0.02,afade=t=out:st=0.15:d=0.65,lowpass=f=3000",
        },
        "airhorn": {
            # Stacked sine waves with vibrato for that MLG airhorn feel
            "filter": (
                "sine=f=750:d=0.6,aformat=sample_rates=44100"
            ),
            "filter_complex": (
                "[0:a]volume=0.4[a];"
                "anoisesrc=d=0.6:c=pink:a=0.1,aformat=sample_rates=44100[n];"
                "sine=f=950:d=0.6,aformat=sample_rates=44100,volume=0.3[b];"
                "[a][n]amix=inputs=2[m1];[m1][b]amix=inputs=2,afade=t=in:d=0.02,afade=t=out:st=0.3:d=0.3[out]"
            ),
        },
        "static": {
            # Pink noise crackle
            "filter": "anoisesrc=d=0.5:c=pink:a=0.25,afade=t=in:d=0.1,afade=t=out:st=0.3:d=0.2,highpass=f=500",
        },
        "reverb_hit": {
            # Impact with reverb tail
            "filter": "anoisesrc=d=0.05:c=white:a=0.8,apad=whole_dur=1,aecho=0.8:0.88:60:0.4,afade=t=out:st=0.3:d=0.7",
        },
        "sweep": {
            # Frequency sweep
            "filter": "sine=f=200:d=0.4,aformat=sample_rates=44100,afade=t=in:d=0.05,afade=t=out:st=0.2:d=0.2,vibrato=f=20:d=1.0",
        },
    }
    
    for name, spec in sfx_specs.items():
        outfile = SFX_DIR / f"{name}.wav"
        if outfile.exists():
            continue
            
        if "filter_complex" in spec:
            cmd = [
                "ffmpeg", "-y",
                "-f", "lavfi", "-i", spec["filter"],
                "-filter_complex", spec["filter_complex"],
                "-map", "[out]",
                "-ar", "44100", "-ac", "2",
                str(outfile)
            ]
        else:
            cmd = [
                "ffmpeg", "-y",
                "-f", "lavfi", "-i", spec["filter"],
                "-ar", "44100", "-ac", "2",
                str(outfile)
            ]
        
        subprocess.run(cmd, capture_output=True)
        print(f"  Generated SFX: {name}")
    
    return {f.stem: str(f) for f in SFX_DIR.glob("*.wav")}


def tts_segment(text, voice, outfile):
    """Generate TTS for a text segment using edge-tts."""
    cmd = [
        str(EDGE_TTS),
        "--voice", voice,
        "--rate", "+10%",  # Slightly faster for radio energy
        "--text", text,
        "--write-media", outfile,
    ]
    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode != 0:
        print(f"  TTS error: {result.stderr}")
        return False
    return True


def stitch_jingle(segments, sfx_files, output_path):
    """
    Stitch TTS segments and SFX together with background ambience.
    segments: list of (type, path) where type is 'voice' or 'sfx'
    """
    if not segments:
        return False
    
    # Build ffmpeg concat filter
    inputs = []
    filter_parts = []
    
    for i, (stype, path) in enumerate(segments):
        inputs.extend(["-i", path])
        # Normalize all inputs to same format
        filter_parts.append(f"[{i}:a]aformat=sample_rates=44100:channel_layouts=stereo,aresample=44100[s{i}]")
    
    # Concat all segments
    concat_inputs = "".join(f"[s{i}]" for i in range(len(segments)))
    filter_parts.append(f"{concat_inputs}concat=n={len(segments)}:v=0:a=1[main]")
    
    # Generate background static layer
    filter_parts.append(
        "anoisesrc=d=30:c=pink:a=0.04,aformat=sample_rates=44100:channel_layouts=stereo,"
        "highpass=f=800,lowpass=f=4000[bg]"
    )
    
    # Mix main audio with background, trim to main length
    filter_parts.append(
        "[bg][main]amix=inputs=2:duration=shortest:weights=1 3,"
        "afade=t=in:d=0.1,volume=2.0[out]"
    )
    
    filter_complex = ";\n".join(filter_parts)
    
    cmd = [
        "ffmpeg", "-y",
        *inputs,
        "-filter_complex", filter_complex,
        "-map", "[out]",
        "-ar", "44100", "-ac", "2", "-b:a", "192k",
        str(output_path)
    ]
    
    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode != 0:
        print(f"  FFmpeg error: {result.stderr[-500:]}")
        return False
    return True


def parse_phrase(phrase):
    """Parse a phrase into segments: text parts and {sfx} markers."""
    parts = re.split(r'\{(\w+)\}', phrase)
    segments = []
    for i, part in enumerate(parts):
        part = part.strip()
        if not part:
            continue
        if i % 2 == 0:
            segments.append(("text", part))
        else:
            segments.append(("sfx", part))
    return segments


def generate_one_jingle(phrase, voice, sfx_files, output_path):
    """Generate a single jingle from a phrase template."""
    parsed = parse_phrase(phrase)
    
    audio_segments = []  # (type, filepath)
    tmpfiles = []
    
    try:
        for stype, content in parsed:
            if stype == "text":
                tmpfile = tempfile.mktemp(suffix=".mp3", dir="/tmp")
                tmpfiles.append(tmpfile)
                if tts_segment(content, voice, tmpfile):
                    audio_segments.append(("voice", tmpfile))
                else:
                    print(f"  Failed TTS for: {content}")
                    return False
            elif stype == "sfx":
                if content in sfx_files:
                    audio_segments.append(("sfx", sfx_files[content]))
                else:
                    print(f"  Unknown SFX: {content}")
        
        if not audio_segments:
            return False
            
        return stitch_jingle(audio_segments, sfx_files, output_path)
    finally:
        for f in tmpfiles:
            try:
                os.unlink(f)
            except:
                pass


def main():
    JINGLES_DIR.mkdir(parents=True, exist_ok=True)
    
    if "--list-sfx" in sys.argv:
        sfx_files = generate_sfx()
        print("Available SFX:", ", ".join(sorted(sfx_files.keys())))
        return
    
    # Clear old jingles before regenerating
    for old in JINGLES_DIR.glob("jingle_*.mp3"):
        old.unlink()
    
    print("Generating SFX...")
    sfx_files = generate_sfx()
    print(f"  {len(sfx_files)} SFX ready: {', '.join(sorted(sfx_files.keys()))}")
    
    # Generate multiple jingles with different voice/phrase combos
    print("\nGenerating jingles...")
    
    generated = 0
    for i, phrase in enumerate(PHRASES):
        voice = random.choice(VOICES)
        output = JINGLES_DIR / f"jingle_{i+1:02d}.mp3"
        
        print(f"\n  [{i+1}/{len(PHRASES)}] Voice: {voice}")
        print(f"  Phrase: {phrase}")
        
        if generate_one_jingle(phrase, voice, sfx_files, output):
            # Get duration
            probe = subprocess.run(
                ["ffprobe", "-hide_banner", "-show_entries", "format=duration",
                 "-of", "csv=p=0", str(output)],
                capture_output=True, text=True
            )
            dur = float(probe.stdout.strip()) if probe.stdout.strip() else 0
            print(f"  ✓ {output.name} ({dur:.1f}s)")
            generated += 1
        else:
            print(f"  ✗ Failed")
    
    print(f"\nDone! {generated}/{len(PHRASES)} jingles generated in {JINGLES_DIR}")
    print("Reload liquidsoap to pick them up: sudo systemctl restart liquidsoap")


if __name__ == "__main__":
    main()