Initial commit: Radio Susan scripts, configs, and SFX

announce_tracks.py

@@ -0,0 +1,202 @@
+#!/usr/bin/env python3
+"""
+Generate a track announcement TTS clip for Radio Susan.
+Called by liquidsoap via request.dynamic between tracks.
+Reads track_state.json (written by liquidsoap on_track) and
+track_history.json (maintained here) to announce what just played.
+
+Outputs an MP3 file path to stdout, or nothing if no announcement needed.
+"""
+
+import hashlib
+import json
+import os
+import random
+import subprocess
+import tempfile
+from pathlib import Path
+
+VENV_BIN = Path("/var/lib/radio/venv/bin")
+EDGE_TTS = VENV_BIN / "edge-tts"
+ANNOUNCE_DIR = Path("/var/lib/radio/announcements")
+ANNOUNCE_DIR.mkdir(parents=True, exist_ok=True)
+STATE_FILE = Path("/var/lib/radio/track_state.json")
+HISTORY_FILE = Path("/var/lib/radio/track_history.json")
+
+VOICES = [
+    "en-US-GuyNeural",
+    "en-US-ChristopherNeural",
+    "en-GB-RyanNeural",
+    "en-US-AriaNeural",
+    "en-US-JennyNeural",
+]
+
+# Station ID variations
+STATION_IDS = [
+    "Radio Susan, ninety nine point oh.",
+    "You're listening to Radio Susan.",
+    "Radio Susan.",
+    "Susan Radio, always on.",
+    "This is Radio Susan.",
+]
+
+
+def pick_voice(text):
+    h = int(hashlib.md5(text.encode()).hexdigest(), 16)
+    return VOICES[h % len(VOICES)]
+
+
+def generate_tts(text, outfile, voice=None):
+    if voice is None:
+        voice = pick_voice(text)
+    cmd = [
+        str(EDGE_TTS),
+        "--voice", voice,
+        "--rate", "+10%",
+        "--text", text,
+        "--write-media", outfile,
+    ]
+    try:
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
+        return result.returncode == 0
+    except subprocess.TimeoutExpired:
+        return False
+
+
+def normalize_audio(infile, outfile):
+    cmd = [
+        "ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
+        "-i", infile,
+        "-af", "loudnorm=I=-14:TP=-1:LRA=11,aformat=sample_rates=44100:channel_layouts=stereo",
+        "-b:a", "192k",
+        outfile,
+    ]
+    try:
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
+        return result.returncode == 0
+    except subprocess.TimeoutExpired:
+        return False
+
+
+def clean_title(s):
+    if not s or s == "Unknown":
+        return None
+    for ext in ('.flac', '.mp3', '.ogg', '.m4a', '.opus', '.wav'):
+        s = s.replace(ext, '')
+    s = s.strip()
+    # Strip leading track numbers like "01 -", "03.", "001 "
+    if len(s) > 3 and s[:2].isdigit() and s[2] in '. -':
+        s = s[3:].strip()
+    if len(s) > 4 and s[:3].isdigit() and s[3] in '. -':
+        s = s[4:].strip()
+    # Strip "Artist - " prefix if title already contains it
+    return s.strip() if s.strip() else None
+
+
+def fmt_track(entry):
+    """Format a history entry as 'Artist - Title'."""
+    artist = entry.get("artist", "")
+    title = entry.get("title", "")
+    if artist and title:
+        return f"{artist} - {title}"
+    return clean_title(title or artist) or None
+
+
+def load_history():
+    if HISTORY_FILE.exists():
+        try:
+            return json.loads(HISTORY_FILE.read_text())
+        except Exception:
+            pass
+    return []
+
+
+def save_history(history):
+    HISTORY_FILE.write_text(json.dumps(history[-20:]))
+
+
+def main():
+    import argparse
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--state", action="store_true")
+    args = parser.parse_args()
+
+    if not args.state:
+        return
+
+    # Read current track from liquidsoap's state file
+    current = None
+    if STATE_FILE.exists():
+        try:
+            current = json.loads(STATE_FILE.read_text())
+        except Exception:
+            return
+
+    if not current:
+        return
+
+    history = load_history()
+
+    # Check if this is the same track we already announced (avoid repeats)
+    current_key = f"{current.get('artist', '')}|{current.get('title', '')}"
+    if history and f"{history[-1].get('artist', '')}|{history[-1].get('title', '')}" == current_key:
+        return  # Already announced this one
+
+    # Add current to history
+    history.append(current)
+    save_history(history)
+
+    # Need at least 2 tracks in history to announce (we announce what JUST played)
+    if len(history) < 2:
+        return
+
+    # Announce the two most recent tracks before this one.
+    # With 2:1 rotation these roughly match what the listener just heard.
+    last = fmt_track(history[-2]) if len(history) >= 2 else None
+    prev = fmt_track(history[-3]) if len(history) >= 3 else None
+
+    if not last:
+        return
+
+    # Build announcement — keep it vague ("recently" not "just played")
+    station_id = random.choice(STATION_IDS)
+
+    if prev:
+        text = f"You've been listening to {last}, and {prev}. {station_id}"
+    else:
+        text = f"You've been listening to {last}. {station_id}"
+
+    # Cache based on content hash
+    content_hash = hashlib.md5(text.encode()).hexdigest()[:12]
+    cached = ANNOUNCE_DIR / f"announce_{content_hash}.mp3"
+
+    if cached.exists():
+        print(str(cached))
+        return
+
+    # Generate TTS
+    with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as tmp:
+        tmp_path = tmp.name
+
+    try:
+        if not generate_tts(text, tmp_path):
+            return
+        if normalize_audio(tmp_path, str(cached)):
+            print(str(cached))
+        else:
+            os.rename(tmp_path, str(cached))
+            print(str(cached))
+    finally:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+
+    # Cleanup old announcements (keep last 200)
+    announcements = sorted(ANNOUNCE_DIR.glob("announce_*.mp3"), key=lambda f: f.stat().st_mtime)
+    for old in announcements[:-200]:
+        old.unlink()
+
+
+if __name__ == "__main__":
+    main()