12.8. Widżety i telemetria z CBORChannel

Backendy z dwóch poprzednich stron przenoszą surowe bajty, a kanał surowych bajtów wymaga programu hosta, który wie, jak je zdekodować. Najczęściej to, co kamera chce publikować, jest prostsze: kilka nazwanych odczytów, kilka nazwanych elementów sterujących, przebieg, mapa głębi. Na ten przypadek pakiet protocol dostarcza protocol.CBORChannel – gotowy backend, który przechowuje nazwane pola, serializuje je jako rekordy CBOR w układzie SenML i dekoduje zapisy hosta z powrotem na wartości pól. Jego zaletą jest to, że OpenMV IDE już go rozumie: zarejestruj CBORChannel, a widok Kanały w IDE wyrenderuje każde pole jako żywy widżet – etykiety, przełączniki, suwaki, wykresy, mapy – bez żadnego kodu hosta. Te same rekordy to zwykły CBOR, więc własny host dekoduje je dowolną biblioteką CBOR, gdy wyrośniesz z IDE.

Kanał przykładu controls_channel.py w postaci renderowanej przez IDE: odczyty, przełączniki i suwaki dla FLIR Lepton. Każdy element sterujący zapisuje wartość z powrotem do skryptu.

12.8.1. Odczyty i elementy sterujące

CBORChannel to słownik typowanych pól. add deklaruje każde z nich z nazwą, typem widżetu type i argumentami, których ten typ potrzebuje; przypisanie do ch["name"] aktualizuje pole; wywołanie zwrotne on_write otrzymuje zmiany od hosta:

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)

Uruchom go w IDE, przełącz panel pod buforem ramki na Kanały, a widżety się pojawią: label to wartość tylko do odczytu (z unit obok), text to blok statycznego tekstu sformatowanego, toggle to przełącznik, slider i spinbox ustawiają liczbę w zakresie min / max / step (pole liczbowe do precyzyjnego wpisywania), radio i select wybierają jedną z options, lineedit to pole dowolnego tekstu, a pushbutton to chwilowa akcja, która wywołuje on_write z True. Skrypt może w dowolnej chwili odczytać swoje elementy sterujące przez ch["Threshold"] – kanał przechowuje najnowszą wartość – a on_write jest hakiem dla tych, które wymagają natychmiastowej reakcji. Krotka (min, max, value) przypisana do suwaka lub pola liczbowego przesuwa jego zakres wraz z wartością, i tak właśnie element sterujący podąża za czujnikiem, którego granice zależą od innego ustawienia.

Opcjonalne wywołanie zwrotne on_read(channel) uruchamia się tuż przed serializacją kanału dla hosta – to miejsce, by spróbkować czujnik do etykiety, tak aby odczyt był pobierany tylko wtedy, gdy ktoś patrzy:

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

Przykład controls_channel.py w File → Examples → 12-Protocol używa dokładnie tego kształtu, aby udostępnić tryb pomiaru, zakres temperatur i ustawienia odbicia / obrotu FLIR Lepton podczas strumieniowania z kamery.

12.8.2. Przebiegi

Pole waveform przenosi blok próbek z częstotliwością próbkowania. Próbki to surowe bajty array w typecode pola (domyślnie 16-bitowe bez znaku); dla kilku sygnałów naraz – trzech osi akcelerometru – przepleć je i podaj, ile jest series oraz jak je nazwać:

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)

IDE wykreśla każdy przebieg jako przewijany wykres z jedną linią na serię, znakuje czasem każdy fragment, dzięki czemu przerwy między aktualizacjami są widoczne jako przerwy, i oferuje widok widma, wyzwalanie, znaczniki oraz nagrywanie do plików CSV, WAV, NumPy lub Edge Impulse. Zastąp syntetyczną sinusoidę próbkami audio lub odczytami sterownika IMU, a skrypt stanie się narzędziem do zbierania danych.

12.8.3. Mapy głębi

Pole depth to siatka width x height 32-bitowych liczb zmiennoprzecinkowych, po jednej odległości na komórkę – to, co produkuje czujnik czasu przelotu. min / max deklarują zakres, który host powinien odwzorować na kolor; IDE może też dobrać zakres kolorów automatycznie na podstawie danych:

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)

Przykład sensors_channel.py łączy taki kanał głębi z kanałem odczytów i detektorem twarzy na tej samej kamerze. Wywołanie send_event() jest opcjonalne – IDE samo odpytuje kanały – ale informuje host sterowany zdarzeniami, że nowa ramka jest gotowa, bez konieczności pytania z jego strony.

12.8.4. Dekodowanie na własnym hoście

Rekordy to standardowy CBOR: każdy odczyt kanału zwraca tablicę map kluczowanych całkowitymi kluczami SenML (0 nazwa, 1 jednostka, 2 wartość liczbowa, 3 wartość tekstowa, 8 wartość danych oraz klucze ujemne dla typu widżetu, opcji, zakresu i wymiarów). Host, który wyrósł z IDE, odczytuje kanał przez channel_read() i dekoduje go dowolną biblioteką CBOR (pip install cbor2); zapis elementu sterującego z powrotem to tablica CBOR map {0: name, 2: value} wysłana przez channel_write(). Dokumentacja protocol.CBORChannel wymienia każdy klucz. W większości projektów jednak hostem jest widok Kanały w IDE, a jedynym kodem jest powyższy skrypt po stronie kamery.