`protocol` — Канали протоколу OpenMV#

Модуль protocol відкриває хост-протокол OpenMV для Python. Він дозволяє ініціалізувати та налаштовувати стек протоколу на стороні мікропрограми, а також дає змогу користувацькому коду реєструвати власні логічні канали на основі Python-об’єкта, що реалізує інтерфейс каналу (read, write, size, poll тощо). Саме з цим взаємодіють настільні допоміжні інструменти при потоковій передачі даних зображень або відображенні інтерактивних віджетів підключеної камери.

Приклади#

Потокова передача зображення RGB565 на хост за допомогою власного бекенду, що реалізує низькорівневий інтерфейс каналу (backend.size(), backend.shape(), backend.poll(), backend.read()):

import csi
import protocol

csi0 = csi.CSI()
csi0.reset()
csi0.pixformat(csi.RGB565)
csi0.framesize(csi.HD)

img = csi0.snapshot()
img_mv = memoryview(img.bytearray())
frame_ready = True


class FrameChannel:
    def size(self):
        return len(img_mv)

    def shape(self):
        return (img.height(), img.width(), len(img_mv))

    def poll(self):
        return frame_ready

    def read(self, offset, size):
        global frame_ready
        end = offset + size
        chunk = img_mv[offset:end]
        if end >= len(img_mv):
            frame_ready = False
        return chunk


protocol.register(name="frame", backend=FrameChannel())

while True:
    if not frame_ready:
        img = csi0.snapshot()
        img_mv = memoryview(img.bytearray())
        frame_ready = True

Відповідний скрипт на стороні хоста, що використовує пакет openmv для Python (pip install openmv) для підключення, надсилання скрипта на камеру та отримання кожного кадру:

import cv2
import numpy as np
from openmv.camera import Camera

# The on-cam script above, stored as a string (or read from a file).
SCRIPT = open("frame_streamer_on_cam.py").read()

with Camera("/dev/ttyACM0", baudrate=921600) as cam:
    cam.stop()
    cam.exec(SCRIPT)

    while True:
        status = cam.read_status()
        if not cam.has_channel("frame") or not status.get("frame"):
            continue

        h, w, size = cam._channel_shape(cam.get_channel(name="frame"))
        if cam.channel_size("frame") < size:
            continue

        data = cam.channel_read("frame", size)
        rgb565 = np.frombuffer(data, dtype="<u2").reshape(h, w)

        # Unpack RGB565 to an HxWx3 uint8 RGB image.
        r = ((rgb565 >> 11) & 0x1F) << 3
        g = ((rgb565 >>  5) & 0x3F) << 2
        b = ( rgb565        & 0x1F) << 3
        frame = np.dstack([r, g, b]).astype(np.uint8)

        # Display with OpenCV (cv2 expects BGR, not RGB).
        cv2.imshow("OpenMV", cv2.cvtColor(frame, cv2.COLOR_RGB2BGR))
        if cv2.waitKey(1) == ord("q"):
            break

cv2.destroyAllWindows()

Замість /dev/ttyACM0 вкажіть послідовний порт камери (наприклад, COM3 у Windows). Конструктор openmv.camera.Camera приймає ті самі параметри протоколу, що й init (crc / seq / ack / events / max_payload / max_retry / timeout), якщо стек на стороні камери переналаштований відповідним чином.

Функції#

protocol.init(crc: bool = True, seq: bool = True, ack: bool = True, events: bool = True, max_payload: int = ..., rtx_retries: int = 3, rtx_timeout_ms: int = 500, lock_interval_ms: int = 10, poll_ms: int = 0) → None#

Ініціалізує (або переналаштовує) стек протоколу та реєструє стандартні логічні канали даних (stdin, stdout, stream і, якщо скомпільовано, profile). Викликає RuntimeError, якщо ініціалізація не вдалася. Мікропрограма завантажується з уже запущеним стандартним стеком протоколу USB, тому цей виклик потрібен лише для зміни транспорту або перевизначення стандартних параметрів фреймування. Повторний виклик повторно ініціалізує стек: попередній таймер опитування видаляється, а канали реєструються заново.

