10.3. Živé streamování – jeden divák

Prohlížeče dokážou vykreslit vícedílné Motion JPEG (MJPEG) streamy přímo uvnitř tagu <img>. Předejte prohlížeči jednu HTTP odpověď, která nikdy neskončí, zapisujte JPEGy oddělené multipart hranicí a prohlížeč zobrazí každý snímek, jakmile dorazí.

Přenos je přímočarý: jedna hlavička odpovědi, Content-Type: multipart/x-mixed-replace; boundary=frame, poté řádek --frame, Content-Type: image/jpeg, prázdný řádek, JPEG bajty, \r\n a opakovat. Prohlížeč spojení uzavře, když je <img> odebrán nebo je zavřena karta.

10.3.1. Zachytávání bez blokování

Doposud používané blokující csi0.snapshot() zastaví celou event loop, dokud senzor nedodá snímek. To bylo v pořádku, když jeden požadavek spustil jeden snímek a nic jiného neběželo. Jakmile je stream otevřen, server musí pokračovat v obsluze ostatních požadavků, zatímco se zachytává další snímek – volání zachycení musí předat řízení event loopu, zatímco čeká na senzor.

Vzorem je tenký wrapper AsyncCSI, který dotazuje csi.CSI.snapshot() v neblokujícím režimu a mezi dotazy korutinu uspí. Kapitola o asyncio tento vzor prošla v AsyncCSI; pro teď jej vložte přímo do skriptu:

import asyncio

class AsyncCSI:
    def __init__(self, *args, **kwargs):
        self._csi = csi.CSI(*args, **kwargs)

    def __getattr__(self, name):
        return getattr(self._csi, name)

    async def snapshot(self):
        while True:
            img = self._csi.snapshot(blocking=False)
            if img is not None:
                return img
            await asyncio.sleep_ms(0)

Každá další metoda CSI (reset(), pixformat(), framesize(), gain_db(), …) je přeposlána přes __getattr__; pouze snapshot() je nahrazena awaitable verzí, která umožňuje event loopu naplánovat mezi dotazy jiné korutiny.

Vyměňte holé csi.CSI() z cesty snapshot za AsyncCSI():

csi0 = AsyncCSI()
csi0.reset()
csi0.pixformat(csi.RGB565)
csi0.framesize(csi.QVGA)

10.3.2. Těla streamů jsou iterátory založené na třídách

Tělo streamovací odpovědi je jen objekt, který microdot iteruje pomocí async for a posílá každý vyprodukovaný kus do socketu. V CPythonu je to běžně async generátorová funkce – async def s yield. MicroPython to nepodporuje:

Poznámka

asyncio v MicroPythonu nepodporuje async-generátorové funkce (async def name(): ... yield ...). Těla streamovacích odpovědí musí být async iterátory založené na třídách s __aiter__ vracejícím self a __anext__ definovaným jako async def.

Pro MJPEG stream to znamená třídu, jejíž __anext__ počká na jeden snímek a vrátí jej zabalený v multipart obalu:

BOUNDARY = b'frame'

class FrameStream:
    def __aiter__(self):
        return self

    async def __anext__(self):
        img = await csi0.snapshot()
        jpeg = bytes(img.compress(quality=85).bytearray())
        return (b'--' + BOUNDARY + b'\r\n'
                b'Content-Type: image/jpeg\r\n\r\n'
                + jpeg + b'\r\n')

@app.get('/stream.jpg')
async def stream(request):
    return Response(
        body=FrameStream(),
        headers={
            'Content-Type':
                b'multipart/x-mixed-replace; boundary=' + BOUNDARY,
        },
    )

Instance je pro každý požadavek čerstvá, takže každý připojený klient dostane svůj vlastní iterátor. Když se prohlížeč odpojí, microdot přestane čekat na __anext__ a iterátor je uvolněn garbage collectorem.

Poznámka

Obal bytes(...) kolem JPEGu je obranný. bytearray() vrací pohled do obrazového bufferu kamery a další volání snapshot() tento buffer přepíše na místě. Zabalení do bytes zkopíruje JPEG ven, takže kus, který microdot právě zapisuje, zůstane stabilní, i kdyby zapisovač nedokončil vyprázdnění, než se __anext__ znovu spustí.

10.3.3. Spuštění serveru uvnitř asyncio

Dřívější volání app.run(host=..., port=...) je blokující. MJPEG handler potřebuje sdílet smyčku s dotazy AsyncCSI snapshot, takže vyměňte app.run za start_server() uvnitř asyncio.run():

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

asyncio.run(main())

Wrapper asyncio.run() umožňuje, aby byl server jednou úlohou z několika – korutina main je pak přirozeným místem pro spuštění zachytávání, detekce pohybu a čehokoli dalšího, co musí sdílet smyčku s HTTP serverem.

10.3.4. Jeden divák najednou

Každý připojený klient spouští svůj vlastní iterátor FrameStream, což znamená, že každý klient spouští své vlastní volání csi0.snapshot(). Dva prohlížeče znamenají dvě čtení senzoru na interval snímku, tři znamenají tři a tak dále. Senzor ve skutečnosti nedokáže dodávat snímky rychleji než je jeho vlastní snímková frekvence, takže se požadavky řadí za sebe a stream každého se zpomalí.

Řešením je jediná sdílená smyčka zachytávání, která publikuje jeden snímek mnoha čtenářům.