10.4. Compartir un único bucle de captura entre varios espectadores

Que cada cliente conectado llame a csi0.snapshot() de forma independiente es un derroche, y una vez que hay dos transmisiones abiertas a la vez la cosa empeora: el sensor entrega fotogramas a su propia tasa, y cada captura duplicada ralentiza a todos. El enfoque correcto es una sola corrutina de captura que publica «el fotograma más reciente» en una ranura compartida, más iteradores por cliente que leen de la ranura.

10.4.1. La tarea de captura

Una corrutina en segundo plano toma fotogramas tan rápido como el sensor los entrega, comprime cada uno a JPEG en un bytes compartido, y emite un pulso de evento para que cualquier cliente en espera se despierte:

latest_jpeg = None
new_frame = asyncio.Event()

async def capture_loop():
    global latest_jpeg
    while True:
        img = await csi0.snapshot()
        latest_jpeg = bytes(img.compress(quality=85).bytearray())
        new_frame.set()
        new_frame.clear()

El par set() / clear() es el patrón de pulso. set() desbloquea de golpe a todas las corrutinas que esperan actualmente al evento; clear() reinicia inmediatamente el evento para que el siguiente wait() vuelva a bloquear. Con múltiples consumidores (un espectador, otro espectador, cualquier otra corrutina que necesite reaccionar a un nuevo fotograma), ningún consumidor concreto es responsable de reiniciar el evento, y nadie le roba un despertar a nadie.

Nota

La envoltura bytes(...) alrededor del JPEG es fundamental aquí. bytearray() devuelve una vista del búfer de imagen de la cámara; la siguiente llamada a snapshot() reescribe ese búfer in situ con el siguiente fotograma. latest_jpeg sobrevive al img local, así que sin la copia cada lector vería la ranura cambiar bajo sus pies en cada captura.

10.4.2. Los iteradores por cliente leen de la ranura

El manejador de la transmisión MJPEG deja de llamar a csi0.snapshot() por sí mismo. En su lugar, cada instancia de FrameStream espera al evento compartido y lee de los bytes compartidos:

class FrameStream:
    # One instance per connected client. Each one independently
    # waits on the shared new_frame pulse; the capture loop is
    # responsible for resetting the event between frames.

    def __aiter__(self):
        return self

    async def __anext__(self):
        await new_frame.wait()
        if latest_jpeg is None:
            return b''
        return (b'--' + BOUNDARY + b'\r\n'
                b'Content-Type: image/jpeg\r\n\r\n'
                + latest_jpeg + b'\r\n')

La ruta de captura también cambia: ya no dispara una captura, sino que devuelve lo que sea que latest_jpeg contenga en ese momento:

@app.get('/snapshot.jpg')
async def snapshot(request):
    if latest_jpeg is None:
        return 'no frame yet', 503
    return Response(
        body=latest_jpeg,
        headers={'Content-Type': 'image/jpeg'},
    )

La tupla (body, status) es la forma abreviada de microdot para establecer un código de estado HTTP sin construir una microdot.Response. 503 dice estoy aquí pero no listo – el código correcto para «vuelve a preguntar en un momento».

10.4.3. Ejecutar la captura junto al servidor

main tiene ahora dos corrutinas de nivel superior: el bucle de captura y el servidor HTTP. asyncio.gather() ejecuta ambas, y si alguna falla la otra se cancela:

async def main():
    await asyncio.gather(
        capture_loop(),
        app.start_server(host='0.0.0.0', port=80),
    )

asyncio.run(main())

Ahora el sensor lee un fotograma por ciclo sin importar cuántos espectadores estén conectados. El primer navegador en ir a /stream.jpg ve fotogramas; también lo hace el segundo, el tercero, el décimo – todos comparten la misma captura, y la cámara sigue igual de receptiva en sus otras rutas.