10.3. Live-стримінг — один глядач

Браузери можуть відтворювати потоки Motion JPEG (MJPEG) безпосередньо всередині тегу <img>. Надайте браузеру одну HTTP-відповідь, яка ніколи не завершується, записуйте JPEG-файли, розділені межею multipart, і браузер відображатиме кожен кадр у міру його надходження.

Протокол передачі простий: один заголовок відповіді Content-Type: multipart/x-mixed-replace; boundary=frame, потім рядок --frame, Content-Type: image/jpeg, порожній рядок, байти JPEG, \r\n, і повтор. Браузер закриває з’єднання, коли тег <img> видаляється або вкладка закривається.

10.3.1. Захоплення без блокування

Блокуючий csi0.snapshot(), який використовувався раніше, зупиняє весь цикл подій, доки датчик не доставить кадр. Це було прийнятно, коли один запит запускав один знімок і більше нічого не виконувалось. Після відкриття потоку сервер повинен продовжувати обробляти інші запити, поки захоплюється наступний кадр — виклик захоплення повинен передавати керування циклу подій під час очікування датчика.

Шаблон — це тонка обгортка AsyncCSI, яка опитує csi.CSI.snapshot() у неблокуючому режимі і призупиняє корутину між опитуваннями. Розділ про asyncio описав цей шаблон у AsyncCSI; вбудуйте його в скрипт поки що:

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)

Усі інші методи CSI (reset(), pixformat(), framesize(), gain_db(), …) перенаправляються через __getattr__; лише snapshot() замінюється очікуваною версією, яка дозволяє циклу подій планувати інші корутини між опитуваннями.

Замініть простий csi.CSI() зі snapshot-маршруту на AsyncCSI():

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

10.3.2. Тіла потокових відповідей — це класові ітератори

Тіло потокової відповіді — це просто об’єкт, по якому microdot ітерується за допомогою async for, надсилаючи кожен отриманий фрагмент у сокет. У CPython це зазвичай функція асинхронного генератора — async def з yield. MicroPython це не підтримує:

Примітка

asyncio MicroPython не підтримує функції асинхронних генераторів (async def name(): ... yield ...). Тіла потокових відповідей мають бути класовими асинхронними ітераторами з __aiter__, що повертає self, та __anext__, визначеним як async def.

Для MJPEG-потоку це означає клас, чий __anext__ очікує один кадр і повертає його, обрамленим у multipart-обгортку:

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

Екземпляр є новим для кожного запиту, тому кожен підключений клієнт отримує власний ітератор. Коли браузер відключається, microdot перестає очікувати __anext__ і ітератор видаляється збирачем сміття.

Примітка

Обгортка bytes(...) навколо JPEG є захисним заходом. bytearray() повертає вигляд у кадровий буфер камери, а наступний виклик snapshot() перезаписує цей буфер на місці. Обгортання в bytes копіює JPEG, щоб фрагмент, який microdot перебуває у процесі запису, залишався стабільним, навіть якщо flush записувача не завершився на момент повторного запуску __anext__.

10.3.3. Запуск сервера всередині asyncio

Попередній виклик app.run(host=..., port=...) є блокуючим. Обробнику MJPEG потрібно ділити цикл подій з опитуваннями знімків AsyncCSI, тому замініть app.run на start_server() всередині asyncio.run():

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

asyncio.run(main())

Обгортка asyncio.run() дозволяє серверу бути одним завданням серед кількох — корутина main тоді є природним місцем для запуску захоплення, виявлення руху та всього іншого, що повинно ділити цикл подій з HTTP-сервером.

10.3.4. Один глядач за раз

Кожен підключений клієнт запускає власний ітератор FrameStream, що означає, що кожен клієнт ініціює власний виклик csi0.snapshot(). Два браузери означають два зчитування датчика на інтервал кадру, три — три, і так далі. Датчик насправді не може доставляти кадри швидше, ніж його власна частота кадрів, тому запити шикуються в чергу один за одним і потік у всіх сповільнюється.

Виправлення полягає в єдиному спільному циклі захоплення, який публікує один кадр для багатьох читачів.