406 lines
14 KiB
Python
406 lines
14 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]
|
||
|
||
|
||
def make_chime(rate: int, volume: float = 0.35) -> bytes:
|
||
"""To-tone «ding-dong» som 16-bit mono PCM i gitt samplerate."""
|
||
out = bytearray()
|
||
for freq, dur in ((784.0, 0.28), (523.25, 0.42)):
|
||
n = int(rate * dur)
|
||
for i in range(n):
|
||
t = i / rate
|
||
env = math.exp(-5.5 * t) # rask utklang
|
||
s = (
|
||
math.sin(2 * math.pi * freq * t)
|
||
+ 0.3 * math.sin(2 * math.pi * freq * 2 * t)
|
||
) * env * volume
|
||
out += struct.pack("<h", int(max(-1.0, min(1.0, s)) * 32767))
|
||
out += b"\x00\x00" * int(rate * 0.35) # pause før talen
|
||
return bytes(out)
|
||
|
||
|
||
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: bool) -> str:
|
||
h = hashlib.sha1(f"{voice}|{speed:.2f}|{int(chime)}|{text}".encode()).hexdigest()
|
||
return h[:20]
|
||
|
||
|
||
def synthesize(text: str, voice_name: str, speed: float, chime: bool) -> 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
|
||
if chime:
|
||
pcm += make_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"])
|
||
if chime:
|
||
pcm += make_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: bool
|
||
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: bool, 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, bool]:
|
||
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 = bool(data.get("chime", True))
|
||
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,
|
||
"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()
|