scripts/examples: Add Protocol examples.

Signed-off-by: iabdalkader <i.abdalkader@gmail.com>
This commit is contained in:
iabdalkader 2025-09-15 18:50:52 +02:00
parent dd12dd2bc3
commit 759df69770
2 changed files with 153 additions and 0 deletions

View File

@ -0,0 +1,72 @@
import time
import protocol
class StaticChannel:
def __init__(self):
pass
def size(self):
return len("HelloWorld!")
def read(self, offset, size):
print(f"StaticChannel read {size} bytes from offset {offset}")
return "HelloWorld!"
class BufferChannel:
"""A simple channel backed by a buffer that supports read/write operations."""
def __init__(self, buffer_size=1024):
self.buffer = bytearray(buffer_size)
self.data_size = 0 # Actual amount of valid data in buffer
def init(self):
"""Initialize the channel. Return None for success."""
print(f"BufferChannel initialized with {len(self.buffer)} bytes")
def deinit(self):
"""Deinitialize the channel. Return None for success."""
print("BufferChannel deinitialized")
def poll(self):
"""Check if channel has data available. Return bool."""
return self.data_size > 0
def size(self):
"""Return the amount of valid data in the buffer."""
return self.data_size
def read(self, offset, size):
"""Read data from the buffer. Return bytes/bytearray."""
if offset >= self.data_size:
return b"" # No data at this offset
end_pos = min(offset + size, self.data_size)
data = bytes(self.buffer[offset:end_pos])
print(f"BufferChannel read {len(data)} bytes from offset {offset}")
return data
def write(self, offset, data):
"""Write data to the buffer. data is a bytearray. Return bytes written or None."""
if offset + len(data) > len(self.buffer):
# Truncate if it would exceed buffer size
available = len(self.buffer) - offset
data = data[:available] if available > 0 else b""
if len(data) > 0:
self.buffer[offset : offset + len(data)] = data
# Update data_size if we wrote beyond current end
self.data_size = max(self.data_size, offset + len(data))
print(f"BufferChannel wrote {len(data)} bytes to offset {offset}")
return len(data) # Return bytes written (or None for default success)
# Register custom channels
ch1 = protocol.register(name="static", backend=StaticChannel())
ch2 = protocol.register(name="buffer", backend=BufferChannel())
while True:
ch1.send_event(0xAA, wait_ack=False)
time.sleep_ms(100)

View File

@ -0,0 +1,81 @@
"""
UART Transport Example for OpenMV Protocol
This example shows how to create a UART-based transport channel and register
built-in protocol channels in Python.
The transport implements the physical layer interface required by the protocol:
- read(): Physical read from UART with timeout
- write(): Physical write to UART with timeout
- is_active(): Check if UART connection is available
- size(): Return number of bytes available to read
"""
import time
import protocol
from machine import UART
class StaticChannel:
def __init__(self):
pass
def size(self):
return len("HelloWorld!")
def read(self, offset, size):
print(f"StaticChannel read {size} bytes from offset {offset}")
return "HelloWorld!"
class UartTransport:
"""UART-based transport channel for OpenMV Protocol"""
def __init__(self, uart_id=1, baudrate=115200, timeout=1000, rxbuf=1024):
self.uart = UART(
uart_id, baudrate, rxbuf=rxbuf, timeout=timeout, timeout_char=500
)
self.buf = memoryview(bytearray(rxbuf))
def is_active(self):
return True
def size(self):
return self.uart.any()
def read(self, offset, size):
size = self.uart.readinto(self.buf, size)
return None if size is None else self.buf[:size]
def write(self, offset, data):
return self.uart.write(data)
def flush(self):
self.uart.flush()
if __name__ == "__main__":
MAX_PAYLOAD = 4096 - 10 - 2
# Initialize and configure the protocol
protocol.init(
crc=True, # Enable CRC
seq=True, # Enable sequence checking
ack=True, # Wait for CKs
events=True, # Enable async-events
soft_reboot=False, # Disable soft-reboots (required)
max_payload=MAX_PAYLOAD, # Max packet payload
rtx_retries=3, # Retransmission retry count
rtx_timeout_ms=500, # Timeout before retransmission (doubled after each try)
lock_interval_ms=10, # Minimum locking interval
timer_ms=10, # Schedules the protocol task every 10ms
)
# Register the transport
protocol.register(
name="uart",
flags=protocol.CHANNEL_FLAG_PHYSICAL,
backend=UartTransport(7, timeout=5000, rxbuf=8 * 1024, baudrate=921600),
)
# Register custom channel
protocol.register(name="static", backend=StaticChannel())