10.3. Élő streamelés – egy néző

A böngészők közvetlenül egy <img> címkén belül tudják megjeleníteni a többrészes Motion JPEG (MJPEG) streameket. Adj a böngészőnek egy HTTP-választ, amely soha nem fejeződik be, írj ki többrészes határolóval elválasztott JPEG-eket, és a böngésző minden képkockát megjelenít, amint megérkezik.

A vezeték egyszerű: egy válaszfejléc, Content-Type: multipart/x-mixed-replace; boundary=frame, majd egy --frame sor, Content-Type: image/jpeg, egy üres sor, a JPEG-bájtok, \r\n, és ismétlés. A böngésző lezárja a kapcsolatot, amikor az <img> eltávolításra kerül, vagy a fület bezárják.

10.3.1. Rögzítés blokkolás nélkül

Az eddig használt blokkoló csi0.snapshot() megakasztja az egész eseményhurkot, amíg az érzékelő kézbesít egy képkockát. Ez rendben volt, amikor egy kérés egy pillanatképet indított, és semmi más nem futott. Amint egy stream nyitva van, a szervernek továbbra is kezelnie kell a többi kérést, miközben a következő képkocka rögzítés alatt áll – a rögzítési hívásnak át kell adnia a vezérlést az eseményhuroknak, amíg az érzékelőre vár.

A minta egy vékony AsyncCSI burkoló, amely nem blokkoló módban lekérdezi a csi.CSI.snapshot() metódust, és a lekérdezések között alvásra teszi a coroutine-t. Az asyncio fejezet végigvette ezt a mintát a AsyncCSI oldalon; egyelőre illeszd be a szkriptbe:

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)

Minden más CSI-metódus (reset(), pixformat(), framesize(), gain_db(), …) a __getattr__ metóduson keresztül van továbbítva; csak a snapshot() van lecserélve egy await-elhető változatra, amely lehetővé teszi az eseményhurok számára, hogy más coroutine-okat ütemezzen a lekérdezések között.

Cseréld le a csupasz csi.CSI() hívást a pillanatkép-útvonalból egy AsyncCSI() hívásra:

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

10.3.2. A streamelő törzsek osztályalapú iterátorok

Egy streamelő válasz törzse csak egy objektum, amelyet a microdot async for segítségével iterál, és minden visszaadott darabot leküld a socketen. CPythonon ez általában egy aszinkron generátorfüggvény – async def egy yield utasítással. A MicroPython ezt nem támogatja:

Megjegyzés

A MicroPython asyncio modulja nem támogatja az aszinkron generátorfüggvényeket (async def name(): ... yield ...). A streamelő válasz törzseknek osztályalapú aszinkron iterátoroknak kell lenniük, ahol az __aiter__ a self-et adja vissza, és az __anext__ async def-ként van definiálva.

Egy MJPEG-stream esetén ez egy olyan osztályt jelent, amelynek __anext__ metódusa megvár egy képkockát, és a többrészes burkolóba foglalva adja vissza:

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

A példány kérésenként friss, így minden csatlakozott kliens a saját iterátorát kapja meg. Amikor a böngésző lecsatlakozik, a microdot abbahagyja az __anext__ várását, és az iterátort a szemétgyűjtő begyűjti.

Megjegyzés

A JPEG köré tett bytes(...) burkolás védekező jellegű. A bytearray() egy nézetet ad vissza a kamera képpufferébe, és a következő snapshot() hívás helyben felülírja azt a puffert. A bytes burkolás kimásolja a JPEG-et, így a darab, amelynek microdot a kiírása közben tart, stabil marad még akkor is, ha az író ürítése nem fejeződött be, mire az __anext__ újra lefut.

10.3.3. A szerver futtatása asyncion belül

A korábbi app.run(host=..., port=...) hívás blokkoló. Az MJPEG-kezelőnek meg kell osztania az eseményhurkot az AsyncCSI pillanatkép-lekérdezésekkel, ezért cseréld le az app.run hívást a start_server() metódusra egy asyncio.run() belsejében:

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

asyncio.run(main())

Az asyncio.run() burkoló lehetővé teszi, hogy a szerver egyike legyen több feladatnak – a main coroutine ekkor a természetes hely a rögzítés, a mozgásérzékelés és minden más elindítására, aminek meg kell osztania az eseményhurkot a HTTP-szerverrel.

10.3.4. Egy néző egyszerre

Minden csatlakozott kliens a saját FrameStream iterátorát futtatja, ami azt jelenti, hogy minden kliens a saját csi0.snapshot() hívását váltja ki. Két böngésző két érzékelő-olvasást jelent képkocka-időközönként, három hármat, és így tovább. Az érzékelő valójában nem tud gyorsabban képkockákat kézbesíteni, mint a saját képkockasebessége, így a kérések egymás mögött sorba állnak, és mindenki streamje lelassul.

A megoldás egyetlen megosztott rögzítési hurok, amely egy képkockát tesz közzé sok olvasó számára.