10.3. Transmisión en vivo – un solo espectador

Los navegadores pueden renderizar transmisiones multiparte de Motion JPEG (MJPEG) directamente dentro de una etiqueta <img>. Entrégale al navegador una respuesta HTTP que nunca termina, escribe JPEGs separados por un límite multiparte, y el navegador muestra cada fotograma a medida que llega.

El canal es sencillo: una cabecera de respuesta, Content-Type: multipart/x-mixed-replace; boundary=frame, luego una línea --frame, Content-Type: image/jpeg, una línea en blanco, los bytes JPEG, \r\n, y se repite. El navegador cierra la conexión cuando se elimina el <img> o se cierra la pestaña.

10.3.1. Capturar sin bloquear

La llamada bloqueante csi0.snapshot() usada hasta ahora detiene todo el bucle de eventos hasta que el sensor entrega un fotograma. Eso estaba bien cuando una petición disparaba una captura y nada más se estaba ejecutando. Una vez que una transmisión está abierta, el servidor tiene que seguir atendiendo otras peticiones mientras se captura el siguiente fotograma – la llamada de captura necesita ceder el control al bucle de eventos mientras espera al sensor.

El patrón es un fino envoltorio AsyncCSI que sondea csi.CSI.snapshot() en modo no bloqueante y duerme la corrutina entre sondeos. El capítulo de asyncio recorrió este patrón en AsyncCSI; por ahora, incorpóralo directamente al script:

import asyncio

class AsyncCSI:
    def __init__(self, *args, **kwargs):
        self._csi = csi.CSI(*args, **kwargs)

    def __getattr__(self, name):
        return getattr(self._csi, name)

    async def snapshot(self):
        while True:
            img = self._csi.snapshot(blocking=False)
            if img is not None:
                return img
            await asyncio.sleep_ms(0)

Cualquier otro método de CSI (reset(), pixformat(), framesize(), gain_db(), …) se reenvía a través de __getattr__; solo snapshot() se reemplaza por una versión esperable (awaitable) que permite al bucle de eventos planificar otras corrutinas entre sondeos.

Sustituye el csi.CSI() desnudo de la ruta de captura por un AsyncCSI():

csi0 = AsyncCSI()
csi0.reset()
csi0.pixformat(csi.RGB565)
csi0.framesize(csi.QVGA)

10.3.2. Los cuerpos de transmisión son iteradores basados en clases

Un cuerpo de respuesta de transmisión es simplemente un objeto sobre el que microdot itera con async for, enviando cada fragmento producido por el socket. En CPython esto normalmente es una función generadora asíncrona – async def con yield. MicroPython no admite eso:

Nota

El módulo asyncio de MicroPython no admite funciones generadoras asíncronas (async def name(): ... yield ...). Los cuerpos de respuesta de transmisión deben ser iteradores asíncronos basados en clases con __aiter__ devolviendo self y __anext__ definido como async def.

Para una transmisión MJPEG eso significa una clase cuyo __anext__ espera un fotograma y lo devuelve enmarcado en el envoltorio multiparte:

BOUNDARY = b'frame'

class FrameStream:
    def __aiter__(self):
        return self

    async def __anext__(self):
        img = await csi0.snapshot()
        jpeg = bytes(img.compress(quality=85).bytearray())
        return (b'--' + BOUNDARY + b'\r\n'
                b'Content-Type: image/jpeg\r\n\r\n'
                + jpeg + b'\r\n')

@app.get('/stream.jpg')
async def stream(request):
    return Response(
        body=FrameStream(),
        headers={
            'Content-Type':
                b'multipart/x-mixed-replace; boundary=' + BOUNDARY,
        },
    )

La instancia es nueva por cada petición, así que cada cliente conectado obtiene su propio iterador. Cuando el navegador se desconecta, microdot deja de esperar a __anext__ y el iterador es recolectado por el recolector de basura.

Nota

La envoltura bytes(...) alrededor del JPEG es defensiva. bytearray() devuelve una vista del búfer de imagen de la cámara, y la siguiente llamada a snapshot() reescribe ese búfer in situ. Envolver en bytes copia el JPEG hacia fuera, de modo que el fragmento que microdot está escribiendo permanece estable incluso si el vaciado del escritor no ha terminado para cuando __anext__ se ejecuta de nuevo.

10.3.3. Ejecutar el servidor dentro de asyncio

La llamada anterior app.run(host=..., port=...) es bloqueante. El manejador de MJPEG necesita compartir el bucle con los sondeos de captura de AsyncCSI, así que sustituye app.run por start_server() dentro de un asyncio.run():

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

asyncio.run(main())

El envoltorio asyncio.run() permite que el servidor sea una tarea entre varias – la corrutina main es entonces el lugar natural para lanzar la captura, la detección de movimiento y cualquier otra cosa que tenga que compartir el bucle con el servidor HTTP.

10.3.4. Un espectador a la vez

Cada cliente conectado ejecuta su propio iterador FrameStream, lo que significa que cada cliente dispara su propia llamada csi0.snapshot(). Dos navegadores significan dos lecturas del sensor por intervalo de fotograma, tres significan tres, y así sucesivamente. El sensor no puede realmente entregar fotogramas más rápido que su propia tasa de fotogramas, así que las peticiones se encolan unas detrás de otras y la transmisión de todos se ralentiza.

La solución es un único bucle de captura compartido que publica un fotograma para muchos lectores.