diff --git a/tests/04a-realtime-oneshot-cs/Program.cs b/tests/04a-realtime-oneshot-cs/Program.cs new file mode 100644 index 0000000..eca0d35 --- /dev/null +++ b/tests/04a-realtime-oneshot-cs/Program.cs @@ -0,0 +1,491 @@ +using System.Net.WebSockets; +using System.Runtime.InteropServices; +using System.Text; +using System.Text.Json; +using System.Threading.Channels; +using PortAudioSharp; +using Stream = PortAudioSharp.Stream; + +// .NET-side mirror for any readers that go through Environment.GetEnvironmentVariable, +// AND libc setenv so PortAudio (which uses getenv()) sees the value. +Environment.SetEnvironmentVariable("PA_ALSA_PLUGHW", "1"); +Libc.setenv("PA_ALSA_PLUGHW", "1", 1); + +string apiKey = LoadApiKey(); + +PortAudio.Initialize(); + +using var cts = new CancellationTokenSource(); +Console.CancelKeyPress += (_, e) => { e.Cancel = true; cts.Cancel(); }; + +var upstreamChannel = Channel.CreateBounded(new BoundedChannelOptions(C.UpstreamChannelCapacity) +{ + FullMode = BoundedChannelFullMode.Wait, + SingleWriter = true, + SingleReader = true, +}); +var downstreamChannel = Channel.CreateBounded(new BoundedChannelOptions(C.DownstreamChannelCapacity) +{ + FullMode = BoundedChannelFullMode.Wait, + SingleWriter = false, + SingleReader = true, +}); + +var doneFlag = new ManualResetEventSlim(false); +var state = new SharedState(); + +var ws = new ClientWebSocket(); +ws.Options.SetRequestHeader("Authorization", $"Bearer {apiKey}"); +ws.Options.SetRequestHeader("OpenAI-Beta", "realtime=v1"); + +Stream? inputStream = null; +Stream? outputStream = null; +Task? receiveTask = null; +Task? sendTask = null; + +try +{ + var uri = new Uri($"wss://api.openai.com/v1/realtime?model={C.ModelName}"); + Console.WriteLine($"[ws] connecting to {uri}"); + try + { + await ws.ConnectAsync(uri, cts.Token); + } + catch (Exception ex) + { + Console.Error.WriteLine($"[error] WS connect failed: {ex.Message}"); + if (ex.InnerException is HttpRequestException httpEx) + Console.Error.WriteLine($"[error] inner: HTTP {(int?)httpEx.StatusCode}"); + Environment.ExitCode = 3; + return; + } + Console.WriteLine("[ws] connected"); + + receiveTask = Task.Run(() => ReceiveLoop(ws, downstreamChannel, cts, state)); + + await SendSessionUpdate(ws, cts.Token); + + int device = FindUsbDevice(); + Console.WriteLine($"Using device {device} ('{PortAudio.GetDeviceInfo(device).name}')"); + + inputStream = OpenInputStream(device, upstreamChannel, state); + try { inputStream.Start(); } + catch (Exception ex) + { + Console.Error.WriteLine($"[error] input stream start failed: {ex.Message}"); + Environment.ExitCode = 5; + return; + } + + outputStream = OpenOutputStream(device, downstreamChannel, state, doneFlag); + try { outputStream.Start(); } + catch (Exception ex) + { + Console.Error.WriteLine($"[error] output stream start failed: {ex.Message}"); + Environment.ExitCode = 6; + return; + } + + sendTask = Task.Run(() => UpstreamSendLoop(ws, upstreamChannel, cts, state)); + + // Startup beep — queued onto downstreamChannel before the receive loop produces any audio. + short[] beep = MakeBeep(C.BeepHz, C.BeepSeconds, C.SampleRate); + Console.WriteLine("[beep]"); + await downstreamChannel.Writer.WriteAsync(beep, cts.Token); + + // Wait for beep to drain (200 ms beep + 100 ms margin) so it doesn't get captured + // by the mic and shipped to OpenAI. + Thread.Sleep((int)(C.BeepSeconds * 1000) + C.BeepDrainMs); + state.MicArmed = true; + Console.WriteLine("Speak now."); + + try + { + doneFlag.Wait(cts.Token); + } + catch (OperationCanceledException) { /* Ctrl-C path */ } + + Console.WriteLine("[ws] closing"); +} +finally +{ + // Cancel any background work first so receive/send loops exit cleanly. + cts.Cancel(); + + if (ws.State == WebSocketState.Open || ws.State == WebSocketState.CloseReceived) + { + try + { + using var closeTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(2)); + await ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "done", closeTimeout.Token); + } + catch (Exception ex) + { + Console.Error.WriteLine($"[ws] close failed: {ex.Message}"); + } + } + + try { inputStream?.Stop(); } catch { } + try { outputStream?.Stop(); } catch { } + try { inputStream?.Dispose(); } catch { } + try { outputStream?.Dispose(); } catch { } + try { ws.Dispose(); } catch { } + try { PortAudio.Terminate(); } catch { } + + Console.WriteLine("bye"); +} + +static string LoadApiKey() +{ + string home = Environment.GetEnvironmentVariable("HOME") ?? "/root"; + string path = Path.Combine(home, ".openai_key"); + if (!File.Exists(path)) + { + Console.Error.WriteLine($"[error] API key file not found at {path}"); + Environment.Exit(2); + } + + // Non-fatal mode-600 check. + try + { + var mode = File.GetUnixFileMode(path); + var leaky = mode & (UnixFileMode.GroupRead | UnixFileMode.GroupWrite | UnixFileMode.GroupExecute + | UnixFileMode.OtherRead | UnixFileMode.OtherWrite | UnixFileMode.OtherExecute); + if (leaky != UnixFileMode.None) + Console.Error.WriteLine($"[warn] {path} mode includes group/other perms ({mode}); expected 600"); + } + catch (PlatformNotSupportedException) { /* not Unix */ } + catch (Exception ex) + { + Console.Error.WriteLine($"[warn] could not check {path} mode: {ex.Message}"); + } + + string key = File.ReadAllText(path).Trim(); + if (string.IsNullOrEmpty(key)) + { + Console.Error.WriteLine($"[error] API key file empty"); + Environment.Exit(2); + } + if (!key.StartsWith("sk-")) + { + Console.Error.WriteLine($"[error] API key doesn't start with sk-"); + Environment.Exit(2); + } + return key; +} + +static int FindUsbDevice() +{ + int count = PortAudio.DeviceCount; + for (int i = 0; i < count; i++) + { + var info = PortAudio.GetDeviceInfo(i); + if (info.name.ToLowerInvariant().Contains("usb") && info.maxInputChannels >= 1) + return i; + } + Console.Error.WriteLine("USB audio device not found. Devices:"); + for (int i = 0; i < count; i++) + { + var info = PortAudio.GetDeviceInfo(i); + Console.Error.WriteLine( + $" [{i}] {info.name} in={info.maxInputChannels} out={info.maxOutputChannels}"); + } + Environment.Exit(1); + return -1; // unreachable +} + +static short[] MakeBeep(double freqHz, double durationS, int sampleRate) +{ + int n = (int)(sampleRate * durationS); + var buf = new short[n]; + for (int i = 0; i < n; i++) + { + double t = i / (double)sampleRate; + double v = 0.3 * Math.Sin(2.0 * Math.PI * freqHz * t); + buf[i] = (short)(v * short.MaxValue); + } + return buf; +} + +static async Task SendSessionUpdate(ClientWebSocket ws, CancellationToken ct) +{ + var msg = new + { + type = "session.update", + session = new + { + modalities = new[] { "audio", "text" }, + instructions = C.Instructions, + voice = C.Voice, + input_audio_format = "pcm16", + output_audio_format = "pcm16", + turn_detection = new + { + type = "server_vad", + threshold = 0.5, + prefix_padding_ms = 300, + silence_duration_ms = 500, + create_response = true, + }, + }, + }; + byte[] bytes = JsonSerializer.SerializeToUtf8Bytes(msg); + await ws.SendAsync(new ArraySegment(bytes), WebSocketMessageType.Text, endOfMessage: true, ct); +} + +static Stream OpenInputStream(int device, Channel upstream, SharedState state) +{ + var inParams = new StreamParameters + { + device = device, + channelCount = C.Channels, + sampleFormat = SampleFormat.Int16, + suggestedLatency = PortAudio.GetDeviceInfo(device).defaultLowInputLatency, + hostApiSpecificStreamInfo = IntPtr.Zero, + }; + + Stream.Callback callback = (IntPtr input, IntPtr _, uint frameCount, + ref StreamCallbackTimeInfo _2, StreamCallbackFlags status, IntPtr _3) => + { + if (status.HasFlag(StreamCallbackFlags.InputOverflow)) + Console.Error.WriteLine("[status] input overflow"); + + if (!state.MicArmed || state.StopSending) + return StreamCallbackResult.Continue; + + if (frameCount != C.MicBlockFrames) + { + Console.Error.WriteLine($"[status] unexpected input frameCount={frameCount} (want {C.MicBlockFrames})"); + return StreamCallbackResult.Continue; + } + + var frame = new short[C.MicBlockFrames]; + unsafe + { + short* src = (short*)input.ToPointer(); + fixed (short* dst = &frame[0]) + Buffer.MemoryCopy(src, dst, C.MicBlockFrames * sizeof(short), C.MicBlockFrames * sizeof(short)); + } + if (!upstream.Writer.TryWrite(frame)) + Console.Error.WriteLine("[status] upstream behind, dropping mic frame"); + return StreamCallbackResult.Continue; + }; + + return new Stream(inParams, null, C.SampleRate, C.MicBlockFrames, StreamFlags.NoFlag, callback, IntPtr.Zero); +} + +static Stream OpenOutputStream(int device, Channel downstream, SharedState state, ManualResetEventSlim doneFlag) +{ + var outParams = new StreamParameters + { + device = device, + channelCount = C.Channels, + sampleFormat = SampleFormat.Int16, + suggestedLatency = PortAudio.GetDeviceInfo(device).defaultLowOutputLatency, + hostApiSpecificStreamInfo = IntPtr.Zero, + }; + + Stream.Callback callback = (IntPtr _, IntPtr output, uint frameCount, + ref StreamCallbackTimeInfo _2, StreamCallbackFlags _3, IntPtr _4) => + { + int fill = 0; + + while (fill < frameCount) + { + // Pull next chunk if we have no working one. + if (state.CurrentChunk == null) + { + if (!downstream.Reader.TryRead(out var nextChunk)) + break; // ring empty + state.CurrentChunk = nextChunk; + state.ChunkOffset = 0; + } + // Local non-null alias so the compiler doesn't warn on dereference inside `fixed`. + short[] chunk = state.CurrentChunk!; + int slack = (int)frameCount - fill; + int remaining = chunk.Length - state.ChunkOffset; + int take = Math.Min(remaining, slack); + unsafe + { + short* dst = (short*)output.ToPointer(); + fixed (short* src = &chunk[state.ChunkOffset]) + Buffer.MemoryCopy(src, dst + fill, take * sizeof(short), take * sizeof(short)); + } + state.ChunkOffset += take; + fill += take; + if (state.ChunkOffset >= chunk.Length) + state.CurrentChunk = null; + } + + unsafe + { + short* dst = (short*)output.ToPointer(); + for (int i = fill; i < (int)frameCount; i++) dst[i] = 0; + } + + // Completion = working chunk drained AND channel empty AND writer completed. + if (state.CurrentChunk == null + && downstream.Reader.Count == 0 + && downstream.Reader.Completion.IsCompleted) + { + doneFlag.Set(); + return StreamCallbackResult.Complete; + } + return StreamCallbackResult.Continue; + }; + + return new Stream(null, outParams, C.SampleRate, C.OutBlockFrames, StreamFlags.NoFlag, callback, IntPtr.Zero); +} + +static async Task UpstreamSendLoop(ClientWebSocket ws, Channel upstream, CancellationTokenSource cts, SharedState state) +{ + try + { + await foreach (var frame in upstream.Reader.ReadAllAsync(cts.Token)) + { + if (state.StopSending) continue; + ReadOnlySpan bytes = MemoryMarshal.AsBytes(frame.AsSpan()); + string base64 = Convert.ToBase64String(bytes); + var msg = new { type = "input_audio_buffer.append", audio = base64 }; + byte[] json = JsonSerializer.SerializeToUtf8Bytes(msg); + await ws.SendAsync(new ArraySegment(json), WebSocketMessageType.Text, endOfMessage: true, cts.Token); + } + } + catch (OperationCanceledException) { /* expected on Cancel */ } + catch (Exception ex) + { + Console.Error.WriteLine($"[error] upstream send: {ex.Message}"); + cts.Cancel(); + } +} + +static async Task ReceiveLoop(ClientWebSocket ws, Channel downstream, CancellationTokenSource cts, SharedState state) +{ + var transcript = new StringBuilder(); + var buffer = new byte[4096]; + bool firstDeltaSeen = false; + DateTime speechStoppedTime = DateTime.MinValue; + using var msgBuffer = new MemoryStream(); + + try + { + while (!cts.IsCancellationRequested) + { + msgBuffer.SetLength(0); + WebSocketReceiveResult result; + do + { + result = await ws.ReceiveAsync(new ArraySegment(buffer), cts.Token); + if (result.MessageType == WebSocketMessageType.Close) return; + msgBuffer.Write(buffer, 0, result.Count); + } while (!result.EndOfMessage); + + msgBuffer.Position = 0; + using var doc = JsonDocument.Parse(msgBuffer); + string type = doc.RootElement.GetProperty("type").GetString() ?? ""; + + switch (type) + { + case "session.created": + { + string? sid = doc.RootElement.TryGetProperty("session", out var sess) + && sess.TryGetProperty("id", out var idEl) ? idEl.GetString() : null; + Console.WriteLine($"[ws] session.created {sid}"); + break; + } + case "session.updated": + Console.WriteLine("[ws] session.updated — session pinned"); + break; + case "input_audio_buffer.speech_started": + Console.WriteLine("[vad] speech started"); + break; + case "input_audio_buffer.speech_stopped": + Console.WriteLine("[vad] speech stopped"); + speechStoppedTime = DateTime.UtcNow; + state.StopSending = true; + break; + case "input_audio_buffer.committed": + Console.WriteLine("[ws] input_audio_buffer.committed"); + break; + case "response.created": + Console.WriteLine("[ws] response.created"); + break; + case "response.audio_transcript.delta": + if (doc.RootElement.TryGetProperty("delta", out var tEl)) + transcript.Append(tEl.GetString()); + break; + case "response.audio.delta": + { + string b64 = doc.RootElement.GetProperty("delta").GetString() ?? ""; + byte[] audioBytes = Convert.FromBase64String(b64); + short[] samples = new short[audioBytes.Length / 2]; + Buffer.BlockCopy(audioBytes, 0, samples, 0, audioBytes.Length); + if (!firstDeltaSeen) + { + firstDeltaSeen = true; + double elapsed = speechStoppedTime == DateTime.MinValue + ? -1 + : (DateTime.UtcNow - speechStoppedTime).TotalSeconds; + Console.WriteLine($"[audio] first delta received (t = {elapsed:0.00} s since speech_stopped)"); + } + await downstream.Writer.WriteAsync(samples, cts.Token); + break; + } + case "response.audio.done": + Console.WriteLine("[ws] response.audio.done"); + break; + case "response.done": + state.NoMoreDeltas = true; + downstream.Writer.Complete(); + Console.WriteLine("[ws] response.done"); + Console.WriteLine($"[transcript] {transcript}"); + break; + case "error": + Console.Error.WriteLine($"[error] WS error event: {doc.RootElement.GetRawText()}"); + cts.Cancel(); + break; + default: + Console.Error.WriteLine($"[ws] ignored type={type}"); + break; + } + } + } + catch (OperationCanceledException) { /* expected on Cancel */ } + catch (Exception ex) + { + Console.Error.WriteLine($"[error] receive loop: {ex.Message}"); + cts.Cancel(); + } +} + +sealed class SharedState +{ + public volatile bool MicArmed; + public volatile bool StopSending; + public volatile bool NoMoreDeltas; + // Output-callback-thread-local; do NOT touch from anywhere else. + public short[]? CurrentChunk; + public int ChunkOffset; +} + +static class C +{ + public const int SampleRate = 24_000; + public const int Channels = 1; + public const uint MicBlockFrames = 1920; // 40 ms @ 24 kHz + public const uint OutBlockFrames = 1024; // ~43 ms @ 24 kHz + public const string ModelName = "gpt-realtime"; + public const string Voice = "alloy"; + public const string Instructions = "Reply in one short sentence."; + public const double BeepHz = 880.0; + public const double BeepSeconds = 0.2; + public const int BeepDrainMs = 100; + public const int UpstreamChannelCapacity = 64; + public const int DownstreamChannelCapacity = 64; +} + +static class Libc +{ + [DllImport("libc", EntryPoint = "setenv")] + public static extern int setenv(string name, string value, int overwrite); +}