365 lines
12 KiB
Python
365 lines
12 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. Modellene lastes ned fra rhasspy/piper-voices
|
||
# ved første oppstart om de mangler i VOICE_DIR. «speaker» er bare for
|
||
# flerstemmemodeller (nvcc har ti norske talere i én modell).
|
||
def _v(id: str, name: str, group: str, model: str, speaker: int | None = None) -> dict:
|
||
return {"id": id, "name": name, "group": group, "model": model, "speaker": speaker}
|
||
|
||
|
||
VOICES_LIST: list[dict] = [
|
||
_v("no-tale", "Talesyntese (nøytral)", "Norsk", "no_NO-talesyntese-medium"),
|
||
_v("no-knn", "Kvinne – KNN", "Norsk", "no_NO-nvcc-medium", 0),
|
||
_v("no-ksv", "Kvinne – KSV", "Norsk", "no_NO-nvcc-medium", 1),
|
||
_v("no-kon", "Kvinne – KON", "Norsk", "no_NO-nvcc-medium", 3),
|
||
_v("no-kmn", "Kvinne – KMN", "Norsk", "no_NO-nvcc-medium", 8),
|
||
_v("no-knv", "Kvinne – KNV", "Norsk", "no_NO-nvcc-medium", 9),
|
||
_v("no-mmn", "Mann – MMN", "Norsk", "no_NO-nvcc-medium", 2),
|
||
_v("no-mnn", "Mann – MNN", "Norsk", "no_NO-nvcc-medium", 4),
|
||
_v("no-msv", "Mann – MSV", "Norsk", "no_NO-nvcc-medium", 5),
|
||
_v("no-mon", "Mann – MON", "Norsk", "no_NO-nvcc-medium", 6),
|
||
_v("no-mnv", "Mann – MNV", "Norsk", "no_NO-nvcc-medium", 7),
|
||
_v("en-ryan", "Ryan (US, mann)", "English", "en_US-ryan-high"),
|
||
_v("en-amy", "Amy (US, kvinne)", "English", "en_US-amy-medium"),
|
||
_v("en-lessac", "Lessac (US, nøytral)", "English", "en_US-lessac-medium"),
|
||
_v("en-alan", "Alan (UK, mann)", "English", "en_GB-alan-medium"),
|
||
_v("en-north", "Northern English (UK, mann)", "English", "en_GB-northern_english_male-medium"),
|
||
_v("en-jenny", "Jenny (UK, kvinne)", "English", "en_GB-jenny_dioco-medium"),
|
||
_v("sv-nst", "Svensk (NST)", "Nabospråk", "sv_SE-nst-medium"),
|
||
_v("da-tale", "Dansk (talesyntese)", "Nabospråk", "da_DK-talesyntese-medium"),
|
||
]
|
||
VOICES: dict[str, dict] = {v["id"]: v for v in VOICES_LIST}
|
||
DEFAULT_VOICE = "no-tale"
|
||
|
||
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 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 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]
|
||
voice = load_voice(spec["model"])
|
||
rate = voice.config.sample_rate
|
||
cfg = SynthesisConfig(length_scale=1.0 / speed, speaker_id=spec["speaker"])
|
||
|
||
pcm = bytearray()
|
||
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"]} 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()
|
||
path = synthesize(text, voice, speed, chime)
|
||
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(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(VOICES[DEFAULT_VOICE]["model"]) # varm opp
|
||
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()
|