12.8. Widgetar och telemetri med CBORChannel

Backenderna på de två föregående sidorna flyttar råa byte, och en rå bytekanal behöver ett värdprogram som vet hur de ska avkodas. För det mesta är det en kamera vill publicera enklare än så: några namngivna avläsningar, några namngivna kontroller, en vågform, en djupkarta. För det fallet levererar paketet protocol protocol.CBORChannel, en färdig backend som håller namngivna fält, serialiserar dem som CBOR-poster i SenML-layouten och avkodar värdens skrivningar tillbaka till fältvärden. Vinsten är att OpenMV IDE redan förstår den: registrera en CBORChannel så renderar IDE:ns Channels-vy varje fält som en live-widget – etiketter, brytare, skjutreglage, grafer, kartor – helt utan värdkod. Samma poster är vanlig CBOR, så en anpassad värd avkodar dem med vilket CBOR-bibliotek som helst när du växer ur IDE:n.

Kanalen i exemplet controls_channel.py så som IDE:n renderar den: avläsningar, växlare och skjutreglage för en FLIR Lepton. Varje kontroll skriver tillbaka till skriptet.

12.8.1. Avläsningar och kontroller

En CBORChannel är en ordbok av typade fält. add deklarerar vart och ett med ett namn, en widget-type och de argument typen behöver; tilldelning till ch["name"] uppdaterar ett fält; on_write-återanropet tar emot värdens ändringar:

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)

Kör det i IDE:n, växla rutan under bildbufferten till Channels, så visas widgetarna: en label är ett skrivskyddat värde (med sin unit bredvid), text är ett block med statisk formaterad text, en toggle är en brytare, slider och spinbox sätter ett tal inom min / max / step (spinnrutan för exakt inmatning), radio och select väljer ett av options, lineedit är ett fritextfält, och pushbutton är en tillfällig åtgärd som anropar on_write med True. Skriptet läser tillbaka sina kontroller med ch["Threshold"] närhelst det vill – kanalen håller det senaste värdet – och on_write är kroken för dem som behöver en omedelbar reaktion. En (min, max, value)-tupel tilldelad ett skjutreglage eller en spinnruta flyttar dess intervall tillsammans med dess värde, vilket är hur en kontroll följer en sensor vars gränser beror på en annan inställning.

Det valfria återanropet on_read(channel) körs precis innan kanalen serialiseras för värden – platsen att sampla en sensor in i en label så att en avläsning bara görs när någon tittar:

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

Exemplet controls_channel.py under File → Examples → 12-Protocol använder exakt denna form för att exponera en FLIR Leptons mätläge, temperaturintervall och spegel- / vändinställningar medan kameran strömmar.

12.8.2. Vågformer

Ett waveform-fält bär ett block av sampel med en samplingsfrekvens. Samplen är de råa byten i en array med fältets typecode (osignerad 16-bitars som standard); för flera signaler samtidigt – en accelerometers tre axlar – fläta samman dem och ange hur många series det finns och vad de ska heta:

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:n ritar varje vågform som en rullande graf med ett spår per serie, tidsstämplar varje block så att luckor mellan uppdateringar visas som luckor, och erbjuder en spektrumvy, triggrar, markörer och inspelning till CSV-, WAV-, NumPy- eller Edge Impulse-filer. Ersätt den syntetiska sinusen med audio-sampel eller en IMU-drivrutins avläsningar så är skriptet ett datainsamlingsverktyg.

12.8.3. Djupkartor

Ett depth-fält är ett width x height-rutnät av 32-bitars flyttal, ett avstånd per cell – vad en time-of-flight-sensor producerar. min / max deklarerar intervallet värden ska mappa till färg; IDE:n kan också sätta färgintervallet automatiskt utifrån datan:

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)

Exemplet sensors_channel.py parar ihop en depth-kanal som denna med en avläsningskanal och en ansiktsdetektor på samma kamera. Anropet send_event() är valfritt – IDE:n pollar kanaler på egen hand – men det talar om för en händelsedriven värd att en ny bildruta är klar utan att den behöver fråga.

12.8.4. Avkodning på en anpassad värd

Posterna är standard-CBOR: varje kanalläsning returnerar en array av mappningar med SenML:s heltalsnycklar (0 namn, 1 enhet, 2 numeriskt värde, 3 strängvärde, 8 datavärde, och negativa nycklar för widgettyp, alternativ, intervall och dimensioner). En värd som vuxit ur IDE:n läser kanalen med channel_read() och avkodar den med vilket CBOR-bibliotek som helst (pip install cbor2); att skriva tillbaka en kontroll är en CBOR-array av {0: name, 2: value}-mappningar som skickas med channel_write(). Referensen för protocol.CBORChannel listar varje nyckel. För de flesta projekt är dock IDE:ns Channels-vy värden, och den enda koden är kameraskriptet ovan.