10.4. Egy rögzítési hurok megosztása a nézők között

Ha minden csatlakozott kliens önállóan hívja a csi0.snapshot() metódust, az pazarló, és amint egyszerre két stream van nyitva, még rosszabb lesz: az érzékelő a saját sebességén kézbesíti a képkockákat, és minden megduplázott rögzítés mindenkit lelassít. A helyes megközelítés egyetlen rögzítő coroutine, amely „a legfrissebb képkockát” egy megosztott helyre teszi közzé, plusz kliensenkénti iterátorok, amelyek a helyről olvasnak.

10.4.1. A rögzítő feladat

Egy háttér-coroutine olyan gyorsan ragadja meg a képkockákat, amilyen gyorsan az érzékelő kézbesíti őket, mindegyiket egy megosztott bytes objektumba JPEG-tömöríti, és impulzust ad egy eseménynek, hogy bármely várakozó kliens felébredjen:

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

A set() / clear() páros az impulzus minta. A set() egyszerre oldja fel minden olyan coroutine blokkolását, amely jelenleg az eseményre vár; a clear() azonnal visszaállítja az eseményt, így a következő wait() ismét blokkol. Több fogyasztó esetén (egy néző, egy másik néző, bármely más coroutine, amelynek reagálnia kell egy új képkockára) egyetlen fogyasztó sem felelős az esemény visszaállításáért, és senki sem lop el egy ébresztést senki mástól.

Megjegyzés

A JPEG köré tett bytes(...) burkolás itt teherhordó. A bytearray() egy nézetet ad vissza a kamera képpufferébe; a rögtön következő snapshot() hívás helyben felülírja azt a puffert a következő képkockával. A latest_jpeg túléli a lokális img változót, így a másolás nélkül minden olvasó azt látná, hogy a hely minden rögzítéskor elmozdul alóla.

10.4.2. A kliensenkénti iterátorok a helyről olvasnak

Az MJPEG-stream kezelője már nem hívja maga a csi0.snapshot() metódust. Ehelyett minden FrameStream példány a megosztott eseményre vár, és a megosztott bájtokból olvas:

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

A pillanatkép-útvonal is változik: már nem vált ki rögzítést, hanem azt adja vissza, amit a latest_jpeg éppen tartalmaz:

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

A (body, status) tuple a microdot rövidítése egy HTTP-állapotkód beállítására anélkül, hogy egy microdot.Response objektumot építenénk. Az 503 azt jelenti, hogy itt vagyok, de nem vagyok kész – a megfelelő kód a „kérdezd meg újra egy pillanat múlva” üzenethez.

10.4.3. A rögzítés futtatása a szerver mellett

A main mostantól két felső szintű coroutine-nal rendelkezik: a rögzítési hurokkal és a HTTP-szerverrel. Az asyncio.gather() mindkettőt futtatja, és ha bármelyik összeomlik, a másik megszakad:

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

asyncio.run(main())

Most az érzékelő ciklusonként egy képkockát olvas, függetlenül attól, hány néző csatlakozik. Az első böngésző, amely a /stream.jpg címet kéri, képkockákat lát; ugyanúgy a második, a harmadik, a tizedik – mindannyian ugyanazon a rögzítésen osztoznak, és a kamera ugyanolyan reszponzív marad a többi útvonalán.