From facbfe6a5c2dfbd4640f9421830b1f9bc61606ea Mon Sep 17 00:00:00 2001 From: Mute Date: Thu, 13 Aug 2026 09:58:33 +0000 Subject: [PATCH] v0.8: TCP transport via gatuna tunnel, session IDs, pre-synth playback, server Rust rewrite --- BACKLOG.md | 21 + DhcpTunnelTest/Program.cs | 106 ---- .../Robovoice.DhcpTunnelTest.csproj | 10 - Robovoice.App/AppConfig.cs | 2 +- Robovoice.App/MainForm.Designer.cs | 17 +- Robovoice.App/MainForm.cs | 170 +++---- Robovoice.App/Orchestrator.cs | 40 +- Robovoice.App/Robovoice.App.csproj | 2 +- Robovoice.Stt.Dhcp/DhcpSttSource.cs | 277 ---------- .../Robovoice.Stt.Tcp.csproj | 0 Robovoice.Stt.Tcp/TcpSttSource.cs | 256 ++++++++++ Robovoice.slnx | 3 +- Server/PROTOCOL.md | 166 +++--- Server/server/Cargo.toml | 18 + Server/server/build.rs | 27 + Server/server/src/main.rs | 478 ++++++++++++++++++ 16 files changed, 955 insertions(+), 638 deletions(-) create mode 100644 BACKLOG.md delete mode 100644 DhcpTunnelTest/Program.cs delete mode 100644 DhcpTunnelTest/Robovoice.DhcpTunnelTest.csproj delete mode 100644 Robovoice.Stt.Dhcp/DhcpSttSource.cs rename Robovoice.Stt.Dhcp/Robovoice.Stt.Dhcp.csproj => Robovoice.Stt.Tcp/Robovoice.Stt.Tcp.csproj (100%) create mode 100644 Robovoice.Stt.Tcp/TcpSttSource.cs create mode 100644 Server/server/Cargo.toml create mode 100644 Server/server/build.rs create mode 100644 Server/server/src/main.rs diff --git a/BACKLOG.md b/BACKLOG.md new file mode 100644 index 0000000..0aba70b --- /dev/null +++ b/BACKLOG.md @@ -0,0 +1,21 @@ +# Robovoice Backlog + +## Second-order PTT + +When a game is running, it has its own PTT button (e.g. a voip push-to-talk +key). Robovoice should simulate pressing the game's PTT key before TTS audio +output begins, and release it after playback finishes. + +This lets the TTS audio be transmitted through the game's voip channel to +other players. + +### Considerations + +- Needs a configurable "game PTT key" (separate from Robovoice's own PTT key) +- Use `SendInput` or `keybd_event` to synthesize the keypress +- Press the game PTT key right before buffered audio starts playing +- Release it after `AudioOutput` finishes playback (need a playback-complete + signal — currently `Flush()` doesn't provide one) +- Edge cases: what if the user presses Robovoice PTT while game PTT is still + held from a previous utterance? Flush should release game PTT too. +- Should this be a per-output-device setting? (CABLE Output vs speakers) diff --git a/DhcpTunnelTest/Program.cs b/DhcpTunnelTest/Program.cs deleted file mode 100644 index adfef86..0000000 --- a/DhcpTunnelTest/Program.cs +++ /dev/null @@ -1,106 +0,0 @@ -using System.Net; -using System.Net.NetworkInformation; -using System.Net.Sockets; -using System.Text; - -// Args: [interface_ip] [interval_ms] -// Defaults: auto-detect LAN IP, 50ms - -string localIp = args.Length > 0 ? args[0] : GetLanInterfaceIp() ?? "0.0.0.0"; -int intervalMs = args.Length > 1 && int.TryParse(args[1], out int iv) ? iv : 50; - -using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp); -sock.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true); -sock.EnableBroadcast = true; - -var localEp = new IPEndPoint(IPAddress.Parse(localIp), 68); -sock.Bind(localEp); - -Console.WriteLine($"Bound to {localEp} (SO_REUSEADDR)"); -Console.WriteLine("Sending NOPs + listening. Press Ctrl+C to stop."); -Console.WriteLine(); - -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 {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 s={session} n={nonce} {sent} bytes"); - } - catch (Exception ex) - { - Console.WriteLine($"[{DateTime.Now:HH:mm:ss.fff}] SEND FAILED: {ex.Message}"); - } - - try { Thread.Sleep(intervalMs); } - catch (OperationCanceledException) { break; } -} - -cts.Cancel(); -await recvTask; - -static string? GetLanInterfaceIp() -{ - foreach (var nic in NetworkInterface.GetAllNetworkInterfaces()) - { - if (nic.OperationalStatus != OperationalStatus.Up) - continue; - if (nic.NetworkInterfaceType == NetworkInterfaceType.Loopback) - continue; - if (nic.Description.Contains("WireGuard", StringComparison.OrdinalIgnoreCase)) - continue; - - foreach (var addr in nic.GetIPProperties().UnicastAddresses) - { - if (addr.Address.AddressFamily == AddressFamily.InterNetwork) - { - var ip = addr.Address.ToString(); - if (ip.StartsWith("192.168.") || ip.StartsWith("10.") || ip.StartsWith("172.")) - return ip; - } - } - } - return null; -} - -static void ReceiveLoop(Socket sock, CancellationToken ct) -{ - byte[] buffer = new byte[4096]; - EndPoint fromEp = new IPEndPoint(IPAddress.Any, 0); - - while (!ct.IsCancellationRequested) - { - int received; - try - { - if (!sock.Poll(500_000, SelectMode.SelectRead)) - continue; - received = sock.ReceiveFrom(buffer, ref fromEp); - } - catch (Exception ex) - { - Console.WriteLine($"[{DateTime.Now:HH:mm:ss.fff}] RECV ERROR: {ex.Message}"); - continue; - } - - string text = Encoding.UTF8.GetString(buffer, 0, received).TrimEnd('\n', '\r'); - if (!text.StartsWith("HKMSTR")) - { - Console.WriteLine($"[{DateTime.Now:HH:mm:ss.fff}] RECV {received} bytes from {fromEp} (no magic)"); - continue; - } - - Console.WriteLine($"[{DateTime.Now:HH:mm:ss.fff}] RECV {received} bytes from {fromEp}: {text}"); - } -} diff --git a/DhcpTunnelTest/Robovoice.DhcpTunnelTest.csproj b/DhcpTunnelTest/Robovoice.DhcpTunnelTest.csproj deleted file mode 100644 index dfb40ca..0000000 --- a/DhcpTunnelTest/Robovoice.DhcpTunnelTest.csproj +++ /dev/null @@ -1,10 +0,0 @@ - - - - Exe - net10.0 - enable - enable - - - diff --git a/Robovoice.App/AppConfig.cs b/Robovoice.App/AppConfig.cs index f410f81..68c5166 100644 --- a/Robovoice.App/AppConfig.cs +++ b/Robovoice.App/AppConfig.cs @@ -12,7 +12,7 @@ public sealed class AppConfig public int LengthScale { get; set; } = 100; public int NoiseWScale { get; set; } = 800; public bool MinimizeToTray { get; set; } = true; - public string InterfaceIp { get; set; } = string.Empty; + public string SttEndpoint { get; set; } = "127.0.0.1:6996"; public static string AppDataDir => Path.Combine( Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData), diff --git a/Robovoice.App/MainForm.Designer.cs b/Robovoice.App/MainForm.Designer.cs index 7513f79..c81541a 100644 --- a/Robovoice.App/MainForm.Designer.cs +++ b/Robovoice.App/MainForm.Designer.cs @@ -17,7 +17,7 @@ partial class MainForm private Button btnBrowseFile = null!; private Label lblFileName = null!; private Label lblServer = null!; - private ComboBox cmbInterface = null!; + private TextBox txtSttEndpoint = null!; private Label lblLineStatus = null!; private RichTextBox txtLog = null!; private CheckBox chkMinimizeToTray = null!; @@ -57,7 +57,7 @@ partial class MainForm btnBrowseFile = new Button(); lblFileName = new Label(); lblServer = new Label(); - cmbInterface = new ComboBox(); + txtSttEndpoint = new TextBox(); lblLineStatus = new Label(); txtLog = new RichTextBox(); chkMinimizeToTray = new CheckBox(); @@ -143,15 +143,14 @@ partial class MainForm lblFileName.ForeColor = Color.Gray; // lblServer - lblServer.Text = "Interface:"; + lblServer.Text = "STT:"; lblServer.Location = new Point(440, 48); - lblServer.Size = new Size(55, 23); + lblServer.Size = new Size(35, 23); lblServer.TextAlign = ContentAlignment.MiddleLeft; - // cmbInterface - cmbInterface.Location = new Point(498, 45); - cmbInterface.Size = new Size(165, 23); - cmbInterface.DropDownStyle = ComboBoxStyle.DropDownList; + // txtSttEndpoint + txtSttEndpoint.Location = new Point(478, 45); + txtSttEndpoint.Size = new Size(185, 23); // lblLineStatus lblLineStatus.Text = ""; @@ -270,7 +269,7 @@ partial class MainForm Controls.Add(btnBrowseFile); Controls.Add(lblFileName); Controls.Add(lblServer); - Controls.Add(cmbInterface); + Controls.Add(txtSttEndpoint); Controls.Add(lblLineStatus); Controls.Add(lblNoise); Controls.Add(trkNoise); diff --git a/Robovoice.App/MainForm.cs b/Robovoice.App/MainForm.cs index ebbe1b6..930c2c9 100644 --- a/Robovoice.App/MainForm.cs +++ b/Robovoice.App/MainForm.cs @@ -2,7 +2,7 @@ using NAudio.Wave; using Robovoice.App; using Robovoice.Core; using Robovoice.Core.Voices; -using Robovoice.Stt.Dhcp; +using Robovoice.Stt.Tcp; using Robovoice.Tts.LibPiper; using System.Diagnostics; @@ -16,7 +16,7 @@ internal sealed partial class MainForm : Form private LibPiperTtsEngine? _tts; private AudioOutput? _audioOutput; - private DhcpSttSource? _sttSource; + private TcpSttSource? _sttSource; private Orchestrator? _orchestrator; private PttHotkey? _pttHotkey; private NotifyIcon? _trayIcon; @@ -36,57 +36,64 @@ internal sealed partial class MainForm : Form private async void OnLoad(object? sender, EventArgs e) { - ActiveControl = txtLog; - PopulateVoices(); - PopulateOutputDevices(); + try + { + ActiveControl = txtLog; + PopulateVoices(); + PopulateOutputDevices(); - _pttKey = (Keys)_config.PttKey; - if (_pttKey == Keys.None) - _pttKey = Keys.F8; - txtPttKey.Text = KeyToDisplayString(_pttKey); + _pttKey = (Keys)_config.PttKey; + if (_pttKey == Keys.None) + _pttKey = Keys.F8; + txtPttKey.Text = KeyToDisplayString(_pttKey); - if (!string.IsNullOrEmpty(_config.Voice) && cmbVoice.Items.Contains(_config.Voice)) - cmbVoice.SelectedItem = _config.Voice; - else if (cmbVoice.Items.Count > 0) - cmbVoice.SelectedIndex = 0; + if (!string.IsNullOrEmpty(_config.Voice) && cmbVoice.Items.Contains(_config.Voice)) + cmbVoice.SelectedItem = _config.Voice; + else if (cmbVoice.Items.Count > 0) + cmbVoice.SelectedIndex = 0; - if (!string.IsNullOrEmpty(_config.OutputDevice) && cmbOutput.Items.Contains(_config.OutputDevice)) - cmbOutput.SelectedItem = _config.OutputDevice; - else - AutoSelectCableOutput(); + if (!string.IsNullOrEmpty(_config.OutputDevice) && cmbOutput.Items.Contains(_config.OutputDevice)) + cmbOutput.SelectedItem = _config.OutputDevice; + else + AutoSelectCableOutput(); - trkNoise.Value = _config.NoiseScale; - trkSpeed.Value = _config.LengthScale; - trkNoiseW.Value = _config.NoiseWScale; - OnSliderScroll(null, EventArgs.Empty); + trkNoise.Value = _config.NoiseScale; + trkSpeed.Value = _config.LengthScale; + trkNoiseW.Value = _config.NoiseWScale; + OnSliderScroll(null, EventArgs.Empty); - chkMinimizeToTray.Checked = _config.MinimizeToTray; - PopulateInterfaces(); + chkMinimizeToTray.Checked = _config.MinimizeToTray; + txtSttEndpoint.Text = _config.SttEndpoint; - btnBrowseFile.Click += OnBrowseFile; - btnTestVoice.Click += OnTestVoice; - btnManageVoices.Click += OnManageVoices; - btnClearLog.Click += (_, _) => txtLog.Clear(); - txtPttKey.Enter += OnPttKeyFocus; - txtPttKey.KeyDown += OnPttKeyDown; - cmbOutput.SelectedIndexChanged += OnOutputChanged; - cmbVoice.SelectedIndexChanged += OnVoiceChanged; - cmbInterface.SelectedIndexChanged += OnInterfaceChanged; + btnBrowseFile.Click += OnBrowseFile; + btnTestVoice.Click += OnTestVoice; + btnManageVoices.Click += OnManageVoices; + btnClearLog.Click += (_, _) => txtLog.Clear(); + txtPttKey.Enter += OnPttKeyFocus; + txtPttKey.KeyDown += OnPttKeyDown; + cmbOutput.SelectedIndexChanged += OnOutputChanged; + cmbVoice.SelectedIndexChanged += OnVoiceChanged; + txtSttEndpoint.Leave += OnSttEndpointChanged; - trkNoise.Scroll += OnSliderScroll; - trkSpeed.Scroll += OnSliderScroll; - trkNoiseW.Scroll += OnSliderScroll; - trkNoise.MouseUp += OnSliderReleased; - trkSpeed.MouseUp += OnSliderReleased; - trkNoiseW.MouseUp += OnSliderReleased; + trkNoise.Scroll += OnSliderScroll; + trkSpeed.Scroll += OnSliderScroll; + trkNoiseW.Scroll += OnSliderScroll; + trkNoise.MouseUp += OnSliderReleased; + trkSpeed.MouseUp += OnSliderReleased; + trkNoiseW.MouseUp += OnSliderReleased; - chkMinimizeToTray.CheckedChanged += (_, _) => SaveConfig(); + chkMinimizeToTray.CheckedChanged += (_, _) => SaveConfig(); - Resize += OnResize; + Resize += OnResize; - SetupTray(); + SetupTray(); + } + catch (Exception ex) + { + System.Diagnostics.Debug.WriteLine($"OnLoad failed: {ex}"); + } - await InitializeEngineAsync(); + BeginInvoke(async () => await InitializeEngineAsync()); } private Keys _pttKey = Keys.F8; @@ -232,7 +239,8 @@ internal sealed partial class MainForm : Form Log($"Loading voice: {voiceName}..."); _audioOutput?.Dispose(); - _tts?.DisposeAsync().AsTask().Wait(); + if (_tts is not null) + await _tts.DisposeAsync(); _tts = new LibPiperTtsEngine( modelPath, @@ -242,21 +250,21 @@ internal sealed partial class MainForm : Form noiseWScale: trkNoiseW.Value / 1000.0f); _audioOutput = new AudioOutput(); - if (_sttSource is null) - { - string ifaceIp = cmbInterface.SelectedItem as string ?? ""; - _sttSource = new DhcpSttSource { InterfaceIp = ifaceIp, Log = Log }; - await _sttSource.StartAsync(); - } - - _orchestrator?.DisposeAsync().AsTask().Wait(); - _orchestrator = new Orchestrator(_tts, _audioOutput, _sttSource, Log) - { - OutputDeviceName = cmbOutput.SelectedItem as string ?? string.Empty, - }; - try { + if (_sttSource is null) + { + _sttSource = new TcpSttSource { Endpoint = txtSttEndpoint.Text, Log = Log }; + await _sttSource.StartAsync(); + } + + if (_orchestrator is not null) + await _orchestrator.DisposeAsync(); + _orchestrator = new Orchestrator(_tts, _audioOutput, _sttSource, Log) + { + OutputDeviceName = cmbOutput.SelectedItem as string ?? string.Empty, + }; + await _orchestrator.InitializeTtsAsync(); Log("Engine ready. Press PTT to send to STT server."); SetupHotkey(); @@ -426,59 +434,19 @@ internal sealed partial class MainForm : Form SaveConfig(); } - private void PopulateInterfaces() + private async void OnSttEndpointChanged(object? sender, EventArgs e) { - cmbInterface.Items.Clear(); - foreach (var (ip, name) in DhcpSttSource.GetAvailableInterfaces()) - { - cmbInterface.Items.Add(name); - cmbInterface.Items[^1] = name; - cmbInterface.Items[cmbInterface.Items.Count - 1] = name; - } - - if (!string.IsNullOrEmpty(_config.InterfaceIp)) - { - for (int i = 0; i < cmbInterface.Items.Count; i++) - { - if (cmbInterface.Items[i] is string s && s.Contains(_config.InterfaceIp)) - { - cmbInterface.SelectedIndex = i; - return; - } - } - } - - if (cmbInterface.Items.Count > 0) - cmbInterface.SelectedIndex = 0; - } - - private void OnInterfaceChanged(object? sender, EventArgs e) - { - if (cmbInterface.SelectedItem is not string selected) - return; - - string? ip = ExtractIpFromDisplay(selected); - if (ip is null) return; - - _config.InterfaceIp = ip; + _config.SttEndpoint = txtSttEndpoint.Text; SaveConfig(); if (_sttSource is not null) { - _sttSource.DisposeAsync().AsTask().Wait(2000); + await _sttSource.DisposeAsync(); _sttSource = null; - _ = InitializeEngineAsync(); + await InitializeEngineAsync(); } } - private static string? ExtractIpFromDisplay(string display) - { - int start = display.IndexOf('('); - int end = display.IndexOf(')'); - if (start < 0 || end <= start) return null; - return display[(start + 1)..end]; - } - private void SetupTray() { if (_trayInit) return; @@ -544,7 +512,7 @@ internal sealed partial class MainForm : Form _config.LengthScale = trkSpeed.Value; _config.NoiseWScale = trkNoiseW.Value; _config.MinimizeToTray = chkMinimizeToTray.Checked; - _config.InterfaceIp = ExtractIpFromDisplay(cmbInterface.SelectedItem as string ?? "") ?? ""; + _config.SttEndpoint = txtSttEndpoint.Text; _config.Save(); } diff --git a/Robovoice.App/Orchestrator.cs b/Robovoice.App/Orchestrator.cs index 8c01670..c696619 100644 --- a/Robovoice.App/Orchestrator.cs +++ b/Robovoice.App/Orchestrator.cs @@ -19,8 +19,6 @@ internal sealed class Orchestrator : IAsyncDisposable 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; @@ -51,7 +49,6 @@ internal sealed class Orchestrator : IAsyncDisposable _log($"SEGMENT: \"{msg.Text}\" ({msg.Text.Length} chars)"); _pendingTexts.Enqueue(msg.Text); EnsureSynthTask(); - ResetSegmentTimer(); } } else @@ -67,33 +64,11 @@ internal sealed class Orchestrator : IAsyncDisposable _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) @@ -104,13 +79,11 @@ internal sealed class Orchestrator : IAsyncDisposable if (_synthTask is null || _synthTask.IsCompleted) { _log("TTS: nothing to play"); - CancelSegmentTimer(); } return; } _playing = true; - CancelSegmentTimer(); int sampleRate = _bufferSampleRate; var chunks = _audioBuffer.ToList(); @@ -189,7 +162,6 @@ internal sealed class Orchestrator : IAsyncDisposable { lock (_stateLock) { - CancelSegmentTimer(); _synthCts?.Cancel(); _synthCts?.Dispose(); _synthTask = null; @@ -209,7 +181,7 @@ internal sealed class Orchestrator : IAsyncDisposable public async Task InitializeTtsAsync() { _log("Initializing TTS engine..."); - await _tts.InitializeAsync(); + await Task.Run(() => _tts.InitializeAsync()); _log($"TTS ready (sample rate: {_tts.SampleRate} Hz)"); } @@ -269,15 +241,17 @@ internal sealed class Orchestrator : IAsyncDisposable public async ValueTask DisposeAsync() { if (_disposed) return; - _currentCts?.Cancel(); + _disposed = true; + + try { _currentCts?.Cancel(); } catch { } _currentCts?.Dispose(); - _synthCts?.Cancel(); + + try { _synthCts?.Cancel(); } catch { } _synthCts?.Dispose(); - CancelSegmentTimer(); + _sttSource.TranscriptReceived -= OnTranscript; await _sttSource.DisposeAsync(); await _tts.DisposeAsync(); _audioOutput.Dispose(); - _disposed = true; } } diff --git a/Robovoice.App/Robovoice.App.csproj b/Robovoice.App/Robovoice.App.csproj index fbbe61c..d50c566 100644 --- a/Robovoice.App/Robovoice.App.csproj +++ b/Robovoice.App/Robovoice.App.csproj @@ -3,7 +3,7 @@ - + diff --git a/Robovoice.Stt.Dhcp/DhcpSttSource.cs b/Robovoice.Stt.Dhcp/DhcpSttSource.cs deleted file mode 100644 index a74514a..0000000 --- a/Robovoice.Stt.Dhcp/DhcpSttSource.cs +++ /dev/null @@ -1,277 +0,0 @@ -using System.Net; -using System.Net.NetworkInformation; -using System.Net.Sockets; -using System.Text; -using Robovoice.Core; - -namespace Robovoice.Stt.Dhcp; - -public sealed class DhcpSttSource : ISttSource -{ - private const string Magic = "HKMSTR"; - private const int DhcpClientPort = 68; - private const int DhcpServerPort = 67; - private static readonly TimeSpan NopInterval = TimeSpan.FromMilliseconds(50); - - private Socket? _sock; - private CancellationTokenSource? _cts; - private Task? _receiveTask; - 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; - - public Action? Log { get; set; } - - public event TranscriptEventHandler? TranscriptReceived; - - public Task StartAsync(CancellationToken ct = default) - { - ObjectDisposedException.ThrowIf(_disposed, this); - if (_cts is not null) - return Task.CompletedTask; - - string ip = string.IsNullOrEmpty(InterfaceIp) ? AutoDetectInterfaceIp() ?? "0.0.0.0" : InterfaceIp; - - _sock = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp); - _sock.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true); - _sock.EnableBroadcast = true; - _sock.Bind(new IPEndPoint(IPAddress.Parse(ip), DhcpClientPort)); - - Log?.Invoke($"STT: bound to {ip}:{DhcpClientPort}, broadcasting to :{DhcpServerPort}"); - - _cts = CancellationTokenSource.CreateLinkedTokenSource(ct); - _receiveTask = ReceiveLoopAsync(_cts.Token); - return Task.CompletedTask; - } - - public async Task StopAsync(CancellationToken ct = default) - { - StopNop(); - - if (_cts is not null) - _cts.Cancel(); - - _sock?.Dispose(); - _sock = null; - - if (_receiveTask is not null) - { - try { await _receiveTask.WaitAsync(ct); } - catch { } - _receiveTask = null; - } - - _cts?.Dispose(); - _cts = null; - } - - public void SendOn() - { - if (_cts is null) - return; - - _session++; - _nonce = 0; - Log?.Invoke($"STT: session {_session} started"); - SendNop(); - _nopTask = NopLoopAsync(_cts.Token); - } - - public void SendOff() - { - StopNop(); - SendControl($"HKMSTR:OFF {_session} {_nonce}"); - } - - private void StopNop() - { - if (_nopTask is not null) - { - try { _nopTask.Wait(2000); } catch { } - _nopTask = null; - } - } - - private async Task NopLoopAsync(CancellationToken ct) - { - while (!ct.IsCancellationRequested) - { - try { await Task.Delay(NopInterval, ct); } - catch (OperationCanceledException) { break; } - - SendNop(); - } - } - - private void SendNop() - { - _nonce++; - SendControl($"HKMSTR {_session} {_nonce}"); - } - - private void SendControl(string message) - { - if (_sock is null) - return; - - try - { - byte[] payload = Encoding.UTF8.GetBytes(message + "\n"); - _sock.SendTo(payload, _broadcastEp); - } - catch (Exception ex) - { - Log?.Invoke($"STT: send failed: {ex.Message}"); - } - } - - private async Task ReceiveLoopAsync(CancellationToken ct) - { - byte[] buffer = new byte[4096]; - EndPoint fromEp = new IPEndPoint(IPAddress.Any, 0); - - while (!ct.IsCancellationRequested) - { - int received; - try - { - if (!_sock!.Poll(500_000, SelectMode.SelectRead)) - continue; - received = _sock.ReceiveFrom(buffer, ref fromEp); - } - catch (OperationCanceledException) - { - break; - } - catch (ObjectDisposedException) - { - break; - } - catch (Exception ex) - { - Log?.Invoke($"STT: receive error: {ex.Message}"); - continue; - } - - string text = Encoding.UTF8.GetString(buffer, 0, received).TrimEnd('\n', '\r'); - if (!text.StartsWith(Magic)) - continue; - - TranscriptMessage? message = ParseReply(text); - if (message is null) - continue; - - TranscriptReceived?.Invoke(this, new TranscriptEventArgs - { - 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)>(); - - foreach (var nic in NetworkInterface.GetAllNetworkInterfaces()) - { - if (nic.OperationalStatus != OperationalStatus.Up) - continue; - if (nic.NetworkInterfaceType == NetworkInterfaceType.Loopback) - continue; - - string desc = nic.Description; - if (desc.Contains("WireGuard", StringComparison.OrdinalIgnoreCase)) - continue; - - foreach (var addr in nic.GetIPProperties().UnicastAddresses) - { - if (addr.Address.AddressFamily != AddressFamily.InterNetwork) - continue; - - string ip = addr.Address.ToString(); - if (ip.StartsWith("192.168.") || ip.StartsWith("10.") || - ip.StartsWith("172.16.") || ip.StartsWith("172.17.") || - ip.StartsWith("172.18.") || ip.StartsWith("172.19.") || - ip.StartsWith("172.20.") || ip.StartsWith("172.21.") || - ip.StartsWith("172.22.") || ip.StartsWith("172.23.") || - ip.StartsWith("172.24.") || ip.StartsWith("172.25.") || - ip.StartsWith("172.26.") || ip.StartsWith("172.27.") || - ip.StartsWith("172.28.") || ip.StartsWith("172.29.") || - ip.StartsWith("172.30.") || ip.StartsWith("172.31.")) - { - result.Add((ip, $"{nic.Name} ({ip})")); - } - } - } - - return result; - } - - private static string? AutoDetectInterfaceIp() - { - foreach (var (ip, _) in GetAvailableInterfaces()) - return ip; - return null; - } - - public async ValueTask DisposeAsync() - { - if (_disposed) return; - await StopAsync(); - _disposed = true; - } -} diff --git a/Robovoice.Stt.Dhcp/Robovoice.Stt.Dhcp.csproj b/Robovoice.Stt.Tcp/Robovoice.Stt.Tcp.csproj similarity index 100% rename from Robovoice.Stt.Dhcp/Robovoice.Stt.Dhcp.csproj rename to Robovoice.Stt.Tcp/Robovoice.Stt.Tcp.csproj diff --git a/Robovoice.Stt.Tcp/TcpSttSource.cs b/Robovoice.Stt.Tcp/TcpSttSource.cs new file mode 100644 index 0000000..3af3864 --- /dev/null +++ b/Robovoice.Stt.Tcp/TcpSttSource.cs @@ -0,0 +1,256 @@ +using System.Net; +using System.Net.Sockets; +using System.Text; +using Robovoice.Core; + +namespace Robovoice.Stt.Tcp; + +public sealed class TcpSttSource : ISttSource +{ + private TcpClient? _tcp; + private NetworkStream? _stream; + private StreamReader? _reader; + private StreamWriter? _writer; + private CancellationTokenSource? _cts; + private Task? _runTask; + private readonly object _sendLock = new(); + private uint _session; + private bool _disposed; + + public string Endpoint { get; set; } = "127.0.0.1:6996"; + + public Action? Log { get; set; } + + public event TranscriptEventHandler? TranscriptReceived; + + public Task StartAsync(CancellationToken ct = default) + { + ObjectDisposedException.ThrowIf(_disposed, this); + if (_cts is not null) + return Task.CompletedTask; + + _cts = CancellationTokenSource.CreateLinkedTokenSource(ct); + _runTask = RunAsync(_cts.Token); + return Task.CompletedTask; + } + + public async Task StopAsync(CancellationToken ct = default) + { + if (_cts is not null) + _cts.Cancel(); + + CleanupConnection(); + + if (_runTask is not null) + { + try { await _runTask.WaitAsync(ct); } + catch { } + _runTask = null; + } + + _cts?.Dispose(); + _cts = null; + } + + private async Task RunAsync(CancellationToken ct) + { + while (!ct.IsCancellationRequested) + { + IPEndPoint? endpoint = ParseEndpoint(Endpoint); + if (endpoint is null) + { + Log?.Invoke($"STT: invalid endpoint '{Endpoint}'"); + try { await Task.Delay(3000, ct); } catch { break; } + continue; + } + + try + { + _tcp = new TcpClient(); + using var connectCts = CancellationTokenSource.CreateLinkedTokenSource(ct); + connectCts.CancelAfter(TimeSpan.FromSeconds(5)); + await _tcp.ConnectAsync(endpoint.Address, endpoint.Port, connectCts.Token); + + _stream = _tcp.GetStream(); + _reader = new StreamReader(_stream, Encoding.UTF8); + _writer = new StreamWriter(_stream, Encoding.UTF8) { AutoFlush = true }; + + Log?.Invoke($"STT: connected to {Endpoint}"); + + await ReceiveLoopAsync(ct); + } + catch (OperationCanceledException) + { + break; + } + catch (Exception ex) + { + Log?.Invoke($"STT: connection failed ({ex.Message}), retrying..."); + } + finally + { + CleanupConnection(); + } + + if (!ct.IsCancellationRequested) + { + try { await Task.Delay(3000, ct); } + catch (OperationCanceledException) { break; } + } + } + } + + private async Task ReceiveLoopAsync(CancellationToken ct) + { + while (!ct.IsCancellationRequested && _reader is not null) + { + string? line; + try + { + line = await _reader.ReadLineAsync(ct); + } + catch + { + break; + } + + if (line is null) + break; + + TranscriptMessage? message = ParseReply(line); + if (message is null) + continue; + + TranscriptReceived?.Invoke(this, new TranscriptEventArgs + { + Message = message, + }); + } + } + + public void SendOn() + { + _session++; + Send($"ON {_session}"); + } + + public void SendOff() + { + Send($"OFF {_session}"); + } + + private void Send(string message) + { + lock (_sendLock) + { + if (_writer is null) + return; + + try + { + _writer.WriteLine(message); + } + catch + { + Log?.Invoke($"STT: failed to send '{message}' (not connected?)"); + } + } + } + + private void CleanupConnection() + { + lock (_sendLock) + { + _writer?.Dispose(); + _reader?.Dispose(); + _stream?.Dispose(); + _tcp?.Dispose(); + _writer = null; + _reader = null; + _stream = null; + _tcp = null; + } + } + + private static IPEndPoint? ParseEndpoint(string endpoint) + { + int colon = endpoint.LastIndexOf(':'); + if (colon <= 0) + return null; + + string host = endpoint[..colon]; + if (!int.TryParse(endpoint[(colon + 1)..], out int port)) + return null; + + if (IPAddress.TryParse(host, out var addr)) + return new IPEndPoint(addr, port); + + try + { + var addresses = Dns.GetHostAddresses(host); + addr = addresses.FirstOrDefault(a => a.AddressFamily == AddressFamily.InterNetwork); + if (addr is null) + return null; + return new IPEndPoint(addr, port); + } + catch + { + return null; + } + } + + private TranscriptMessage? ParseReply(string line) + { + if (line.StartsWith("P ", StringComparison.Ordinal)) + { + string rest = line["P ".Length..]; + int space = rest.IndexOf(' '); + if (space < 0) + return null; + + if (!uint.TryParse(rest[..space], out uint session)) + return null; + + if (session != _session) + { + Log?.Invoke($"STT: dropping stale reply (session {session} != current {_session})"); + return null; + } + + return new TranscriptMessage(TranscriptType.Partial, rest[(space + 1)..]); + } + + if (line.StartsWith("F ", StringComparison.Ordinal)) + { + string rest = line["F ".Length..]; + int space = rest.IndexOf(' '); + if (space < 0) + return null; + + if (!uint.TryParse(rest[..space], out uint session)) + return null; + + if (session != _session) + { + Log?.Invoke($"STT: dropping stale reply (session {session} != current {_session})"); + return null; + } + + return new TranscriptMessage(TranscriptType.Final, rest[(space + 1)..]); + } + + if (line == "F") + { + return new TranscriptMessage(TranscriptType.Final, string.Empty); + } + + return null; + } + + public async ValueTask DisposeAsync() + { + if (_disposed) return; + await StopAsync(); + _disposed = true; + } +} diff --git a/Robovoice.slnx b/Robovoice.slnx index dccac60..8d92e14 100644 --- a/Robovoice.slnx +++ b/Robovoice.slnx @@ -2,7 +2,6 @@ - + - diff --git a/Server/PROTOCOL.md b/Server/PROTOCOL.md index 1e6dbfa..19c0530 100644 --- a/Server/PROTOCOL.md +++ b/Server/PROTOCOL.md @@ -1,136 +1,106 @@ -# Robovoice DHCP Tunnel Protocol +# Robovoice STT Protocol ## Overview -Robovoice communicates with a remote STT server by tunneling through the -DHCP UDP ports (68→67). This exploits a common killswitch exception: VPN -software (e.g. WireGuard) blocks all traffic except DHCP, which is allowed -for network connectivity maintenance. +Robovoice connects to the STT server over TCP (typically through a gatuna +L2 tunnel). The server captures audio, runs Moonshine STT, and sends +transcript segments back. The client pre-synthesizes TTS on segments and +plays audio on final. ``` -[Robovoice client] --broadcast UDP :68→:67--> [STT server] -[Robovoice client] <--unicast UDP :67→:68-- [STT server] +[Robovoice client] --TCP--> [STT server 127.0.0.1:6996] + │ │ + ├── ON \n ────────►│ (abort old, start new session) + ├── OFF \n ────────►│ (stop, final STT pass) + │◄── P \n ──┤ (completed segment) + │◄── F \n ──┤ (all done; text may be empty) ``` -The client broadcasts NOP heartbeats while PTT is held. The server starts -recording on the first NOP and stops when it receives OFF or when 150ms -pass with no NOPs. - ## Transport -- **Protocol:** UDP (connectionless, unreliable) -- **Client → Server:** broadcast, source port 68, dest port 67 -- **Server → Client:** unicast, source port 67, dest port 68 -- **Client binds:** to a specific LAN interface IP on port 68 (with - `SO_REUSEADDR` to coexist with the Windows DHCP service) -- **No connection state** — purely fire-and-forget datagrams +- **Protocol:** TCP (reliable, ordered, connection-oriented) +- **Server:** `127.0.0.1:6996` (hardcoded loopback) +- **Framing:** newline-delimited text (`\n`), UTF-8 +- **Auto-reconnect:** client retries every 3s if connection drops ## Wire format -All messages are plain text, newline-terminated (`\n`). Every message starts -with the 6-byte magic `HKMSTR` to distinguish our traffic from real DHCP. - ### Client → Server -**NOP (heartbeat while PTT held):** +**ON (PTT pressed):** ``` -HKMSTR \n +ON \n ``` -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." +Starts a new STT session. The server aborts any active session and starts +recording. `` is an incrementing unsigned integer chosen by the +client. Replies from the server echo this session ID. **OFF (PTT released):** ``` -HKMSTR:OFF \n +OFF \n ``` -Sent once when PTT is released. This is the fast-stop signal. If lost, the -150ms timeout acts as a backstop. +Stops the session. The server does a final STT pass on remaining audio and +sends any new segments followed by `F`. ### Server → Client -The server echoes the session ID from the NOPs in all replies. The client -drops any reply with a stale session ID. - -**Partial segment (completed VAD segment):** +**Segment (completed VAD segment):** ``` -HKMSTR:P \n +P \n ``` 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). +synthesis immediately and buffers the audio (does not play yet). **Final (all done):** ``` -HKMSTR:F \n +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. +Signals all segments have been sent. `` may be empty (`F \n`). +Triggers playback of all buffered audio on the client. If text 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. +## Session IDs + +- Client increments session ID on each PTT press +- Server echoes the session ID in all replies for that session +- Client drops any reply with a stale session ID (handles the race where + stale segments from an aborted session are still in the TCP buffer) +- Server aborts old session on receiving `ON` with a new session ID + +## Client playback model + +1. `P` arrives → start TTS synthesis immediately, buffer audio (don't play) +2. More `P` arrive → keep synthesizing and buffering +3. `F` arrives → play all buffered audio immediately +4. PTT pressed → flush: stop playback, cancel synthesis, clear buffers + +The purpose of pre-synthesis is to minimize latency between PTT release +and audio playback. By the time `F` arrives, audio is already buffered. ## Server state machine ``` - ┌──────────────────────────────────────────┐ - │ │ - ▼ │ - ┌──────────┐ first NOP ┌──────────────┐ │ - │ IDLE │ ──────────► │ RECORDING │ │ - └──────────┘ └──────────────┘ │ - │ │ │ - OFF │ │ 150ms │ - recv'd │ │ silence │ - ▼ ▼ │ - ┌─────────────┐ │ - │ PROCESSING │ │ - └─────────────┘ │ - │ │ - send │ │ - :P/:F │ │ - ▼ │ - back to IDLE ──────────┘ + ┌──────────┐ ON ┌──────────────┐ + │ IDLE │ ──────────────► │ RECORDING │ + └──────────┘ └──────────────┘ + │ │ + OFF │ │ + recv'd │ │ + ▼ │ + ┌─────────────┐ + │ PROCESSING │ + └─────────────┘ + │ + send │ + P/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:** send remaining segments as `:P`, then `HKMSTR:F` +- **IDLE → RECORDING:** `ON ` received, start mic capture +- **RECORDING:** Moonshine streaming produces completed segments → send `P` +- **RECORDING → PROCESSING:** `OFF ` received +- **PROCESSING → IDLE:** final STT pass, send remaining `P` + `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 - -| Parameter | Value | Purpose | -|-----------|-------|---------| -| NOP interval | 50ms | Heartbeat frequency while PTT held | -| Silence timeout | 150ms | Stop recording if no NOPs (3 missed = lost OFF) | -| NOP bandwidth | ~20 msg/s × ~20 bytes | ~400 bytes/s — negligible | - -## Why this works - -1. **Outbound broadcast `:68→:67` to `255.255.255.255`** passes the - WireGuard WFP killswitch (DHCP exception matches this exact pattern) -2. **Inbound `:67→:68`** has no address restriction in the WFP rule, so - unicast replies pass through -3. **Binding to a specific interface IP** (not `0.0.0.0`) wins unicast - delivery over the Windows DHCP client service -4. **NOP spam** ensures the ON message gets through even at 5% packet loss - (3 consecutive NOPs = ~0.01% drop probability) -5. **150ms timeout** is the backstop for lost OFF — at 50ms intervals, 3 - consecutive NOPs must all be lost to false-stop +If `ON` arrives while recording, the current session is aborted (no final +flush) and a new session starts immediately. diff --git a/Server/server/Cargo.toml b/Server/server/Cargo.toml new file mode 100644 index 0000000..6dfa3c0 --- /dev/null +++ b/Server/server/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "robovoice-stt-server" +version = "0.1.0" +edition = "2021" + +[[bin]] +name = "robovoice-stt-server" +path = "src/main.rs" + +[dependencies] +anyhow = "1" +cpal = "0.15" + +[build-dependencies] +bindgen = "0.71" + +[profile.release] +opt-level = 3 diff --git a/Server/server/build.rs b/Server/server/build.rs new file mode 100644 index 0000000..e29bcd5 --- /dev/null +++ b/Server/server/build.rs @@ -0,0 +1,27 @@ +use std::path::PathBuf; + +fn main() { + let manifest_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")); + let lib_dir = manifest_dir.join("..").join("bench").join("moonshine-voice").join("lib"); + let include_dir = manifest_dir.join("..").join("bench").join("moonshine-voice").join("include"); + let header = include_dir.join("moonshine-c-api.h"); + + println!("cargo:rerun-if-changed={}", header.display()); + println!("cargo:rustc-link-search=native={}", lib_dir.display()); + println!("cargo:rustc-link-lib=dylib=moonshine"); + println!("cargo:rustc-link-arg=-Wl,-rpath,{}", lib_dir.display()); + + let bindings = bindgen::Builder::default() + .header(header.to_str().unwrap()) + .allowlist_function("moonshine_.*") + .allowlist_var("MOONSHINE_.*") + .allowlist_type("transcript.*|moonshine_option_t|speaker_span_t|transcript_word_t") + .derive_default(true) + .generate() + .expect("Unable to generate moonshine bindings"); + + let out_path = PathBuf::from(std::env::var("OUT_DIR").unwrap()); + bindings + .write_to_file(out_path.join("moonshine_bindings.rs")) + .expect("Couldn't write bindings"); +} diff --git a/Server/server/src/main.rs b/Server/server/src/main.rs new file mode 100644 index 0000000..7e93524 --- /dev/null +++ b/Server/server/src/main.rs @@ -0,0 +1,478 @@ +use anyhow::{anyhow, Result}; +use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; +use cpal::{SampleFormat, SampleRate}; +use std::collections::HashSet; +use std::ffi::CStr; +use std::io::{BufRead, BufReader, Write}; +use std::net::{TcpListener, TcpStream}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +include!(concat!(env!("OUT_DIR"), "/moonshine_bindings.rs")); + +const SAMPLE_RATE: i32 = 16000; +const HEADER_VERSION: i32 = 30000; +const ARCH: u32 = 5; // MOONSHINE_MODEL_ARCH_MEDIUM_STREAMING +const BIND_ADDR: &str = "127.0.0.1:6996"; +const MAX_TEXT_BYTES: usize = 1380; + +// ─── helpers ────────────────────────────────────────────────────────────── + +fn ts() -> String { + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default(); + let secs = now.as_secs() % 86400; + let h = secs / 3600; + let m = (secs % 3600) / 60; + let s = secs % 60; + let ms = now.subsec_millis(); + format!("{:02}:{:02}:{:02}.{:03}", h, m, s, ms) +} + +fn log(msg: &str) { + eprintln!("[{}] {}", ts(), msg); +} + +fn err_str(code: i32) -> String { + unsafe { + let s = moonshine_error_to_string(code); + if s.is_null() { + format!("error {}", code) + } else { + CStr::from_ptr(s).to_string_lossy().into_owned() + } + } +} + +fn truncate_to_word(text: &str, max_bytes: usize) -> &str { + if text.len() <= max_bytes { + return text; + } + let cut = &text[..max_bytes.min(text.len())]; + match cut.rfind(' ') { + Some(pos) => &text[..pos], + None => cut, + } +} + +fn line_text(line: &transcript_line_t) -> String { + if line.text.is_null() { + return String::new(); + } + unsafe { CStr::from_ptr(line.text) } + .to_string_lossy() + .into_owned() +} + +// ─── shared state ───────────────────────────────────────────────────────── + +struct Shared { + writer: Mutex, + session_id: u64, + transcriber_handle: i32, +} + +impl Shared { + fn send_msg(&self, prefix: &str, text: &str) { + let text = truncate_to_word(text, MAX_TEXT_BYTES); + let line = if text.is_empty() { + format!("{} {}\n", prefix, self.session_id) + } else { + format!("{} {} {}\n", prefix, self.session_id, text) + }; + let mut writer = self.writer.lock().unwrap(); + match writer.write_all(line.as_bytes()) { + Ok(_) => log(&format!("TX {} {} {}", prefix, self.session_id, text)), + Err(e) => log(&format!("TX failed: {}", e)), + } + } +} + +// ─── session ────────────────────────────────────────────────────────────── + +struct Session { + shared: Arc, + stop_signal: Arc, + aborted: Arc, + transcriber: thread::JoinHandle<()>, + cpal_stream: cpal::Stream, + stream_handle: i32, +} + +impl Session { + fn stop(self) { + self.stop_signal.store(true, Ordering::SeqCst); + drop(self.cpal_stream); + self.transcriber.join().ok(); + unsafe { moonshine_free_stream(self.shared.transcriber_handle, self.stream_handle) }; + } + + fn abort(self) { + self.stop_signal.store(true, Ordering::SeqCst); + self.aborted.store(true, Ordering::SeqCst); + drop(self.cpal_stream); + self.transcriber.join().ok(); + unsafe { moonshine_free_stream(self.shared.transcriber_handle, self.stream_handle) }; + } +} + +fn start_session(shared: Arc) -> Option { + let stream_handle = unsafe { moonshine_create_stream(shared.transcriber_handle, 0) }; + if stream_handle < 0 { + log(&format!("create_stream failed: {}", err_str(stream_handle))); + return None; + } + let rc = unsafe { moonshine_start_stream(shared.transcriber_handle, stream_handle) }; + if rc != 0 { + log(&format!("start_stream failed: {}", err_str(rc))); + unsafe { moonshine_free_stream(shared.transcriber_handle, stream_handle) }; + return None; + } + + let audio_buf: Arc>> = Arc::new(Mutex::new(Vec::new())); + let stop_signal = Arc::new(AtomicBool::new(false)); + let aborted = Arc::new(AtomicBool::new(false)); + + let cpal_stream = match start_cpal(audio_buf.clone(), stop_signal.clone()) { + Ok(s) => s, + Err(e) => { + log(&format!("cpal failed: {}", e)); + unsafe { moonshine_free_stream(shared.transcriber_handle, stream_handle) }; + return None; + } + }; + + let shared_clone = shared.clone(); + let stop_signal_clone = stop_signal.clone(); + let aborted_clone = aborted.clone(); + + let transcriber = thread::spawn(move || { + transcriber_loop(shared_clone, audio_buf, stop_signal_clone, aborted_clone, stream_handle); + }); + + Some(Session { + shared, + stop_signal, + aborted, + transcriber, + cpal_stream, + stream_handle, + }) +} + +fn transcriber_loop( + shared: Arc, + audio_buf: Arc>>, + stop_signal: Arc, + aborted: Arc, + stream_handle: i32, +) { + let handle = shared.transcriber_handle; + let mut sent_ids: HashSet = HashSet::new(); + + while !stop_signal.load(Ordering::SeqCst) { + let chunk = { + let mut buf = audio_buf.lock().unwrap(); + if buf.is_empty() { + drop(buf); + thread::sleep(Duration::from_millis(5)); + continue; + } + std::mem::take(&mut *buf) + }; + + unsafe { + moonshine_transcribe_add_audio_to_stream( + handle, stream_handle, + chunk.as_ptr(), chunk.len() as u64, + SAMPLE_RATE, 0, + ); + } + + let mut t_ptr: *mut transcript_t = std::ptr::null_mut(); + let rc = unsafe { moonshine_transcribe_stream(handle, stream_handle, 0, &mut t_ptr) }; + if rc != 0 || t_ptr.is_null() { + continue; + } + + send_new_segments(&shared, t_ptr, &mut sent_ids, "P"); + } + + // If aborted (new session took over), skip final flush entirely + if aborted.load(Ordering::SeqCst) { + unsafe { moonshine_stop_stream(handle, stream_handle) }; + log(&format!("Session {} aborted, skipping final flush", shared.session_id)); + return; + } + + // Drain remaining audio + let remaining = { + let mut buf = audio_buf.lock().unwrap(); + std::mem::take(&mut *buf) + }; + if !remaining.is_empty() { + unsafe { + moonshine_transcribe_add_audio_to_stream( + handle, stream_handle, + remaining.as_ptr(), remaining.len() as u64, + SAMPLE_RATE, 0, + ); + } + } + + // Final flush + unsafe { moonshine_stop_stream(handle, stream_handle) }; + let mut t_ptr: *mut transcript_t = std::ptr::null_mut(); + let rc = unsafe { moonshine_transcribe_stream(handle, stream_handle, 0, &mut t_ptr) }; + + if rc == 0 && !t_ptr.is_null() { + let t = unsafe { &*t_ptr }; + let mut new_segments: Vec = Vec::new(); + + for i in 0..t.line_count as usize { + let line = unsafe { &*t.lines.add(i) }; + if line.text.is_null() || line.is_complete == 0 { + continue; + } + if !sent_ids.insert(line.id) { + continue; + } + let text = line_text(line); + if !text.is_empty() { + new_segments.push(text); + } + } + + if new_segments.is_empty() { + shared.send_msg("F", ""); + } else { + let last = new_segments.len() - 1; + for (i, text) in new_segments.iter().enumerate() { + let prefix = if i == last { "F" } else { "P" }; + shared.send_msg(prefix, text); + } + } + } else { + shared.send_msg("F", ""); + } +} + +fn send_new_segments( + shared: &Shared, + t_ptr: *const transcript_t, + sent_ids: &mut HashSet, + prefix: &str, +) { + let t = unsafe { &*t_ptr }; + + for i in 0..t.line_count as usize { + let line = unsafe { &*t.lines.add(i) }; + if line.text.is_null() || line.is_complete == 0 { + continue; + } + if !sent_ids.insert(line.id) { + continue; + } + let text = line_text(line); + if text.is_empty() { + continue; + } + shared.send_msg(prefix, &text); + } +} + +// ─── cpal ───────────────────────────────────────────────────────────────── + +fn start_cpal( + audio_buf: Arc>>, + stop_signal: Arc, +) -> Result { + let host = cpal::default_host(); + let dev = host + .default_input_device() + .ok_or_else(|| anyhow!("no input device"))?; + + let supported = dev + .supported_input_configs()? + .filter(|c| c.channels() <= 2 && c.min_sample_rate().0 <= 16000) + .min_by_key(|c| match c.sample_format() { + SampleFormat::F32 => 0, + SampleFormat::I16 => 1, + SampleFormat::U8 => 2, + _ => 99, + }) + .ok_or_else(|| anyhow!("no suitable input config"))?; + + let fmt = supported.sample_format(); + let mut config = supported.with_max_sample_rate().config(); + if config.channels > 1 { + config.channels = 1; + } + config.sample_rate = SampleRate(16000); + + let err_fn = |e: cpal::StreamError| log(&format!("cpal error: {}", e)); + + let stream = match fmt { + SampleFormat::F32 => dev.build_input_stream( + &config, + move |data: &[f32], _: &_| { + if !stop_signal.load(Ordering::Relaxed) { + audio_buf.lock().unwrap().extend_from_slice(data); + } + }, + err_fn, + None, + )?, + SampleFormat::I16 => dev.build_input_stream( + &config, + move |data: &[i16], _: &_| { + if !stop_signal.load(Ordering::Relaxed) { + audio_buf.lock().unwrap().extend(data.iter().map(|&x| x as f32 / 32768.0)); + } + }, + err_fn, + None, + )?, + SampleFormat::U8 => dev.build_input_stream( + &config, + move |data: &[u8], _: &_| { + if !stop_signal.load(Ordering::Relaxed) { + audio_buf.lock().unwrap().extend(data.iter().map(|&x| (x as f32 - 128.0) / 128.0)); + } + }, + err_fn, + None, + )?, + _ => return Err(anyhow!("unsupported sample format {:?}", fmt)), + }; + + stream.play()?; + Ok(stream) +} + +// ─── main ───────────────────────────────────────────────────────────────── + +fn main() -> Result<()> { + let mut args = std::env::args().skip(1); + let mut model_dir = String::from("../bench/medium-streaming-en"); + + while let Some(a) = args.next() { + match a.as_str() { + "--model-dir" | "-m" => { + model_dir = args.next().unwrap_or(model_dir); + } + "--help" | "-h" => { + println!("Usage: robovoice-stt-server [--model-dir DIR]"); + println!("Listens on TCP {}", BIND_ADDR); + println!("Model: medium-streaming (Moonshine)"); + return Ok(()); + } + _ => return Err(anyhow!("unknown arg: {}", a)), + } + } + + let model_path = std::fs::canonicalize(&model_dir) + .unwrap_or_else(|_| std::path::PathBuf::from(&model_dir)); + + log(&format!("Loading model from {}...", model_path.display())); + let c_dir = std::ffi::CString::new(model_path.to_str().unwrap()).unwrap(); + let transcriber_handle = unsafe { + moonshine_load_transcriber_from_files( + c_dir.as_ptr(), + ARCH, + std::ptr::null(), + 0, + HEADER_VERSION, + ) + }; + if transcriber_handle < 0 { + return Err(anyhow!("failed to load model: {}", err_str(transcriber_handle))); + } + log(&format!("Model loaded (handle {})", transcriber_handle)); + + let listener = TcpListener::bind(BIND_ADDR)?; + log(&format!("STT server listening on TCP {}", BIND_ADDR)); + + let mut session_id_counter: u64 = 0; + let mut current_session: Option = None; + + for stream in listener.incoming() { + let stream = match stream { + Ok(s) => s, + Err(e) => { + log(&format!("accept failed: {}", e)); + continue; + } + }; + + stream.set_nodelay(true).ok(); + log(&format!("Client connected: {}", stream.peer_addr().unwrap_or_default())); + + let writer_stream = stream.try_clone()?; + let reader = BufReader::new(stream); + + for line in reader.lines() { + let line = match line { + Ok(l) => l, + Err(_) => break, + }; + + let line = line.trim(); + log(&format!("RX {}", line)); + + // ON + if let Some(rest) = line.strip_prefix("ON ") { + let new_session_id: u64 = rest.parse().unwrap_or(0); + + if let Some(s) = current_session.take() { + log(&format!("Aborting session {} for new session {}", s.shared.session_id, new_session_id)); + s.abort(); + } + + session_id_counter = new_session_id; + let shared = Arc::new(Shared { + writer: Mutex::new(writer_stream.try_clone()?), + session_id: session_id_counter, + transcriber_handle, + }); + + log(&format!("PTT on session {}", session_id_counter)); + match start_session(shared) { + Some(s) => current_session = Some(s), + None => log("Failed to start session"), + } + } + // OFF + else if let Some(rest) = line.strip_prefix("OFF ") { + let off_session: u64 = rest.parse().unwrap_or(0); + + if let Some(s) = current_session.as_ref() { + if s.shared.session_id == off_session { + log(&format!("OFF session {}", off_session)); + if let Some(s) = current_session.take() { + s.stop(); + } + } else { + log(&format!("OFF session {} (stale, current={}), ignoring", off_session, s.shared.session_id)); + } + } else { + log(&format!("OFF session {} (no active session), ignoring", off_session)); + } + } + else { + log(&format!("Unknown command: {}", line)); + } + } + + log("Client disconnected"); + + if let Some(s) = current_session.take() { + s.abort(); + } + } + + Ok(()) +}