12.8. Widgets en telemetrie met CBORChannel

De backends op de vorige twee pagina’s verplaatsen ruwe bytes, en een ruw bytekanaal heeft een hostprogramma nodig dat weet hoe het die moet decoderen. Meestal is wat een cam wil publiceren eenvoudiger dan dat: een paar benoemde metingen, een paar benoemde bedieningselementen, een golfvorm, een dieptekaart. Voor dat geval levert het protocol-pakket protocol.CBORChannel, een kant-en-klare backend die benoemde velden bevat, ze serialiseert als CBOR-records in de SenML-indeling, en hostschrijfbewerkingen terugdecodeert naar veldwaarden. De winst is dat OpenMV IDE het al begrijpt: registreer een CBORChannel en de Channels-weergave van de IDE geeft elk veld weer als een live widget – labels, schakelaars, schuifregelaars, grafieken, kaarten – zonder enige hostcode. Dezelfde records zijn gewone CBOR, dus een aangepaste host decodeert ze met elke CBOR-bibliotheek wanneer je de IDE ontgroeit.

Het kanaal van het voorbeeld controls_channel.py zoals de IDE het weergeeft: uitlezingen, schakelaars en schuifregelaars voor een FLIR Lepton. Elk bedieningselement schrijft terug naar het script.

12.8.1. Uitlezingen en bedieningselementen

Een CBORChannel is een dictionary van getypeerde velden. add declareert elk veld met een naam, een widget-type en de argumenten die dat type nodig heeft; toewijzen aan ch["name"] werkt een veld bij; de on_write-callback ontvangt de wijzigingen van de host:

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)

Voer het uit in de IDE, schakel het paneel onder de framebuffer naar Channels, en de widgets verschijnen: een label is een alleen-lezen waarde (met zijn unit ernaast), text is een blok statische opgemaakte tekst, een toggle is een schakelaar, slider en spinbox stellen een getal in binnen min / max / step (de spinbox voor nauwkeurige invoer), radio en select kiezen een van de options, lineedit is een vrij tekstveld, en pushbutton is een kortstondige actie die on_write aanroept met True. Het script leest zijn bedieningselementen terug met ch["Threshold"] wanneer het maar wil – het kanaal bewaart de laatste waarde – en on_write is de hook voor de elementen die een onmiddellijke reactie nodig hebben. Een (min, max, value)-tuple toegewezen aan een schuifregelaar of spinbox verplaatst het bereik samen met de waarde, en zo volgt een bedieningselement een sensor waarvan de grenzen van een andere instelling afhangen.

De optionele on_read(channel)-callback draait vlak voordat het kanaal voor de host wordt geserialiseerd – de plek om een sensor in een label te bemonsteren, zodat er alleen een meting wordt gedaan wanneer iemand kijkt:

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

Het voorbeeld controls_channel.py onder File → Examples → 12-Protocol gebruikt precies deze vorm om de meetmodus, het temperatuurbereik en de spiegel- / kantelinstellingen van een FLIR Lepton beschikbaar te maken terwijl de camera streamt.

12.8.2. Golfvormen

Een waveform-veld draagt een blok samples met een samplefrequentie. De samples zijn de ruwe bytes van een array in de typecode van het veld (standaard unsigned 16-bits); voor meerdere signalen tegelijk – de drie assen van een accelerometer – verweef je ze en geef je aan hoeveel series er zijn en hoe ze heten:

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)

De IDE plot elke golfvorm als een scrollende grafiek met één spoor per reeks, voorziet elk chunk van een tijdstempel zodat gaten tussen updates als gaten zichtbaar zijn, en biedt een spectrumweergave, triggers, markers en opnemen naar CSV-, WAV-, NumPy- of Edge Impulse-bestanden. Vervang de synthetische sinus door audio-samples of de metingen van een IMU-driver en het script is een gereedschap voor gegevensverzameling.

12.8.3. Dieptekaarten

Een depth-veld is een raster van width x height 32-bits floats, één afstand per cel – wat een time-of-flight-sensor produceert. min / max declareren het bereik dat de host op kleur moet afbeelden; de IDE kan de kleuren ook automatisch uit de gegevens afleiden:

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)

Het voorbeeld sensors_channel.py combineert een dieptekanaal als dit met een metingenkanaal en een gezichtsdetector op dezelfde camera. De aanroep send_event() is optioneel – de IDE polt kanalen uit zichzelf – maar vertelt een gebeurtenisgestuurde host dat er een nieuw frame klaarstaat zonder dat die erom hoeft te vragen.

12.8.4. Decoderen op een aangepaste host

De records zijn standaard CBOR: elke kanaallezing levert een array van maps op, gesleuteld op de integersleutels van SenML (0 naam, 1 eenheid, 2 numerieke waarde, 3 tekstwaarde, 8 datawaarde, en negatieve sleutels voor het widgettype, de opties, het bereik en de afmetingen). Een host die de IDE is ontgroeid leest het kanaal met channel_read() en decodeert het met elke CBOR-bibliotheek (pip install cbor2); een bedieningselement terugschrijven is een CBOR-array van {0: name, 2: value}-maps, verzonden met channel_write(). De referentie van protocol.CBORChannel somt elke sleutel op. Voor de meeste projecten is de Channels-weergave van de IDE echter de host, en is de enige code het cam-script hierboven.