10.4. Partajarea unei singure bucle de captare între spectatori

Fiecare client conectat care apelează csi0.snapshot() independent este o risipă, iar odată ce două stream-uri sunt deschise simultan situația se înrăutățește: senzorul livrează cadre la propria rată, iar fiecare captare duplicată îi încetinește pe toți. Abordarea corectă este o singură corutină de captare care publică „cel mai recent cadru” într-un slot partajat, plus iteratoare per client care citesc din slot.

10.4.1. Sarcina de captare

O corutină de fundal preia cadre cât de repede le livrează senzorul, comprimă fiecare cadru în JPEG într-un bytes partajat și pulsează un eveniment astfel încât orice client în așteptare să se trezească:

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

Perechea set() / clear() este modelul de puls. set() deblochează dintr-o singură mișcare fiecare corutină care așteaptă în prezent evenimentul; clear() resetează imediat evenimentul astfel încât următorul wait() să se blocheze din nou. Cu mai mulți consumatori (un spectator, alt spectator, orice altă corutină care trebuie să reacționeze la un cadru nou), niciun consumator nu este responsabil pentru resetarea evenimentului și nimeni nu fură o trezire de la altcineva.

Notă

Învelirea bytes(...) în jurul JPEG-ului este esențială aici. bytearray() returnează o vedere asupra tamponului de imagine al camerei; chiar următorul apel snapshot() rescrie acel tampon pe loc cu următorul cadru. latest_jpeg supraviețuiește variabilei locale img, așa că fără copie fiecare cititor ar vedea slotul deplasându-se sub el la fiecare captare.

10.4.2. Iteratoarele per client citesc din slot

Handlerul de stream MJPEG nu mai apelează el însuși csi0.snapshot(). În schimb, fiecare instanță FrameStream așteaptă evenimentul partajat și citește din octeții partajați:

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

Ruta de instantaneu se schimbă și ea: nu mai declanșează o captare, ci returnează ceea ce conține în prezent latest_jpeg:

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

Tuplul (body, status) este prescurtarea din microdot pentru setarea unui cod de status HTTP fără a construi un microdot.Response. 503 spune sunt aici, dar nu sunt gata – codul potrivit pentru „mai întreabă peste o clipă”.

10.4.3. Rularea captării alături de server

main are acum două corutine de nivel superior: bucla de captare și serverul HTTP. asyncio.gather() le rulează pe ambele, iar dacă una se prăbușește, cealaltă este anulată:

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

asyncio.run(main())

Acum senzorul citește un cadru per ciclu indiferent de câți spectatori sunt conectați. Primul browser care accesează /stream.jpg vede cadre; la fel și al doilea, al treilea, al zecelea – toți partajează aceeași captare, iar camera rămâne la fel de receptivă pe celelalte rute ale sale.