10.4. Partager une seule boucle de capture entre les spectateurs

Chaque client connecté appelant csi0.snapshot() indépendamment est un gaspillage, et dès que deux flux sont ouverts en même temps, cela empire : le capteur fournit des trames à sa propre cadence, et chaque capture dupliquée ralentit tout le monde. La bonne approche est une seule coroutine de capture qui publie « la dernière trame » dans un emplacement partagé, plus des itérateurs par client qui lisent depuis cet emplacement.

10.4.1. La tâche de capture

Une coroutine d’arrière-plan saisit les trames aussi vite que le capteur les fournit, compresse chacune en JPEG dans un bytes partagé, et déclenche une impulsion sur un événement afin que tout client en attente se réveille :

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

La paire set() / clear() est le motif d”impulsion. set() débloque d’un coup toutes les coroutines actuellement en attente sur l’événement ; clear() réinitialise immédiatement l’événement afin que le prochain wait() bloque à nouveau. Avec plusieurs consommateurs (un spectateur, un autre spectateur, toute autre coroutine qui doit réagir à une nouvelle trame), aucun consommateur unique n’est responsable de la réinitialisation de l’événement, et personne ne dérobe un réveil à un autre.

Note

L’enrobage bytes(...) autour du JPEG est porteur ici. bytearray() renvoie une vue sur le tampon d’image de la caméra ; l’appel suivant à snapshot() réécrit immédiatement ce tampon sur place avec la trame suivante. latest_jpeg survit à la variable locale img, donc sans la copie chaque lecteur verrait l’emplacement changer sous ses pieds à chaque capture.

10.4.2. Les itérateurs par client lisent depuis l’emplacement

Le gestionnaire de flux MJPEG cesse d’appeler lui-même csi0.snapshot(). À la place, chaque instance FrameStream attend l’événement partagé et lit depuis les octets partagés :

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

La route snapshot change elle aussi : elle ne déclenche plus de capture, elle renvoie ce que latest_jpeg contient à cet instant :

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

Le tuple (body, status) est le raccourci de microdot pour définir un code de statut HTTP sans construire de microdot.Response. 503 signifie je suis là mais pas prêt – le code adéquat pour « redemandez dans un instant ».

10.4.3. Exécuter la capture aux côtés du serveur

main possède désormais deux coroutines de premier niveau : la boucle de capture et le serveur HTTP. asyncio.gather() exécute les deux, et si l’une plante l’autre est annulée :

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

asyncio.run(main())

Désormais, le capteur lit une trame par cycle quel que soit le nombre de spectateurs connectés. Le premier navigateur à atteindre /stream.jpg voit les trames ; le deuxième aussi, le troisième, le dixième – ils partagent tous la même capture, et la caméra reste tout aussi réactive sur ses autres routes.