`protocol` — Canaux du protocole OpenMV#

Le module protocol expose le protocole hôte OpenMV à Python. Il permet d’initialiser et de configurer la pile de protocole côté micrologiciel, et permet au code utilisateur d’enregistrer des canaux logiques personnalisés adossés à un objet Python qui implémente l’interface de canal (read, write, size, poll, etc.). C’est ce avec quoi les outils compagnons de bureau communiquent lorsqu’ils diffusent des données d’image ou exposent des widgets interactifs à une caméra connectée.

Exemples#

Diffusez une image RGB565 vers un outil hôte à l’aide d’un backend personnalisé qui implémente l’interface de canal brute (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

Le script côté hôte correspondant, utilisant le paquet Python openmv (pip install openmv) pour se connecter, envoyer le script embarqué sur la caméra et récupérer chaque trame

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

Remplacez /dev/ttyACM0 par le port série de la caméra (par ex. COM3 sous Windows). Le constructeur openmv.camera.Camera accepte les mêmes paramètres de protocole que init (crc / seq / ack / events / max_payload / max_retry / timeout) lorsque la pile côté caméra a été reconfigurée en conséquence.

Fonctions#

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#

Initialise (ou reconfigure) la pile de protocole et enregistre les canaux de données logiques par défaut (stdin, stdout, stream et, s’il est compilé, profile). Lève RuntimeError si l’initialisation échoue. Le micrologiciel démarre avec une pile de protocole USB par défaut déjà active, l’appel de cette fonction n’est donc nécessaire que pour changer de transport ou remplacer les paramètres de tramage par défaut.

crc active la validation CRC sur les trames du protocole.

seq active le suivi des numéros de séquence.

ack active les accusés de réception par trame.

events active les notifications d’événements de canal.

max_payload est la taille maximale de la charge utile en octets. S’il est omis, la valeur par défaut propre à chaque caméra ci-dessous est utilisée ; elle est dérivée de la taille du tampon de protocole de chaque carte selon buffer - 10 (header) - 4 (CRC).

Caméra	Taille du tampon	Charge utile max
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 est le nombre de tentatives de retransmission. Par défaut 3.

rtx_timeout_ms est le délai d’expiration de retransmission en millisecondes (doublé après chaque expiration). Par défaut 500.

lock_interval_ms est l’intervalle de verrouillage minimal en millisecondes. Par défaut 10.

poll_ms est l’intervalle d’interrogation en millisecondes. 0 (la valeur par défaut) désactive l’interrogation par minuteur.

protocol.is_active() → bool#: Renvoie True si un hôte est actuellement connecté et que la pile de protocole est active, sinon False.

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

Enregistre un objet Python backend comme nouveau canal logique et renvoie un descripteur ProtocolChannel. Les méthodes disponibles de l’objet backend (voir Interface du backend ci-dessous) déterminent les capacités du canal ; protocol.CHANNEL_FLAG_READ, protocol.CHANNEL_FLAG_WRITE et protocol.CHANNEL_FLAG_LOCK sont ajoutés automatiquement à flags lorsque les méthodes correspondantes sont implémentées.

name est le nom du canal sous forme de chaîne. Tronqué à la taille du tampon de nom de canal du micrologiciel. Obligatoire.

backend est l’objet Python qui implémente l’interface du backend. Obligatoire. Généralement passé par mot-clé (backend=...).

flags correspond à des bits d’indicateur de canal supplémentaires (voir les constantes CHANNEL_FLAG_*). Facultatif ; vaut 0 par défaut.

Lève RuntimeError si le canal ne peut pas être enregistré (par ex. aucun emplacement de canal libre).

Classes#

class protocol.ProtocolChannel#

Descripteur renvoyé par protocol.register. Les instances ne sont pas construites directement.

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

Envoie une notification d’événement de canal à l’hôte.

event est l’identifiant de l’événement (entier).

wait_ack si True bloque jusqu’à ce que l’hôte accuse réception de l’événement.

Lève RuntimeError si l’envoi de l’événement échoue.

Interface du backend#

Un objet backend passé à protocol.register peut implémenter n’importe quel sous-ensemble des méthodes suivantes. Seules les méthodes présentes sur l’objet sont reliées à la couche de protocole C ; les méthodes manquantes laissent la capacité correspondante désactivée.

class protocol.backend#

Objet backend de canal passé à protocol.register. Les méthodes ci-dessous décrivent l’interface optionnelle qu’un backend Python peut implémenter.

init() → object#: Appelée une seule fois lorsque le canal est initialisé. Renvoie toute valeur autre que None en cas de succès ; une exception ou l’absence de valeur de retour est considérée comme une erreur.

poll() → bool#: Renvoie True si le canal a des données prêtes à être lues par l’hôte.

lock() → bool#: Acquiert le canal pour un transfert. Renvoie True en cas de succès.

unlock() → bool#: Libère le canal après un transfert. Renvoie True en cas de succès.

size() → int#: Renvoie le nombre d’octets actuellement lisibles depuis le canal.

shape() → tuple#: Renvoie un tuple d’au plus quatre entiers décrivant la forme des données (par ex. les dimensions d’une image). Jusqu’à quatre éléments sont consommés par la couche de protocole.

flush() → object#: Vide toutes les données en attente. Renvoie toute valeur autre que None en cas de succès.

read(offset: int, size: int) → bytes#: Renvoie jusqu’à size octets à partir de offset sous forme d’objet de type bytes prenant en charge le protocole tampon.

readp(offset: int, size: int) → bytes#: Variante sans copie de read. Renvoie un tampon dont la mémoire sous-jacente est lue directement par la couche de protocole ; le tampon doit rester valide pendant toute la durée du transfert.

write(offset: int, data: bytearray) → int#: Écrit data à offset. data est un bytearray référençant directement le tampon C. Renvoie le nombre d’octets écrits, ou 0 en cas de succès par défaut.

ioctl(cmd: int, length: int, arg: bytearray | None) → int#: Traite un ioctl. arg vaut None si length est nul, sinon un bytearray référençant le tampon C. Renvoie 0 ou None en cas de succès, ou un entier négatif en cas d’erreur.

is_active() → bool#: Pour les canaux de transport, renvoie True si le transport sous-jacent est actuellement connecté.

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

Un backend Python de plus haut niveau (fourni par le paquet figé protocol) qui sérialise des champs nommés en CBOR à l’aide de clés entières compatibles SenML. Prend en charge les widgets d’affichage (label, depth) et les contrôles interactifs (toggle, slider, select) avec des fonctions de rappel on_read/on_write.

on_read est un appelable optionnel on_read(channel) invoqué avant que le canal ne soit sérialisé pour l’hôte. Utilisez-le pour rafraîchir les valeurs des champs.

on_write est un appelable optionnel on_write(channel, name, value) invoqué lorsque l’hôte écrit une nouvelle valeur pour un champ nommé.

Ajoute un champ nommé au canal.

name est le nom d’affichage ; il doit être unique au sein de ce canal.

type est le type de widget : "label", "toggle", "slider", "select" ou "depth".

value est la valeur initiale. La valeur par défaut dépend de type.

unit est la chaîne d’unité pour label/slider (par ex. "Cel", "%RH").

min est la valeur minimale (plage du slider ou plage de depth).

max est la valeur maximale (plage du slider ou plage de depth).

step est le pas (slider).

options est la liste des chaînes d’options (select).

width est la largeur en pixels (depth).

height est la hauteur en pixels (depth).

__getitem__(name: str) → object#: Renvoie la valeur actuelle du champ nommé. Pour les champs depth, le tampon de données binaires est renvoyé, sinon la valeur scalaire.

__setitem__(name: str, value: Any) → None#: Définit la valeur du champ nommé. Pour les champs slider, un tuple (min, max, value) met à jour simultanément la plage et la valeur actuelle. Pour les champs depth, value est le tampon de données binaires.

poll() → bool#: Méthode de l’interface du backend. Renvoie True lorsque des données sérialisées sont disponibles pour l’hôte.

size() → int#: Méthode de l’interface du backend. Invoque on_read (s’il est défini) et renvoie la taille du tampon sérialisé.

read(offset: int, size: int) → bytes#: Méthode de l’interface du backend. Renvoie une tranche du tampon sérialisé.

write(offset: int, data: bytearray) → int#: Méthode de l’interface du backend. Décode une liste de mises à jour CBOR et applique les valeurs aux champs nommés correspondants, en invoquant on_write pour chacun.

Constantes#

Bits d’indicateur de canal (combinés au niveau du bit ; passés à protocol.register via flags ou définis automatiquement en fonction des méthodes du backend).

protocol.CHANNEL_FLAG_READ: int#: Le canal prend en charge les lectures.

protocol.CHANNEL_FLAG_WRITE: int#: Le canal prend en charge les écritures.

protocol.CHANNEL_FLAG_LOCK: int#: Le canal implémente lock/unlock.

protocol.CHANNEL_FLAG_PHYSICAL: int#: Le canal représente un transport physique (par opposition à un canal de données logique).

Identifiants de canaux intégrés.

protocol.CHANNEL_ID_TRANSPORT: int#: ID de canal réservé pour le transport actif.

protocol.CHANNEL_ID_STDIN: int#: ID de canal du canal stdin intégré.

protocol.CHANNEL_ID_STDOUT: int#: ID de canal du canal stdout intégré.

protocol.CHANNEL_ID_STREAM: int#: ID de canal du canal stream intégré.

protocol.CHANNEL_ID_PROFILE: int#: ID de canal du canal de profilage intégré (présent uniquement lorsque le micrologiciel est compilé avec le profileur activé).

protocol — Canaux du protocole OpenMV#