12.8. Widgeturi și telemetrie cu CBORChannel

Backend-urile din cele două pagini anterioare mută octeți bruți, iar un canal de octeți bruți are nevoie de un program gazdă care știe cum să îi decodifice. De cele mai multe ori, ceea ce vrea să publice o cameră este mai simplu de atât: câteva citiri denumite, câteva comenzi denumite, o formă de undă, o hartă de adâncime. Pentru acest caz, pachetul protocol livrează protocol.CBORChannel, un backend gata făcut care conține câmpuri denumite, le serializează ca înregistrări CBOR în formatul SenML și decodifică scrierile gazdei înapoi în valori de câmp. Avantajul său este că OpenMV IDE îl înțelege deja: înregistrați un CBORChannel și vizualizarea Canale a IDE-ului redă fiecare câmp ca un widget în timp real – etichete, comutatoare, glisoare, grafice, hărți – fără niciun cod pe gazdă. Aceleași înregistrări sunt CBOR simplu, așa că o gazdă personalizată le decodifică cu orice bibliotecă CBOR atunci când depășiți IDE-ul.

Canalul exemplului controls_channel.py așa cum îl redă IDE-ul: afișaje, comutatoare și glisoare pentru un FLIR Lepton. Fiecare comandă scrie înapoi către script.

12.8.1. Afișaje și comenzi

Un CBORChannel este un dicționar de câmpuri tipizate. add declară fiecare câmp cu un nume, un type de widget și argumentele de care are nevoie acel tip; atribuirea către ch["name"] actualizează un câmp; callback-ul on_write primește modificările gazdei:

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)

Rulați-l în IDE, comutați panoul de sub tamponul de cadre la Canale, și widgeturile apar: un label este o valoare doar pentru citire (cu unit-ul său alături), text este un bloc de text formatat static, un toggle este un comutator, slider și spinbox stabilesc un număr în min / max / step (caseta de incrementare pentru introducere precisă), radio și select aleg una dintre options, lineedit este un câmp de text liber, iar pushbutton este o acțiune momentană care apelează on_write cu True. Scriptul își citește comenzile înapoi cu ch["Threshold"] oricând dorește – canalul păstrează ultima valoare – iar on_write este cârligul pentru cele care au nevoie de o reacție imediată. Un tuplu (min, max, value) atribuit unui glisor sau unei casete de incrementare îi mută intervalul odată cu valoarea, ceea ce este modul în care o comandă urmărește un senzor ale cărui limite depind de o altă setare.

Callback-ul opțional on_read(channel) rulează chiar înainte ca acel canal să fie serializat pentru gazdă – locul în care să eșantionați un senzor într-o etichetă, astfel încât o citire să fie luată doar când se uită cineva:

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

Exemplul controls_channel.py din File → Examples → 12-Protocol folosește exact această formă pentru a expune modul de măsurare, intervalul de temperatură și setările de oglindire / răsturnare ale unui FLIR Lepton în timp ce camera transmite.

12.8.2. Forme de undă

Un câmp waveform transportă un bloc de eșantioane cu o rată de eșantionare. Eșantioanele sunt octeții bruți ai unui array în typecode-ul câmpului (16 biți fără semn în mod implicit); pentru mai multe semnale deodată – cele trei axe ale unui accelerometru – intercalați-le și spuneți câte series există și cum să fie numite:

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-ul reprezintă fiecare formă de undă ca un grafic derulant cu câte o urmă per serie, marchează temporal fiecare bloc, astfel încât golurile dintre actualizări apar ca goluri, și oferă o vizualizare de spectru, declanșatoare, marcatoare și înregistrare în fișiere CSV, WAV, NumPy sau Edge Impulse. Înlocuiți sinusoida sintetică cu eșantioane audio sau cu citirile unui driver IMU, iar scriptul devine un instrument de colectare a datelor.

12.8.3. Hărți de adâncime

Un câmp depth este o grilă width x height de numere în virgulă mobilă pe 32 de biți, o distanță per celulă – ceea ce produce un senzor time-of-flight. min / max declară intervalul pe care gazda ar trebui să îl mapeze la culoare; IDE-ul poate de asemenea să stabilească automat intervalul culorilor din date:

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)

Exemplul sensors_channel.py combină un canal de adâncime ca acesta cu un canal de citiri și un detector de fețe pe aceeași cameră. Apelul send_event() este opțional – IDE-ul interoghează canalele de unul singur – dar îi spune unei gazde bazate pe evenimente că un cadru nou este gata fără ca aceasta să fie nevoită să întrebe.

12.8.4. Decodificarea pe o gazdă personalizată

Înregistrările sunt CBOR standard: fiecare citire a canalului returnează un tablou de hărți indexate după cheile întregi ale SenML (0 nume, 1 unitate, 2 valoare numerică, 3 valoare șir, 8 valoare de date și chei negative pentru tipul widgetului, opțiuni, interval și dimensiuni). O gazdă care a depășit IDE-ul citește canalul cu channel_read() și îl decodifică cu orice bibliotecă CBOR (pip install cbor2); scrierea înapoi a unei comenzi este un tablou CBOR de hărți {0: name, 2: value} trimis cu channel_write(). Referința protocol.CBORChannel listează fiecare cheie. Pentru majoritatea proiectelor, însă, vizualizarea Canale a IDE-ului este gazda, iar singurul cod este scriptul de pe cameră de mai sus.