EasyUDPClient

EasyUDPClient 提供了一种以事件驱动方式创建 UDP 客户端的简单方法。

UiFlow2 示例

简单客户端

在 UiFlow2 中打开 udp_client_core2_example.m5f2 项目。

本示例创建一个 UDP 客户端，该客户端连接到服务器并发送数据。

UiFlow2 代码块：

示例输出：

None

MicroPython 示例

简单客户端

本示例创建一个 UDP 客户端，该客户端连接到服务器并发送数据。

MicroPython 代码块：

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

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


title0 = None
label2 = None
label0 = None
label1 = None
wlan_sta = None
udpc_0 = None


import random

received_data = None
client_address_port = None


def udpc_0_received_event(args):
    global title0, label2, label0, label1, wlan_sta, udpc_0, received_data, client_address_port
    client, client_address_port, received_data = args
    label1.setText(str("Receive Msg:"))
    label2.setText(str((str(received_data) + str(client_address_port))))


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

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

    wlan_sta = network.WLAN(network.STA_IF)
    wlan_sta.active(True)
    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])))))
    udpc_0 = EasyUDPClient(
        "Please enter the UDP server IP address.", 8000, EasyUDPClient.MODE_UNICAST
    )
    udpc_0.on_data_received(udpc_0_received_event)


def loop():
    global title0, label2, label0, label1, wlan_sta, udpc_0, received_data, client_address_port
    M5.update()
    udpc_0.check_event(timeout=1)
    if BtnA.wasPressed():
        udpc_0.send(str((random.randint(1, 100))))


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_client.EasyUDPClient(remote_host, remote_port, mode=0)

基类：object

创建一个 EasyUDPClient 对象。

参数:

remote_host (str) – 远程主机地址。
remote_port (int) – 远程端口号。
mode (int) – UDP模式（单播、广播、组播）。默认为单播。

备注

对象创建时，连接会在后台启动。

备注

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

UiFlow2 代码块：

MicroPython 代码块：

from easysocket import EasyUDPClient

udp_client = EasyUDPClient("192.168.1.100", 8080, mode=EasyUDPClient.MODE_UNICAST)

connect()

连接到远程服务器。

MicroPython 代码块：

udp_client.connect()

on_data_received(callback)

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

参数:: callback – 回调函数。

UiFlow2 代码块：

MicroPython 代码块：

def on_data_received_cb(client, data):
    print("Received:", data)

udp_client.on_data_received(on_data_received_cb)

check_event(timeout=-1)

检查事件。

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

UiFlow2 代码块：

MicroPython 代码块：

udp_client.check_event()

send(data)

向远程服务器发送数据。

参数:: data (bytes) – 要发送的数据。
返回:: 已发送的字节数。

UiFlow2 代码块：

MicroPython 代码块：

udp_client.send(b"Hello")

sendto(data, address=None)

向远程服务器发送数据。

参数:

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

返回:

已发送的字节数。

UiFlow2 代码块：

MicroPython 代码块：

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

close()

关闭套接字连接。

UiFlow2 代码块：

MicroPython 代码块：

udp_client.close()

getsockname(*args, **kwargs)

返回套接字自身的地址。

返回:: 套接字自身的地址。格式为（主机，端口）。
返回类型:: tuple

UiFlow2 代码块：

MicroPython 代码块：

# get local ip address
client_socket.getsockname()[0]

getpeername(*args, **kwargs)

返回套接字所连接的远程地址。

返回:: 远程地址。格式为（主机，端口）。
返回类型:: tuple

UiFlow2 代码块：

MicroPython 代码块：

# get remote ip address
client_socket.getpeername()[0]