Emit state, context and audio levels from the voice loop

speaking comes from the reply thread, not the speaker: the tool
announcement would otherwise replace the tool state while the tool is
still running. arecord gets a 33 ms period, because it writes its file
in periods and the default left the user's waves at 8 levels a second;
the context size is read at startup so the first hello carries it.

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

1 files changed, 71 insertions, 6 deletions

diff --git a/assistant.py b/assistant.py
index fa03f2f..8ccc6ff 100644
--- a/assistant.py
+++ b/assistant.py
@@ -400,7 +400,7 @@ turn = 0          # bumped by every button press: older replies and audio are st
 player = None     # the aplay speaking now, killed by a new press
 
 
-def speaker(q):
+def speaker(q, lq):
     global player
     from kokoro_onnx import Kokoro
     k = Kokoro(CFG["kokoro"]["model"], CFG["kokoro"]["voices"])
@@ -408,6 +408,7 @@ def speaker(q):
     voices = {l: sum(w * k.get_voice_style(n) for n, w in v["blend"].items())
               for l, v in CFG["voice"].items()}
     play, playing = None, None            # aplay and the turn it speaks for
+    start, written = 0.0, 0               # playback clock of the open stream
     while True:
         t, item = q.get()
         if t != turn:                     # from a reply the user cut off
@@ -421,10 +422,13 @@ def speaker(q):
                     play.stdin.close()
                     play.wait()
                     play = None
+                if t == turn:
+                    emit({"type": "state", "state": "idle"})
                 continue
             if isinstance(item, float):   # seconds of silence
                 if play:
                     play.stdin.write(bytes(int(sr * item) * 2))
+                    written += int(sr * item)
                 continue
             text, lang = item
             audio, sr = k.create(text, voice=voices[lang], lang=KLANG[lang])
@@ -437,12 +441,51 @@ def speaker(q):
                 # A new stream wakes the suspended sink and the first ~0.2 s is
                 # lost: lead with silence so it eats that, not the first word.
                 play.stdin.write(bytes(int(sr * 0.3) * 2))
+                start, written = time.monotonic() + 0.3, 0
+            # This sentence plays after everything written so far. If synthesis
+            # fell behind (an underrun), playback restarts from now.
+            # ponytail: no sound-server latency correction; add an offset here
+            # if the waves visibly lead the voice.
+            now = time.monotonic()
+            if start + written / sr < now:
+                start = now - written / sr
+            due = start + written / sr
+            for i, value in enumerate(envelope(audio, sr)):
+                lq.put((t, due + i * FRAME, value))
+            written += len(audio)
             # The pipe buffer lets synthesis of the next sentence overlap playback.
             play.stdin.write((audio.clip(-1, 1) * 32767).astype("<i2").tobytes())
         except BrokenPipeError:           # killed by a new press mid-write
             play = None
 
 
+def user_levels(rec, mine):
+    # The user's waves: the loudness of what arecord wrote since last time.
+    pos = 0
+    while rec.poll() is None and turn == mine:
+        time.sleep(FRAME)
+        try:
+            with open(RAW, "rb") as f:
+                f.seek(pos)
+                data = f.read()
+        except OSError:
+            continue
+        if len(data) >= 2:
+            pos += len(data) // 2 * 2
+            emit({"type": "level", "who": "user", "value": pcm_level(data)})
+
+
+def level_emitter(lq):
+    # The reply's waves, each frame sent when its audio plays rather than when
+    # it was synthesised (synthesis runs ahead of playback).
+    while True:
+        t, due, value = lq.get()
+        while t == turn and (wait := due - time.monotonic()) > 0:
+            time.sleep(min(wait, 0.05))
+        if t == turn:
+            emit({"type": "level", "who": "assistant", "value": value})
+
+
 def die_with_parent():
     # Runs in the child before exec: the kernel sends SIGTERM when we die,
     # however we die, so no whisper-server or amixer is ever orphaned.
@@ -477,6 +520,7 @@ def reply(history, q, mine):
     # garbled question at worst; per-turn file names if that ever shows.
     pcm = open(RAW, "rb").read()
     if len(pcm) < 16000 * 2 * 0.3:               # under 0.3 s: a stray press
