528 lines
18 KiB
Python
528 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
"""SG Announcer – tekst inn, forhåndslytt i nettleseren, les ut på PC-ens lydutgang.
|
||
|
||
Kjøres på PC-en som er koblet til høyttaleranlegget. Alle på nettet kan åpne
|
||
sida og sende meldinger; selve avspillingen skjer alltid lokalt på denne
|
||
maskinen via pw-play (PipeWire).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import io
|
||
import logging
|
||
import math
|
||
import os
|
||
import queue
|
||
import struct
|
||
import subprocess
|
||
import threading
|
||
import time
|
||
import wave
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
|
||
from flask import Flask, Response, abort, jsonify, request, send_file
|
||
|
||
HERE = Path(__file__).resolve().parent
|
||
VOICE_DIR = Path(os.environ.get("ANNOUNCER_VOICES", HERE / "voices"))
|
||
CACHE_DIR = Path(os.environ.get("ANNOUNCER_CACHE", HERE / "cache"))
|
||
PIN = os.environ.get("ANNOUNCER_PIN", "").strip()
|
||
SINK = os.environ.get("ANNOUNCER_SINK", "").strip() # pw-play --target
|
||
MAX_CHARS = 600
|
||
|
||
# Stemmene som tilbys i menyen. To motorer:
|
||
# edge – Microsofts nevrale stemmer via edge-tts (gratis, krever internett,
|
||
# klart mest naturlig; ~2–3 s per melding)
|
||
# piper – lokalt, ingen nett, robotaktig; reserve hvis nettet faller ut.
|
||
# Piper-modellene lastes ned fra rhasspy/piper-voices ved første oppstart.
|
||
# «speaker» er bare for Piper-flerstemmemodeller (nvcc har ti norske talere).
|
||
def _edge(id: str, name: str, group: str, voice: str) -> dict:
|
||
return {"id": id, "name": name, "group": group, "engine": "edge", "model": voice, "speaker": None}
|
||
|
||
|
||
def _piper(id: str, name: str, group: str, model: str, speaker: int | None = None) -> dict:
|
||
return {"id": id, "name": name, "group": group, "engine": "piper", "model": model, "speaker": speaker}
|
||
|
||
|
||
VOICES_LIST: list[dict] = [
|
||
_edge("finn", "Finn (mann)", "Norsk", "nb-NO-FinnNeural"),
|
||
_edge("pernille", "Pernille (kvinne)", "Norsk", "nb-NO-PernilleNeural"),
|
||
_edge("andrew", "Andrew (US, mann)", "English", "en-US-AndrewMultilingualNeural"),
|
||
_edge("ava", "Ava (US, kvinne)", "English", "en-US-AvaMultilingualNeural"),
|
||
_edge("brian", "Brian (US, mann)", "English", "en-US-BrianMultilingualNeural"),
|
||
_edge("emma", "Emma (US, kvinne)", "English", "en-US-EmmaMultilingualNeural"),
|
||
_edge("ryan-uk", "Ryan (UK, mann)", "English", "en-GB-RyanNeural"),
|
||
_edge("sonia", "Sonia (UK, kvinne)", "English", "en-GB-SoniaNeural"),
|
||
_edge("sv", "Svensk – Mattias", "Nabospråk", "sv-SE-MattiasNeural"),
|
||
_edge("da", "Dansk – Christel", "Nabospråk", "da-DK-ChristelNeural"),
|
||
_piper("no-tale", "Talesyntese (nøytral)", "Lokal reserve (uten nett)", "no_NO-talesyntese-medium"),
|
||
_piper("no-knn", "Kvinne – KNN", "Lokal reserve (uten nett)", "no_NO-nvcc-medium", 0),
|
||
_piper("no-msv", "Mann – MSV", "Lokal reserve (uten nett)", "no_NO-nvcc-medium", 5),
|
||
_piper("en-ryan", "Ryan (US)", "Lokal reserve (uten nett)", "en_US-ryan-high"),
|
||
_piper("en-jenny", "Jenny (UK)", "Lokal reserve (uten nett)", "en_GB-jenny_dioco-medium"),
|
||
]
|
||
VOICES: dict[str, dict] = {v["id"]: v for v in VOICES_LIST}
|
||
DEFAULT_VOICE = "finn"
|
||
EDGE_RATE = 24000
|
||
|
||
log = logging.getLogger("announcer")
|
||
app = Flask(__name__, static_folder=None)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# TTS
|
||
# --------------------------------------------------------------------------
|
||
|
||
_voice_cache: dict[str, object] = {}
|
||
_voice_lock = threading.Lock()
|
||
|
||
|
||
def ensure_voices() -> None:
|
||
import piper.download_voices as dl
|
||
|
||
VOICE_DIR.mkdir(parents=True, exist_ok=True)
|
||
for name in sorted({v["model"] for v in VOICES_LIST if v["engine"] == "piper"}):
|
||
if not (VOICE_DIR / f"{name}.onnx").exists():
|
||
log.info("laster ned stemme %s", name)
|
||
dl.download_voice(name, VOICE_DIR)
|
||
|
||
|
||
def load_voice(name: str):
|
||
from piper import PiperVoice
|
||
|
||
with _voice_lock:
|
||
if name not in _voice_cache:
|
||
_voice_cache[name] = PiperVoice.load(VOICE_DIR / f"{name}.onnx")
|
||
return _voice_cache[name]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Jingler (chimes). Alle de innebygde genereres i Python; egne lydfiler
|
||
# (wav/mp3/ogg/flac) i CHIME_DIR dukker opp automatisk i menyen.
|
||
# --------------------------------------------------------------------------
|
||
|
||
CHIME_DIR = Path(os.environ.get("ANNOUNCER_CHIMES", HERE / "chimes"))
|
||
|
||
|
||
def _tone(rate: int, freq: float, dur: float, vol: float = 0.35, wave_: str = "sine",
|
||
decay: float = 5.5, attack: float = 0.005, harm2: float = 0.0, slide: float = 0.0) -> list[float]:
|
||
"""Én tone som liste av floats. slide = frekvensendring (multiplikator) over tonen."""
|
||
n = int(rate * dur)
|
||
out = []
|
||
phase = 0.0
|
||
for i in range(n):
|
||
t = i / rate
|
||
f = freq * (slide ** (t / dur)) if slide else freq
|
||
phase += 2 * math.pi * f / rate
|
||
if wave_ == "square":
|
||
v = 1.0 if math.sin(phase) >= 0 else -1.0
|
||
elif wave_ == "saw":
|
||
v = 2.0 * ((phase / (2 * math.pi)) % 1.0) - 1.0
|
||
else:
|
||
v = math.sin(phase) + harm2 * math.sin(2 * phase)
|
||
env = min(1.0, t / attack) * math.exp(-decay * t)
|
||
out.append(v * env * vol)
|
||
return out
|
||
|
||
|
||
def _mix(*parts: list[float], gap: float = 0.0, rate: int = 22050) -> list[float]:
|
||
out: list[float] = []
|
||
for p in parts:
|
||
out += p
|
||
out += [0.0] * int(rate * gap)
|
||
return out
|
||
|
||
|
||
def _overlay(base: list[float], layer: list[float], at: float, rate: int) -> list[float]:
|
||
start = int(at * rate)
|
||
if len(base) < start + len(layer):
|
||
base = base + [0.0] * (start + len(layer) - len(base))
|
||
for i, v in enumerate(layer):
|
||
base[start + i] += v
|
||
return base
|
||
|
||
|
||
def _dingdong(rate: int) -> list[float]:
|
||
return _mix(_tone(rate, 784.0, 0.28, harm2=0.3), _tone(rate, 523.25, 0.42, harm2=0.3), rate=rate)
|
||
|
||
|
||
def _airport(rate: int) -> list[float]:
|
||
"""Tre stigende toner med lang klang, som på en flyplass."""
|
||
out: list[float] = []
|
||
for i, f in enumerate((523.25, 659.25, 783.99)):
|
||
out = _overlay(out, _tone(rate, f, 1.2, vol=0.3, decay=2.2, harm2=0.2), i * 0.38, rate)
|
||
return out
|
||
|
||
|
||
def _chiptune(rate: int) -> list[float]:
|
||
"""8-bit arpeggio, som et power-up i et gammelt spill."""
|
||
notes = (523.25, 659.25, 783.99, 1046.5, 1318.5, 1567.98)
|
||
parts = [_tone(rate, f, 0.09, vol=0.3, wave_="square", decay=3.0) for f in notes]
|
||
parts.append(_tone(rate, 2093.0, 0.35, vol=0.3, wave_="square", decay=6.0))
|
||
return _mix(*parts, rate=rate)
|
||
|
||
|
||
def _scifi(rate: int) -> list[float]:
|
||
"""Sveip opp + to blipp, som en kommunikator."""
|
||
sweep = _tone(rate, 300.0, 0.35, vol=0.3, decay=1.5, slide=5.0)
|
||
blips = _mix(_tone(rate, 1800.0, 0.07, vol=0.25, decay=10), _tone(rate, 2400.0, 0.12, vol=0.25, decay=10), gap=0.03, rate=rate)
|
||
return _mix(sweep, blips, rate=rate)
|
||
|
||
|
||
def _horn(rate: int) -> list[float]:
|
||
"""Air horn: to lag av sagtann med litt vibrato, derav grovheten."""
|
||
out: list[float] = []
|
||
for f, v in ((233.08, 0.22), (349.23, 0.16), (466.16, 0.1)):
|
||
out = _overlay(out, _tone(rate, f, 0.9, vol=v, wave_="saw", decay=1.3, attack=0.03), 0.0, rate)
|
||
return out
|
||
|
||
|
||
def _gong(rate: int) -> list[float]:
|
||
"""Dyp gong med uharmoniske overtoner."""
|
||
out: list[float] = []
|
||
for f, v, d in ((110.0, 0.3, 1.0), (164.8, 0.18, 1.4), (263.0, 0.12, 2.0), (411.0, 0.07, 3.0), (650.0, 0.04, 4.0)):
|
||
out = _overlay(out, _tone(rate, f, 2.5, vol=v, decay=d, attack=0.004), 0.0, rate)
|
||
return out
|
||
|
||
|
||
BUILTIN_CHIMES: dict[str, tuple[str, object]] = {
|
||
"dingdong": ("Ding-dong", _dingdong),
|
||
"airport": ("Flyplass (tre toner)", _airport),
|
||
"chiptune": ("8-bit power-up", _chiptune),
|
||
"scifi": ("Sci-fi kommunikator", _scifi),
|
||
"horn": ("Air horn", _horn),
|
||
"gong": ("Gong", _gong),
|
||
}
|
||
CHIME_EXT = {".wav", ".mp3", ".ogg", ".flac", ".m4a"}
|
||
|
||
|
||
def file_chimes() -> dict[str, Path]:
|
||
if not CHIME_DIR.is_dir():
|
||
return {}
|
||
return {f"file:{p.stem}": p for p in sorted(CHIME_DIR.iterdir()) if p.suffix.lower() in CHIME_EXT}
|
||
|
||
|
||
def list_chimes() -> list[dict]:
|
||
out = [{"id": "none", "name": "Ingen"}]
|
||
out += [{"id": k, "name": v[0]} for k, v in BUILTIN_CHIMES.items()]
|
||
out += [{"id": k, "name": f"{p.stem} (egen fil)"} for k, p in file_chimes().items()]
|
||
return out
|
||
|
||
|
||
def chime_pcm(chime: str, rate: int) -> bytes:
|
||
"""Jingle som 16-bit mono PCM i gitt samplerate, etterfulgt av en kort pause."""
|
||
if chime == "none":
|
||
return b""
|
||
if chime in BUILTIN_CHIMES:
|
||
samples = BUILTIN_CHIMES[chime][1](rate)
|
||
pcm = b"".join(struct.pack("<h", int(max(-1.0, min(1.0, v)) * 32767)) for v in samples)
|
||
else:
|
||
path = file_chimes().get(chime)
|
||
if path is None:
|
||
raise ValueError("Ukjent jingle")
|
||
pcm = subprocess.run(
|
||
["ffmpeg", "-v", "error", "-i", str(path), "-t", "6", "-f", "s16le", "-ac", "1", "-ar", str(rate),
|
||
"-af", "loudnorm=I=-18,afade=t=out:st=5:d=1", "-"],
|
||
capture_output=True, check=True,
|
||
).stdout
|
||
return pcm + b"\x00\x00" * int(rate * 0.35)
|
||
|
||
|
||
def edge_pcm(text: str, voice: str, speed: float) -> bytes:
|
||
"""Henter MP3 fra edge-tts og dekoder til 16-bit mono PCM i EDGE_RATE."""
|
||
import asyncio
|
||
import tempfile
|
||
|
||
import edge_tts
|
||
|
||
rate_pct = int(round((speed - 1.0) * 100))
|
||
rate = f"{rate_pct:+d}%"
|
||
with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False, dir=CACHE_DIR) as tmp:
|
||
mp3 = Path(tmp.name)
|
||
try:
|
||
asyncio.run(edge_tts.Communicate(text, voice, rate=rate).save(str(mp3)))
|
||
out = subprocess.run(
|
||
["ffmpeg", "-v", "error", "-i", str(mp3), "-f", "s16le", "-ac", "1", "-ar", str(EDGE_RATE), "-"],
|
||
capture_output=True,
|
||
check=True,
|
||
)
|
||
return out.stdout
|
||
finally:
|
||
mp3.unlink(missing_ok=True)
|
||
|
||
|
||
def synth_key(text: str, voice: str, speed: float, chime: str) -> str:
|
||
h = hashlib.sha1(f"{voice}|{speed:.2f}|{chime}|{text}".encode()).hexdigest()
|
||
return h[:20]
|
||
|
||
|
||
def synthesize(text: str, voice_name: str, speed: float, chime: str) -> Path:
|
||
"""Returnerer sti til ferdig WAV (bufres på disk etter innhold)."""
|
||
from piper.config import SynthesisConfig
|
||
|
||
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
||
path = CACHE_DIR / f"{synth_key(text, voice_name, speed, chime)}.wav"
|
||
if path.exists():
|
||
return path
|
||
|
||
spec = VOICES[voice_name]
|
||
pcm = bytearray()
|
||
if spec["engine"] == "edge":
|
||
rate = EDGE_RATE
|
||
pcm += chime_pcm(chime, rate)
|
||
pcm += edge_pcm(text, spec["model"], speed)
|
||
else:
|
||
voice = load_voice(spec["model"])
|
||
rate = voice.config.sample_rate
|
||
cfg = SynthesisConfig(length_scale=1.0 / speed, speaker_id=spec["speaker"])
|
||
pcm += chime_pcm(chime, rate)
|
||
for chunk in voice.synthesize(text, cfg):
|
||
pcm += chunk.audio_int16_bytes
|
||
pcm += b"\x00\x00" * int(rate * 0.3)
|
||
|
||
tmp = path.with_suffix(".tmp")
|
||
with wave.open(str(tmp), "wb") as w:
|
||
w.setnchannels(1)
|
||
w.setsampwidth(2)
|
||
w.setframerate(rate)
|
||
w.writeframes(bytes(pcm))
|
||
tmp.rename(path)
|
||
return path
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Avspillingskø
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass
|
||
class Item:
|
||
id: int
|
||
text: str
|
||
voice: str
|
||
speed: float
|
||
chime: str
|
||
sender: str
|
||
created: float = field(default_factory=time.time)
|
||
status: str = "queued" # queued | playing | done | stopped | failed
|
||
error: str = ""
|
||
|
||
def public(self) -> dict:
|
||
return {
|
||
"id": self.id,
|
||
"text": self.text,
|
||
"voice": VOICES[self.voice]["name"] if self.voice in VOICES else self.voice,
|
||
"sender": self.sender,
|
||
"created": self.created,
|
||
"status": self.status,
|
||
"error": self.error,
|
||
}
|
||
|
||
|
||
class Player:
|
||
def __init__(self) -> None:
|
||
self.q: queue.Queue[Item] = queue.Queue()
|
||
self.history: list[Item] = []
|
||
self.current: Item | None = None
|
||
self.proc: subprocess.Popen | None = None
|
||
self.lock = threading.Lock()
|
||
self.seq = 0
|
||
threading.Thread(target=self._loop, daemon=True, name="player").start()
|
||
|
||
def enqueue(self, text: str, voice: str, speed: float, chime: str, sender: str) -> Item:
|
||
with self.lock:
|
||
self.seq += 1
|
||
item = Item(self.seq, text, voice, speed, chime, sender)
|
||
self.history.append(item)
|
||
del self.history[:-50]
|
||
self.q.put(item)
|
||
return item
|
||
|
||
def stop(self) -> None:
|
||
with self.lock:
|
||
if self.proc and self.proc.poll() is None:
|
||
self.proc.terminate()
|
||
if self.current:
|
||
self.current.status = "stopped"
|
||
|
||
def clear(self) -> None:
|
||
while True:
|
||
try:
|
||
item = self.q.get_nowait()
|
||
except queue.Empty:
|
||
break
|
||
item.status = "stopped"
|
||
self.stop()
|
||
|
||
def snapshot(self) -> dict:
|
||
with self.lock:
|
||
return {
|
||
"current": self.current.public() if self.current else None,
|
||
"queued": [i.public() for i in self.history if i.status == "queued"],
|
||
"history": [i.public() for i in reversed(self.history) if i.status not in ("queued", "playing")][:15],
|
||
}
|
||
|
||
def _loop(self) -> None:
|
||
while True:
|
||
item = self.q.get()
|
||
if item.status != "queued":
|
||
continue
|
||
item.status = "playing"
|
||
with self.lock:
|
||
self.current = item
|
||
try:
|
||
path = synthesize(item.text, item.voice, item.speed, item.chime)
|
||
cmd = ["pw-play"]
|
||
if SINK:
|
||
cmd += ["--target", SINK]
|
||
cmd.append(str(path))
|
||
with self.lock:
|
||
self.proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
|
||
_, err = self.proc.communicate()
|
||
if item.status == "playing":
|
||
if self.proc.returncode == 0:
|
||
item.status = "done"
|
||
else:
|
||
item.status = "failed"
|
||
item.error = err.decode(errors="replace").strip()[-200:]
|
||
except Exception as e: # noqa: BLE001
|
||
log.exception("avspilling feilet")
|
||
item.status = "failed"
|
||
item.error = str(e)
|
||
finally:
|
||
with self.lock:
|
||
self.current = None
|
||
self.proc = None
|
||
|
||
|
||
player = Player()
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# HTTP
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def check_pin() -> None:
|
||
if PIN and request.headers.get("X-Pin", "") != PIN:
|
||
abort(401)
|
||
|
||
|
||
def parse_request() -> tuple[str, str, float, str]:
|
||
data = request.get_json(silent=True) or {}
|
||
text = " ".join(str(data.get("text", "")).split())
|
||
if not text:
|
||
abort(400, "Tom melding")
|
||
if len(text) > MAX_CHARS:
|
||
abort(400, f"Maks {MAX_CHARS} tegn")
|
||
voice = str(data.get("voice", DEFAULT_VOICE))
|
||
if voice not in VOICES:
|
||
abort(400, "Ukjent stemme")
|
||
try:
|
||
speed = float(data.get("speed", 1.0))
|
||
except (TypeError, ValueError):
|
||
abort(400, "Ugyldig hastighet")
|
||
speed = max(0.6, min(1.6, speed))
|
||
chime = data.get("chime", "dingdong")
|
||
if chime is True:
|
||
chime = "dingdong"
|
||
elif chime is False or chime is None:
|
||
chime = "none"
|
||
chime = str(chime)
|
||
if chime != "none" and chime not in BUILTIN_CHIMES and chime not in file_chimes():
|
||
abort(400, "Ukjent jingle")
|
||
return text, voice, speed, chime
|
||
|
||
|
||
@app.get("/")
|
||
def index() -> Response:
|
||
return send_file(HERE / "index.html")
|
||
|
||
|
||
@app.get("/api/config")
|
||
def config() -> Response:
|
||
return jsonify(
|
||
{
|
||
"voices": [{"id": v["id"], "name": v["name"], "group": v["group"], "engine": v["engine"]} for v in VOICES_LIST],
|
||
"default_voice": DEFAULT_VOICE,
|
||
"chimes": list_chimes(),
|
||
"default_chime": "dingdong",
|
||
"max_chars": MAX_CHARS,
|
||
"pin_required": bool(PIN),
|
||
}
|
||
)
|
||
|
||
|
||
@app.post("/api/preview")
|
||
def preview() -> Response:
|
||
check_pin()
|
||
text, voice, speed, chime = parse_request()
|
||
try:
|
||
path = synthesize(text, voice, speed, chime)
|
||
except Exception as e: # noqa: BLE001
|
||
log.exception("syntese feilet")
|
||
abort(502, f"Syntese feilet ({VOICES[voice]['name']}): {e}. Prøv en lokal reserve-stemme.")
|
||
return send_file(path, mimetype="audio/wav", max_age=3600)
|
||
|
||
|
||
@app.post("/api/announce")
|
||
def announce() -> Response:
|
||
check_pin()
|
||
text, voice, speed, chime = parse_request()
|
||
sender = request.remote_addr or "?"
|
||
item = player.enqueue(text, voice, speed, chime, sender)
|
||
log.info("kø #%d fra %s: %s", item.id, sender, text)
|
||
return jsonify(item.public())
|
||
|
||
|
||
@app.post("/api/stop")
|
||
def stop() -> Response:
|
||
check_pin()
|
||
player.stop()
|
||
return jsonify({"ok": True})
|
||
|
||
|
||
@app.post("/api/clear")
|
||
def clear() -> Response:
|
||
check_pin()
|
||
player.clear()
|
||
return jsonify({"ok": True})
|
||
|
||
|
||
@app.get("/api/status")
|
||
def status() -> Response:
|
||
return jsonify(player.snapshot())
|
||
|
||
|
||
@app.errorhandler(400)
|
||
def bad_request(e): # noqa: ANN001
|
||
return jsonify({"error": e.description}), 400
|
||
|
||
|
||
@app.errorhandler(502)
|
||
def bad_gateway(e): # noqa: ANN001
|
||
return jsonify({"error": e.description}), 502
|
||
|
||
|
||
@app.errorhandler(401)
|
||
def unauthorized(e): # noqa: ANN001
|
||
return jsonify({"error": "Feil PIN"}), 401
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||
ap.add_argument("--host", default="0.0.0.0")
|
||
ap.add_argument("--port", type=int, default=int(os.environ.get("ANNOUNCER_PORT", 8090)))
|
||
args = ap.parse_args()
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
||
logging.getLogger("werkzeug").setLevel(logging.WARNING)
|
||
ensure_voices()
|
||
load_voice("no_NO-talesyntese-medium") # varm opp reserven
|
||
log.info("klar på http://%s:%d (PIN: %s, sink: %s)", args.host, args.port, "på" if PIN else "av", SINK or "standard")
|
||
app.run(host=args.host, port=args.port, threaded=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|