`protocol` --- Các kênh giao thức OpenMV#

Mô-đun protocol cung cấp giao thức máy chủ OpenMV cho Python. Nó cho phép ngăn xếp giao thức phía firmware được khởi tạo và cấu hình, đồng thời cho phép mã người dùng đăng ký các kênh logic tùy chỉnh được hỗ trợ bởi một đối tượng Python triển khai giao diện kênh (read, write, size, poll, v.v.). Đây là thứ mà các công cụ đồng hành trên máy tính để bàn nói chuyện khi chúng truyền trực tiếp dữ liệu ảnh hoặc hiển thị các tiện ích tương tác cho camera đã kết nối.

Ví dụ#

Truyền trực tiếp ảnh RGB565 tới công cụ máy chủ sử dụng backend tùy chỉnh triển khai giao diện kênh thô (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

Tập lệnh phía máy chủ tương ứng, sử dụng gói Python openmv (pip install openmv) để kết nối, đẩy tập lệnh trên camera và kéo từng khung hình:

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

Thay /dev/ttyACM0 bằng cổng serial của camera (ví dụ: COM3 trên Windows). Hàm khởi tạo openmv.camera.Camera chấp nhận các tham số giao thức tương tự như init (crc / seq / ack / events / max_payload / max_retry / timeout) khi ngăn xếp phía camera đã được cấu hình lại để khớp.

Hàm#

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#

Khởi tạo (hoặc cấu hình lại) ngăn xếp giao thức và đăng ký các kênh dữ liệu logic mặc định (stdin, stdout, stream và, nếu được biên dịch kèm, profile). Ném ra RuntimeError nếu khởi tạo thất bại. Firmware khởi động với một ngăn xếp giao thức USB mặc định đã chạy sẵn, nên chỉ cần gọi hàm này khi muốn đổi transport hoặc ghi đè các tham số đóng khung mặc định. Gọi lại hàm này sẽ khởi tạo lại ngăn xếp: bộ hẹn giờ polling trước đó bị gỡ bỏ và các kênh được đăng ký lại từ đầu.

crc bật kiểm tra CRC trên các khung giao thức.

seq bật theo dõi số thứ tự.

ack bật xác nhận từng khung.

events bật thông báo sự kiện kênh.

max_payload là kích thước tải trọng tối đa tính bằng byte. Nếu bỏ qua, giá trị mặc định của từng camera bên dưới sẽ được sử dụng; nó được lấy từ kích thước bộ đệm giao thức của mỗi bo mạch theo công thức buffer - 10 (header) - 4 (CRC).

Camera	Kích thước bộ đệm	Tải trọng tối đa
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 là số lần thử truyền lại. Mặc định là 3.

rtx_timeout_ms là thời gian chờ truyền lại tính bằng mili giây (tăng gấp đôi sau mỗi lần hết thời gian chờ). Mặc định là 500.

lock_interval_ms là khoảng thời gian khóa tối thiểu tính bằng mili giây. Mặc định là 10.

poll_ms là khoảng thời gian thăm dò tính bằng mili giây. 0 (mặc định) vô hiệu hóa thăm dò bằng bộ định thời.

protocol.is_active() → bool#: Trả về True nếu máy chủ đang kết nối và ngăn xếp giao thức đang hoạt động, ngược lại trả về False.

protocol.poll() → int#

Chạy một vòng lặp của tác vụ giao thức: xả bộ đệm nhận của transport, phân tích mọi khung đã hoàn chỉnh, điều phối các lệnh và xử lý các sự kiện kênh. Thông thường firmware tự làm việc này cho bạn từ ngắt USB và từ bộ hẹn giờ poll_ms, nên hàm này chỉ cần thiết cho các transport tùy chỉnh không được điều khiển bởi cả hai cơ chế đó -- ví dụ khi poll_ms=0 và transport được poll từ vòng lặp chính của tập lệnh.

Trả về 0 khi thành công và -1 nếu không có transport đang hoạt động nào được đăng ký hoặc lời gọi được thực hiện từ ngữ cảnh ngắt.

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

Đăng ký đối tượng Python backend như một kênh logic mới và trả về handle ProtocolChannel. Các phương thức khả dụng của đối tượng backend (xem Giao diện Backend bên dưới) xác định khả năng của kênh; protocol.CHANNEL_FLAG_READ, protocol.CHANNEL_FLAG_WRITE và protocol.CHANNEL_FLAG_LOCK được tự động thêm vào flags khi các phương thức tương ứng được triển khai.

name là tên kênh dưới dạng chuỗi. Bị cắt ngắn theo kích thước bộ đệm tên kênh của firmware. Bắt buộc.

backend là đối tượng Python triển khai giao diện backend. Bắt buộc. Thường được truyền theo từ khóa (backend=...).

flags là các bit cờ kênh bổ sung (xem hằng số CHANNEL_FLAG_*). Tùy chọn; mặc định là 0.

Phát sinh RuntimeError nếu kênh không thể được đăng ký (ví dụ: không còn khe kênh trống).

Lớp#

class protocol.ProtocolChannel#

Handle được trả về bởi protocol.register. Các thực thể không được khởi tạo trực tiếp.

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

Gửi thông báo sự kiện kênh tới máy chủ.

event là định danh sự kiện (số nguyên).

wait_ack nếu là True sẽ chặn cho đến khi máy chủ xác nhận sự kiện.

Phát sinh RuntimeError nếu việc gửi sự kiện thất bại.

Giao diện Backend#

Đối tượng backend được truyền tới protocol.register có thể triển khai bất kỳ tập hợp con nào của các phương thức sau. Chỉ các phương thức hiện diện trên đối tượng mới được kết nối với lớp giao thức C; các phương thức bị thiếu sẽ để khả năng tương ứng bị vô hiệu hóa.

class protocol.backend#

Đối tượng backend kênh được truyền tới protocol.register. Các phương thức bên dưới mô tả giao diện tùy chọn mà một backend Python có thể triển khai.

init() → object#: Được gọi một lần khi kênh được khởi tạo. Trả về bất kỳ giá trị nào khác None khi thành công; ngoại lệ hoặc thiếu giá trị trả về được coi là lỗi.

poll() → bool#: Trả về True nếu kênh có dữ liệu sẵn sàng để máy chủ đọc.

lock() → bool#: Chiếm quyền kênh cho một lần truyền. Trả về True khi thành công.

unlock() → bool#: Giải phóng kênh sau khi truyền. Trả về True khi thành công.

size() → int#: Trả về số byte hiện có thể đọc từ kênh.

shape() → tuple#: Trả về một tuple gồm tối đa bốn số nguyên mô tả hình dạng dữ liệu (ví dụ: kích thước ảnh). Tối đa bốn phần tử được lớp giao thức sử dụng.

flush() → object#: Xả dữ liệu đang chờ. Trả về bất kỳ giá trị nào khác None khi thành công.

read(offset: int, size: int) → bytes#: Trả về tối đa size byte bắt đầu từ offset dưới dạng đối tượng giống bytes hỗ trợ giao thức bộ đệm.

readp(offset: int, size: int) → bytes#: Biến thể không sao chép của read. Trả về bộ đệm có bộ nhớ cơ bản được lớp giao thức đọc trực tiếp; bộ đệm phải hợp lệ trong suốt quá trình truyền.

write(offset: int, data: bytearray) → int#: Ghi data tại offset. data là bytearray tham chiếu trực tiếp tới bộ đệm C. Trả về số byte đã ghi, hoặc 0 khi thành công mặc định.

ioctl(cmd: int, length: int, arg: bytearray | None) → int#: Xử lý một ioctl. arg là None nếu length bằng không, ngược lại là bytearray tham chiếu tới bộ đệm C. Trả về 0 hoặc None khi thành công, hoặc số nguyên âm khi có lỗi.

is_active() → bool#: Đối với các kênh truyền tải, trả về True nếu phương tiện truyền tải cơ bản hiện đang kết nối.

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

Một backend Python cấp cao hơn (do gói protocol đóng băng cung cấp) tuần tự hóa các trường có tên sang CBOR bằng các khóa số nguyên tương thích SenML. Hỗ trợ các widget hiển thị (label, text, depth, waveform) và các điều khiển tương tác (toggle, pushbutton, slider, spinbox, select, radio, lineedit) với các callback on_read/on_write.

on_read là callable tùy chọn on_read(channel) được gọi trước khi kênh được tuần tự hóa cho máy chủ. Dùng để làm mới giá trị trường.

on_write là callable tùy chọn on_write(channel, name, value) được gọi khi máy chủ ghi giá trị mới cho một trường có tên.

Thêm một trường có tên vào kênh.

name là tên hiển thị; phải là duy nhất trong kênh này.

type là loại widget:

"label" -- hiển thị số chỉ đọc với unit tùy chọn.
"text" -- hiển thị chuỗi chỉ đọc.
"toggle" -- công tắc boolean.
"pushbutton" -- hành động tức thời (kích hoạt, hiệu chuẩn, ...). Máy chủ ghi True khi nhấp và không hiển thị trạng thái nào.
"slider" -- điều khiển số với min/max/step.
"spinbox" -- nhập số chính xác theo bước; dùng chung các trường của slider và dạng cập nhật bằng tuple (min, max, value).
"select" -- danh sách thả xuống gồm các chuỗi options.
"radio" -- options được hiển thị dưới dạng một nhóm nút radio.
"lineedit" -- ô nhập chuỗi có thể ghi.
"depth" -- hiển thị bản đồ độ sâu 2D kích thước width x height.
"waveform" -- đồ thị chuỗi thời gian 1D của các mẫu xen kẽ.

value là giá trị ban đầu. Giá trị mặc định phụ thuộc vào type: 0 cho label, "" cho text/lineedit/select/ radio, False cho toggle/pushbutton, min cho slider/spinbox.

unit là chuỗi đơn vị cho label/slider/spinbox/ waveform (ví dụ "Cel", "%RH").

min là giá trị nhỏ nhất (phạm vi của slider/spinbox, hoặc phạm vi hiển thị của depth/waveform).

max là giá trị lớn nhất (phạm vi của slider/spinbox, hoặc phạm vi hiển thị của depth/waveform).

step là kích thước bước (slider/spinbox).

options là danh sách các chuỗi tùy chọn (select/radio), hoặc danh sách tên của từng chuỗi dữ liệu (waveform).

width là chiều rộng tính bằng điểm ảnh (depth).

height là chiều cao tính bằng điểm ảnh (depth).

sample_rate là tốc độ lấy mẫu tính bằng số mẫu mỗi giây (waveform). Được gửi tới máy chủ dưới dạng chu kỳ lấy mẫu.

series là số chuỗi dữ liệu xen kẽ trong mỗi khối mẫu (waveform), ví dụ 3 cho một đường ghi gia tốc kế XYZ.

typecode là typecode array của các mẫu waveform (ví dụ "H", "h", "f"). Mặc định là "H" (16-bit không dấu).

__getitem__(name: str) → object#: Trả về giá trị hiện tại của trường có tên đã cho. Với các trường depth và waveform thì bộ đệm dữ liệu nhị phân được trả về, ngược lại là giá trị vô hướng.

__setitem__(name: str, value: Any) → None#: Đặt giá trị của trường có tên đã cho. Với các trường slider và spinbox, một tuple (min, max, value) cập nhật đồng thời phạm vi và giá trị hiện tại. Với các trường depth, value là bộ đệm dữ liệu nhị phân. Với các trường waveform, value là một khối dạng bytes gồm các mẫu xen kẽ theo typecode của trường; số mẫu trên mỗi chuỗi dữ liệu được suy ra từ độ dài của khối và một dấu thời gian đơn điệu (tính bằng micro giây, tích lũy từ time.ticks_us()) được đính kèm để máy chủ có thể đặt khối dữ liệu lên trục thời gian của nó.

poll() → bool#: Phương thức giao diện backend. Trả về True khi dữ liệu đã tuần tự hóa sẵn sàng cho máy chủ.

size() → int#: Phương thức giao diện backend. Gọi on_read (nếu được đặt) và trả về kích thước của bộ đệm đã tuần tự hóa.

read(offset: int, size: int) → bytes#: Phương thức giao diện backend. Trả về một lát của bộ đệm đã tuần tự hóa.

write(offset: int, data: bytearray) → int#: Phương thức giao diện backend. Giải mã danh sách cập nhật CBOR và áp dụng các giá trị cho các trường có tên khớp, gọi on_write cho mỗi trường.

Hằng số#

Các bit cờ kênh (kết hợp theo bit; được truyền tới protocol.register qua flags hoặc được đặt tự động dựa trên các phương thức của backend).

protocol.CHANNEL_FLAG_READ: int#: Kênh hỗ trợ đọc.

protocol.CHANNEL_FLAG_WRITE: int#: Kênh hỗ trợ ghi.

protocol.CHANNEL_FLAG_LOCK: int#: Kênh triển khai lock/unlock.

protocol.CHANNEL_FLAG_PHYSICAL: int#: Kênh đại diện cho một phương tiện truyền tải vật lý (khác với kênh dữ liệu logic).

Định danh kênh tích hợp sẵn.

protocol.CHANNEL_ID_TRANSPORT: int#: ID kênh dành riêng cho phương tiện truyền tải đang hoạt động.

protocol.CHANNEL_ID_STDIN: int#: ID kênh của kênh stdin tích hợp sẵn.

protocol.CHANNEL_ID_STDOUT: int#: ID kênh của kênh stdout tích hợp sẵn.

protocol.CHANNEL_ID_STREAM: int#: ID kênh của kênh stream tích hợp sẵn.

protocol.CHANNEL_ID_PROFILE: int#: ID kênh của kênh profiler tích hợp sẵn (chỉ có khi firmware được xây dựng với profiler được bật).

protocol --- Các kênh giao thức OpenMV#