From 01682c62c2c9ea2f7f498544ee3aaa299c0c2423 Mon Sep 17 00:00:00 2001 From: Caine Date: Sat, 7 Mar 2026 12:52:55 +0000 Subject: Initial commit: Radio Susan scripts, configs, and SFX --- announce_tracks.py | 202 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 202 insertions(+) create mode 100755 announce_tracks.py (limited to 'announce_tracks.py') diff --git a/announce_tracks.py b/announce_tracks.py new file mode 100755 index 0000000..c129040 --- /dev/null +++ b/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() -- cgit v1.2.3