SG Announcer: Piper-TTS-utroper med forhåndslytt og kø
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
commit
1c3e17afa0
8 changed files with 971 additions and 0 deletions
342
announcer.py
Normal file
342
announcer.py
Normal file
|
|
@ -0,0 +1,342 @@
|
|||
#!/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. Lastes ned fra rhasspy/piper-voices ved
|
||||
# første oppstart om de mangler i VOICE_DIR.
|
||||
VOICES: dict[str, str] = {
|
||||
"no_NO-talesyntese-medium": "Norsk (talesyntese)",
|
||||
"en_US-lessac-medium": "English US (Lessac)",
|
||||
"en_GB-alan-medium": "English UK (Alan)",
|
||||
}
|
||||
DEFAULT_VOICE = "no_NO-talesyntese-medium"
|
||||
|
||||
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 VOICES:
|
||||
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
|
||||
|
||||
voice = load_voice(voice_name)
|
||||
rate = voice.config.sample_rate
|
||||
cfg = SynthesisConfig(length_scale=1.0 / speed)
|
||||
|
||||
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.get(self.voice, 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": k, "name": v} for k, v in VOICES.items()],
|
||||
"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(DEFAULT_VOICE) # 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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue