12.8. Widgets und Telemetrie mit CBORChannel#

Die Backends auf den beiden vorherigen Seiten bewegen rohe Bytes, und ein Rohbyte-Kanal braucht ein Host-Programm, das sie zu dekodieren weiß. Meistens ist das, was eine Kamera veröffentlichen möchte, einfacher: ein paar benannte Messwerte, ein paar benannte Steuerelemente, eine Wellenform, eine Tiefenkarte. Für diesen Fall liefert das protocol-Paket protocol.CBORChannel, ein fertiges Backend, das benannte Felder hält, sie als CBOR-Records im SenML-Layout serialisiert und Host-Schreibvorgänge zurück in Feldwerte dekodiert. Sein Vorteil ist, dass OpenMV IDE es bereits versteht: Registrieren Sie einen CBORChannel, und die Kanäle-Ansicht der IDE stellt jedes Feld als Live-Widget dar – Labels, Schalter, Schieberegler, Diagramme, Karten – ganz ohne Host-Code. Dieselben Records sind einfaches CBOR, sodass ein eigener Host sie mit jeder CBOR-Bibliothek dekodiert, wenn Sie der IDE entwachsen.

The Channels view showing a thermal camera script's controls channel: FPA and AUX temperature readouts with their unit, Measurement and High Temp toggles, Temp Min and Temp Max sliders, and H-Mirror and V-Flip toggles — Der Kanal des Beispiels `controls_channel.py`, wie die IDE ihn darstellt: Anzeigen, Schalter und Schieberegler für eine FLIR Lepton. Jedes Steuerelement schreibt zurück an das Skript.#

12.8.1. Anzeigen und Steuerelemente#

Ein CBORChannel ist ein Dictionary typisierter Felder. add deklariert jedes mit einem Namen, einem Widget-type und den Argumenten, die dieser Typ benötigt; eine Zuweisung an ch["name"] aktualisiert ein Feld; der on_write-Callback empfängt die Änderungen des Hosts:

import time
import protocol
from protocol import CBORChannel

def on_write(ch, name, value):
    # Called for every control the host changes.
    print(name, "=", value)
    if name == "Reset":
        ch["Count"] = 0

ch = CBORChannel(on_write=on_write)
ch.add("Status", type="label", value="starting")
ch.add("Count", type="label", value=0)
ch.add("Enable", type="toggle", value=True)
ch.add("Threshold", type="slider", min=0, max=100, step=1,
       value=50, unit="%")
ch.add("Gap", type="spinbox", min=0.0, max=15.0, step=0.1,
       value=1.5, unit="mm")
ch.add("Mode", type="radio", options=["Idle", "Track", "Record"],
       value="Idle")
ch.add("Quality", type="select", options=["Low", "Medium", "High"],
       value="Medium")
ch.add("Name", type="lineedit", value="openmv-cam")
ch.add("Reset", type="pushbutton")
protocol.register(name="controls", backend=ch)

count = 0
while True:
    if ch["Enable"]:
        count += 1
        ch["Count"] = count
    ch["Status"] = "threshold %d%%" % ch["Threshold"]
    time.sleep_ms(100)

Führen Sie es in der IDE aus, schalten Sie den Bereich unter dem Framebuffer auf Kanäle um, und die Widgets erscheinen: Ein Label ist ein schreibgeschützter Wert (mit seiner unit daneben), text ist ein Block statischen Rich Texts, ein toggle ist ein Schalter, slider und spinbox stellen eine Zahl innerhalb von min / max / step ein (das Drehfeld für präzise Eingabe), radio und select wählen eine der options, lineedit ist ein Freitextfeld, und pushbutton ist eine einmalige Aktion, die on_write mit True aufruft. Das Skript liest seine Steuerelemente mit ch["Threshold"] zurück, wann immer es möchte – der Kanal hält den letzten Wert –, und on_write ist der Haken für diejenigen, die eine sofortige Reaktion brauchen. Ein einem Schieberegler oder Drehfeld zugewiesenes (min, max, value)-Tupel verschiebt dessen Bereich zusammen mit dem Wert; so folgt ein Steuerelement einem Sensor, dessen Grenzen von einer anderen Einstellung abhängen.

Der optionale Callback on_read(channel) läuft unmittelbar bevor der Kanal für den Host serialisiert wird – der Ort, um einen Sensor in ein Label abzutasten, sodass ein Messwert nur genommen wird, wenn jemand hinsieht:

def on_read(ch):
    ch["FPA Temp"] = round(csi0.ioctl(csi.IOCTL_LEPTON_GET_FPA_TEMP), 1)