crc вмикає перевірку CRC для кадрів протоколу.

seq вмикає відстеження порядкових номерів.

ack вмикає підтвердження для кожного кадру.

events вмикає сповіщення про події каналу.

max_payload — максимальний розмір корисного навантаження в байтах. Якщо не вказано, використовується стандартне значення для конкретної камери; воно визначається на основі розміру буфера протоколу кожної плати: buffer - 10 (header) - 4 (CRC).

Camera	Buffer size	Max payload
OpenMV Cam M4 (`OPENMV2`)	512	498
OpenMV Cam M7 (`OPENMV3`)	512	498
OpenMV Cam H7 (`OPENMV4`)	512	498
OpenMV Cam H7 Plus (`OPENMV4P`)	4096	4082
OpenMV Pure Thermal (`OPENMVPT`)	4096	4082
OpenMV Cam RT1062 (`OPENMV_RT1060`)	4096	4082
OpenMV Cam N6 (`OPENMV_N6`)	8192	8178
OpenMV AE3 (`OPENMV_AE3`)	8192	8178
Arduino Portenta H7 (`ARDUINO_PORTENTA_H7`)	4096	4082
Arduino Giga (`ARDUINO_GIGA`)	4096	4082
Arduino Nicla Vision (`ARDUINO_NICLA_VISION`)	4096	4082

rtx_retries — кількість спроб повторної передачі. Стандартне значення: 3.

rtx_timeout_ms — тайм-аут повторної передачі в мілісекундах (подвоюється після кожного тайм-ауту). Стандартне значення: 500.

lock_interval_ms — мінімальний інтервал блокування в мілісекундах. Стандартне значення: 10.

poll_ms — інтервал опитування в мілісекундах. 0 (стандартне значення) вимикає таймерне опитування.

protocol.is_active() → bool#: Повертає True, якщо хост наразі підключений і стек протоколу активний, інакше False.

protocol.poll() → int#

Виконує одну ітерацію задачі протоколу: спорожняє приймальний буфер транспорту, розбирає всі повні фрейми, диспетчеризує команди та обслуговує події каналів. Зазвичай мікропрограма робить це за вас із переривання USB та з таймера poll_ms, тому це потрібно лише для власних транспортів, які не керуються жодним із них – наприклад, коли poll_ms=0 і транспорт опитується з головного циклу скрипта.

Повертає 0 у разі успіху та -1, якщо не зареєстровано активного транспорту або виклик було зроблено з контексту переривання.

protocol.register(name: str, *, backend: object, flags: int = 0) → ProtocolChannel#

Реєструє Python-об’єкт backend як новий логічний канал і повертає дескриптор ProtocolChannel. Доступні методи об’єкта backend (див. Інтерфейс бекенду нижче) визначають можливості каналу; protocol.CHANNEL_FLAG_READ, protocol.CHANNEL_FLAG_WRITE та protocol.CHANNEL_FLAG_LOCK автоматично додаються до flags, якщо реалізовано відповідні методи.

name — назва каналу у вигляді рядка. Обрізається до розміру буфера назви каналу мікропрограми. Обов’язково.

backend — Python-об’єкт, що реалізує інтерфейс бекенду. Обов’язково. Зазвичай передається як іменований аргумент (backend=...).

flags — додаткові біти прапорців каналу (див. константи CHANNEL_FLAG_*). Необов’язково; стандартне значення: 0.

Генерує RuntimeError, якщо канал не вдається зареєструвати (наприклад, немає вільних слотів для каналів).

Класи#

class protocol.ProtocolChannel#

Дескриптор, що повертається protocol.register. Екземпляри не створюються безпосередньо.

send_event(event: int, wait_ack: bool = False) → None#

