12.8. Widget e telemetria con CBORChannel

I backend delle due pagine precedenti spostano byte grezzi, e un canale di byte grezzi ha bisogno di un programma host che sappia decodificarli. Il più delle volte ciò che una cam vuole pubblicare è più semplice: qualche lettura con nome, qualche controllo con nome, una forma d’onda, una mappa di profondità. Per questo caso il pacchetto protocol include protocol.CBORChannel, un backend pronto all’uso che contiene campi con nome, li serializza come record CBOR nel layout SenML e decodifica le scritture dell’host riportandole in valori dei campi. Il suo vantaggio è che OpenMV IDE lo capisce già: registra un CBORChannel e la vista Canali dell’IDE visualizza ogni campo come un widget in tempo reale – etichette, interruttori, cursori, grafici, mappe – senza alcun codice host. Gli stessi record sono semplice CBOR, così un host personalizzato li decodifica con qualsiasi libreria CBOR quando l’IDE non ti basta più.

Il canale dell’esempio controls_channel.py come lo visualizza l’IDE: letture, interruttori e cursori per un FLIR Lepton. Ogni controllo scrive di ritorno sullo script.

12.8.1. Letture e controlli

Un CBORChannel è un dizionario di campi tipizzati. add dichiara ciascuno con un nome, un type di widget e gli argomenti richiesti da quel tipo; l’assegnazione a ch["name"] aggiorna un campo; il callback on_write riceve le modifiche dell’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)

Eseguilo nell’IDE, passa il pannello sotto il frame buffer a Canali, e i widget compaiono: una label è un valore di sola lettura (con la sua unit accanto), text è un blocco di rich text statico, un toggle è un interruttore, slider e spinbox impostano un numero entro min / max / step (la casella numerica per l’inserimento preciso), radio e select scelgono una delle options, lineedit è un campo di testo libero, e pushbutton è un’azione momentanea che chiama on_write con True. Lo script rilegge i suoi controlli con ch["Threshold"] quando vuole – il canale conserva l’ultimo valore – e on_write è il gancio per quelli che richiedono una reazione immediata. Una tupla (min, max, value) assegnata a un cursore o a una casella numerica ne sposta l’intervallo insieme al valore, che è il modo in cui un controllo segue un sensore i cui limiti dipendono da un’altra impostazione.

Il callback opzionale on_read(channel) viene eseguito subito prima che il canale sia serializzato per l’host – il posto in cui campionare un sensore in una label, così che una lettura venga effettuata solo quando qualcuno sta guardando:

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

L’esempio controls_channel.py in File → Examples → 12-Protocol usa esattamente questa forma per esporre la modalità di misura, l’intervallo di temperatura e le impostazioni di specchiatura / capovolgimento di un FLIR Lepton mentre la camera trasmette.

12.8.2. Forme d’onda

Un campo waveform trasporta un blocco di campioni con una frequenza di campionamento. I campioni sono i byte grezzi di un array nel typecode del campo (a 16 bit senza segno per impostazione predefinita); per più segnali insieme – i tre assi di un accelerometro – interlacciali e indica quante series ci sono e come chiamarle:

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)

L’IDE traccia ogni forma d’onda come un grafico a scorrimento con una traccia per serie, marca temporalmente ogni blocco così che le pause tra gli aggiornamenti appaiano come vuoti, e offre una vista spettro, trigger, marcatori e la registrazione su file CSV, WAV, NumPy o Edge Impulse. Sostituisci la sinusoide sintetica con campioni audio o con le letture di un driver IMU e lo script è uno strumento di raccolta dati.

12.8.3. Mappe di profondità

Un campo depth è una griglia width x height di float a 32 bit, una distanza per cella – ciò che produce un sensore time-of-flight. min / max dichiarano l’intervallo che l’host dovrebbe mappare in colore; l’IDE può anche determinare automaticamente l’intervallo dei colori dai dati:

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)

L’esempio sensors_channel.py abbina un canale depth come questo a un canale di letture e a un rilevatore di volti sulla stessa camera. La chiamata send_event() è opzionale – l’IDE interroga i canali da sé – ma comunica a un host guidato dagli eventi che un nuovo frame è pronto senza che debba chiederlo.

12.8.4. Decodifica su un host personalizzato

I record sono CBOR standard: ogni lettura del canale restituisce un array di mappe con le chiavi intere di SenML (0 nome, 1 unità, 2 valore numerico, 3 valore stringa, 8 valore dati, e chiavi negative per il tipo di widget, le opzioni, l’intervallo e le dimensioni). Un host a cui l’IDE non basta più legge il canale con channel_read() e lo decodifica con qualsiasi libreria CBOR (pip install cbor2); scrivere di ritorno un controllo è un array CBOR di mappe {0: name, 2: value} inviato con channel_write(). Il riferimento di protocol.CBORChannel elenca ogni chiave. Per la maggior parte dei progetti, però, l’host è la vista Canali dell’IDE, e l’unico codice è lo script lato cam qui sopra.