如何在Python中从服务器以音频流发送ICY格式消息？

Question

我使用来自异步生成器源的音频字节在 FastAPI 中实现了 StreamingResponse。但除了需要在流中插入一些客户端音频播放器（目前是 React Native）的消息。 了解 ICY 格式，它看起来适合于此。那么流端点需要什么标头以及什么格式以及如何编码（base64 或 utf-8）音频播放器的消息来触发事件（如

Event.MetadataCommonReceived

）？

@router.get(
    "/stream/{session_id}",
    response_class=StreamingResponse,
    responses={200: {"content": {"audio/mpeg": {}}, "description": "An audio file in MP3 format"}},
)
async def stream_audio(session_id: str):
...
    return StreamingResponse(
        stream_from_queue(<some asyncio.Queue>, session_id),
        headers={
            "content-type": "audio/mpeg",
            "icy-metaint": "16000",
            "icy-name": "MyAudioStream",
            "icy-genre": "Podcast",
            "icy-url": "http://localhost:8000"
        },
        media_type="audio/mpeg",
    )

async def stream_from_queue(queue: Queue, session_id: str):
    ...
    yield audio_bytes # send an audio chunk from queue
    ...
    <I need periodically send a message for audio Event>
    yield "some message".encode("utf-8") # ??

Answer 1

阅读完https://gist.github.com/niko/2a1d7b2d109ebe7f7ca2f860c3505ef0后，您的方法

stream_from_queue

似乎需要用您想要发送的元数据信息填充您将

yield

的初始消息。

1. 计算元数据中使用了多少字节。
2. 将它们添加到下一条消息
3. 发送后，您可以继续传输数据

我知道编码是ascii