network_transport.py

"""
Network Transport Example for OpenMV Protocol (UDP)

This example shows how to create a network transport channel using UDP.
The camera connects to a network, sets its mDNS hostname (reachable at
<hostname>.local), and listens for UDP datagrams on a known port.

The IDE (or any client) sends datagrams to <hostname>.local:<port> and
communicates using the OpenMV protocol over UDP.

The transport implements the physical layer interface required by the protocol:
- read(): Read from UDP socket
- write(): Buffer data for the next datagram
- is_active(): Check if a client address is known
- size(): Peek at available bytes without consuming
- flush(): Send buffered writes as a single UDP datagram
"""
import time
import errno
import network
import socket
import protocol
from machine import LED

LED_B = LED("LED_BLUE")

SSID = ""  # 网络 SSID
KEY = ""  # 网络密钥

HOST = ""  # 使用第一个可用的接口
PORT = 5555  # 协议UDP端口
HOSTNAME = "openmv"  # mDNS主机名（可通过openmv.local访问）


class NetworkTransport:
    """UDP-based network transport for OpenMV Protocol"""

    def __init__(self, host="", port=5555, max_payload=4096):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
        self.sock.bind((host, port))
        self.sock.setblocking(False)

        self.client = None
        self.txbuf = bytearray(max_payload + 14)
        self.txlen = 0
        self.rxbuf = bytearray(max_payload + 14)
        self.rxlen = 0
        self.rxpos = 0
        LED_B.off()

    def is_active(self):
        return self.client is not None or self.size() > 0

    def size(self):
        avail = self.rxlen - self.rxpos
        if avail > 0:
            return avail
        try:
            data, addr = self.sock.recvfrom(len(self.rxbuf))
        except OSError as e:
            if e.args[0] != errno.EAGAIN:
                self.client = None
            return 0
        if not data:
            return 0
        n = len(data)

        self.rxpos = 0
        self.rxlen = n
        self.rxbuf[:n] = data
        self.client = addr
        return n

    def read(self, offset, size):
        avail = self.rxlen - self.rxpos
        if avail <= 0:
            return None
        n = min(size, avail)
        end = self.rxpos + n
        data = self.rxbuf[self.rxpos:end]
        self.rxpos = end
        return data

    def write(self, offset, data):
        if self.client is None:
            return -1
        n = len(data)
        self.txbuf[self.txlen:self.txlen + n] = data
        self.txlen += n
        return n

    def flush(self):
        if self.client is None or self.txlen == 0:
            return 0
        mv = memoryview(self.txbuf)
        try:
            self.sock.sendto(mv[:self.txlen], self.client)
        except OSError as e:
            self.txlen = 0
            if e.args[0] != errno.EAGAIN:
                self.client = None
            return -1
        self.txlen = 0
        return 0


if __name__ == "__main__":
    # 最大载荷必须能装入一个UDP数据报，以避免IP分片。
    # MTU(1500) - IP(20) - UDP(8) - Header(10) - CRC(4) = 1458
    MAX_PAYLOAD = 1400

    # 先连接WiFi（在protocol.init启动轮询定时器之前）
    network.hostname(HOSTNAME)
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    wlan.connect(SSID, KEY)
    wlan.config(pm=wlan.PM_NONE)

    while not wlan.isconnected():
        time.sleep_ms(100)
        LED_B.toggle()

    LED_B.off()

    # 初始化并配置协议
    protocol.init(
        crc=True,  # 启用CRC（UDP没有完整性保证）
        seq=True,  # 启用序号检查（UDP不保证顺序）
        ack=False,  # 禁用ACK（避免阻塞主线程）
        events=True,  # 启用异步事件
        max_payload=MAX_PAYLOAD,  # 最大数据包载荷
        rtx_retries=3,  # 重传次数
        rtx_timeout_ms=100,  # 重传前的超时时间（WiFi往返时延约5毫秒）
        lock_interval_ms=10,  # 最小锁定间隔
        poll_ms=10,  # 每10毫秒调度一次协议任务
    )

    # 注册网络传输层
    protocol.register(
        name="network",
        flags=protocol.CHANNEL_FLAG_PHYSICAL,
        backend=NetworkTransport(host=HOST, port=PORT, max_payload=MAX_PAYLOAD),
    )