10.4. Att dela en fångstloop mellan tittare

Att varje ansluten klient anropar csi0.snapshot() oberoende är slöseri, och när väl två strömmar är öppna samtidigt blir det värre: sensorn levererar bildrutor i sin egen takt, och varje duplicerad fångst saktar ner alla. Det rätta tillvägagångssättet är en fångst-coroutine som publicerar ”den senaste bildrutan” till en delad plats, plus per-klient-iteratorer som läser från platsen.

10.4.1. Fångstuppgiften

En bakgrunds-coroutine fångar bildrutor så fort sensorn levererar dem, JPEG-komprimerar var och en till en delad bytes och pulsar en händelse så att alla väntande klienter vaknar:

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

Paret set() / clear() är puls-mönstret. set() avblockerar varje coroutine som just nu väntar på händelsen på en gång; clear() återställer omedelbart händelsen så att nästa wait() blockerar igen. Med flera konsumenter (en tittare, en annan tittare, vilken annan coroutine som helst som behöver reagera på en ny bildruta) är ingen enskild konsument ansvarig för att återställa händelsen, och ingen stjäl en uppvakning från någon annan.

Anteckning

Omslutningen bytes(...) runt JPEG:en är bärande här. bytearray() returnerar en vy in i kamerans bildbuffert; det allra nästa snapshot()-anropet skriver om den bufferten på plats med nästa bildruta. latest_jpeg lever längre än den lokala img, så utan kopian skulle varje läsare se platsen förskjutas under dem vid varje fångst.

10.4.2. Per-klient-iteratorer läser från platsen

MJPEG-strömhanteraren slutar anropa csi0.snapshot() själv. Istället väntar varje FrameStream-instans på den delade händelsen och läser från de delade byten:

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

Stillbildsrutten ändras också: den utlöser inte längre en fångst, den returnerar vad latest_jpeg än för närvarande innehåller:

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

Tupeln (body, status) är microdots kortform för att sätta en HTTP-statuskod utan att konstruera en microdot.Response. 503 säger jag är här men inte redo – rätt kod för ”fråga igen om en stund”.

10.4.3. Att köra fångst jämsides med servern

main har nu två coroutiner på toppnivå: fångstloopen och HTTP-servern. asyncio.gather() kör dem båda, och om endera kraschar avbryts den andra:

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

asyncio.run(main())

Nu läser sensorn en bildruta per cykel oavsett hur många tittare som är anslutna. Den första webbläsaren till /stream.jpg ser bildrutor; det gör även den andra, den tredje, den tionde – de delar alla samma fångst, och kameran förblir lika responsiv på sina andra rutter.