TCP 服务端

UIFLOW2 应用示例

回显服务器

在 UiFlow2 中打开 cores3_tcp_server_example.m5f2 项目。

此示例创建一个回显接收数据的 TCP 服务器。

UiFlow2 代码块：

示例输出：

None

MicroPython 应用示例

回显服务器

此示例创建一个回显接收数据的 TCP 服务器。

MicroPython 代码块：

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

import os, sys, io
import M5
from M5 import *
import network
import socket


label0 = None
label1 = None
label2 = None
label3 = None
label4 = None
label5 = None
label6 = None
label7 = None
label8 = None
label9 = None
title0 = None
wlan = None
tcps = None


con_info = None
con_sock = None
address = None
data = None


def setup():
    global \
        label0, \
        label1, \
        label2, \
        label3, \
        label4, \
        label5, \
        label6, \
        label7, \
        label8, \
        label9, \
        title0, \
        wlan, \
        tcps, \
        con_info, \
        con_sock, \
        address, \
        data

    M5.begin()
    Widgets.fillScreen(0x222222)
    label0 = Widgets.Label("Local IP:", 4, 32, 1.0, 0xFFFFFF, 0x222222, Widgets.FONTS.DejaVu18)
    label1 = Widgets.Label("Local Port:", 4, 60, 1.0, 0xFFFFFF, 0x222222, Widgets.FONTS.DejaVu18)
    label2 = Widgets.Label("Remote IP:", 4, 88, 1.0, 0xFFFFFF, 0x222222, Widgets.FONTS.DejaVu18)
    label3 = Widgets.Label("Remote Port:", 4, 118, 1.0, 0xFFFFFF, 0x222222, Widgets.FONTS.DejaVu18)
    label4 = Widgets.Label("Data:", 4, 144, 1.0, 0xFFFFFF, 0x222222, Widgets.FONTS.DejaVu18)
    label5 = Widgets.Label("label5", 136, 32, 1.0, 0xFFFFFF, 0x222222, Widgets.FONTS.DejaVu18)
    label6 = Widgets.Label("label6", 136, 60, 1.0, 0xFFFFFF, 0x222222, Widgets.FONTS.DejaVu18)
    label7 = Widgets.Label("label7", 136, 88, 1.0, 0xFFFFFF, 0x222222, Widgets.FONTS.DejaVu18)
    label8 = Widgets.Label("label8", 136, 118, 1.0, 0xFFFFFF, 0x222222, Widgets.FONTS.DejaVu18)
    label9 = Widgets.Label("label9", 136, 144, 1.0, 0xFFFFFF, 0x222222, Widgets.FONTS.DejaVu18)
    title0 = Widgets.Title("TCP Server", 3, 0xFFFFFF, 0x0000FF, Widgets.FONTS.DejaVu24)

    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    tcps = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    tcps.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    tcps.bind(("0.0.0.0", 8001))
    tcps.listen(5)
    tcps.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    label5.setText(str(wlan.ifconfig()[0]))
    label6.setText(str("8001"))


def loop():
    global \
        label0, \
        label1, \
        label2, \
        label3, \
        label4, \
        label5, \
        label6, \
        label7, \
        label8, \
        label9, \
        title0, \
        wlan, \
        tcps, \
        con_info, \
        con_sock, \
        address, \
        data
    M5.update()
    con_info = tcps.accept()
    con_sock = con_info[0]
    address = con_info[1]
    label7.setText(str(address[0]))
    label8.setText(str(address[1]))
    while True:
        data = con_sock.recv(1024)
        label9.setText(str(data))
        con_sock.send(data)


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参考

socket.socket()

socket.bind(address)

socket.listen([backlog])

创建一个套接字对象并将其绑定到一个地址。

参数:

address (tuple) – 包含服务器 IP 地址和端口号的元组。
backlog (int) – 排队连接的最大数量。默认值为 5。

返回:

一个套接字对象。

返回类型:

socket.socket

UiFlow2 代码块：

MicroPython 代码块：

import socket

tcps = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
tcps.bind(('0.0.0.0', 8001))
tcps.listen(5)

socket.close()

关闭套接字连接。

UiFlow2 代码块：

MicroPython 代码块：

tcps.close()

socket.accept()

接受连接。套接字必须绑定到地址并监听连接。返回值是一对 (conn, address)，其中 conn 是可用于在连接上发送和接收数据的新套接字对象，address 是绑定到连接另一端套接字的地址。

返回:: 包含客户端套接字和客户端地址的元组。
返回类型:: tuple

UiFlow2 代码块：

MicroPython 代码块：

client_socket, addr = tcps.accept()
print('Connected by', addr)

socket.recv(bufsize)

从套接字接收数据。

参数:: bufsize (int) – 一次可接收的最大数据量。
返回:: 接收到的数据。
返回类型:: bytes

UiFlow2 代码块：

MicroPython 代码块：

data = tcps.recv(1024)
print(data)

socket.read()

从套接字读取数据。

返回:: 接收到的数据。
返回类型:: bytes

UiFlow2 代码块：

MicroPython 代码块：

data = tcps.read()
print(data)

socket.send(data)

向套接字发送数据。该套接字必须连接到远程套接字。

参数:: data (bytes | str) – 要发送的数据。
返回:: 已发送的字节数。
返回类型:: int

UiFlow2 代码块：

MicroPython 代码块：

tcps.send(b'Hello, World!')
tcps.send('Hello, World!')

socket.write(data)

将字节缓冲区写入套接字。此函数将尝试将所有数据写入套接字（无“短写入”）。不过，对于非阻塞套接字，这可能无法实现，并且返回的值将小于 buf 的长度。

参数:: data (bytes | str) – 要发送的数据。
返回:: 已发送的字节数。
返回类型:: int

UiFlow2 代码块：

MicroPython 代码块：

tcps.write(b'Hello, World!')
tcps.write('Hello, World!')

socket.setblocking(flag)

设置套接字的阻塞或非阻塞模式：如果标志为假，则套接字设置为非阻塞，否则设置为阻塞模式。

此方法是某些 settimeout() 调用的简写：

sock.setblocking(True) 等同于 sock.settimeout(None)
sock.setblocking(False) 等同于 sock.settimeout(0)

param bool flag:

如果为 True，套接字将处于阻塞模式。如果为 False，套接字将处于非阻塞模式。

return:

None

rtype:

None

UiFlow2 代码块：

MicroPython 代码块：
tcps.setblocking(True)
tcps.setblocking(False)

socket.settimeout(timeout)

设置阻塞套接字操作的超时时间。值参数可以是表示秒数的非负浮点数，也可以是 None。如果给出了非零值，则在操作完成之前如果超时时间值已过，则后续套接字操作将引发 OSError 异常。如果给出了零，则套接字将处于非阻塞模式。如果给出了 None，则套接字将处于阻塞模式。

参数:: timeout (float) – 超时值（以秒为单位）。如果为 None，则套接字将处于阻塞模式。
返回:: None
返回类型:: None

UiFlow2 代码块：

MicroPython 代码块：

tcps.settimeout(5)
tcps.settimeout(None)

socket.setsockopt(level, optname, value)

设置给定套接字选项的值。所需的符号常量在套接字模块中定义（SO_* 等）。该值可以是整数或表示缓冲区的字节类对象。

参数:

level (int) – 定义选项的级别。
optname (int) – 要设置的选项的名称。
value – 为选项设置的值。

UiFlow2 代码块：

MicroPython 代码块：

tcps.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 0)
tcps.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)