+        emit({"type": "state", "state": "idle"})
         return
     # Trailing silence: whisper drops the last word when audio ends mid-word,
     # which it does when the button is pressed while still talking.
@@ -485,11 +529,16 @@ def reply(history, q, mine):
         w.setsampwidth(2)
         w.setframerate(16000)
         w.writeframes(pcm + bytes(16000 * 2 // 2))
+    emit({"type": "state", "state": "transcribing"})
     t = time.perf_counter()
     lang, text = stt()
-    if not text or text.startswith("[") or turn != mine:   # [BLANK_AUDIO] and friends
+    if turn != mine:
+        return
+    if not text or text.startswith("["):         # [BLANK_AUDIO] and friends
+        emit({"type": "state", "state": "idle"})
         return
     log(f"you ({lang}, {time.perf_counter() - t:.2f}s): {text}")
+    emit({"type": "state", "state": "thinking"})
     question = {"role": "user", "content": text}
     # ponytail: last 20 messages, summarise if long sessions need memory
     # The model has no clock: without the date it makes one up for tools.
@@ -501,6 +550,8 @@ def reply(history, q, mine):
         nonlocal voice
         if not said:
             log(f"first sentence after {time.perf_counter() - t:.2f}s")
+        if now_state["state"] != "speaking":
+            emit({"type": "state", "state": "speaking"})
         said.append(s)
         voice = lang_of(s, voice)
         q.put((mine, (re.sub(r"[*_#`]", "", s), voice)))
@@ -533,9 +584,12 @@ def reply(history, q, mine):
             announced = True
         # ponytail: a press does not kill a running tool, its result is dropped
         for c in calls.values():
+            emit({"type": "state", "state": "tool", "tool": c["name"]})
             result = run_tool(TOOLS, c["name"], c["arguments"])
             log(f"tool {c['name']}({c['arguments']}): {result[:120]!r}")
             msgs.append({"role": "tool", "tool_call_id": c["id"], "content": result})
+        if turn == mine:
+            emit({"type": "state", "state": "thinking"})
     if turn != mine:
         log("interrupted")
         return
@@ -551,8 +605,14 @@ def main():
     # SIGTERM must run the cleanup below too, or the recording is left behind.
     signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))
     start_whisper()
-    q = queue.Queue()
-    threading.Thread(target=speaker, args=(q,), daemon=True).start()
+    try:
+        context_size(llm_model())         # so the overlay's first hello has it
+    except OSError:
+        pass                              # llama-server down: asked again per request
+    serve()
+    q, lq = queue.Queue(), queue.Queue()
+    threading.Thread(target=speaker, args=(q, lq), daemon=True).start()
+    threading.Thread(target=level_emitter, args=(lq,), daemon=True).start()
     history, rec = [], None
     events = subprocess.Popen(["stdbuf", "-oL", "amixer", "-c", CFG["card"], "events"],
                               stdout=subprocess.PIPE, text=True, preexec_fn=die_with_parent)
@@ -575,8 +635,13 @@ def main():
                 turn += 1
                 if player:
                     player.kill()
+                emit({"type": "state", "state": "listening"})
+                # A 33 ms period: arecord writes the file in periods, and the
+                # default (~125 ms) left the user's waves at 8 levels a second.
                 rec = subprocess.Popen(["arecord", "-q", "-t", "raw", "-f", "S16_LE", "-r", "16000",
-                                        "-c", "1", RAW], stderr=subprocess.DEVNULL)
+                                        "-c", "1", "--period-time=33333", RAW],
+                                       stderr=subprocess.DEVNULL)
+                threading.Thread(target=user_levels, args=(rec, turn), daemon=True).start()
             elif rec:
                 rec.send_signal(signal.SIGINT)
                 rec.wait()
@@ -585,7 +650,7 @@ def main():
     except KeyboardInterrupt:
         pass
     finally:
-        for f in (RAW, WAV):                    # the user's voice: never keep it
+        for f in (RAW, WAV, SOCK):              # the user's voice: never keep it
             if os.path.exists(f):
                 os.remove(f)