12.8. Виджеты и телеметрия с CBORChannel

Бэкенды на двух предыдущих страницах передают сырые байты, а каналу сырых байтов нужна хост-программа, которая умеет их декодировать. Чаще всего то, что камера хочет публиковать, проще: несколько именованных показаний, несколько именованных элементов управления, осциллограмма, карта глубины. Для этого случая пакет protocol поставляет protocol.CBORChannel – готовый бэкенд, который хранит именованные поля, сериализует их как записи CBOR в раскладке SenML и декодирует записи хоста обратно в значения полей. Его выигрыш в том, что OpenMV IDE уже его понимает: зарегистрируйте CBORChannel, и представление Channels IDE отобразит каждое поле как живой виджет – метки, переключатели, ползунки, графики, карты – вообще без кода на хосте. Те же записи – обычный CBOR, поэтому собственный хост декодирует их любой библиотекой CBOR, когда вы перерастёте IDE.

Канал примера controls_channel.py в том виде, в каком его отображает IDE: показания, переключатели и ползунки для FLIR Lepton. Каждый элемент управления записывает изменения обратно в скрипт.

12.8.1. Показания и элементы управления

CBORChannel – это словарь типизированных полей. add объявляет каждое из них с именем, type виджета и аргументами, которые нужны этому типу; присваивание ch["name"] обновляет поле; функция обратного вызова on_write получает изменения от хоста:

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)

Запустите его в IDE, переключите панель под буфером кадра на Channels, и появятся виджеты: label – это значение только для чтения (с его unit рядом), text – блок статического форматированного текста, toggle – переключатель, slider и spinbox задают число в пределах min / max / step (счётчик – для точного ввода), radio и select выбирают один из options, lineedit – поле свободного ввода текста, а pushbutton – мгновенное действие, вызывающее on_write с True. Скрипт считывает свои элементы управления обратно через ch["Threshold"] когда угодно – канал хранит последнее значение, – а on_write – это хук для тех, которым нужна немедленная реакция. Кортеж (min, max, value), присвоенный ползунку или счётчику, перемещает его диапазон вместе со значением – так элемент управления следует за датчиком, пределы которого зависят от другой настройки.

Необязательная функция обратного вызова on_read(channel) выполняется непосредственно перед сериализацией канала для хоста – место, где можно опросить датчик в метку, чтобы показание снималось только тогда, когда кто-то смотрит:

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

Пример controls_channel.py в File → Examples → 12-Protocol использует именно эту форму, чтобы открыть доступ к режиму измерения, диапазону температур и настройкам зеркалирования / переворота FLIR Lepton, пока камера ведёт трансляцию.

12.8.2. Осциллограммы

Поле waveform несёт блок выборок с частотой дискретизации. Выборки – это сырые байты array в typecode поля (по умолчанию беззнаковые 16-битные); для нескольких сигналов сразу – трёх осей акселерометра – чередуйте их и укажите, сколько есть series и как их называть:

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 строит каждую осциллограмму как прокручивающийся график с одной трассой на серию, снабжает каждый фрагмент временной меткой, чтобы промежутки между обновлениями отображались как промежутки, и предлагает представление спектра, триггеры, маркеры и запись в файлы CSV, WAV, NumPy или Edge Impulse. Замените синтетическую синусоиду выборками audio или показаниями драйвера IMU – и скрипт станет инструментом сбора данных.

12.8.3. Карты глубины

Поле depth – это сетка width x height 32-битных чисел с плавающей точкой, по одному расстоянию на ячейку – то, что выдаёт времяпролётный датчик. min / max объявляют диапазон, который хост должен отобразить в цвет; IDE также может подобрать диапазон цветов автоматически по данным:

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)

Пример sensors_channel.py объединяет такой канал глубины с каналом показаний и детектором лиц на одной камере. Вызов send_event() необязателен – IDE опрашивает каналы сама, – но он сообщает событийно-управляемому хосту, что новый кадр готов, без необходимости спрашивать.

12.8.4. Декодирование на собственном хосте

Записи – это стандартный CBOR: каждое чтение канала возвращает массив отображений с целочисленными ключами SenML (0 – имя, 1 – единица измерения, 2 – числовое значение, 3 – строковое значение, 8 – значение данных, а отрицательные ключи – для типа виджета, вариантов, диапазона и размерностей). Хост, переросший IDE, читает канал с помощью channel_read() и декодирует его любой библиотекой CBOR (pip install cbor2); запись элемента управления обратно – это массив CBOR из отображений {0: name, 2: value}, отправляемый с помощью channel_write(). Справочник protocol.CBORChannel перечисляет каждый ключ. Впрочем, для большинства проектов хостом является представление Channels IDE, а единственный код – это скрипт на стороне камеры выше.