`protocol` — Canali del protocollo OpenMV#

Il modulo protocol espone il protocollo host di OpenMV a Python. Consente di inizializzare e configurare lo stack di protocollo lato firmware e permette al codice utente di registrare canali logici personalizzati supportati da un oggetto Python che implementa l’interfaccia del canale (read, write, size, poll, ecc.). È ciò con cui dialogano gli strumenti companion desktop quando trasmettono dati immagine in streaming o espongono widget interattivi a una camera collegata.

Esempi#

Trasmette in streaming un’immagine RGB565 a uno strumento host usando un backend personalizzato che implementa l’interfaccia raw del canale (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

Lo script lato host corrispondente, che usa il pacchetto Python openmv (pip install openmv) per connettersi, inviare lo script sulla camera e recuperare ogni frame:

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()

Sostituisci /dev/ttyACM0 con la porta seriale della camera (ad esempio COM3 su Windows). Il costruttore openmv.camera.Camera accetta gli stessi parametri di protocollo di init (crc / seq / ack / events / max_payload / max_retry / timeout) quando lo stack lato camera è stato riconfigurato in modo corrispondente.

Funzioni#

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#

Inizializza (o riconfigura) lo stack di protocollo e registra i canali dati logici predefiniti (stdin, stdout, stream e, se compilato, profile). Solleva RuntimeError se l’inizializzazione fallisce. Il firmware si avvia con uno stack di protocollo USB predefinito già in esecuzione, quindi chiamare questa funzione è necessario solo per cambiare il trasporto o sovrascrivere i parametri di framing predefiniti. Chiamarla di nuovo reinizializza lo stack: il timer di polling precedente viene rimosso e i canali vengono registrati da capo.

crc abilita la validazione CRC sui frame di protocollo.

seq abilita il tracciamento dei numeri di sequenza.

ack abilita gli acknowledgement per ogni frame.

events abilita le notifiche di evento del canale.

max_payload è la dimensione massima del payload in byte. Se omesso viene usato il valore predefinito per camera indicato di seguito; deriva dalla dimensione del buffer di protocollo di ciascuna scheda secondo la formula buffer - 10 (header) - 4 (CRC).

Camera	Dimensione buffer	Payload massimo
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 è il numero di tentativi di ritrasmissione. Predefinito 3.

rtx_timeout_ms è il timeout di ritrasmissione in millisecondi (raddoppiato dopo ogni timeout). Predefinito 500.

lock_interval_ms è l’intervallo di lock minimo in millisecondi. Predefinito 10.

poll_ms è l’intervallo di polling in millisecondi. 0 (il valore predefinito) disabilita il polling tramite timer.

protocol.is_active() → bool#: Restituisce True se un host è attualmente connesso e lo stack di protocollo è attivo, altrimenti False.

protocol.poll() → int#

Esegue un’iterazione del task di protocollo: svuota il buffer di ricezione del trasporto, analizza i frame completi, smista i comandi e gestisce gli eventi dei canali. Normalmente il firmware lo fa per te dall’interrupt USB e dal timer poll_ms, quindi è necessario solo per trasporti personalizzati che non sono pilotati da nessuno dei due – ad esempio quando poll_ms=0 e il trasporto viene interrogato in polling dal loop principale dello script.

Restituisce 0 in caso di successo e -1 se non è registrato alcun trasporto attivo o se la chiamata è stata effettuata da un contesto di interrupt.

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

Registra un oggetto Python backend come nuovo canale logico e restituisce un handle ProtocolChannel. I metodi disponibili dell’oggetto backend (vedi Interfaccia del backend di seguito) determinano le capacità del canale; protocol.CHANNEL_FLAG_READ, protocol.CHANNEL_FLAG_WRITE e protocol.CHANNEL_FLAG_LOCK vengono aggiunti automaticamente a flags quando i metodi corrispondenti sono implementati.

name è il nome del canale come stringa. Viene troncato alla dimensione del buffer del nome canale del firmware. Obbligatorio.

backend è l’oggetto Python che implementa l’interfaccia del backend. Obbligatorio. Tipicamente passato per parola chiave (backend=...).

flags sono bit di flag aggiuntivi del canale (vedi le costanti CHANNEL_FLAG_*). Opzionale; il valore predefinito è 0.

Solleva RuntimeError se il canale non può essere registrato (ad esempio se non ci sono slot di canale liberi).

Classi#

class protocol.ProtocolChannel#

Handle restituito da protocol.register. Le istanze non vengono costruite direttamente.

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

Invia una notifica di evento del canale all’host.

event è l’identificatore dell’evento (intero).

wait_ack se True blocca fino a quando l’host conferma l’evento.

Solleva RuntimeError se l’invio dell’evento fallisce.

Interfaccia del backend#

Un oggetto backend passato a protocol.register può implementare un qualsiasi sottoinsieme dei metodi seguenti. Solo i metodi presenti nell’oggetto vengono collegati al livello di protocollo C; i metodi mancanti lasciano disabilitata la capacità corrispondente.

class protocol.backend#

Oggetto backend del canale passato a protocol.register. I metodi seguenti descrivono l’interfaccia opzionale che un backend Python può implementare.

init() → object#: Richiamato una volta sola quando il canale viene inizializzato. Restituisce un qualsiasi valore diverso da None in caso di successo; un’eccezione o un valore di ritorno mancante viene trattato come errore.

poll() → bool#: Restituisce True se il canale ha dati pronti per essere letti dall’host.

lock() → bool#: Acquisisce il canale per un trasferimento. Restituisce True in caso di successo.

unlock() → bool#: Rilascia il canale dopo un trasferimento. Restituisce True in caso di successo.

size() → int#: Restituisce il numero di byte attualmente leggibili dal canale.

shape() → tuple#: Restituisce una tupla di un massimo di quattro interi che descrivono la forma dei dati (ad esempio le dimensioni dell’immagine). Il livello di protocollo consuma fino a quattro elementi.

flush() → object#: Svuota eventuali dati in sospeso. Restituisce un qualsiasi valore diverso da None in caso di successo.

read(offset: int, size: int) → bytes#: Restituisce fino a size byte a partire da offset come oggetto di tipo bytes che supporta il protocollo buffer.

readp(offset: int, size: int) → bytes#: Variante zero-copy di read. Restituisce un buffer la cui memoria sottostante viene letta direttamente dal livello di protocollo; il buffer deve rimanere valido per tutta la durata del trasferimento.

write(offset: int, data: bytearray) → int#: Scrive data a offset. data è un bytearray che fa riferimento direttamente al buffer C. Restituisce il numero di byte scritti, oppure 0 come successo predefinito.

ioctl(cmd: int, length: int, arg: bytearray | None) → int#: Gestisce una ioctl. arg è None se length è zero, altrimenti è un bytearray che fa riferimento al buffer C. Restituisce 0 o None in caso di successo, oppure un intero negativo in caso di errore.

is_active() → bool#: Per i canali di trasporto, restituisce True se il trasporto sottostante è attualmente connesso.

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

Un backend Python di più alto livello (fornito dal pacchetto frozen protocol) che serializza campi con nome in CBOR usando chiavi intere compatibili con SenML. Supporta widget di visualizzazione (label, text, depth, waveform) e controlli interattivi (toggle, pushbutton, slider, spinbox, select, radio, lineedit) con callback on_read/on_write.

on_read è un callable opzionale on_read(channel) invocato prima che il canale venga serializzato per l’host. Usalo per aggiornare i valori dei campi.

on_write è un callable opzionale on_write(channel, name, value) invocato quando l’host scrive un nuovo valore per un campo denominato.

Aggiunge un campo denominato al canale.

name è il nome visualizzato; deve essere univoco all’interno di questo canale.

type è il tipo di widget:

"label" – visualizzazione numerica in sola lettura con una unit opzionale.
"text" – visualizzazione di stringa in sola lettura.
"toggle" – interruttore booleano.
"pushbutton" – azione momentanea (trigger, calibrazione, …). L’host scrive True al clic e non visualizza alcuno stato.
"slider" – controllo numerico con min/max/step.
"spinbox" – inserimento numerico preciso a passi; condivide i campi dello slider e la forma di aggiornamento con tupla (min, max, value).
"select" – menu a tendina delle stringhe options.
"radio" – options visualizzate come un gruppo di pulsanti radio.
"lineedit" – campo di inserimento di stringa scrivibile.
"depth" – visualizzazione 2D di una mappa di profondità di width x height.
"waveform" – grafico 1D di serie temporali di campioni interlacciati.

value è il valore iniziale. Il valore predefinito dipende da type: 0 per label, "" per text/lineedit/select/ radio, False per toggle/pushbutton, min per slider/spinbox.

unit è la stringa dell’unità di misura per label/slider/spinbox/ waveform (ad es. "Cel", "%RH").

min è il valore minimo (intervallo di slider/spinbox, o intervallo di visualizzazione di depth/waveform).

max è il valore massimo (intervallo di slider/spinbox, o intervallo di visualizzazione di depth/waveform).

step è la dimensione del passo (slider/spinbox).

options è la lista delle stringhe delle opzioni (select/radio), oppure la lista dei nomi delle singole serie (waveform).

width è la larghezza in pixel (depth).

height è l’altezza in pixel (depth).

sample_rate è la frequenza di campionamento in campioni al secondo (waveform). Viene inviata all’host come periodo di campionamento.

series è il numero di serie interlacciate in ogni blocco di campioni (waveform), ad es. 3 per una traccia XYZ di un accelerometro.

typecode è il typecode array dei campioni della waveform (ad es. "H", "h", "f"). Il valore predefinito è "H" (16 bit senza segno).

__getitem__(name: str) → object#: Restituisce il valore corrente del campo con il nome indicato. Per i campi depth e waveform viene restituito il buffer di dati binari, altrimenti il valore scalare.

__setitem__(name: str, value: Any) → None#: Imposta il valore del campo con il nome indicato. Per i campi slider e spinbox, una tupla (min, max, value) aggiorna contemporaneamente l’intervallo e il valore corrente. Per i campi depth, value è il buffer di dati binari. Per i campi waveform, value è un blocco bytes-like di campioni interlacciati nel typecode del campo; il numero di campioni per serie viene ricavato dalla sua lunghezza e viene allegato un timestamp monotonico (in microsecondi, accumulato da time.ticks_us()) in modo che l’host possa collocare il blocco sul proprio asse temporale.

poll() → bool#: Metodo dell’interfaccia del backend. Restituisce True quando ci sono dati serializzati disponibili per l’host.

size() → int#: Metodo dell’interfaccia del backend. Invoca on_read (se impostato) e restituisce la dimensione del buffer serializzato.

read(offset: int, size: int) → bytes#: Metodo dell’interfaccia del backend. Restituisce una porzione del buffer serializzato.

write(offset: int, data: bytearray) → int#: Metodo dell’interfaccia del backend. Decodifica un elenco di aggiornamenti CBOR e applica i valori ai campi denominati corrispondenti, invocando on_write per ciascuno.

Costanti#

Bit di flag del canale (combinati bit a bit; passati a protocol.register tramite flags o impostati automaticamente in base ai metodi del backend).

protocol.CHANNEL_FLAG_READ: int#: Il canale supporta le letture.

protocol.CHANNEL_FLAG_WRITE: int#: Il canale supporta le scritture.

protocol.CHANNEL_FLAG_LOCK: int#: Il canale implementa lock/unlock.

protocol.CHANNEL_FLAG_PHYSICAL: int#: Il canale rappresenta un trasporto fisico (in contrapposizione a un canale di dati logico).

Identificatori dei canali integrati.

protocol.CHANNEL_ID_TRANSPORT: int#: ID di canale riservato per il trasporto attivo.

protocol.CHANNEL_ID_STDIN: int#: ID del canale stdin integrato.

protocol.CHANNEL_ID_STDOUT: int#: ID del canale stdout integrato.

protocol.CHANNEL_ID_STREAM: int#: ID del canale stream integrato.

protocol.CHANNEL_ID_PROFILE: int#: ID del canale del profiler integrato (presente solo quando il firmware è compilato con il profiler abilitato).

protocol — Canali del protocollo OpenMV#