Simple solution for streaming audio from Gqrx to Firefox.

gqrx_audio_stream.py

""" Simple solution for streaming audio from Gqrx to Firefox.

Requires the opusenc command line tool.

This program uses the fact that Opus files can be concatenated to form a valid
stream and that Firefox can play these streams natively. The drawback is the
overhead created by repeatedly inserting containers and metadata into the
stream.

- Check https://fastapi.tiangolo.com/tutorial/ to see how to run the server.
- Receive some audio with Gqrx, activate "Mute" and "UDP" in the Audio view.
- Open the server URL in Firefox.
"""

import asyncio
import io
import subprocess

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

class Encoder(asyncio.DatagramProtocol):
    chunk_duration = 1.0

    def __init__(self):
        self.audio_buffer_lock = asyncio.Lock()
        self.audio_buffer = io.BytesIO()
        self.new_chunk = asyncio.Condition()
        self.opus_chunk = None
        self.worker = asyncio.create_task(self.encode())

    async def encode(self):
        while True:
            await asyncio.sleep(self.chunk_duration)
            # Gqrx output format is documented in
            # https://gqrx.dk/doc/streaming-audio-over-udp
            proc = await asyncio.create_subprocess_shell(
                ('opusenc --raw --raw-bits 16 --raw-rate 48000 --raw-chan 1 '
                 '--raw-endianness 0 - -'),
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE)
            async with self.audio_buffer_lock:
                proc.stdin.write(self.audio_buffer.getvalue())
                proc.stdin.close()
                self.audio_buffer.close()
                self.audio_buffer = io.BytesIO()
            await proc.wait()
            async with self.new_chunk:
                self.opus_chunk = await proc.stdout.read()
                self.new_chunk.notify_all()

    async def write(self, data):
        async with self.audio_buffer_lock:
            self.audio_buffer.write(data)

    def datagram_received(self, data, addr):
        asyncio.create_task(self.write(data))

    async def stream(self):
        while True:
            async with self.new_chunk:
                await self.new_chunk.wait()
                yield self.opus_chunk

ENCODER = None

app = FastAPI()

@app.on_event('startup')
async def startup():
    global ENCODER
    ENCODER = Encoder()
    await asyncio.get_event_loop().create_datagram_endpoint(
        lambda: ENCODER, local_addr=('localhost', 7355))

@app.on_event('shutdown')
async def shutdown():
    ENCODER.worker.cancel()

@app.get('/')
async def stream():
   return StreamingResponse(ENCODER.stream(), media_type='audio/ogg')