Das Beispiel controls_channel.py unter File → Examples → 12-Protocol verwendet genau diese Form, um Messmodus, Temperaturbereich und Spiegel-/Flip-Einstellungen einer FLIR Lepton freizugeben, während die Kamera streamt.

12.8.2. Wellenformen#

Ein waveform-Feld trägt einen Block von Samples mit einer Abtastrate. Die Samples sind die rohen Bytes eines array im typecode des Feldes (standardmäßig vorzeichenlos 16 Bit); für mehrere Signale auf einmal – die drei Achsen eines Beschleunigungssensors – verschachteln Sie sie und geben an, wie viele series es gibt und wie sie heißen sollen:

import math
import time
import protocol
from array import array
from protocol import CBORChannel

RATE = 1000           # samples per second
CHUNK = 100           # samples per update (100 ms)

ch = CBORChannel()
ch.add("Mic", type="waveform", sample_rate=RATE,
       min=0, max=65535)                           # uint16
ch.add("IMU", type="waveform", sample_rate=RATE, series=3,
       typecode="f", min=-2.0, max=2.0,
       options=["X", "Y", "Z"], unit="g")          # float32 x 3
protocol.register(name="signals", backend=ch)

t = 0
while True:
    mic = array("H", (int(32768 + 20000 * math.sin(2 * math.pi * 50 * (t + i) / RATE))
                      for i in range(CHUNK)))
    imu = array("f")
    for i in range(CHUNK):
        phase = 2 * math.pi * 2 * (t + i) / RATE
        imu.extend((math.sin(phase), math.cos(phase), 1.0))
    ch["Mic"] = bytes(mic)    # the raw sample bytes
    ch["IMU"] = bytes(imu)
    t += CHUNK
    time.sleep_ms(100)

Die IDE zeichnet jede Wellenform als laufendes Diagramm mit einer Spur pro Reihe, versieht jeden Chunk mit einem Zeitstempel, sodass Lücken zwischen Aktualisierungen als Lücken erscheinen, und bietet eine Spektrumansicht, Trigger, Marker und Aufzeichnung in CSV-, WAV-, NumPy- oder Edge-Impulse-Dateien. Ersetzen Sie den synthetischen Sinus durch audio-Samples oder die Messwerte eines IMU-Treibers, und das Skript ist ein Werkzeug zur Datensammlung.

12.8.3. Tiefenkarten#

Ein depth-Feld ist ein width x height-Raster aus 32-Bit-Floats, eine Entfernung pro Zelle – das, was ein Time-of-Flight-Sensor erzeugt. min / max deklarieren den Bereich, den der Host auf Farbe abbilden soll; die IDE kann die Farben auch automatisch aus den Daten skalieren:

import struct
import time
import tof
import protocol
from protocol import CBORChannel

tof.init()
ch = CBORChannel()
ch.add("depth", type="depth", width=tof.width(), height=tof.height(),
       min=0, max=1000)
reg = protocol.register(name="ToF", backend=ch)

while True:
    try:
        d, dmin, dmax = tof.read_depth(vflip=True, hmirror=True)
    except RuntimeError:
        continue
    ch["depth"] = struct.pack("<%df" % len(d), *d)
    reg.send_event(0xFFFF)
    time.sleep_ms(50)

Das Beispiel sensors_channel.py kombiniert einen solchen Tiefenkanal mit einem Messwertkanal und einem Gesichtsdetektor auf derselben Kamera. Der Aufruf send_event() ist optional – die IDE fragt Kanäle von sich aus ab –, teilt aber einem ereignisgesteuerten Host mit, dass ein neuer Frame bereit ist, ohne dass er nachfragen muss.

12.8.4. Dekodieren auf einem eigenen Host#

Die Records sind Standard-CBOR: Jedes Lesen eines Kanals liefert ein Array von Maps mit den Integer-Schlüsseln von SenML (0 Name, 1 Einheit, 2 numerischer Wert, 3 String-Wert, 8 Datenwert und negative Schlüssel für Widget-Typ, Optionen, Bereich und Dimensionen). Ein Host, der der IDE entwachsen ist, liest den Kanal mit channel_read() und dekodiert ihn mit jeder CBOR-Bibliothek (pip install cbor2); das Zurückschreiben eines Steuerelements ist ein CBOR-Array aus {0: name, 2: value}-Maps, gesendet mit channel_write(). Die Referenz zu protocol.CBORChannel listet jeden Schlüssel auf. Für die meisten Projekte ist jedoch die Kanäle-Ansicht der IDE der Host, und der einzige Code ist das Kameraskript oben.