Linux 开发板接入OpenAI 实时语音聊天实战

.ConfigureAppConfiguration((hostingContext, config) =>
{
    var env = hostingContext.HostingEnvironment;
    config.AddJsonFile("appsettings.json", optional: true, reloadOnChange: true)
          .AddJsonFile($"appsettings.{env.EnvironmentName}.json", optional: true, reloadOnChange: true);
    config.AddEnvironmentVariables();
})

.ConfigureServices((context, services) =>
{
    services.Configure<RealtimeAPIOptions>(context.Configuration.GetSection("RealtimeAPI"));
    services.Configure<SessionUpdateOptions>(context.Configuration.GetSection("SessionUpdate"));
    services.Configure<AudioSettings>(context.Configuration.GetSection("AudioSettings"));
    services.Configure<LcdSettings>(context.Configuration.GetSection("LcdSettings"));
    services.AddSingleton<AudioService>();
    services.AddSingleton<WebSocketService>();
    services.AddSingleton<LcdService>();
    services.AddHostedService<Worker>();
})

private async Task ReceiveLoopAsync(CancellationToken cancellationToken)
{
    var buffer = new byte[1024 * 4];
    var messageBuffer = new List<byte>();

    while (_client.State == WebSocketState.Open && !cancellationToken.IsCancellationRequested)
    {
        var result = await _client.ReceiveAsync(new ArraySegment<byte>(buffer), cancellationToken);
        if (result.MessageType == WebSocketMessageType.Close)
        {
            await _client.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", cancellationToken);
            _logger.LogInformation("WebSocket closed!");

            await OnDisconnected.Invoke();
        }
        else
        {
            messageBuffer.AddRange(buffer.Take(result.Count));

            if (result.EndOfMessage)
            {
                var message = Encoding.UTF8.GetString(messageBuffer.ToArray());
                await OnMessageReceived.Invoke(message);
                _logger.LogDebug($"Received: {message}");
                messageBuffer.Clear();
            }
        }
    }
}

public async Task StartRecordingAsync()
{
    if (_isRecording)
    {
        return;
    }
    _cancellationTokenSourceRecording = new CancellationTokenSource();
    _isRecording = true;
    await Task.Run(() => _alsaDevice.Record(async (data) => 
    {
        if (OnAudioDataAvailable != null)
        {
            if (!hasDiscardedWavHeader)
            {
                hasDiscardedWavHeader = true;
                return;
            }
            if (!_isRecording)
            {
                return;
            }
            data = ConvertStereoToMono(data); // 将数据转换为单声道
            _packetQueue.Enqueue(data); // 将数据包加入队列

            if (_packetQueue.Count >= PacketThreshold)
            {
                await SendPackets(); // 当队列中的数据包数量达到阈值时发送数据包
            }
        }
    }, _cancellationTokenSourceRecording.Token), _cancellationTokenSourceRecording.Token);
}

private async Task SendPackets()
{
    if (!_isRecording)
    {
        return;
    }

    int totalSize = PacketThreshold * PacketSize;
    byte[] combinedPacket = new byte[totalSize];
    int offset = 0;

    while (_packetQueue.Count > 0 && offset + PacketSize <= totalSize)
    {
        byte[] packet = _packetQueue.Dequeue();
        if (packet.Length == PacketSize)
        {
            Buffer.BlockCopy(packet, 0, combinedPacket, offset, PacketSize);
            offset += PacketSize;
        }
    }

    if (OnAudioDataAvailable != null && offset > 0)
    {
        await OnAudioDataAvailable(combinedPacket);
    }
}

{
    "type":"error",
    "event_id":"event_BDv6vl8eCfBloxsnlJ74U",
    "error":{
        "type":"invalid_request_error",
        "code":"invalid_value",
        "message":"Invalid 'audio'. Expected base64-encoded audio bytes (mono PCM16 at 24kHz) but got an invalid value.",
        "param":"audio.audio",
        "event_id":null
    }
}

import asyncio
import websockets
import json
import base64
import wave
import numpy as np

