Compute audio levels for the overlay

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 38 insertions, 0 deletions

diff --git a/assistant.py b/assistant.py
index 4dc11db..7f8fd1d 100644
--- a/assistant.py
+++ b/assistant.py
@@ -20,9 +20,11 @@ ask the loaded llama-server model, speak the reply (Kokoro).
 Run: python3 assistant.py        (--test runs the self-check)
 """
 
+import array
 import ctypes
 import glob
 import json
+import math
 import os
 import queue
 import random
@@ -110,6 +112,7 @@ KLANG = {l: v["lang"] for l, v in CFG["voice"].items()}
 TOOLS_DIR = os.path.join(CONFIG_DIR, "tools")
 TOOLS = load_tools(TOOLS_DIR)
 TOOL_SCHEMA = schema(TOOLS) or None
+FRAME = 1 / 30                            # one audio level per frame, seconds
 
 
 def log(*a):
@@ -142,6 +145,30 @@ def latin(text):
     return all(unicodedata.name(c, "").startswith("LATIN") for c in text if c.isalpha())
 
 
+def scale(rms):
+    # Loudness as the overlay draws it: -60 dBFS and below is 0, full scale
+    # is 1. Logarithmic, so quiet speech still moves the ring.
+    if rms <= 0:
+        return 0.0
+    return max(0.0, min(1.0, (20 * math.log10(rms / 32768) + 60) / 60))
+
+
+def pcm_level(data):
+    # 16-bit little-endian mono, as arecord writes it; a trailing odd byte is
+    # half a sample and ignored.
+    samples = array.array("h", data[: len(data) // 2 * 2])
+    if not samples:
+        return 0.0
+    return scale(math.sqrt(sum(s * s for s in samples) / len(samples)))
+
+
+def envelope(audio, sr):
+    # One level per FRAME of float audio in -1..1 (Kokoro's output).
+    n = int(sr * FRAME)
+    frames = audio[: len(audio) // n * n].reshape(-1, n)
+    return [scale(float(r) * 32768) for r in (frames.astype("float64") ** 2).mean(axis=1) ** 0.5]
+
+
 def transcribe(lang="auto"):
     # Always send language: whisper-server's own default is en, not auto.
     out = subprocess.run(["curl", "-s", "-F", f"file=@{WAV}", "-F", f"language={lang}",
@@ -536,6 +563,17 @@ def test():
     assert announcement(tools, {0: {"name": "now"}}, "it") is None
     assert announcement(tools, {0: {"name": "now"}, 1: {"name": "search"}}, "en") == "Searching."
     assert announcement(tools, {0: {"name": "unknown"}}, "it") is None
+    import numpy as np
+    assert scale(0) == 0.0 and scale(1) == 0.0            # -90 dBFS clamps to 0
+    assert scale(32768) == 1.0
+    assert abs(scale(32768 * 10 ** (-30 / 20)) - 0.5) < 1e-9
+    assert pcm_level(b"") == 0.0
+    assert pcm_level(array.array("h", [1000] * 100).tobytes()) == scale(1000)
+    assert pcm_level(b"\x00") == 0.0                      # half a sample is ignored
+    env = envelope(np.zeros(24000, dtype=np.float32), 24000)
+    assert len(env) == 30 and max(env) == 0.0             # 1 s at 30 frames/s
+    env = envelope(np.full(24000, 0.5, dtype=np.float32), 24000)
+    assert all(abs(v - scale(16384)) < 1e-6 for v in env)
     print("ok")