10.3. Livestreaming – en tittare

Webbläsare kan rendera flerdelade Motion JPEG-strömmar (MJPEG) direkt inuti en <img>-tagg. Ge webbläsaren ett HTTP-svar som aldrig avslutas, skriv JPEG-bilder åtskilda av en flerdelad gräns, och webbläsaren visar varje bildruta allteftersom den anländer.

Ledningen är okomplicerad: en svarsheader, Content-Type: multipart/x-mixed-replace; boundary=frame, sedan en --frame-rad, Content-Type: image/jpeg, en tom rad, JPEG-byten, \r\n och så vidare. Webbläsaren stänger anslutningen när <img> tas bort eller fliken stängs.

10.3.1. Att fånga utan att blockera

Det blockerande csi0.snapshot() som använts hittills stoppar hela händelseloopen tills sensorn levererar en bildruta. Det var bra när en begäran utlöste en stillbild och inget annat kördes. När en ström väl är öppen måste servern fortsätta hantera andra begäranden medan nästa bildruta fångas – fångstanropet behöver ge efter till händelseloopen medan det väntar på sensorn.

Mönstret är en tunn AsyncCSI-omslutning som pollar csi.CSI.snapshot() i icke-blockerande läge och får coroutinen att sova mellan pollningar. Asyncio-kapitlet gick igenom det här mönstret i AsyncCSI; bädda in det i skriptet tills vidare:

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)

Varje annan CSI-metod (reset(), pixformat(), framesize(), gain_db(), …) vidarebefordras genom __getattr__; endast snapshot() ersätts med en awaitable version som låter händelseloopen schemalägga andra coroutiner mellan pollningar.

Byt ut det bara csi.CSI() från stillbildsrutten mot ett AsyncCSI():

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

10.3.2. Strömmande kroppar är klassbaserade iteratorer

En strömmande svarskropp är bara ett objekt som microdot itererar med async for och skickar varje genererad bit ner i socketen. På CPython är detta normalt en asynkron generatorfunktion – async def med yield. MicroPython stöder inte det:

Anteckning

MicroPythons asyncio stöder inte asynkrona generatorfunktioner (async def name(): ... yield ...). Strömmande svarskroppar måste vara klassbaserade asynkrona iteratorer med __aiter__ som returnerar self och __anext__ definierad som async def.

För en MJPEG-ström betyder det en klass vars __anext__ inväntar en bildruta och returnerar den inramad i den flerdelade omslutningen:

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

Instansen är ny per begäran, så varje ansluten klient får sin egen iterator. När webbläsaren kopplar från slutar microdot invänta __anext__ och iteratorn skräpsamlas.

Anteckning

Omslutningen bytes(...) runt JPEG:en är defensiv. bytearray() returnerar en vy in i kamerans bildbuffert, och nästa snapshot()-anrop skriver om den bufferten på plats. Att omsluta i bytes kopierar ut JPEG:en så att biten som microdot är mitt i att skriva förblir stabil även om skrivarens tömning inte har slutförts när __anext__ körs igen.

10.3.3. Att köra servern inuti asyncio

Det tidigare app.run(host=..., port=...)-anropet är blockerande. MJPEG-hanteraren behöver dela loopen med AsyncCSI:s stillbildspollningar, så byt ut app.run mot start_server() inuti en asyncio.run():

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

asyncio.run(main())

Omslutningen asyncio.run() låter servern vara en uppgift bland flera – main-coroutinen är då det naturliga stället att starta fångst, rörelsedetektering och allt annat som måste dela loopen med HTTP-servern.

10.3.4. En tittare i taget

Varje ansluten klient kör sin egen FrameStream-iterator, vilket betyder att varje klient utlöser sitt eget csi0.snapshot()-anrop. Två webbläsare betyder två sensorläsningar per bildruteintervall, tre betyder tre, och så vidare. Sensorn kan faktiskt inte leverera bildrutor snabbare än sin egen bildfrekvens, så begärandena köar upp sig bakom varandra och allas ström saktar ner.

Lösningen är en enda delad fångstloop som publicerar en bildruta till många läsare.