buffer = []
save_count = 0
async def process_audio_data():
    global buffer
    global save_count
    if len(buffer) >= 50:
        print("Processing audio data...")
        # 组合缓冲区中的数据
        audio_data = b''.join(buffer)

        # 存储一个buffer 测试
        with open(f'data/output{save_count}.pcm', 'wb') as f:
            f.write(audio_data)

        buffer = []  # 清空缓冲区

        # 将 PCM 数据写入 WAV 文件
        with wave.open(f'data/output{save_count}.wav', 'wb') as wf:
            wf.setnchannels(1)  # 单声道
            wf.setsampwidth(2)  # 16位
            wf.setframerate(24000)  # 24000 Hz
            wf.writeframes(audio_data)
            save_count += 1
        print("Audio data saved to output.wav")

async def handler(websocket):
    global buffer
    print("Client connected")

    async def ping():
        while True:
            try:
                await websocket.send(json.dumps({"type": "ping"}))
                await asyncio.sleep(10)  # 每10秒发送一次ping
            except websockets.ConnectionClosed:
                break

    asyncio.create_task(ping())

    async for message in websocket:
        # 记录接收到的数据到文件
        with open('data/received_data.txt', 'a') as f:
            f.write(message + '\n')
        data = json.loads(message)
        if data['type'] == 'input_audio_buffer.append':
            print("Received audio buffer data")
            pcm_data = base64.b64decode(data['audio'])
            buffer.append(pcm_data)
            await process_audio_data()
    print("Client disconnected")

async def main():
    print("Starting server...")
    async with websockets.serve(handler, "0.0.0.0", 8765):
        print("Server started on ws://0.0.0.0:8765")
        await asyncio.Future()  # run forever

if __name__ == "__main__":
    asyncio.run(main())

amixer cset name='ADC HPF Cut-off' 'On'

/// <summary>
/// 播放队列
/// </summary>
private readonly ConcurrentQueue<byte[]> _playbackQueue = new ConcurrentQueue<byte[]>();
private bool _isPlaying = false;

public async Task PlayAudioAsync(byte[] pcmData)
{
    var monoData = ConvertMonoToStereo(pcmData);
    _playbackQueue.Enqueue(monoData);

    if (!_isPlaying)
    {
        _isPlaying = true;
        await Task.Run(() => ProcessPlaybackQueue());
    }
}

private void ProcessPlaybackQueue()
{
    while (_playbackQueue.TryDequeue(out var pcmData))
    {
        Play(pcmData);
    }
    _isPlaying = false;
    OnPlaybackStopped?.Invoke();
}

private void Play(byte[] pcmData)
{
    var wavData = CreateWavHeader(24000, 16, 2).Concat(pcmData).ToArray();
    //using var alsaDevice = AlsaDeviceBuilder.Create(_settings);
    _alsaDevice.Play(new MemoryStream(wavData), _cancellationTokenSourcePlayback.Token);
}

private void StopPlayback()
{
    _cancellationTokenSourcePlayback?.Cancel();
    _cancellationTokenSourcePlayback = new CancellationTokenSource();
    _isPlaying = false;
    _playbackQueue.Clear();
}

public async Task SaveAudio(string audioId, byte[] pcmData)
{
    string filePath = $"audio/{audioId}.wav";
    var monoData = ConvertMonoToStereo(pcmData);
    byte[] wavData;
    if (!File.Exists(filePath))
    {
        byte[] wavHeader = CreateWavHeader(24000, 16, 2);
        wavData = wavHeader.Concat(monoData).ToArray();
        await File.WriteAllBytesAsync(filePath, wavData);
    }
    else
    {
        using (var fileStream = new FileStream(filePath, FileMode.Append, FileAccess.Write))
        {
            await fileStream.WriteAsync(monoData, 0, monoData.Length);
        }
    }
}

public async Task PlayAudio(string audioId)
{
    string filePath = $"audio/{audioId}.wav";
    if (!File.Exists(filePath))
    {
        return;
    }
    await Task.Run(() =>{
        //using var alsaDevice = AlsaDeviceBuilder.Create(_settings);
        _alsaDevice.Play(filePath, _cancellationTokenSourcePlayback.Token);
        OnPlaybackStopped?.Invoke();
    }, _cancellationTokenSourcePlayback.Token); 
}