Надіслати хосту сповіщення про подію каналу.

event — ідентифікатор події (ціле число).

wait_ack, якщо True, блокує виконання до отримання підтвердження від хоста.

Генерує RuntimeError у разі помилки надсилання події.

Інтерфейс бекенду#

Об’єкт бекенду, переданий у protocol.register, може реалізовувати будь-яку підмножину наведених нижче методів. До рівня протоколу C підключаються лише методи, наявні в об’єкті; відсутні методи залишають відповідну можливість вимкненою.

class protocol.backend#

Об’єкт бекенду каналу, переданий у protocol.register. Наведені нижче методи описують необов’язковий інтерфейс, який може реалізувати Python-бекенд.

init() → object#: Викликається один раз під час ініціалізації каналу. Повертає будь-яке значення, відмінне від None, у разі успіху; виняток або відсутність значення, що повертається, вважається помилкою.

poll() → bool#: Повертає True, якщо канал має дані, готові для читання хостом.

lock() → bool#: Захоплює канал для передачі. Повертає True у разі успіху.

unlock() → bool#: Звільняє канал після передачі. Повертає True у разі успіху.

size() → int#: Повертає кількість байтів, доступних для читання з каналу на даний момент.

shape() → tuple#: Повертає кортеж із не більше ніж чотирьох цілих чисел, що описують форму даних (наприклад, розміри зображення). Рівень протоколу використовує до чотирьох елементів.

flush() → object#: Скидає всі незбережені дані. Повертає будь-яке значення, відмінне від None, у разі успіху.

read(offset: int, size: int) → bytes#: Повертає до size байтів починаючи з offset у вигляді bytes-подібного об’єкта, що підтримує буферний протокол.

readp(offset: int, size: int) → bytes#: Варіант read без копіювання. Повертає буфер, пам’ять якого безпосередньо зчитується рівнем протоколу; буфер повинен залишатися дійсним протягом усього часу передачі.

write(offset: int, data: bytearray) → int#: Записує data за зміщенням offset. data — це bytearray, що безпосередньо посилається на C-буфер. Повертає кількість записаних байтів або 0 у разі стандартного успіху.

ioctl(cmd: int, length: int, arg: bytearray | None) → int#: Обробляє ioctl. arg дорівнює None, якщо length рівний нулю, інакше — bytearray, що безпосередньо посилається на C-буфер. Повертає 0 або None у разі успіху, або від’ємне ціле число у разі помилки.

is_active() → bool#: Для транспортних каналів повертає True, якщо базовий транспорт наразі підключений.

class protocol.CBORChannel(on_read: Callable | None = None, on_write: Callable | None = None)#

Високорівневий Python-бекенд (надається замороженим пакетом protocol), який серіалізує іменовані поля в CBOR, використовуючи цілочисельні ключі, сумісні з SenML. Підтримує віджети відображення (label, text, depth, waveform) та інтерактивні елементи керування (toggle, pushbutton, slider, spinbox, select, radio, lineedit) зі зворотними викликами on_read/on_write.

on_read — необов’язковий виклик on_read(channel), що виконується перед серіалізацією каналу для хоста. Використовується для оновлення значень полів.

on_write — необов’язковий виклик on_write(channel, name, value), що виконується, коли хост записує нове значення для іменованого поля.

Додати іменоване поле до каналу.

name — відображувана назва; повинна бути унікальною в межах цього каналу.

type — тип віджета:

"label" – числовий індикатор лише для читання з необов’язковим unit.
"text" – рядковий індикатор лише для читання.
"toggle" – логічний перемикач.
"pushbutton" – миттєва дія (запуск, калібрування, …). Хост записує True при натисканні і не відображає жодного стану.
"slider" – числовий елемент керування з min/max/step.
"spinbox" – точне покрокове числове введення; має ті самі поля, що й повзунок, і ту саму форму оновлення кортежем (min, max, value).
"select" – розкривний список рядків options.
"radio" – options, відображені як група перемикачів (radio buttons).
"lineedit" – рядкове поле введення з можливістю запису.
"depth" – відображення 2D-карти глибини розміром width x height.
"waveform" – 1D-графік часового ряду з чергованих вибірок.

