8.14. AsyncCSI

Типичный скрипт OpenMV Cam заканчивается циклом while True: img = csi0.snapshot() – блокирующим циклом снимков, которому asyncio вообще не нужен. Но как только приложению требуется делать что-то ещё параллельно с захватом кадров – слушать кнопку, отправлять данные другому устройству, выполнять фоновую задачу – блокирующий вызов начинает мешать. Пока snapshot ожидает следующий кадр, цикл событий не работает, поэтому все остальные сопрограммы программы заморожены до прихода кадра.

На этой странице мы создаём небольшую обёртку вокруг CSI, которая превращает snapshot в сопрограмму, поддерживающую await. В результате получается полноценная замена, позволяющая циклу захвата сосуществовать с остальной частью asyncio-программы.

8.14.1. Составные части

Большую часть работы выполняет одна часть API CSI – snapshot() в неблокирующем режиме. Вызов snapshot(blocking=False) либо возвращает следующий кадр (если он готов), либо None (если нет). Первый неблокирующий вызов также запускает DMA-захват камеры, если он ещё не был запущен, поэтому обёртке не нужно делать ничего особенного для инициализации.

Вторая часть – это asyncio.sleep_ms(). Обёртка в цикле опрашивает неблокирующие снимки, уступая управление циклу событий вызовом await asyncio.sleep_ms(0) между проверками, так что каждая другая готовая сопрограмма получает шанс выполниться до следующего опроса.

8.14.2. Обёртка

import asyncio
import csi


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. __getattr__ перенаправляет каждый атрибут, который обёртка сама не определяет – reset, pixformat, framesize, все настройки датчика – к нижележащему CSI, поэтому обёртка выглядит идентично необёрнутому объекту за исключением одного важного метода.

async def snapshot – это новая часть. Она вызывает snapshot(blocking=False); если вызов возвращает изображение, сопрограмма возвращает его. В противном случае она уступает управление циклу событий вызовом await asyncio.sleep_ms(0), чтобы другие сопрограммы получили шанс выполниться, затем повторяет цикл и пробует снова. Первая итерация запускает DMA; последующие итерации забирают кадры по мере их готовности.

8.14.3. Цикл снимков в компании

С готовой обёрткой цикл снимков вписывается в более крупную asyncio-программу так же, как любая другая сопрограмма. В примере ниже три сопрограммы выполняются параллельно: цикл захвата, мигалка светодиода и пульс, который раз в секунду печатает hello:

import asyncio
import csi
from machine import LED


async def capture_loop(cam):
    while True:
        img = await cam.snapshot()
        # process img here

async def blinker(led, period_ms):
    while True:
        led.on()
        await asyncio.sleep_ms(period_ms)
        led.off()
        await asyncio.sleep_ms(period_ms)

async def hello(period_s):
    while True:
        print("hello")
        await asyncio.sleep(period_s)

async def main():
    cam = AsyncCSI()
    cam.reset()
    cam.pixformat(csi.RGB565)
    cam.framesize(csi.QVGA)

    asyncio.create_task(blinker(LED("LED_BLUE"), 200))
    asyncio.create_task(hello(1))
    await capture_loop(cam)

asyncio.run(main())

Все три сопрограммы продвигаются в одном и том же цикле событий. Пока capture_loop уступает управление между неблокирующими опросами снимков, blinker переключает светодиод, а hello печатает текст. Пока blinker и hello спят, capture_loop опрашивает камеру. Интервал опроса короткий – один такт цикла событий – поэтому он добавляет пренебрежимо малую задержку к моменту, когда приложение видит новый кадр.

Цикл захвата не блокирует цикл событий. Добавление дополнительной параллельной работы – например, UART-клиента – это просто ещё один вызов create_task() внутри main.

Примечание

Настройка буферов кадров по-прежнему важна в этой схеме. Режим одного буфера заставляет snapshot(blocking=False) возвращать None до тех пор, пока не будет захвачен следующий кадр; двойная или тройная буферизация сглаживает это, так что обёртка обычно находит буферизованный кадр уже на первом опросе после того, как предыдущий кадр был обработан.