`protocol` — ערוצי פרוטוקול OpenMV#

המודול protocol חושף את פרוטוקול המארח של OpenMV ל-Python. הוא מאפשר לאתחל ולהגדיר את מחסנית הפרוטוקול בצד הקושחה, ומאפשר לקוד המשתמש לרשום ערוצים לוגיים מותאמים אישית הנתמכים על ידי אובייקט Python המממש את ממשק הערוץ (read, write, size, poll וכו«). זה מה שכלי הלוויין השולחניים מדברים אליו כאשר הם מזרימים נתוני תמונה או חושפים רכיבים אינטראקטיביים למצלמה מחוברת.

דוגמאות#

הזרמת תמונת RGB565 לכלי מארח באמצעות backend מותאם אישית המממש את ממשק הערוץ הגולמי (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

הסקריפט התואם בצד המארח, המשתמש בחבילת ה-Python openmv (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 ברירת מחדל שכבר פועלת, ולכן קריאה לכך נדרשת רק כדי לשנות את התעבורה או לדרוס את פרמטרי המסגור (framing) ברירת המחדל.

crc מפעיל אימות CRC על מסגרות הפרוטוקול.

seq מפעיל מעקב אחר מספרי רצף.

ack מפעיל אישורי קבלה לכל מסגרת.

events מפעיל התראות אירועי ערוץ.

max_payload הוא גודל ה-payload המרבי בבתים. אם מושמט, נעשה שימוש בברירת המחדל לכל מצלמה שלהלן; היא נגזרת מגודל חוצץ (buffer) הפרוטוקול של כל לוח כ-buffer - 10 (header) - 4 (CRC).

מצלמה	גודל חוצץ (buffer)	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.register(name: str, *, backend: object, flags: int = 0) → ProtocolChannel#

רישום אובייקט backend של Python כערוץ לוגי חדש והחזרת ידית ProtocolChannel. השיטות הזמינות של אובייקט ה-backend (ראו ממשק ה-Backend להלן) קובעות את יכולות הערוץ; protocol.CHANNEL_FLAG_READ, protocol.CHANNEL_FLAG_WRITE ו-protocol.CHANNEL_FLAG_LOCK מתווספים ל-flags באופן אוטומטי כאשר השיטות המתאימות ממומשות.

name הוא שם הערוץ כמחרוזת. נחתך לגודל חוצץ שם-הערוץ של הקושחה. חובה.

backend הוא אובייקט ה-Python המממש את ממשק ה-backend. חובה. בדרך כלל מועבר באמצעות מילת מפתח (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 אם שליחת האירוע נכשלת.

ממשק ה-Backend#

אובייקט backend המועבר ל-protocol.register עשוי לממש כל תת-קבוצה של השיטות הבאות. רק השיטות הקיימות באובייקט מחוברות לשכבת פרוטוקול ה-C; שיטות חסרות משאירות את היכולת המתאימה מושבתת.

class protocol.backend#

אובייקט backend של ערוץ המועבר ל-protocol.register. השיטות שלהלן מתארות את הממשק האופציונלי ש-backend של Python עשוי לממש.

init() → object#: נקראת פעם אחת כאשר הערוץ מאותחל. החזר כל ערך שאינו None בהצלחה; חריגה או היעדר ערך מוחזר מטופלים כשגיאה.

poll() → bool#: מחזיר True אם בערוץ יש נתונים מוכנים לקריאה על ידי המארח.

lock() → bool#: השגת הערוץ עבור העברה. מחזיר True בהצלחה.

unlock() → bool#: שחרור הערוץ לאחר העברה. מחזיר True בהצלחה.

size() → int#: מחזיר את מספר הבתים הניתנים כעת לקריאה מהערוץ.

shape() → tuple#: מחזיר tuple של עד ארבעה מספרים שלמים המתארים את צורת הנתונים (למשל ממדי תמונה). עד ארבעה איברים נצרכים על ידי שכבת הפרוטוקול.

flush() → object#: ריקון כל הנתונים הממתינים. החזר כל ערך שאינו None בהצלחה.

read(offset: int, size: int) → bytes#: מחזיר עד size בתים החל מ-offset כאובייקט דמוי bytes התומך בפרוטוקול ה-buffer.

readp(offset: int, size: int) → bytes#: וריאנט zero-copy של read. מחזיר חוצץ (buffer) שהזיכרון הבסיסי שלו נקרא ישירות על ידי שכבת הפרוטוקול; החוצץ חייב להישאר תקף למשך כל ההעברה.

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

backend ברמה גבוהה יותר של Python (מסופק על ידי חבילת protocol הקפואה) המבצע סריאליזציה של שדות בעלי שם ל-CBOR באמצעות מפתחות שלמים תואמי SenML. תומך ברכיבי תצוגה (label, depth) ובבקרות אינטראקטיביות (toggle, slider, select) עם פונקציות callback on_read/on_write.

on_read הוא callable אופציונלי on_read(channel) הנקרא לפני שהערוץ עובר סריאליזציה עבור המארח. השתמש בו כדי לרענן ערכי שדות.

on_write הוא callable אופציונלי on_write(channel, name, value) הנקרא כאשר המארח כותב ערך חדש עבור שדה בעל שם.

הוספת שדה בעל שם לערוץ.

name הוא שם התצוגה; חייב להיות ייחודי בתוך ערוץ זה.

type הוא סוג הרכיב: "label", "toggle", "slider", "select" או "depth".

value הוא הערך ההתחלתי. ברירת המחדל תלויה ב-type.

unit הוא מחרוזת היחידה עבור label/slider (למשל "Cel", "%RH").

min הוא הערך המינימלי (טווח slider או טווח depth).

max הוא הערך המרבי (טווח slider או טווח depth).

step הוא גודל הצעד (slider).

options היא רשימת מחרוזות האפשרויות (select).

width הוא הרוחב בפיקסלים (depth).

height הוא הגובה בפיקסלים (depth).

__getitem__(name: str) → object#: מחזיר את הערך הנוכחי של השדה בעל השם. עבור שדות depth מוחזר חוצץ הנתונים הבינארי, אחרת הערך הסקלרי.

__setitem__(name: str, value: Any) → None#: הגדרת הערך של השדה בעל השם. עבור שדות slider, tuple מסוג (min, max, value) מעדכן את הטווח ואת הערך הנוכחי בו-זמנית. עבור שדות depth, value הוא חוצץ הנתונים הבינארי.

poll() → bool#: שיטת ממשק backend. מחזירה True כאשר נתונים מסוריאליזים זמינים עבור המארח.

size() → int#: שיטת ממשק backend. מפעילה את on_read (אם הוגדר) ומחזירה את גודל החוצץ המסוריאליז.

read(offset: int, size: int) → bytes#: שיטת ממשק backend. מחזירה פלח של החוצץ המסוריאליז.

write(offset: int, data: bytearray) → int#: שיטת ממשק backend. מפענחת רשימת עדכוני CBOR ומחילה ערכים על שדות בעלי שם תואמים, תוך הפעלת on_write עבור כל אחד.

קבועים#

ביטי דגל ערוץ (משולבים בפעולת bitwise; מועברים ל-protocol.register באמצעות flags או נקבעים אוטומטית על סמך השיטות של ה-backend).

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#: מזהה הערוץ של ערוץ ה-profiler המובנה (קיים רק כאשר הקושחה נבנית עם ה-profiler מופעל).

protocol — ערוצי פרוטוקול OpenMV#