10.3. Transmissão em direto – um único espetador

Os browsers conseguem renderizar streams Motion JPEG (MJPEG) diretamente dentro de uma tag <img>. Entregue ao browser uma resposta HTTP que nunca termina, escreva JPEGs separados por um limite multipart, e o browser apresenta cada fotograma à medida que chega.

A transmissão é direta: um cabeçalho de resposta, Content-Type: multipart/x-mixed-replace; boundary=frame, depois uma linha --frame, Content-Type: image/jpeg, uma linha em branco, os bytes JPEG, \r\n, e repetir. O browser fecha a ligação quando a <img> é removida ou o separador é fechado.

10.3.1. Capturar sem bloquear

O csi0.snapshot() bloqueante utilizado até agora paralisa todo o ciclo de eventos até o sensor entregar um fotograma. Isso era adequado quando um único pedido disparava uma captura e mais nada estava em execução. Assim que uma stream está aberta, o servidor tem de continuar a tratar outros pedidos enquanto o próximo fotograma é capturado – a chamada de captura tem de ceder ao ciclo de eventos enquanto aguarda o sensor.

O padrão é um invólucro AsyncCSI simples que consulta csi.CSI.snapshot() em modo não bloqueante e suspende a corrotina entre consultas. O capítulo de asyncio percorreu este padrão em AsyncCSI; integre-o no script por agora:

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)

Todos os outros métodos CSI (reset(), pixformat(), framesize(), gain_db(), …) são reencaminhados através de __getattr__; apenas snapshot() é substituído por uma versão aguardável que permite ao ciclo de eventos agendar outras corrotinas entre consultas.

Substitua o csi.CSI() simples da rota de captura por um AsyncCSI():

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

10.3.2. Os corpos de resposta em streaming são iteradores baseados em classes

Um corpo de resposta em streaming é simplesmente um objeto que o microdot itera com async for, enviando cada bloco produzido pelo socket. No CPython isto é normalmente uma função geradora assíncrona – async def com yield. O MicroPython não suporta isso:

Nota

O asyncio do MicroPython não suporta funções geradoras assíncronas (async def name(): ... yield ...). Os corpos de resposta em streaming têm de ser iteradores assíncronos baseados em classes com __aiter__ a devolver self e __anext__ definido como async def.

Para uma stream MJPEG isso significa uma classe cujo __anext__ aguarda um fotograma e o devolve embrulhado no invólucro multipart:

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,
        },
    )

A instância é nova por pedido, pelo que cada cliente ligado obtém o seu próprio iterador. Quando o browser desliga, o microdot para de aguardar __anext__ e o iterador é eliminado pelo coletor de lixo.

Nota

O embrulho bytes(...) em redor do JPEG é defensivo. bytearray() devolve uma vista para o buffer de imagem da câmara, e a próxima chamada a snapshot() reescreve esse buffer no local. Embrulhar em bytes copia o JPEG para fora, de modo que o bloco que o microdot está a meio de escrever se mantém estável mesmo que o flush do escritor não tenha terminado quando __anext__ voltar a ser executado.

10.3.3. Executar o servidor dentro do asyncio

A chamada anterior app.run(host=..., port=...) é bloqueante. O handler MJPEG precisa de partilhar o ciclo com as consultas de captura do AsyncCSI, por isso substitua app.run por start_server() dentro de um asyncio.run():

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

asyncio.run(main())

O invólucro asyncio.run() permite que o servidor seja uma tarefa entre várias – a corrotina main é então o lugar natural para lançar captura, deteção de movimento e tudo o mais que tenha de partilhar o ciclo com o servidor HTTP.

10.3.4. Um espetador de cada vez

Cada cliente ligado executa o seu próprio iterador FrameStream, o que significa que cada cliente despoleta a sua própria chamada a csi0.snapshot(). Dois browsers significam duas leituras do sensor por intervalo de fotograma, três significam três, e assim por diante. O sensor não consegue realmente entregar fotogramas mais depressa do que a sua própria taxa de fotogramas, pelo que os pedidos se acumulam uns atrás dos outros e a stream de todos abranda.

A solução é um único ciclo de captura partilhado que publica um fotograma para muitos leitores.