12.8. Widgety a telemetrie s CBORChannel

Backendy na předchozích dvou stránkách přenášejí surové bajty a kanál surových bajtů potřebuje hostitelský program, který je umí dekódovat. Většinou je to, co chce kamera publikovat, jednodušší: několik pojmenovaných odečtů, několik pojmenovaných ovládacích prvků, průběh, hloubková mapa. Pro tento případ balíček protocol dodává protocol.CBORChannel, hotový backend, který drží pojmenovaná pole, serializuje je jako záznamy CBOR v rozvržení SenML a dekóduje zápisy hostitele zpět na hodnoty polí. Jeho výhodou je, že OpenMV IDE mu už rozumí: zaregistrujte CBORChannel a zobrazení Channels v IDE vykreslí každé pole jako živý widget – popisky, přepínače, posuvníky, grafy, mapy – bez jakéhokoli hostitelského kódu. Tytéž záznamy jsou obyčejný CBOR, takže až IDE přerostete, vlastní hostitel je dekóduje libovolnou knihovnou CBOR.

Kanál příkladu controls_channel.py tak, jak jej IDE vykresluje: odečty, přepínače a posuvníky pro FLIR Lepton. Každý ovládací prvek zapisuje zpět do skriptu.

12.8.1. Odečty a ovládací prvky

CBORChannel je slovník typovaných polí. add deklaruje každé z nich s názvem, type widgetu a argumenty, které daný typ potřebuje; přiřazení do ch["name"] pole aktualizuje; zpětné volání on_write přijímá změny od hostitele:

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)

Spusťte jej v IDE, přepněte panel pod snímkovým bufferem na Channels a widgety se objeví: label je hodnota jen pro čtení (s unit vedle ní), text je blok statického formátovaného textu, toggle je přepínač, slider a spinbox nastavují číslo v rozsahu min / max / step (číselník pro přesné zadání), radio a select vybírají jednu z options, lineedit je volné textové pole a pushbutton je okamžitá akce, která zavolá on_write s True. Skript si své ovládací prvky může kdykoli přečíst zpět pomocí ch["Threshold"] – kanál drží poslední hodnotu – a on_write je háček pro ty, které potřebují okamžitou reakci. N-tice (min, max, value) přiřazená posuvníku nebo číselníku posune jeho rozsah spolu s hodnotou, což je způsob, jak ovládací prvek sleduje senzor, jehož meze závisejí na jiném nastavení.

Volitelné zpětné volání on_read(channel) běží těsně před serializací kanálu pro hostitele – místo, kde navzorkovat senzor do popisku, aby se odečet prováděl jen tehdy, když se někdo dívá:

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

Příklad controls_channel.py pod File → Examples → 12-Protocol používá přesně tento tvar k vystavení režimu měření, teplotního rozsahu a nastavení zrcadlení / převrácení FLIR Lepton, zatímco kamera streamuje.

12.8.2. Průběhy

Pole waveform nese blok vzorků se vzorkovací frekvencí. Vzorky jsou surové bajty array v typecode daného pole (ve výchozím stavu 16bitové bez znaménka); pro několik signálů najednou – tři osy akcelerometru – je prokládejte a uveďte, kolik series existuje a jak se mají jmenovat:

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 vykresluje každý průběh jako rolující graf s jednou stopou na řadu, každý blok opatřuje časovou značkou, takže mezery mezi aktualizacemi se zobrazí jako mezery, a nabízí spektrální zobrazení, spouště, značky a nahrávání do souborů CSV, WAV, NumPy nebo Edge Impulse. Nahraďte syntetický sinus vzorky audio nebo údaji z ovladače IMU a skript je nástrojem pro sběr dat.

12.8.3. Hloubkové mapy

Pole depth je mřížka width x height 32bitových floatů, jedna vzdálenost na buňku – to, co produkuje time-of-flight senzor. min / max deklarují rozsah, který má hostitel mapovat na barvu; IDE umí rozsah barev určit také automaticky z dat:

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)

Příklad sensors_channel.py spojuje takovýto hloubkový kanál s kanálem odečtů a detektorem obličejů na téže kameře. Volání send_event() je volitelné – IDE kanály dotazuje samo – ale událostně řízenému hostiteli sdělí, že je připraven nový snímek, aniž by se musel ptát.

12.8.4. Dekódování na vlastním hostiteli

Záznamy jsou standardní CBOR: každé čtení kanálu vrací pole map s celočíselnými klíči SenML (0 název, 1 jednotka, 2 číselná hodnota, 3 řetězcová hodnota, 8 datová hodnota a záporné klíče pro typ widgetu, možnosti, rozsah a rozměry). Hostitel, který IDE přerostl, čte kanál pomocí channel_read() a dekóduje jej libovolnou knihovnou CBOR (pip install cbor2); zápis ovládacího prvku zpět je pole CBOR map {0: name, 2: value} odeslané pomocí channel_write(). Reference protocol.CBORChannel uvádí každý klíč. U většiny projektů je však hostitelem zobrazení Channels v IDE a jediným kódem je skript na straně kamery výše.