10.4. Eén vastleglus delen over kijkers

Het is verspillend dat elke verbonden client onafhankelijk csi0.snapshot() aanroept, en zodra er twee streams tegelijk open zijn wordt het erger: de sensor levert frames op zijn eigen snelheid, en elke gedupliceerde vastlegging vertraagt iedereen. De juiste aanpak is één vastleg-coroutine die “het nieuwste frame” naar een gedeelde sleuf publiceert, plus iterators per client die uit de sleuf lezen.

10.4.1. De vastlegtaak

Een achtergrond-coroutine grijpt frames zo snel als de sensor ze levert, JPEG-comprimeert elk frame in een gedeelde bytes, en pulseert een gebeurtenis zodat elke wachtende client wakker wordt:

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()

Het set() / clear() paar is het puls-patroon. set() deblokkeert in één keer elke coroutine die momenteel op de gebeurtenis wacht; clear() reset de gebeurtenis onmiddellijk zodat de volgende wait() weer blokkeert. Met meerdere consumenten (een kijker, nog een kijker, elke andere coroutine die op een nieuw frame moet reageren) is geen enkele consument verantwoordelijk voor het resetten van de gebeurtenis, en niemand steelt een ontwaking van iemand anders.

Notitie

De bytes(...) omhulling rond de JPEG is hier dragend. bytearray() geeft een view in de beeldbuffer van de camera terug; de allereerstvolgende snapshot() aanroep herschrijft die buffer ter plekke met het volgende frame. latest_jpeg leeft langer dan de lokale img, dus zonder de kopie zou elke lezer de sleuf bij elke vastlegging onder zich zien verschuiven.

10.4.2. Iterators per client lezen uit de sleuf

De MJPEG-streamhandler roept niet langer zelf csi0.snapshot() aan. In plaats daarvan wacht elke FrameStream instantie op de gedeelde gebeurtenis en leest uit de gedeelde bytes:

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')

De snapshot-route verandert ook: deze triggert niet langer een vastlegging, hij geeft terug wat latest_jpeg momenteel bevat:

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

De (body, status) tuple is microdot’s afkorting voor het instellen van een HTTP-statuscode zonder een microdot.Response te construeren. 503 zegt ik ben er maar niet klaar – de juiste code voor “vraag het over een momentje nog eens.”

10.4.3. Vastlegging naast de server draaien

main heeft nu twee coroutines op het hoogste niveau: de vastleglus en de HTTP-server. asyncio.gather() draait ze beide, en als een van beide crasht, wordt de ander geannuleerd:

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

asyncio.run(main())

Nu leest de sensor één frame per cyclus, ongeacht hoeveel kijkers er verbonden zijn. De eerste browser naar /stream.jpg ziet frames; de tweede ook, de derde, de tiende – ze delen allemaal dezelfde vastlegging, en de cam blijft net zo responsief op zijn andere routes.