12.8. Widgets e telemetria com CBORChannel

Os backends das duas páginas anteriores movem bytes brutos, e um canal de bytes brutos precisa de um programa de host que saiba decodificá-los. Na maior parte do tempo, o que uma câmera quer publicar é mais simples do que isso: algumas leituras nomeadas, alguns controles nomeados, uma forma de onda, um mapa de profundidade. Para esse caso o pacote protocol traz protocol.CBORChannel, um backend pronto que guarda campos nomeados, os serializa como registros CBOR no layout SenML e decodifica as escritas do host de volta em valores de campo. Sua recompensa é que o OpenMV IDE já o entende: registre um CBORChannel e a visualização Canais da IDE renderiza cada campo como um widget ao vivo – rótulos, chaves, controles deslizantes, gráficos, mapas – sem nenhum código de host. Os mesmos registros são CBOR simples, então um host personalizado os decodifica com qualquer biblioteca CBOR quando você supera a IDE.

O canal do exemplo controls_channel.py como a IDE o renderiza: leituras, chaves e controles deslizantes para um FLIR Lepton. Todo controle escreve de volta no script.

12.8.1. Leituras e controles

Um CBORChannel é um dicionário de campos tipados. add declara cada um com um nome, um type de widget e os argumentos de que esse tipo precisa; atribuir a ch["name"] atualiza um campo; o callback on_write recebe as alterações do 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)

Execute-o na IDE, mude o painel abaixo do frame buffer para Canais, e os widgets aparecem: um label é um valor somente leitura (com sua unit ao lado), text é um bloco de rich text estático, um toggle é uma chave, slider e spinbox definem um número dentro de min / max / step (a caixa de incremento para entrada precisa), radio e select escolhem uma das options, lineedit é um campo de texto livre, e pushbutton é uma ação momentânea que chama on_write com True. O script lê seus controles de volta com ch["Threshold"] sempre que quiser – o canal guarda o valor mais recente – e on_write é o gancho para os que precisam de uma reação imediata. Uma tupla (min, max, value) atribuída a um controle deslizante ou caixa de incremento move sua faixa junto com seu valor, que é como um controle acompanha um sensor cujos limites dependem de outra configuração.

O callback opcional on_read(channel) roda logo antes de o canal ser serializado para o host – o lugar para amostrar um sensor em um label, de modo que uma leitura só seja feita quando alguém está olhando:

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

O exemplo controls_channel.py em File → Examples → 12-Protocol usa exatamente essa forma para expor o modo de medição, a faixa de temperatura e as configurações de espelhamento / inversão de um FLIR Lepton enquanto a câmera transmite.

12.8.2. Formas de onda

Um campo waveform carrega um bloco de amostras com uma taxa de amostragem. As amostras são os bytes brutos de um array no typecode do campo (16 bits sem sinal por padrão); para vários sinais ao mesmo tempo – os três eixos de um acelerômetro – intercale-os e informe quantas series existem e como chamá-las:

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)

A IDE plota cada forma de onda como um gráfico rolante com um traço por série, marca o tempo de cada bloco para que lacunas entre atualizações apareçam como lacunas, e oferece uma visualização de espectro, gatilhos, marcadores e gravação em arquivos CSV, WAV, NumPy ou Edge Impulse. Substitua o seno sintético por amostras de audio ou pelas leituras de um driver de IMU e o script é uma ferramenta de coleta de dados.

12.8.3. Mapas de profundidade

Um campo depth é uma grade de width x height floats de 32 bits, uma distância por célula – o que um sensor time-of-flight produz. min / max declaram a faixa que o host deve mapear em cor; a IDE também pode ajustar as cores automaticamente a partir dos dados:

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)

O exemplo sensors_channel.py combina um canal de profundidade como este com um canal de leituras e um detector de rostos na mesma câmera. A chamada send_event() é opcional – a IDE sonda os canais por conta própria – mas ela avisa a um host orientado a eventos que um novo quadro está pronto sem que ele precise perguntar.

12.8.4. Decodificando em um host personalizado

Os registros são CBOR padrão: cada leitura de canal retorna um array de mapas com as chaves inteiras do SenML (0 nome, 1 unidade, 2 valor numérico, 3 valor de string, 8 valor de dados, e chaves negativas para o tipo de widget, opções, faixa e dimensões). Um host que superou a IDE lê o canal com channel_read() e o decodifica com qualquer biblioteca CBOR (pip install cbor2); escrever um controle de volta é um array CBOR de mapas {0: name, 2: value} enviado com channel_write(). A referência de protocol.CBORChannel lista todas as chaves. Para a maioria dos projetos, porém, a visualização Canais da IDE é o host, e o único código é o script do lado da câmera acima.