1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
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()
|
