8.14. AsyncCSI

Ein typisches OpenMV-Cam-Skript endet mit while True: img = csi0.snapshot() – einer blockierenden Schnappschuss-Schleife, die asyncio überhaupt nicht benötigt. In dem Moment, in dem die Anwendung neben den Aufnahmen etwas anderes tun muss – auf einen Knopf warten, Daten an einen Peer senden, eine Hintergrundaufgabe ausführen – steht der blockierende Aufruf im Weg. Während snapshot auf das nächste Einzelbild wartet, läuft die Ereignisschleife nicht, sodass jede andere Koroutine im Programm eingefroren ist, bis das Einzelbild eintrifft.

Auf dieser Seite wird ein kleiner Wrapper um CSI herum aufgebaut, der snapshot in eine await-fähige Koroutine verwandelt. Das Ergebnis ist ein direkt einsetzbarer Ersatz, mit dem eine Aufnahmeschleife mit dem Rest eines asyncio-Programms koexistieren kann.

8.14.1. Die Bausteine

Ein Teil der CSI-API erledigt den Großteil der Arbeit – snapshot() in seinem nicht-blockierenden Modus. Der Aufruf von snapshot(blocking=False) gibt entweder das nächste Einzelbild zurück (falls eines bereit ist) oder None (falls nicht). Der erste nicht-blockierende Aufruf startet außerdem die DMA-Aufnahme der Kamera, falls sie noch nicht läuft, sodass der Wrapper nichts Besonderes tun muss, um sie zu initialisieren.

Der andere Teil ist asyncio.sleep_ms(). Der Wrapper fragt nicht-blockierende Schnappschüsse in einer Schleife ab und gibt zwischen den Prüfungen mit await asyncio.sleep_ms(0) die Kontrolle an die Ereignisschleife ab, sodass jede andere bereite Koroutine vor der nächsten Abfrage die Gelegenheit erhält, zu laufen.

8.14.2. Der Wrapper

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)

Der Konstruktor umschließt eine CSI-Instanz. __getattr__ leitet jedes Attribut, das der Wrapper nicht selbst definiert – reset, pixformat, framesize, alle Sensor-Stellschrauben – an die zugrunde liegende CSI weiter, sodass der Wrapper bis auf die eine entscheidende Methode identisch mit dem nicht umschlossenen Objekt aussieht.

async def snapshot ist der neue Baustein. Er ruft snapshot(blocking=False) auf; wenn der Aufruf ein Bild zurückgibt, gibt die Koroutine es zurück. Andernfalls gibt sie mit await asyncio.sleep_ms(0) die Kontrolle an die Ereignisschleife zurück, sodass andere Koroutinen die Gelegenheit erhalten, zu laufen, und versucht es dann erneut. Die erste Iteration startet das DMA; nachfolgende Iterationen greifen Einzelbilder ab, sobald sie verfügbar werden.

8.14.3. Eine Schnappschuss-Schleife in Gesellschaft

Mit dem Wrapper an Ort und Stelle fügt sich eine Schnappschuss-Schleife genauso in ein größeres asyncio-Programm ein wie jede andere Koroutine. Das folgende Beispiel führt drei Koroutinen nebenläufig aus: die Aufnahmeschleife, einen LED-Blinker und einen Heartbeat, der einmal pro Sekunde hello ausgibt:

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

Alle drei Koroutinen kommen auf derselben Ereignisschleife voran. Während capture_loop zwischen nicht-blockierenden Schnappschuss-Abfragen die Kontrolle abgibt, schaltet blinker die LED um und hello gibt aus. Während blinker und hello schlafen, fragt capture_loop die Kamera ab. Das Abfrageintervall ist kurz – ein einzelner Ereignisschleifen-Tick – sodass es die Latenz, mit der die Anwendung ein neues Einzelbild sieht, nur vernachlässigbar erhöht.

Die Aufnahmeschleife blockiert die Ereignisschleife nicht. Mehr nebenläufige Arbeit hinzuzufügen – etwa einen UART-Client – ist lediglich ein weiterer create_task()-Aufruf innerhalb von main.

Bemerkung

Die Einstellung framebuffers ist auch in dieser Form weiterhin von Bedeutung. Im Einzelpuffer-Modus gibt snapshot(blocking=False) None zurück, bis das nächste Einzelbild aufgenommen ist; doppeltes oder dreifaches Puffern glättet dies, sodass der Wrapper normalerweise ein gepuffertes Einzelbild vorfindet, das bei der ersten Abfrage nach der Verarbeitung des vorherigen Einzelbilds bereitsteht.