`protocol` — Canais do Protocolo OpenMV#

O módulo protocol expõe o protocolo de host da OpenMV ao Python. Ele permite inicializar e configurar a pilha de protocolo do lado do firmware e possibilita que o código do usuário registre canais lógicos personalizados respaldados por um objeto Python que implementa a interface do canal (read, write, size, poll, etc.). É com isso que as ferramentas complementares de desktop se comunicam quando transmitem dados de imagem ou expõem widgets interativos a uma câmera conectada.

Exemplos#

Transmite uma imagem RGB565 para uma ferramenta de host usando um backend personalizado que implementa a interface de canal bruto (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

O script correspondente do lado do host, usando o pacote Python openmv (pip install openmv) para conectar, enviar o script da câmera e receber cada quadro:

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

Substitua /dev/ttyACM0 pela porta serial da câmera (por exemplo, COM3 no Windows). O construtor openmv.camera.Camera aceita os mesmos parâmetros de protocolo que init (crc / seq / ack / events / max_payload / max_retry / timeout) quando a pilha do lado da câmera foi reconfigurada para corresponder.

Funções#

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#

Inicializa (ou reconfigura) a pilha de protocolo e registra os canais lógicos de dados padrão (stdin, stdout, stream e, se compilado no firmware, profile). Levanta RuntimeError se a inicialização falhar. O firmware inicializa com uma pilha de protocolo USB padrão já em execução, então chamar isto só é necessário para trocar o transporte ou sobrescrever os parâmetros de enquadramento padrão. Chamá-lo novamente reinicializa a pilha: o timer de polling anterior é removido e os canais são registrados novamente.

crc habilita a validação CRC nos quadros do protocolo.

seq habilita o rastreamento de número de sequência.

ack habilita confirmações por quadro.

events habilita notificações de eventos de canal.

max_payload é o tamanho máximo do payload em bytes. Se omitido, é usado o padrão por câmera abaixo; ele é derivado do tamanho do buffer de protocolo de cada placa como buffer - 10 (header) - 4 (CRC).

Câmera	Tamanho do buffer	Payload máximo
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 é o número de tentativas de retransmissão. Padrão 3.

rtx_timeout_ms é o tempo limite de retransmissão em milissegundos (dobrado após cada tempo limite). Padrão 500.

lock_interval_ms é o intervalo mínimo de bloqueio em milissegundos. Padrão 10.

poll_ms é o intervalo de polling em milissegundos. 0 (o padrão) desabilita o polling por timer.

protocol.is_active() → bool#: Retorna True se um host estiver atualmente conectado e a pilha de protocolo estiver ativa, caso contrário False.

protocol.poll() → int#

Executa uma iteração da tarefa do protocolo: esvazia o buffer de recepção do transporte, analisa quaisquer quadros completos, despacha comandos e atende aos eventos de canal. Normalmente o firmware faz isso por você a partir da interrupção USB e do timer poll_ms, então isto só é necessário para transportes personalizados que não são acionados por nenhum dos dois – por exemplo, quando poll_ms=0 e o transporte é consultado (polling) a partir do loop principal do script.

Retorna 0 em caso de sucesso e -1 se nenhum transporte ativo estiver registrado ou se a chamada tiver sido feita de um contexto de interrupção.

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

Registra um objeto Python backend como um novo canal lógico e retorna um manipulador ProtocolChannel. Os métodos disponíveis do objeto backend (consulte Interface do Backend abaixo) determinam as capacidades do canal; protocol.CHANNEL_FLAG_READ, protocol.CHANNEL_FLAG_WRITE e protocol.CHANNEL_FLAG_LOCK são adicionados a flags automaticamente quando os métodos correspondentes são implementados.

name é o nome do canal como uma string. Truncado para o tamanho do buffer de nome de canal do firmware. Obrigatório.

backend é o objeto Python que implementa a interface do backend. Obrigatório. Normalmente passado por palavra-chave (backend=...).

flags são bits adicionais de flag de canal (consulte as constantes CHANNEL_FLAG_*). Opcional; o padrão é 0.

Levanta RuntimeError se o canal não puder ser registrado (por exemplo, se não houver slots de canal livres).

Classes#

class protocol.ProtocolChannel#

Manipulador retornado por protocol.register. As instâncias não são construídas diretamente.

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

Envia uma notificação de evento de canal ao host.

event é o identificador do evento (inteiro).

wait_ack se True bloqueia até que o host confirme o evento.

Levanta RuntimeError se o envio do evento falhar.

Interface do Backend#

Um objeto backend passado a protocol.register pode implementar qualquer subconjunto dos métodos a seguir. Apenas os métodos presentes no objeto são conectados à camada de protocolo em C; métodos ausentes deixam a capacidade correspondente desabilitada.

class protocol.backend#

Objeto backend de canal passado a protocol.register. Os métodos abaixo descrevem a interface opcional que um backend Python pode implementar.

init() → object#: Chamado uma vez quando o canal é inicializado. Retorne qualquer valor diferente de None em caso de sucesso; uma exceção ou a ausência de retorno é tratada como um erro.

poll() → bool#: Retorna True se o canal tiver dados prontos para serem lidos pelo host.

lock() → bool#: Adquire o canal para uma transferência. Retorne True em caso de sucesso.

unlock() → bool#: Libera o canal após uma transferência. Retorne True em caso de sucesso.

size() → int#: Retorna o número de bytes atualmente legíveis do canal.

shape() → tuple#: Retorna uma tupla de até quatro inteiros descrevendo o formato dos dados (por exemplo, as dimensões da imagem). Até quatro elementos são consumidos pela camada de protocolo.

flush() → object#: Descarrega quaisquer dados pendentes. Retorne qualquer valor diferente de None em caso de sucesso.

read(offset: int, size: int) → bytes#: Retorna até size bytes a partir de offset como um objeto do tipo bytes que suporta o protocolo de buffer.

readp(offset: int, size: int) → bytes#: Variante zero-copy de read. Retorna um buffer cuja memória subjacente é lida diretamente pela camada de protocolo; o buffer deve permanecer válido durante toda a transferência.

write(offset: int, data: bytearray) → int#: Escreve data em offset. data é um bytearray que referencia diretamente o buffer em C. Retorne o número de bytes escritos, ou 0 para o sucesso padrão.

ioctl(cmd: int, length: int, arg: bytearray | None) → int#: Trata um ioctl. arg é None se length for zero, caso contrário é um bytearray que referencia o buffer em C. Retorne 0 ou None em caso de sucesso, ou um inteiro negativo em caso de erro.

is_active() → bool#: Para canais de transporte, retorna True se o transporte subjacente estiver atualmente conectado.

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

Um backend Python de nível mais alto (fornecido pelo pacote congelado protocol) que serializa campos nomeados em CBOR usando chaves inteiras compatíveis com SenML. Suporta widgets de exibição (label, text, depth, waveform) e controles interativos (toggle, pushbutton, slider, spinbox, select, radio, lineedit) com callbacks on_read/on_write.

on_read é um callable opcional on_read(channel) invocado antes de o canal ser serializado para o host. Use-o para atualizar os valores dos campos.

on_write é um callable opcional on_write(channel, name, value) invocado quando o host escreve um novo valor para um campo nomeado.

Adiciona um campo nomeado ao canal.

name é o nome de exibição; deve ser único dentro deste canal.

type é o tipo do widget:

"label" – exibição numérica somente leitura com uma unit opcional.
"text" – exibição de string somente leitura.
"toggle" – chave booleana.
"pushbutton" – ação momentânea (disparar, calibrar, …). O host escreve True ao clicar e não renderiza nenhum estado.
"slider" – controle numérico com min/max/step.
"spinbox" – entrada numérica precisa em passos; compartilha os campos do slider e a forma de atualização por tupla (min, max, value).
"select" – lista suspensa de strings de options.
"radio" – options renderizadas como um grupo de botões de opção.
"lineedit" – entrada de string gravável.
"depth" – exibição de mapa de profundidade 2D de width x height.
"waveform" – gráfico 1D de série temporal de amostras intercaladas.

value é o valor inicial. O padrão depende de type: 0 para label, "" para text/lineedit/select/ radio, False para toggle/pushbutton, min para slider/spinbox.

unit é a string de unidade para label/slider/spinbox/ waveform (por exemplo, "Cel", "%RH").

min é o valor mínimo (faixa do slider/spinbox, ou faixa de exibição do depth/waveform).

max é o valor máximo (faixa do slider/spinbox, ou faixa de exibição do depth/waveform).

step é o tamanho do passo (slider/spinbox).

options é a lista de strings de opção (select/radio), ou a lista de nomes por série (waveform).

width é a largura em pixels (depth).

height é a altura em pixels (depth).

sample_rate é a taxa de amostragem em amostras por segundo (waveform). Enviada ao host como o período de amostragem.

series é o número de séries intercaladas em cada bloco de amostras (waveform), por exemplo 3 para um traço de acelerômetro XYZ.

typecode é o typecode de array das amostras da waveform (por exemplo, "H", "h", "f"). O padrão é "H" (16 bits sem sinal).

__getitem__(name: str) → object#: Retorna o valor atual do campo nomeado. Para campos depth e waveform, o buffer de dados binários é retornado; caso contrário, o valor escalar.

__setitem__(name: str, value: Any) → None#: Define o valor do campo nomeado. Para campos slider e spinbox, uma tupla (min, max, value) atualiza a faixa e o valor atual simultaneamente. Para campos depth, value é o buffer de dados binários. Para campos waveform, value é um bloco do tipo bytes com amostras intercaladas no typecode do campo; a contagem de amostras por série é derivada do seu comprimento, e um carimbo de tempo monotônico (em microssegundos, acumulado a partir de time.ticks_us()) é anexado para que o host possa posicionar o bloco em seu eixo de tempo.

poll() → bool#: Método da interface do backend. Retorna True quando há dados serializados disponíveis para o host.

size() → int#: Método da interface do backend. Invoca on_read (se definido) e retorna o tamanho do buffer serializado.

read(offset: int, size: int) → bytes#: Método da interface do backend. Retorna uma fatia do buffer serializado.

write(offset: int, data: bytearray) → int#: Método da interface do backend. Decodifica uma lista de atualizações CBOR e aplica os valores aos campos nomeados correspondentes, invocando on_write para cada um.

Constantes#

Bits de flag de canal (combinados bit a bit; passados a protocol.register via flags ou definidos automaticamente com base nos métodos do backend).

protocol.CHANNEL_FLAG_READ: int#: O canal suporta leituras.

protocol.CHANNEL_FLAG_WRITE: int#: O canal suporta escritas.

protocol.CHANNEL_FLAG_LOCK: int#: O canal implementa lock/unlock.

protocol.CHANNEL_FLAG_PHYSICAL: int#: O canal representa um transporte físico (em oposição a um canal lógico de dados).

Identificadores de canal embutidos.

protocol.CHANNEL_ID_TRANSPORT: int#: ID de canal reservado para o transporte ativo.

protocol.CHANNEL_ID_STDIN: int#: ID de canal do canal stdin embutido.

protocol.CHANNEL_ID_STDOUT: int#: ID de canal do canal stdout embutido.

protocol.CHANNEL_ID_STREAM: int#: ID de canal do canal stream embutido.

protocol.CHANNEL_ID_PROFILE: int#: ID de canal do canal de profiler embutido (presente apenas quando o firmware é compilado com o profiler habilitado).

protocol — Canais do Protocolo OpenMV#