value — початкове значення. Стандартне значення залежить від type: 0 для label, "" для text/lineedit/select/ radio, False для toggle/pushbutton, min для slider/spinbox.

unit — рядок одиниці вимірювання для label/slider/spinbox/ waveform (наприклад, "Cel", "%RH").

min — мінімальне значення (діапазон slider/spinbox або діапазон відображення depth/waveform).

max — максимальне значення (діапазон slider/spinbox або діапазон відображення depth/waveform).

step — розмір кроку (slider/spinbox).

options — список рядків варіантів (select/radio) або список назв окремих серій (waveform).

width — ширина в пікселях (depth).

height — висота в пікселях (depth).

sample_rate — частота дискретизації у вибірках за секунду (waveform). Надсилається хосту як період дискретизації.

series — кількість чергованих серій у кожному блоці вибірок (waveform), наприклад 3 для запису акселерометра XYZ.

typecode — код типу array вибірок сигналу (наприклад, "H", "h", "f"). Стандартно "H" (беззнакове 16-бітне).

__getitem__(name: str) → object#: Повертає поточне значення іменованого поля. Для полів depth і waveform повертається буфер двійкових даних, інакше — скалярне значення.

__setitem__(name: str, value: Any) → None#: Встановлює значення іменованого поля. Для полів slider і spinbox кортеж (min, max, value) оновлює діапазон і поточне значення одночасно. Для полів depth value — це буфер двійкових даних. Для полів waveform value — це байтоподібний блок чергованих вибірок у typecode поля; кількість вибірок на серію визначається з його довжини, і додається монотонна мітка часу (у мікросекундах, накопичена з time.ticks_us()), щоб хост міг розмістити фрагмент на своїй осі часу.

poll() → bool#: Метод інтерфейсу бекенду. Повертає True, якщо серіалізовані дані готові для хоста.

size() → int#: Метод інтерфейсу бекенду. Викликає on_read (якщо встановлено) і повертає розмір серіалізованого буфера.

read(offset: int, size: int) → bytes#: Метод інтерфейсу бекенду. Повертає фрагмент серіалізованого буфера.

write(offset: int, data: bytearray) → int#: Метод інтерфейсу бекенду. Декодує список оновлень CBOR та застосовує значення до відповідних іменованих полів, викликаючи on_write для кожного.

Константи#

Біти прапорців каналу (об’єднуються побітово; передаються в protocol.register через flags або встановлюються автоматично на основі методів бекенду).

protocol.CHANNEL_FLAG_READ: int#: Канал підтримує читання.

protocol.CHANNEL_FLAG_WRITE: int#: Канал підтримує записування.

protocol.CHANNEL_FLAG_LOCK: int#: Канал реалізує lock/unlock.

protocol.CHANNEL_FLAG_PHYSICAL: int#: Канал представляє фізичний транспорт (на відміну від логічного каналу даних).

Вбудовані ідентифікатори каналів.

protocol.CHANNEL_ID_TRANSPORT: int#: Зарезервований ідентифікатор каналу для активного транспорту.

protocol.CHANNEL_ID_STDIN: int#: Ідентифікатор вбудованого каналу stdin.

protocol.CHANNEL_ID_STDOUT: int#: Ідентифікатор вбудованого каналу stdout.

protocol.CHANNEL_ID_STREAM: int#: Ідентифікатор вбудованого каналу stream.

protocol.CHANNEL_ID_PROFILE: int#: Ідентифікатор вбудованого каналу профілювальника (присутній лише тоді, коли мікропрограма зібрана з увімкненим профілювальником).

protocol — Канали протоколу OpenMV#