Test 4a: WS plumbing + audio streams + server VAD round-trip
This commit is contained in:
@@ -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<short[]>(new BoundedChannelOptions(C.UpstreamChannelCapacity)
|
||||
{
|
||||
FullMode = BoundedChannelFullMode.Wait,
|
||||
SingleWriter = true,
|
||||
SingleReader = true,
|
||||
});
|
||||
var downstreamChannel = Channel.CreateBounded<short[]>(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<byte>(bytes), WebSocketMessageType.Text, endOfMessage: true, ct);
|
||||
}
|
||||
|
||||
static Stream OpenInputStream(int device, Channel<short[]> 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<short[]> 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<short[]> upstream, CancellationTokenSource cts, SharedState state)
|
||||
{
|
||||
try
|
||||
{
|
||||
await foreach (var frame in upstream.Reader.ReadAllAsync(cts.Token))
|
||||
{
|
||||
if (state.StopSending) continue;
|
||||
ReadOnlySpan<byte> 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<byte>(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<short[]> 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<byte>(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);
|
||||
}
|
||||
Reference in New Issue
Block a user