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 anfitrião que saiba descodificá-los. Na maior parte das vezes, o que uma câmara quer publicar é mais simples do que isso: algumas leituras nomeadas, alguns controlos nomeados, uma forma de onda, um mapa de profundidade. Para esse caso, o pacote protocol inclui protocol.CBORChannel, um backend pronto a usar que guarda campos nomeados, serializa-os como registos CBOR na disposição SenML e descodifica as escritas do anfitrião de volta em valores de campos. O seu benefício é que o OpenMV IDE já o compreende: registe um CBORChannel e a vista Canais do IDE apresenta cada campo como um widget em direto – etiquetas, interruptores, deslizadores, gráficos, mapas – sem qualquer código no anfitrião. Os mesmos registos são CBOR simples, por isso um anfitrião personalizado descodifica-os com qualquer biblioteca CBOR quando o IDE deixar de ser suficiente.

O canal do exemplo controls_channel.py tal como o IDE o apresenta: leituras, interruptores e deslizadores para uma FLIR Lepton. Cada controlo escreve de volta para o script.

12.8.1. Leituras e controlos

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 anfitrião:

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 no IDE, mude o painel sob o buffer de fotograma para Canais, e os widgets aparecem: um label é um valor só de leitura (com a sua unit ao lado), text é um bloco de texto formatado estático, um toggle é um interruptor, slider e spinbox definem um número dentro de min / max / step (a caixa de incremento para introdução 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ê os seus controlos 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. Um tuplo (min, max, value) atribuído a um deslizador ou caixa de incremento move o seu intervalo juntamente com o seu valor, que é como um controlo acompanha um sensor cujos limites dependem de outra definição.

O callback opcional on_read(channel) corre mesmo antes de o canal ser serializado para o anfitrião – o sítio para amostrar um sensor para um label, de modo a que uma leitura só seja feita quando alguém está a ver:

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

O exemplo controls_channel.py em Ficheiro → Exemplos → 12-Protocol usa exatamente esta forma para expor o modo de medição, o intervalo de temperatura e as definições de espelho / inversão de uma FLIR Lepton enquanto a câmara transmite.

12.8.2. Formas de onda

Um campo waveform transporta 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 predefinição); para vários sinais ao mesmo tempo – os três eixos de um acelerómetro – intercale-os e indique quantas series existem e como se chamam:

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)

O IDE traça cada forma de onda como um gráfico deslizante com um traço por série, marca o tempo de cada bloco para que as lacunas entre atualizações apareçam como lacunas, e oferece uma vista de espetro, disparos, marcadores e gravação em ficheiros CSV, WAV, NumPy ou Edge Impulse. Substitua a sinusoide sintética por amostras de audio ou pelas leituras de um driver de IMU e o script é uma ferramenta de recolha de dados.

12.8.3. Mapas de profundidade

Um campo depth é uma grelha de width x height de floats de 32 bits, uma distância por célula – o que um sensor de tempo de voo produz. min / max declaram o intervalo que o anfitrião deve mapear para cor; o IDE também pode ajustar o intervalo das 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 detetor de rostos na mesma câmara. A chamada a send_event() é opcional – o IDE sonda os canais por si próprio – mas avisa um anfitrião orientado a eventos de que um novo fotograma está pronto sem que ele tenha de perguntar.

12.8.4. Descodificar num anfitrião personalizado

Os registos são CBOR padrão: cada leitura de canal devolve um array de mapas indexados pelas 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, as opções, o intervalo e as dimensões). Um anfitrião para quem o IDE deixou de ser suficiente lê o canal com channel_read() e descodifica-o com qualquer biblioteca CBOR (pip install cbor2); escrever um controlo 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 vista Canais do IDE é o anfitrião, e o único código é o script do lado da câmara acima.