EasyUDPServer

EasyUDPServer 和 EasyUDPClientSocket 提供了一种简单的方法来创建 UDP 服务器并以事件驱动的方式管理客户端连接。

UiFlow2 示例

简单服务器

在 UiFlow2 中打开 udp_server_core2_example.m5f2 项目。

本示例创建一个 UDP 服务器，监听 8000 端口并将接收到的数据显示在屏幕上。

UiFlow2 代码块：

示例输出：

None

MicroPython 示例

简单服务器

本示例创建一个 UDP 服务器，监听 8000 端口并将接收到的数据显示在屏幕上。

MicroPython 代码块：

# SPDX-FileCopyrightText: 2026 M5Stack Technology CO LTD
#
# SPDX-License-Identifier: MIT

import os, sys, io
import M5
from M5 import *
from easysocket import EasyUDPServer
import network
import time


title0 = None
label2 = None
label0 = None
label1 = None
wlan_sta = None
udps_0 = None


received_data = None
client_address_port = None


def udps_0_received_event(args):
    global title0, label2, label0, label1, wlan_sta, udps_0, received_data, client_address_port
    server, client_address_port, received_data = args
    label1.setText(str("Receive Msg:"))
    label2.setText(str((str(received_data) + str(client_address_port))))
    print((str(received_data) + str(client_address_port)))
    udps_0.sendto(received_data, client_address_port)


def setup():
    global title0, label2, label0, label1, wlan_sta, udps_0, received_data, client_address_port

    M5.begin()
    Widgets.setRotation(1)
    Widgets.fillScreen(0x222222)
    title0 = Widgets.Title(
        "UDPServer Core2 Example", 3, 0xFFFFFF, 0x0000FF, Widgets.FONTS.DejaVu18
    )
    label2 = Widgets.Label("", 1, 146, 1.0, 0xFFFFFF, 0x222222, Widgets.FONTS.DejaVu18)
    label0 = Widgets.Label("Local IP:", 2, 68, 1.0, 0xFFFFFF, 0x222222, Widgets.FONTS.DejaVu18)
    label1 = Widgets.Label("Receive Msg:", 2, 109, 1.0, 0xFFFFFF, 0x222222, Widgets.FONTS.DejaVu18)

    wlan_sta = network.WLAN(network.STA_IF)
    print("wait network connecting")
    while not (wlan_sta.isconnected()):
        print(".")
        time.sleep(1)
    print("connect success")
    print(wlan_sta.ifconfig()[0])
    label0.setText(str((str("Local IP:") + str((wlan_sta.ifconfig()[0])))))
    udps_0 = EasyUDPServer(
        host="0.0.0.0",
        port=8000,
        mode=EasyUDPServer.MODE_UNICAST,
        multicast_group=None,
        verbose=False,
    )
    udps_0.on_data_received(udps_0_received_event)


def loop():
    global title0, label2, label0, label1, wlan_sta, udps_0, received_data, client_address_port
    M5.update()
    udps_0.check_event(timeout=-1)


if __name__ == "__main__":
    try:
        setup()
        while True:
            loop()
    except (Exception, KeyboardInterrupt) as e:
        try:
            from utility import print_error_msg

            print_error_msg(e)
        except ImportError:
            print("please update to latest firmware")

示例输出：

None

API参考

class software.easysocket.udp_server.EasyUDPServer(host='0.0.0.0', port=8000, mode=0, multicast_group=None, verbose=False)

基类：object

创建一个 EasyUDPServer 对象。

参数:

host (str) – 要绑定的主机地址。
port (int) – 要绑定的端口号。
mode (int) – UDP模式（单播、广播、组播）。默认为单播。
multicast_group (str) – 多播组地址（如果模式为多播，则为必填项）。
verbose (bool) – 是否打印详细输出。

备注

初始化后自动启动服务。

备注

这个类是非阻塞式的，并且是事件驱动的。你需要定期调用 check_event() 来处理事件。

UiFlow2 代码块：

MicroPython 代码块：

from easysocket import EasyUDPServer

udp_server = EasyUDPServer(host="0.0.0.0", port=8080)

start()

启动服务器。

MicroPython 代码块：

udp_server.start()

stop()

停止服务器。

MicroPython 代码块：

udp_server.stop()

on_data_received(callback)

设置数据接收事件的回调函数。

参数:: callback – 回调函数。

UiFlow2 代码块：

MicroPython 代码块：

def on_data_received_cb(args):
    client, address, data = args
    print("Received:", data, "from", address)

udp_server.on_data_received(on_data_received_cb)

check_event(timeout=-1)

检查事件。

参数:: timeout (int) – 超时时间，单位为毫秒。默认值为 -1（无超时）。

UiFlow2 代码块：

MicroPython 代码块：

udp_server.check_event()

sendto(data, address)

向远程服务器发送数据。

参数:

data (bytes) – 要发送的数据。
address (tuple) – 要发送数据的（主机，端口）元组。

返回:

已发送的字节数。

UiFlow2 代码块：

MicroPython 代码块：

udp_server.sendto(b"Hello", ("192.168.1.100", 8080))