From ecf00c1bc49a72c0a6583c638ff406bec9953f50 Mon Sep 17 00:00:00 2001 From: Mute Date: Wed, 12 Aug 2026 10:30:11 +0000 Subject: [PATCH] v0.7: pre-synth buffering, session IDs, segment timeout playback --- DhcpTunnelTest/Program.cs | 5 +- Robovoice.App/MainForm.cs | 1 + Robovoice.App/Orchestrator.cs | 179 ++++++++++++++- Robovoice.Stt.Dhcp/DhcpSttSource.cs | 79 +++++-- Server/NOTES.md | 332 ---------------------------- Server/PROTOCOL.md | 61 +++-- 6 files changed, 285 insertions(+), 372 deletions(-) delete mode 100644 Server/NOTES.md diff --git a/DhcpTunnelTest/Program.cs b/DhcpTunnelTest/Program.cs index 5efe109..adfef86 100644 --- a/DhcpTunnelTest/Program.cs +++ b/DhcpTunnelTest/Program.cs @@ -24,18 +24,19 @@ var cts = new CancellationTokenSource(); var recvTask = Task.Run(() => ReceiveLoop(sock, cts.Token)); var destEp = new IPEndPoint(IPAddress.Broadcast, 67); +uint session = 1; uint nonce = 0; while (!cts.Token.IsCancellationRequested) { nonce++; - string msg = $"HKMSTR {nonce}\n"; + string msg = $"HKMSTR {session} {nonce}\n"; byte[] payload = Encoding.UTF8.GetBytes(msg); try { int sent = sock.SendTo(payload, destEp); - Console.WriteLine($"[{DateTime.Now:HH:mm:ss.fff}] SENT n={nonce} {sent} bytes"); + Console.WriteLine($"[{DateTime.Now:HH:mm:ss.fff}] SENT s={session} n={nonce} {sent} bytes"); } catch (Exception ex) { diff --git a/Robovoice.App/MainForm.cs b/Robovoice.App/MainForm.cs index 95764c9..ebbe1b6 100644 --- a/Robovoice.App/MainForm.cs +++ b/Robovoice.App/MainForm.cs @@ -280,6 +280,7 @@ internal sealed partial class MainForm : Form private void OnPttPressed(object? sender, EventArgs e) { Log("PTT pressed"); + _orchestrator?.NotifyPttPressed(); try { _sttSource?.SendOn(); diff --git a/Robovoice.App/Orchestrator.cs b/Robovoice.App/Orchestrator.cs index 5ac2e92..8c01670 100644 --- a/Robovoice.App/Orchestrator.cs +++ b/Robovoice.App/Orchestrator.cs @@ -12,6 +12,16 @@ internal sealed class Orchestrator : IAsyncDisposable private CancellationTokenSource? _currentCts; private bool _disposed; + private readonly object _stateLock = new(); + private readonly Queue _pendingTexts = new(); + private readonly List _audioBuffer = new(); + private int _bufferSampleRate; + private bool _playing; + private Task? _synthTask; + private CancellationTokenSource? _synthCts; + private System.Threading.Timer? _segmentTimer; + private static readonly TimeSpan SegmentTimeout = TimeSpan.FromMilliseconds(250); + public string OutputDeviceName { get; set; } = string.Empty; public Orchestrator( @@ -30,9 +40,169 @@ internal sealed class Orchestrator : IAsyncDisposable private void OnTranscript(object? sender, TranscriptEventArgs e) { - if (e.Message.Type == TranscriptType.Final) + var msg = e.Message; + + lock (_stateLock) { - _ = SynthesizeAsync(e.Message.Text); + if (msg.Type == TranscriptType.Partial) + { + if (!string.IsNullOrWhiteSpace(msg.Text)) + { + _log($"SEGMENT: \"{msg.Text}\" ({msg.Text.Length} chars)"); + _pendingTexts.Enqueue(msg.Text); + EnsureSynthTask(); + ResetSegmentTimer(); + } + } + else + { + if (!string.IsNullOrWhiteSpace(msg.Text)) + { + _log($"FINAL: \"{msg.Text}\" ({msg.Text.Length} chars)"); + _pendingTexts.Enqueue(msg.Text); + EnsureSynthTask(); + } + else + { + _log("FINAL: (empty)"); + } + + CancelSegmentTimer(); + TransitionToPlaying(); + } + } + } + + private void ResetSegmentTimer() + { + _segmentTimer?.Dispose(); + _segmentTimer = new System.Threading.Timer(_ => OnSegmentTimeout(), null, SegmentTimeout, Timeout.InfiniteTimeSpan); + } + + private void CancelSegmentTimer() + { + _segmentTimer?.Dispose(); + _segmentTimer = null; + } + + private void OnSegmentTimeout() + { + _log("STT: segment timeout, starting playback early"); + lock (_stateLock) + { + TransitionToPlaying(); + } + } + + private void TransitionToPlaying() + { + if (_playing) + return; + + if (_audioBuffer.Count == 0) + { + if (_synthTask is null || _synthTask.IsCompleted) + { + _log("TTS: nothing to play"); + CancelSegmentTimer(); + } + return; + } + + _playing = true; + CancelSegmentTimer(); + + int sampleRate = _bufferSampleRate; + var chunks = _audioBuffer.ToList(); + _audioBuffer.Clear(); + + _log($"TTS: playing {chunks.Count} buffered chunks ({sampleRate} Hz)"); + + _audioOutput.Start(sampleRate, OutputDeviceName); + foreach (var samples in chunks) + { + _audioOutput.WriteSamples(samples); + } + } + + private void EnsureSynthTask() + { + if (_synthTask is not null && !_synthTask.IsCompleted) + return; + + _synthCts?.Cancel(); + _synthCts = new CancellationTokenSource(); + _synthTask = SynthLoopAsync(_synthCts.Token); + } + + private async Task SynthLoopAsync(CancellationToken ct) + { + while (true) + { + string text; + lock (_stateLock) + { + if (_pendingTexts.Count == 0) + break; + text = _pendingTexts.Dequeue(); + } + + try + { + await foreach (var chunk in _tts.SynthesizeAsync(text, ct)) + { + lock (_stateLock) + { + if (_playing) + { + _audioOutput.WriteSamples(chunk.Samples); + } + else + { + _bufferSampleRate = chunk.SampleRate; + _audioBuffer.Add(chunk.Samples); + } + } + } + } + catch (OperationCanceledException) + { + break; + } + catch (Exception ex) + { + _log($"TTS error: {ex.Message}"); + } + } + + lock (_stateLock) + { + if (_playing) + { + _audioOutput.Flush(); + _log("TTS: synthesis complete, flushed"); + } + } + } + + public void NotifyPttPressed() + { + lock (_stateLock) + { + CancelSegmentTimer(); + _synthCts?.Cancel(); + _synthCts?.Dispose(); + _synthTask = null; + + if (_playing) + { + _audioOutput.Stop(); + _playing = false; + } + + _pendingTexts.Clear(); + _audioBuffer.Clear(); + _bufferSampleRate = 0; } } @@ -50,7 +220,7 @@ internal sealed class Orchestrator : IAsyncDisposable var ct = _currentCts.Token; var sw = System.Diagnostics.Stopwatch.StartNew(); - _log($"FINAL: \"{text}\" ({text.Length} chars)"); + _log($"Speak: \"{text}\" ({text.Length} chars)"); try { @@ -101,6 +271,9 @@ internal sealed class Orchestrator : IAsyncDisposable if (_disposed) return; _currentCts?.Cancel(); _currentCts?.Dispose(); + _synthCts?.Cancel(); + _synthCts?.Dispose(); + CancelSegmentTimer(); _sttSource.TranscriptReceived -= OnTranscript; await _sttSource.DisposeAsync(); await _tts.DisposeAsync(); diff --git a/Robovoice.Stt.Dhcp/DhcpSttSource.cs b/Robovoice.Stt.Dhcp/DhcpSttSource.cs index aefc624..a74514a 100644 --- a/Robovoice.Stt.Dhcp/DhcpSttSource.cs +++ b/Robovoice.Stt.Dhcp/DhcpSttSource.cs @@ -19,6 +19,7 @@ public sealed class DhcpSttSource : ISttSource private Task? _nopTask; private EndPoint _broadcastEp = new IPEndPoint(IPAddress.Broadcast, DhcpServerPort); private uint _nonce; + private uint _session; private bool _disposed; public string InterfaceIp { get; set; } = string.Empty; @@ -73,7 +74,9 @@ public sealed class DhcpSttSource : ISttSource if (_cts is null) return; + _session++; _nonce = 0; + Log?.Invoke($"STT: session {_session} started"); SendNop(); _nopTask = NopLoopAsync(_cts.Token); } @@ -81,7 +84,7 @@ public sealed class DhcpSttSource : ISttSource public void SendOff() { StopNop(); - SendControl("HKMSTR:OFF"); + SendControl($"HKMSTR:OFF {_session} {_nonce}"); } private void StopNop() @@ -107,7 +110,7 @@ public sealed class DhcpSttSource : ISttSource private void SendNop() { _nonce++; - SendControl($"HKMSTR {_nonce}"); + SendControl($"HKMSTR {_session} {_nonce}"); } private void SendControl(string message) @@ -158,25 +161,67 @@ public sealed class DhcpSttSource : ISttSource if (!text.StartsWith(Magic)) continue; - if (text.StartsWith("HKMSTR:P ")) + TranscriptMessage? message = ParseReply(text); + if (message is null) + continue; + + TranscriptReceived?.Invoke(this, new TranscriptEventArgs { - string transcript = text["HKMSTR:P ".Length..]; - TranscriptReceived?.Invoke(this, new TranscriptEventArgs - { - Message = new TranscriptMessage(TranscriptType.Partial, transcript), - }); - } - else if (text.StartsWith("HKMSTR:F ")) - { - string transcript = text["HKMSTR:F ".Length..]; - TranscriptReceived?.Invoke(this, new TranscriptEventArgs - { - Message = new TranscriptMessage(TranscriptType.Final, transcript), - }); - } + Message = message, + }); } } + private TranscriptMessage? ParseReply(string text) + { + // Format: HKMSTR:P or HKMSTR:F + // may be empty. + string prefix; + TranscriptType type; + + if (text.StartsWith("HKMSTR:P ")) + { + prefix = "HKMSTR:P "; + type = TranscriptType.Partial; + } + else if (text.StartsWith("HKMSTR:F ")) + { + prefix = "HKMSTR:F "; + type = TranscriptType.Final; + } + else + { + return null; + } + + string rest = text[prefix.Length..]; + + int spaceIndex = rest.IndexOf(' '); + if (spaceIndex < 0) + { + if (uint.TryParse(rest, out uint sessionOnly)) + { + if (sessionOnly != _session) + return null; + return new TranscriptMessage(type, string.Empty); + } + return null; + } + + string sessionStr = rest[..spaceIndex]; + if (!uint.TryParse(sessionStr, out uint session)) + return null; + + if (session != _session) + { + Log?.Invoke($"STT: dropping stale reply (session {session} != current {_session})"); + return null; + } + + string transcript = rest[(spaceIndex + 1)..]; + return new TranscriptMessage(type, transcript); + } + public static List<(string Ip, string Name)> GetAvailableInterfaces() { var result = new List<(string, string)>(); diff --git a/Server/NOTES.md b/Server/NOTES.md deleted file mode 100644 index 03c31fa..0000000 --- a/Server/NOTES.md +++ /dev/null @@ -1,332 +0,0 @@ -# Server implementation notes - -The STT server listens on UDP port 67, receives NOP heartbeats and OFF -messages from Robovoice, captures audio from a microphone, runs speech -recognition (Moonshine), and sends transcript messages back to the client's -source address. - -## Architecture - -``` - ┌── HKMSTR (broadcast, every 50ms) -Robovoice ──────────────►│ - │ STT Server -Robovoice ◄──────────────┤ - └── HKMSTR:P/F (unicast) -``` - -The server: -1. Listens on UDP :67 -2. First `HKMSTR ` → start recording -3. `HKMSTR:OFF ` → stop recording, run STT -4. 150ms with no NOPs → stop recording, run STT (backstop) -5. Send `HKMSTR:F ` back to the client's source address:port - -## Framing - -All messages are newline-terminated UTF-8 text. No JSON, no binary framing. - -- `HKMSTR ` — NOP heartbeat (client → server) -- `HKMSTR:OFF ` — stop signal (client → server) -- `HKMSTR:P ` — partial transcript (server → client) -- `HKMSTR:F ` — final transcript (server → client) - -The nonce is an incrementing integer for packet uniqueness only. Discard it. - -## Python server with Moonshine - -```python -import socket -import threading -import time -import numpy as np -import sounddevice as sd -import moonshine - -LISTEN_PORT = 67 -CLIENT_PORT = 68 -SAMPLE_RATE = 16000 -SILENCE_TIMEOUT = 0.150 # 150ms - -model = moonshine.MoonshineModel(model="moonshine/base") - -sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) -sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) -sock.bind(("0.0.0.0", LISTEN_PORT)) - -print(f"STT server listening on :{LISTEN_PORT}") - -recording = False -last_nop_time = 0 -client_addr = None -audio_chunks = [] -lock = threading.Lock() - -def monitor_silence(): - """Backstop: stop recording if no NOPs for 150ms.""" - global recording - while True: - time.sleep(0.01) - with lock: - if recording and (time.monotonic() - last_nop_time) > SILENCE_TIMEOUT: - recording = False - threading.Thread(target=process_audio, daemon=True).start() - -threading.Thread(target=monitor_silence, daemon=True).start() - -def process_audio(): - global audio_chunks - with lock: - chunks = audio_chunks - audio_chunks = [] - addr = client_addr - - if not chunks: - return - - audio = np.concatenate(chunks) - print(f"Captured {len(audio)/SAMPLE_RATE:.1f}s") - - text = moonshine.transcribe(model, audio).strip() - - if text: - print(f"Final: {text}") - reply = f"HKMSTR:F {text}\n".encode("utf-8") - sock.sendto(reply, addr) - else: - print("Empty transcript") - -while True: - data, addr = sock.recvfrom(4096) - text = data.decode("utf-8", errors="ignore").strip() - - if not text.startswith("HKMSTR"): - continue - - if text.startswith("HKMSTR:OFF"): - with lock: - if recording: - recording = False - threading.Thread(target=process_audio, daemon=True).start() - continue - - if text.startswith("HKMSTR ") or text == "HKMSTR": - with lock: - client_addr = addr - last_nop_time = time.monotonic() - - if not recording: - recording = True - audio_chunks = [] - print(f"PTT on from {addr}") - - # Capture 50ms of audio - chunk = sd.rec(int(SAMPLE_RATE * 0.05), samplerate=SAMPLE_RATE, - channels=1, dtype="float32") - sd.wait() - audio_chunks.append(chunk.flatten()) -``` - -## Python server with streaming partials - -For live feedback, send partials while recording: - -```python -import socket -import threading -import time -import numpy as np -import sounddevice as sd -import moonshine - -LISTEN_PORT = 67 -SAMPLE_RATE = 16000 -SILENCE_TIMEOUT = 0.150 -PARTIAL_INTERVAL = 0.5 # send partial every 500ms - -model = moonshine.MoonshineModel(model="moonshine/base") - -sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) -sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) -sock.bind(("0.0.0.0", LISTEN_PORT)) - -print(f"STT server listening on :{LISTEN_PORT}") - -recording = False -last_nop_time = 0 -last_partial_time = 0 -client_addr = None -audio_chunks = [] -lock = threading.Lock() - -def capture_and_maybe_partial(): - global last_partial_time - with lock: - if not recording: - return - - chunk = sd.rec(int(SAMPLE_RATE * 0.05), samplerate=SAMPLE_RATE, - channels=1, dtype="float32") - sd.wait() - audio_chunks.append(chunk.flatten()) - - now = time.monotonic() - if now - last_partial_time > PARTIAL_INTERVAL: - last_partial_time = now - partial_audio = np.concatenate(audio_chunks) - partial_text = moonshine.transcribe(model, partial_audio).strip() - if partial_text and client_addr: - reply = f"HKMSTR:P {partial_text}\n".encode("utf-8") - sock.sendto(reply, client_addr) - -while True: - data, addr = sock.recvfrom(4096) - text = data.decode("utf-8", errors="ignore").strip() - - if not text.startswith("HKMSTR"): - continue - - if text.startswith("HKMSTR:OFF"): - with lock: - if recording: - recording = False - chunks = audio_chunks - audio_chunks = [] - - if chunks: - audio = np.concatenate(chunks) - final_text = moonshine.transcribe(model, audio).strip() - if final_text: - reply = f"HKMSTR:F {final_text}\n".encode("utf-8") - sock.sendto(reply, addr) - continue - - if text.startswith("HKMSTR") and not text.startswith("HKMSTR:"): - with lock: - client_addr = addr - last_nop_time = time.monotonic() - - if not recording: - recording = True - audio_chunks = [] - last_partial_time = time.monotonic() - print(f"PTT on from {addr}") - - capture_and_maybe_partial() - - # Check silence timeout - with lock: - if recording and (time.monotonic() - last_nop_time) > SILENCE_TIMEOUT: - recording = False - chunks = audio_chunks - audio_chunks = [] - - if 'chunks' in dir() and chunks: - audio = np.concatenate(chunks) - final_text = moonshine.transcribe(model, audio).strip() - if final_text: - reply = f"HKMSTR:F {final_text}\n".encode("utf-8") - sock.sendto(reply, addr) -``` - -## C# server skeleton - -```csharp -using System.Net; -using System.Net.Sockets; -using System.Text; - -var sock = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp); -sock.Bind(new IPEndPoint(IPAddress.Any, 67)); - -Console.WriteLine("STT server listening on :67"); - -byte[] buffer = new byte[4096]; -EndPoint fromEp = new IPEndPoint(IPAddress.Any, 0); -bool recording = false; -DateTime lastNop = DateTime.MinValue; -List audioChunks = new(); - -while (true) -{ - if (sock.Poll(50_000, SelectMode.SelectRead)) - { - int received = sock.ReceiveFrom(buffer, ref fromEp); - string text = Encoding.UTF8.GetString(buffer, 0, received).TrimEnd('\n', '\r'); - - if (!text.StartsWith("HKMSTR")) - continue; - - if (text.StartsWith("HKMSTR:OFF")) - { - if (recording) - { - recording = false; - ProcessAndReply(audioChunks, fromEp); - audioChunks.Clear(); - } - continue; - } - - // NOP - lastNop = DateTime.UtcNow; - if (!recording) - { - recording = true; - audioChunks.Clear(); - Console.WriteLine($"PTT on from {fromEp}"); - } - - // Capture 50ms audio here... - // audioChunks.Add(capturedChunk); - - // Check silence timeout - if (recording && (DateTime.UtcNow - lastNop).TotalMilliseconds > 150) - { - recording = false; - ProcessAndReply(audioChunks, fromEp); - audioChunks.Clear(); - } - } - else - { - // Timeout check even without incoming data - if (recording && (DateTime.UtcNow - lastNop).TotalMilliseconds > 150) - { - recording = false; - ProcessAndReply(audioChunks, fromEp); - audioChunks.Clear(); - } - } -} - -void ProcessAndReply(List chunks, EndPoint client) -{ - if (chunks.Count == 0) return; - - // Concatenate and run STT... - string text = "recognized text here"; - - if (!string.IsNullOrEmpty(text)) - { - byte[] reply = Encoding.UTF8.GetBytes($"HKMSTR:F {text}\n"); - sock.SendTo(reply, client); - } -} -``` - -## Tips - -- **Reply address:** always reply to the source endpoint of the last NOP. - The client binds to a specific IP on :68. -- **Nonces:** discard them. They exist only to make each datagram unique. - Do not derive any meaning from nonce values. -- **Silence timeout:** 150ms = 3 missed NOPs at 50ms intervals. If you - change the NOP interval on the client, adjust this accordingly. -- **Partials:** optional. Send `HKMSTR:P ` while recording for live - feedback. Client logs them but only `HKMSTR:F` triggers TTS. -- **Broadcast only for C→S:** the WireGuard killswitch only allows - outbound broadcast to 255.255.255.255:67. Unicast from client won't pass. -- **Unicast OK for S→C:** the inbound WFP rule has no address restriction, - so unicast replies to :68 pass through. -- **Moonshine models:** `moonshine/base` or `moonshine/tiny`. diff --git a/Server/PROTOCOL.md b/Server/PROTOCOL.md index 4c57d27..1e6dbfa 100644 --- a/Server/PROTOCOL.md +++ b/Server/PROTOCOL.md @@ -34,33 +34,46 @@ with the 6-byte magic `HKMSTR` to distinguish our traffic from real DHCP. **NOP (heartbeat while PTT held):** ``` -HKMSTR \n +HKMSTR \n ``` -Sent every 50ms while PTT is held. The nonce is an incrementing unsigned -integer that makes each datagram unique. The server discards it — it's -purely for packet uniqueness, not for any protocol logic. +Sent every 50ms while PTT is held. The session is an incrementing unsigned +integer that identifies the current PTT utterance (incremented on each PTT +press). The nonce is an incrementing unsigned integer that makes each +datagram unique. The server should echo the session back in replies. Both +are discarded by the server for protocol logic — the server tracks liveness +via "did anything arrive recently." **OFF (PTT released):** ``` -HKMSTR:OFF \n +HKMSTR:OFF \n ``` Sent once when PTT is released. This is the fast-stop signal. If lost, the 150ms timeout acts as a backstop. ### Server → Client -**Partial transcript:** -``` -HKMSTR:P \n -``` -Intermediate recognition result. Fire-and-forget. Client logs it but does -not act on it. +The server echoes the session ID from the NOPs in all replies. The client +drops any reply with a stale session ID. -**Final transcript:** +**Partial segment (completed VAD segment):** ``` -HKMSTR:F \n +HKMSTR:P \n ``` -Complete utterance. Client feeds this to the TTS engine. +A completed, VAD-separated utterance segment. The client starts TTS +synthesis immediately and buffers the audio output, but does **not** play +it yet. Playback starts when `:F` arrives (or timeout). + +**Final (all done):** +``` +HKMSTR:F \n +``` +Signals that all segments have been sent. May be empty +(`HKMSTR:F \n`). Triggers playback of all buffered audio on the +client. If `` is non-empty, the client synthesizes it before playing. + +The purpose of this design is to minimize latency: TTS synthesis runs in +parallel with recording, so by the time `:F` arrives, audio is already +buffered and playback starts immediately. ## Server state machine @@ -79,15 +92,27 @@ Complete utterance. Client feeds this to the TTS engine. │ PROCESSING │ │ └─────────────┘ │ │ │ - STT │ │ - done │ │ + send │ │ + :P/:F │ │ ▼ │ - send HKMSTR:F ─────────┘ + back to IDLE ──────────┘ ``` - **IDLE → RECORDING:** first NOP received, start mic capture +- **RECORDING:** VAD detects completed segments → send `HKMSTR:P ` - **RECORDING → PROCESSING:** OFF received, OR 150ms since last NOP -- **PROCESSING → IDLE:** STT done, send `HKMSTR:F ` +- **PROCESSING → IDLE:** send remaining segments as `:P`, then `HKMSTR:F` + +## Client playback model + +1. `:P` arrives → start TTS synthesis immediately, buffer audio (don't play). + Reset 250ms segment timer. +2. More `:P` arrive → keep synthesizing and buffering, reset timer each time. +3. `:F` arrives → play all buffered audio immediately, cancel timer. +4. If `:F` doesn't arrive within 250ms of the last `:P` → play buffered audio + early. If more `:P` arrive after early playback, synthesis continues and + new audio is appended to the output — not a failure. +5. `:F` may be empty — it just signals "all segments sent, start/confirm playback." ## Timing