Gabriel Vidal
2f07f3510f
--spoken only became mandatory recently, so most of the log has no line to replay. Rather than leaving those as a bare jingle, read the push itself — minus what only makes sense on a screen: notify.sh's harness/cost footer, bare URLs, and emoji (a French voice reads ✅ as its English name). Asks get the button too; their question is what gets read. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
120 lines
4.8 KiB
Python
120 lines
4.8 KiB
Python
"""Replayable notification sound — the jingle + spoken line, as a WAV.
|
|
|
|
A notification that went to the desk phone was *heard*: the phone service plays
|
|
a per-type procedural jazz earcon and then reads the French ``spoken`` line
|
|
aloud (services/phone/bridge/{jingles,tts}.py). The viewer's notification cards
|
|
can replay that same audio in the browser — this module is the bridge.
|
|
|
|
There is no second synthesizer here: we POST to the phone bridge's ``/render``
|
|
endpoint, which composes exactly what ``/notify`` would have dialled and hands
|
|
back a WAV instead of ringing the handset. One source of truth for what a
|
|
notification sounds like; the browser just plays the bytes.
|
|
|
|
Rendering costs a Pocket-TTS pass (CPU, ~6x realtime), so results are memoized
|
|
in-process, keyed by the audio's inputs — a notification's sound never changes,
|
|
and the same card is usually replayed more than once.
|
|
"""
|
|
|
|
import collections
|
|
import json
|
|
import os
|
|
import re
|
|
import threading
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
# The phone bridge, reachable from this container via the docker host gateway
|
|
# (it publishes 8091 on the host, same as the ai-phone webhook target).
|
|
BRIDGE_URL = os.environ.get(
|
|
"PHONE_BRIDGE_URL", "http://host.docker.internal:8091").rstrip("/")
|
|
|
|
_TIMEOUT_S = 60
|
|
_CACHE_MAX = 64
|
|
|
|
# --spoken only became mandatory recently, so most of the log has no line to
|
|
# replay. Those fall back to reading the push itself — which is written for a
|
|
# screen, so it needs tidying first: the harness/cost footer notify.sh appends
|
|
# ("Claude Code · $4.66 · 10.9M tokens · 14m 56s") and bare URLs are noise
|
|
# aloud, and emoji are read as their (English) names by a French voice.
|
|
_COST_LINE_RE = re.compile(r"^\s*\S.*·.*(\$|tokens)\b.*$", re.M)
|
|
_URL_RE = re.compile(r"https?://\S+")
|
|
_EMOJI_RE = re.compile(
|
|
"[\U0001F000-\U0001FAFF" # the emoji planes
|
|
"\u2190-\u21FF" # arrows
|
|
"\u2300-\u27BF" # misc technical + dingbats (✅ ✔ ⚠ …)
|
|
"\u2B00-\u2BFF" # misc symbols and arrows
|
|
"\uFE0F\u200D]") # variation selector + ZWJ (emoji sequences)
|
|
# Pocket-TTS is ~6x realtime on CPU; a runaway body would make the button hang.
|
|
_MAX_SPEECH_CHARS = 320
|
|
|
|
|
|
def fallback_speech(title: str = "", message: str = "") -> str:
|
|
"""What to read for a push that carried no ``spoken`` line: its own text.
|
|
|
|
Reads the title and message as one sentence, minus the parts that only make
|
|
sense on a screen.
|
|
"""
|
|
text = ". ".join(p for p in ((title or "").strip(), (message or "").strip()) if p)
|
|
text = _COST_LINE_RE.sub("", text)
|
|
text = _URL_RE.sub("", text)
|
|
text = _EMOJI_RE.sub("", text)
|
|
# Newlines and bullet separators become sentence breaks, not silence.
|
|
text = re.sub(r"\s*[\n·|]+\s*", ". ", text)
|
|
text = re.sub(r"(?:\.\s*){2,}", ". ", text)
|
|
text = re.sub(r"\s+", " ", text)
|
|
# Stripping an emoji leaves a gap before the punctuation that followed it.
|
|
text = re.sub(r" +([.,;:!?])", r"\1", text).strip(" .·-—")
|
|
if len(text) > _MAX_SPEECH_CHARS:
|
|
text = text[:_MAX_SPEECH_CHARS].rsplit(" ", 1)[0] + "…"
|
|
return text
|
|
|
|
|
|
_cache: "collections.OrderedDict[tuple, bytes]" = collections.OrderedDict()
|
|
_lock = threading.Lock()
|
|
|
|
|
|
class RenderError(RuntimeError):
|
|
"""The phone bridge could not render this notification's audio."""
|
|
|
|
|
|
def render(spoken: str = "", type_: str = "") -> bytes:
|
|
"""Return WAV bytes for a notification's sound (jingle + spoken line).
|
|
|
|
Either input may be empty: a type with no spoken line renders the bare
|
|
jingle, a spoken line with no type renders speech only. Both empty is a
|
|
caller error — the bridge rejects it.
|
|
"""
|
|
spoken = (spoken or "").strip()
|
|
type_ = (type_ or "").strip().lower()
|
|
key = (spoken, type_)
|
|
with _lock:
|
|
hit = _cache.get(key)
|
|
if hit is not None:
|
|
_cache.move_to_end(key)
|
|
return hit
|
|
|
|
body = json.dumps({"spoken": spoken, "type": type_}).encode()
|
|
req = urllib.request.Request(
|
|
f"{BRIDGE_URL}/render", data=body,
|
|
headers={"Content-Type": "application/json"}, method="POST")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=_TIMEOUT_S) as r:
|
|
wav = r.read()
|
|
except urllib.error.HTTPError as e:
|
|
detail = ""
|
|
try:
|
|
detail = json.loads(e.read()).get("error") or ""
|
|
except (ValueError, OSError):
|
|
pass
|
|
raise RenderError(detail or f"phone bridge returned {e.code}") from e
|
|
except (urllib.error.URLError, OSError) as e:
|
|
raise RenderError(f"phone bridge unreachable: {e}") from e
|
|
if not wav.startswith(b"RIFF"):
|
|
raise RenderError("phone bridge returned something that isn't a WAV")
|
|
|
|
with _lock:
|
|
_cache[key] = wav
|
|
while len(_cache) > _CACHE_MAX:
|
|
_cache.popitem(last=False)
|
|
return wav
|