`protocol` — OpenMV protokolni kanali#

Modul protocol izlaže OpenMV host protokol Pythonu. Omogućuje inicijalizaciju i konfiguraciju protokolnog stoga na strani ugrađenog programa te dopušta korisničkom kodu da registrira prilagođene logičke kanale poduprte Python objektom koji implementira sučelje kanala (read, write, size, poll itd.). To je ono s čime razgovaraju popratni alati na računalu kada prenose podatke slike ili izlažu interaktivne widgete povezanoj kameri.

Primjeri#

Prijenos RGB565 slike host alatu pomoću prilagođenog backenda koji implementira sirovo sučelje kanala (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

Odgovarajuća skripta na strani hosta, koja koristi openmv Python paket (pip install openmv) za povezivanje, slanje skripte na kameru i dohvaćanje svake sličice:

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

Zamijenite /dev/ttyACM0 serijskim priključkom kamere (npr. COM3 na Windowsu). Konstruktor openmv.camera.Camera prihvaća iste parametre protokola kao i init (crc / seq / ack / events / max_payload / max_retry / timeout) kada je stog na strani kamere ponovno konfiguriran tako da im odgovara.

Funkcije#

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#

Inicijalizira (ili ponovno konfigurira) protokolni stog i registrira zadane logičke podatkovne kanale (stdin, stdout, stream i, ako je ugrađen u prijevod, profile). Podiže RuntimeError ako inicijalizacija ne uspije. Ugrađeni program pokreće se s već aktivnim zadanim USB protokolnim stogom, pa je ovo potrebno pozvati samo za promjenu transporta ili nadjačavanje zadanih parametara uokvirivanja. Ponovni poziv ponovno inicijalizira stog: prethodni timer za prozivanje se uklanja, a kanali se iznova registriraju.

crc omogućuje CRC provjeru na okvirima protokola.

seq omogućuje praćenje rednih brojeva.

ack omogućuje potvrde po okviru.

events omogućuje obavijesti o događajima kanala.

max_payload je maksimalna veličina korisnog tereta u bajtovima. Ako se izostavi, koristi se zadana vrijednost po kameri navedena u nastavku; izvodi se iz veličine međuspremnika protokola svake pločice kao buffer - 10 (header) - 4 (CRC).

Kamera	Veličina međuspremnika	Maks. korisni teret
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 je broj pokušaja ponovnog prijenosa. Zadano 3.

rtx_timeout_ms je vremensko ograničenje ponovnog prijenosa u milisekundama (udvostručuje se nakon svakog isteka). Zadano 500.

lock_interval_ms je minimalni interval zaključavanja u milisekundama. Zadano 10.

poll_ms je interval prozivanja u milisekundama. 0 (zadano) onemogućuje prozivanje pomoću mjerača vremena.

protocol.is_active() → bool#: Vraća True ako je host trenutno povezan i protokolni stog aktivan, inače False.

protocol.poll() → int#

Izvodi jednu iteraciju protokolnog zadatka: prazni prijamni međuspremnik transporta, raščlanjuje sve potpune okvire, otprema naredbe i opslužuje događaje kanala. Ugrađeni program to inače obavlja za vas iz USB prekida i iz poll_ms timera, pa je ovo potrebno samo za prilagođene transporte koje ne pokreće ni jedno od toga – na primjer kada je poll_ms=0 i transport se proziva iz glavne petlje skripte.

Vraća 0 u slučaju uspjeha i -1 ako nijedan aktivni transport nije registriran ili je poziv upućen iz konteksta prekida.

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

Registrira Python backend objekt kao novi logički kanal i vraća ProtocolChannel rukovatelj. Dostupne metode objekta backend (vidi Backend sučelje u nastavku) određuju sposobnosti kanala; protocol.CHANNEL_FLAG_READ, protocol.CHANNEL_FLAG_WRITE i protocol.CHANNEL_FLAG_LOCK automatski se dodaju u flags kada su odgovarajuće metode implementirane.

name je naziv kanala kao niz znakova. Skraćuje se na veličinu međuspremnika naziva kanala u ugrađenom programu. Obavezno.

backend je Python objekt koji implementira backend sučelje. Obavezno. Obično se prosljeđuje ključnom riječi (backend=...).

flags su dodatni bitovi zastavica kanala (vidi konstante CHANNEL_FLAG_*). Neobavezno; zadano je 0.

Podiže RuntimeError ako se kanal ne može registrirati (npr. nema slobodnih mjesta za kanale).

Klase#

class protocol.ProtocolChannel#

Rukovatelj koji vraća protocol.register. Instance se ne konstruiraju izravno.

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

Šalje obavijest o događaju kanala hostu.

event je identifikator događaja (cijeli broj).

wait_ack ako je True blokira dok host ne potvrdi događaj.

Podiže RuntimeError ako slanje događaja ne uspije.

Backend sučelje#

Backend objekt proslijeđen funkciji protocol.register može implementirati bilo koji podskup sljedećih metoda. Samo su metode prisutne na objektu povezane s C protokolnim slojem; nedostajuće metode ostavljaju odgovarajuću sposobnost onemogućenom.

class protocol.backend#

Backend objekt kanala proslijeđen funkciji protocol.register. Metode u nastavku opisuju neobavezno sučelje koje Python backend može implementirati.

init() → object#: Poziva se jednom kada se kanal inicijalizira. Vraća bilo koju vrijednost različitu od None u slučaju uspjeha; iznimka ili nedostajuća povratna vrijednost tretira se kao pogreška.

poll() → bool#: Vraća True ako kanal ima podatke spremne za čitanje od strane hosta.

lock() → bool#: Zauzima kanal za prijenos. Vraća True u slučaju uspjeha.

unlock() → bool#: Otpušta kanal nakon prijenosa. Vraća True u slučaju uspjeha.

size() → int#: Vraća broj bajtova koji se trenutno mogu pročitati s kanala.

shape() → tuple#: Vraća n-torku do četiri cijela broja koji opisuju oblik podataka (npr. dimenzije slike). Protokolni sloj koristi do četiri elementa.

flush() → object#: Prazni sve podatke na čekanju. Vraća bilo koju vrijednost različitu od None u slučaju uspjeha.

read(offset: int, size: int) → bytes#: Vraća do size bajtova počevši od offset kao bytes-sličan objekt koji podržava protokol međuspremnika.

readp(offset: int, size: int) → bytes#: Varijanta read bez kopiranja. Vraća međuspremnik čiju temeljnu memoriju protokolni sloj izravno čita; međuspremnik mora ostati valjan tijekom trajanja prijenosa.

write(offset: int, data: bytearray) → int#: Zapisuje data na offset. data je bytearray koji izravno referencira C međuspremnik. Vraća broj zapisanih bajtova ili 0 pri zadanom uspjehu.

ioctl(cmd: int, length: int, arg: bytearray | None) → int#: Obrađuje ioctl. arg je None ako je length nula, inače bytearray koji referencira C međuspremnik. Vraća 0 ili None u slučaju uspjeha ili negativan cijeli broj u slučaju pogreške.

is_active() → bool#: Za transportne kanale, vraća True ako je temeljni transport trenutno povezan.

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

Backend više razine u Pythonu (koji pruža zamrznuti paket protocol) koji serijalizira imenovana polja u CBOR pomoću cjelobrojnih ključeva kompatibilnih sa SenML. Podržava widgete za prikaz (label, text, depth, waveform) i interaktivne kontrole (toggle, pushbutton, slider, spinbox, select, radio, lineedit) s on_read/on_write povratnim pozivima.

on_read je neobavezni pozivni objekt on_read(channel) koji se poziva prije serijalizacije kanala za host. Koristite ga za osvježavanje vrijednosti polja.

on_write je neobavezni pozivni objekt on_write(channel, name, value) koji se poziva kada host upiše novu vrijednost za imenovano polje.

Dodaje imenovano polje u kanal.

name je prikazni naziv; mora biti jedinstven unutar ovog kanala.

type je vrsta widgeta:

"label" – numerički prikaz samo za čitanje s neobaveznom jedinicom unit.
"text" – prikaz niza znakova samo za čitanje.
"toggle" – logička sklopka.
"pushbutton" – trenutna radnja (okidanje, kalibracija, …). Host pri kliku zapisuje True i ne prikazuje stanje.
"slider" – numerička kontrola s min/max/step.
"spinbox" – precizan numerički unos u koracima; dijeli polja klizača i oblik ažuriranja n-torkom (min, max, value).
"select" – padajući popis nizova options.
"radio" – options prikazane kao skupina radijskih gumba.
"lineedit" – zapisivi unos niza znakova.
"depth" – 2D prikaz dubinske karte veličine width x height.
"waveform" – 1D graf vremenskog niza isprepletenih uzoraka.

value je početna vrijednost. Zadana ovisi o type: 0 za label, "" za text/lineedit/select/ radio, False za toggle/pushbutton, min za slider/spinbox.

unit je niz znakova jedinice za label/slider/spinbox/ waveform (npr. "Cel", "%RH").

min je najmanja vrijednost (raspon klizača/spinboxa ili raspon prikaza dubine/valnog oblika).

max je najveća vrijednost (raspon klizača/spinboxa ili raspon prikaza dubine/valnog oblika).

step je veličina koraka (klizač/spinbox).

options je popis nizova opcija (select/radio) ili popis naziva pojedinih serija (waveform).

width je širina u pikselima (depth).

height je visina u pikselima (depth).

sample_rate je brzina uzorkovanja u uzorcima po sekundi (waveform). Hostu se šalje kao razdoblje uzorkovanja.

series je broj isprepletenih serija u svakom bloku uzoraka (waveform), npr. 3 za XYZ trag akcelerometra.

typecode je array kod tipa uzoraka valnog oblika (npr. "H", "h", "f"). Zadano je "H" (16-bitni bez predznaka).

__getitem__(name: str) → object#: Vraća trenutnu vrijednost imenovanog polja. Za polja depth i waveform vraća se binarni podatkovni međuspremnik, inače skalarna vrijednost.

__setitem__(name: str, value: Any) → None#: Postavlja vrijednost imenovanog polja. Za polja slider i spinbox n-torka (min, max, value) istodobno ažurira raspon i trenutnu vrijednost. Za polja depth value je binarni podatkovni međuspremnik. Za polja waveform value je blok isprepletenih uzoraka nalik bajtovima u typecode polja; broj uzoraka po seriji izvodi se iz njegove duljine, a prilaže se monotona vremenska oznaka (u mikrosekundama, akumulirana iz time.ticks_us()) kako bi host mogao smjestiti blok na svoju vremensku os.

poll() → bool#: Metoda backend sučelja. Vraća True kada su serijalizirani podaci dostupni hostu.

size() → int#: Metoda backend sučelja. Poziva on_read (ako je postavljen) i vraća veličinu serijaliziranog međuspremnika.

read(offset: int, size: int) → bytes#: Metoda backend sučelja. Vraća isječak serijaliziranog međuspremnika.

write(offset: int, data: bytearray) → int#: Metoda backend sučelja. Dekodira CBOR popis ažuriranja i primjenjuje vrijednosti na odgovarajuća imenovana polja, pozivajući on_write za svako.

Konstante#

Bitovi zastavica kanala (kombiniraju se bitovno; prosljeđuju se funkciji protocol.register putem flags ili se postavljaju automatski na temelju metoda backenda).

protocol.CHANNEL_FLAG_READ: int#: Kanal podržava čitanja.

protocol.CHANNEL_FLAG_WRITE: int#: Kanal podržava zapisivanja.

protocol.CHANNEL_FLAG_LOCK: int#: Kanal implementira lock/unlock.

protocol.CHANNEL_FLAG_PHYSICAL: int#: Kanal predstavlja fizički transport (za razliku od logičkog podatkovnog kanala).

Ugrađeni identifikatori kanala.

protocol.CHANNEL_ID_TRANSPORT: int#: Rezervirani ID kanala za aktivni transport.

protocol.CHANNEL_ID_STDIN: int#: ID kanala ugrađenog stdin kanala.

protocol.CHANNEL_ID_STDOUT: int#: ID kanala ugrađenog stdout kanala.

protocol.CHANNEL_ID_STREAM: int#: ID kanala ugrađenog stream kanala.

protocol.CHANNEL_ID_PROFILE: int#: ID kanala ugrađenog profiler kanala (prisutan samo kada je ugrađeni program izgrađen s omogućenim profilerom).

protocol — OpenMV protokolni kanali#