From fc970e6e41246a6b72d16640d172b2c09dd6f262 Mon Sep 17 00:00:00 2001 From: iabdalkader Date: Mon, 15 Sep 2025 18:50:52 +0200 Subject: [PATCH] scripts/examples: Add Protocol examples. Signed-off-by: iabdalkader --- .../examples/12-Protocol/custom_channels.py | 79 +++++++++++++++++++ .../examples/12-Protocol/uart_transport.py | 67 ++++++++++++++++ 2 files changed, 146 insertions(+) create mode 100644 scripts/examples/12-Protocol/custom_channels.py create mode 100644 scripts/examples/12-Protocol/uart_transport.py diff --git a/scripts/examples/12-Protocol/custom_channels.py b/scripts/examples/12-Protocol/custom_channels.py new file mode 100644 index 000000000..4ae86390f --- /dev/null +++ b/scripts/examples/12-Protocol/custom_channels.py @@ -0,0 +1,79 @@ +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="time", + backend=StaticChannel() + ) + +ch2 = protocol.register( + name="buffer", + backend=BufferChannel() + ) + +while(True): + ch1.send_event(0xAA, wait_ack=False) + time.sleep_ms(100) + diff --git a/scripts/examples/12-Protocol/uart_transport.py b/scripts/examples/12-Protocol/uart_transport.py new file mode 100644 index 000000000..e8d050be8 --- /dev/null +++ b/scripts/examples/12-Protocol/uart_transport.py @@ -0,0 +1,67 @@ +""" +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 +from machine import UART +import protocol + +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) + ) + + while True: + time.sleep_ms(500)