12.8. Widgets et télémétrie avec CBORChannel

Les backends des deux pages précédentes déplacent des octets bruts, et un canal d’octets bruts a besoin d’un programme hôte qui sait les décoder. La plupart du temps, ce qu’une caméra veut publier est plus simple que cela : quelques relevés nommés, quelques commandes nommées, une forme d’onde, une carte de profondeur. Pour ce cas, le paquet protocol fournit protocol.CBORChannel, un backend prêt à l’emploi qui contient des champs nommés, les sérialise en enregistrements CBOR selon la disposition SenML, et décode les écritures de l’hôte en valeurs de champ. Son avantage est qu”OpenMV IDE le comprend déjà : enregistrez un CBORChannel et la vue Canaux de l’IDE affiche chaque champ sous forme de widget en direct – étiquettes, interrupteurs, curseurs, graphes, cartes – sans aucun code hôte. Les mêmes enregistrements sont du CBOR ordinaire, de sorte qu’un hôte personnalisé les décode avec n’importe quelle bibliothèque CBOR lorsque vous dépassez l’IDE.

Le canal de l’exemple controls_channel.py tel que l’IDE l’affiche : relevés, bascules et curseurs pour un FLIR Lepton. Chaque commande écrit en retour vers le script.

12.8.1. Relevés et commandes

Un CBORChannel est un dictionnaire de champs typés. add déclare chacun d’eux avec un nom, un type de widget et les arguments dont ce type a besoin ; affecter ch["name"] met à jour un champ ; le rappel on_write reçoit les modifications de l’hôte:

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)

Exécutez-le dans l’IDE, basculez le volet sous le tampon d’image sur Canaux, et les widgets apparaissent : un label est une valeur en lecture seule (avec son unit à côté), text est un bloc de texte enrichi statique, un toggle est un interrupteur, slider et spinbox fixent un nombre dans min / max / step (la spin box pour une saisie précise), radio et select choisissent l’une des options, lineedit est un champ de texte libre, et pushbutton est une action momentanée qui appelle on_write avec True. Le script relit ses commandes avec ch["Threshold"] quand il le souhaite – le canal conserve la dernière valeur – et on_write est le point d’accroche pour celles qui exigent une réaction immédiate. Un tuple (min, max, value) affecté à un curseur ou à une spin box déplace sa plage avec sa valeur, ce qui permet à une commande de suivre un capteur dont les limites dépendent d’un autre réglage.

Le rappel facultatif on_read(channel) s’exécute juste avant que le canal ne soit sérialisé pour l’hôte – l’endroit où échantillonner un capteur dans un label pour qu’une mesure ne soit prise que lorsque quelqu’un regarde:

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

L’exemple controls_channel.py sous File → Examples → 12-Protocol utilise exactement cette forme pour exposer le mode de mesure, la plage de température et les réglages de miroir / retournement d’un FLIR Lepton pendant que la caméra diffuse.

12.8.2. Formes d’onde

Un champ waveform transporte un bloc d’échantillons avec une fréquence d’échantillonnage. Les échantillons sont les octets bruts d’un array dans le typecode du champ (16 bits non signé par défaut) ; pour plusieurs signaux à la fois – les trois axes d’un accéléromètre – entrelacez-les et indiquez combien de series il y a et comment les nommer:

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 trace chaque forme d’onde sous forme de graphe défilant avec une courbe par série, horodate chaque bloc pour que les trous entre les mises à jour apparaissent comme des trous, et propose une vue spectrale, des déclencheurs, des marqueurs et l’enregistrement dans des fichiers CSV, WAV, NumPy ou Edge Impulse. Remplacez la sinusoïde synthétique par des échantillons audio ou les mesures d’un pilote d’IMU et le script devient un outil de collecte de données.

12.8.3. Cartes de profondeur

Un champ depth est une grille de width x height flottants 32 bits, une distance par cellule – ce que produit un capteur temps de vol. min / max déclarent la plage que l’hôte doit convertir en couleur ; l’IDE peut aussi étalonner les couleurs automatiquement à partir des données:

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’exemple sensors_channel.py associe un canal depth comme celui-ci à un canal de mesures et à un détecteur de visages sur la même caméra. L’appel send_event() est facultatif – l’IDE interroge les canaux de lui-même – mais il indique à un hôte piloté par événements qu’une nouvelle trame est prête sans qu’il ait à le demander.

12.8.4. Décodage sur un hôte personnalisé

Les enregistrements sont du CBOR standard : chaque lecture de canal renvoie un tableau de maps indexées par les clés entières de SenML (0 nom, 1 unité, 2 valeur numérique, 3 valeur chaîne, 8 valeur de données, et des clés négatives pour le type de widget, les options, la plage et les dimensions). Un hôte qui a dépassé l’IDE lit le canal avec channel_read() et le décode avec n’importe quelle bibliothèque CBOR (pip install cbor2) ; réécrire une commande est un tableau CBOR de maps {0: name, 2: value} envoyé avec channel_write(). La référence protocol.CBORChannel liste chaque clé. Pour la plupart des projets, cependant, la vue Canaux de l’IDE est l’hôte, et le seul code est le script côté caméra ci-dessus.