12.8. Widgets y telemetría con CBORChannel

Los backends de las dos páginas anteriores mueven bytes en bruto, y un canal de bytes en bruto necesita un programa de host que sepa decodificarlos. La mayoría de las veces lo que una cámara quiere publicar es más sencillo que eso: unas cuantas lecturas con nombre, unos cuantos controles con nombre, una forma de onda, un mapa de profundidad. Para ese caso el paquete protocol incluye protocol.CBORChannel, un backend ya preparado que contiene campos con nombre, los serializa como registros CBOR con la disposición SenML y decodifica las escrituras del host de vuelta en valores de campo. Su recompensa es que OpenMV IDE ya lo entiende: registra un CBORChannel y la vista Canales del IDE representa cada campo como un widget en vivo – etiquetas, interruptores, deslizadores, gráficas, mapas – sin ningún código de host. Los mismos registros son CBOR sin más, así que un host personalizado los decodifica con cualquier biblioteca CBOR cuando el IDE se te queda pequeño.

El canal del ejemplo controls_channel.py tal como lo representa el IDE: lecturas, interruptores y deslizadores para un FLIR Lepton. Cada control escribe de vuelta en el script.

12.8.1. Lecturas y controles

Un CBORChannel es un diccionario de campos tipados. add declara cada uno con un nombre, un type de widget y los argumentos que ese tipo necesita; asignar a ch["name"] actualiza un campo; el callback on_write recibe los cambios del 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)

Ejecútalo en el IDE, cambia el panel bajo el búfer de fotogramas a Canales, y los widgets aparecen: una etiqueta es un valor de solo lectura (con su unit al lado), text es un bloque de texto enriquecido estático, un toggle es un interruptor, slider y spinbox fijan un número dentro de min / max / step (el cuadro numérico para una entrada precisa), radio y select eligen una de las options, lineedit es un campo de texto libre, y pushbutton es una acción momentánea que llama a on_write con True. El script lee sus controles de vuelta con ch["Threshold"] cuando quiera – el canal conserva el último valor – y on_write es el gancho para los que necesitan una reacción inmediata. Una tupla (min, max, value) asignada a un deslizador o a un cuadro numérico mueve su rango junto con su valor, que es como un control sigue a un sensor cuyos límites dependen de otro ajuste.

El callback opcional on_read(channel) se ejecuta justo antes de que el canal se serialice para el host – el lugar para muestrear un sensor en una etiqueta, de modo que una lectura solo se tome cuando alguien está mirando:

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

El ejemplo controls_channel.py bajo File → Examples → 12-Protocol usa exactamente esta forma para exponer el modo de medición, el rango de temperatura y los ajustes de espejo / volteo de un FLIR Lepton mientras la cámara transmite.

12.8.2. Formas de onda

Un campo waveform transporta un bloque de muestras con una frecuencia de muestreo. Las muestras son los bytes en bruto de un array en el typecode del campo (entero sin signo de 16 bits por defecto); para varias señales a la vez – los tres ejes de un acelerómetro – intercálalas e indica cuántas series hay y cómo llamarlas:

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)

El IDE representa cada forma de onda como una gráfica con desplazamiento con una traza por serie, marca la hora de cada fragmento para que los huecos entre actualizaciones aparezcan como huecos, y ofrece una vista de espectro, disparos, marcadores y grabación en archivos CSV, WAV, NumPy o Edge Impulse. Sustituye la senoidal sintética por muestras de audio o por las lecturas de un controlador de IMU y el script es una herramienta de recopilación de datos.

12.8.3. Mapas de profundidad

Un campo depth es una cuadrícula de width x height de flotantes de 32 bits, una distancia por celda – lo que produce un sensor de tiempo de vuelo. min / max declaran el rango que el host debería mapear a color; el IDE también puede ajustar el rango de los colores automáticamente a partir de los datos:

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)

El ejemplo sensors_channel.py combina un canal de profundidad como este con un canal de lecturas y un detector de caras en la misma cámara. La llamada a send_event() es opcional – el IDE sondea los canales por su cuenta – pero indica a un host dirigido por eventos que hay un nuevo fotograma listo sin que tenga que preguntar.

12.8.4. Decodificación en un host personalizado

Los registros son CBOR estándar: cada lectura de canal devuelve un array de mapas indexados por las claves enteras de SenML (0 nombre, 1 unidad, 2 valor numérico, 3 valor de cadena, 8 valor de datos, y claves negativas para el tipo de widget, las opciones, el rango y las dimensiones). Un host al que el IDE se le ha quedado pequeño lee el canal con channel_read() y lo decodifica con cualquier biblioteca CBOR (pip install cbor2); escribir un control de vuelta es un array CBOR de mapas {0: name, 2: value} enviado con channel_write(). La referencia de protocol.CBORChannel enumera todas las claves. Para la mayoría de los proyectos, sin embargo, la vista Canales del IDE es el host, y el único código es el script del lado de la cámara de arriba.