1. select module

2. Use the select module to monitor serial port events

# Python env   : MicroPython v1.23.0               
# -*- coding: utf-8 -*-        
# @Time    : 2024/6/27 上午12:08   
# @Author  : 李清水            
# @File    : main.py       
# @Description : Select类实验，主要完成使用select模块进行串口状态读取和数据接收等操作

# ======================================== 导入相关模块 ========================================
# 硬件相关的模块
from machine import UART, Pin
# 时间相关的模块
import time
# 导入select模块
import select

# ======================================== 全局变量 ============================================

# 创建一个串口数据缓冲区
uart_buffer = []
# 串口数据缓冲区最大长度
uart_buffer_max_len = 100

# ======================================== 功能函数 ============================================

def handle_uart_error(event: int) -> None:
    """
    处理串口错误事件。

    当串口发生错误时，调用该方法进行错误处理。它会尝试重置串口连接，
    并重新初始化串口设置以确保通信恢复正常。

    Args:
        event (int): 事件类型标识。只有当事件类型为 `POLLERR` 时才会调用该函数。
    """
    if event & select.POLLERR:
        print("UART error occurred, resetting UART")
        # 错误时熄灭LED指示灯
        LED.off()
        # 可能的恢复操作，关闭串口然后重新初始化
        uart.deinit()
        # 等待1秒后重新初始化
        time.sleep(1)
        uart.init(bits=8, parity=None, stop=1, tx=0, rx=1, timeout=5)
        # 重新亮起LED，表示恢复工作
        LED.on()

def handle_uart_disconnect(event: int) -> None:
    """
    处理串口断开事件。

    当串口连接断开时，调用该方法进行断开处理。它会停止串口操作并熄灭LED指示灯，
    同时准备好重新连接串口。

    Args:
        event (int): 事件类型标识。只有当事件类型为 `POLLHUP` 时才会调用该函数。
    """
    if event & select.POLLHUP:
        print("UART disconnected, resetting UART")
        # 断开时熄灭LED
        LED.off()
        # 停止串口操作
        uart.deinit()
        # 等待1秒，准备重新连接
        time.sleep(1)

# ======================================== 自定义类 ============================================

# ======================================== 初始化配置 ==========================================

# 上电延时3s
time.sleep(3)
# 打印调试消息
print("FreakStudio: Using select module to read UART data")

# 创建串口对象，设置波特率为9600
uart: UART = UART(0, 9600)
# 初始化uart对象，数据位为8，无校验位，停止位为1
# 设置输入引脚为GPIO1，输出引脚为GPIO2
# GPIO1和GPIO2短接，串口0自发自收
# 设置串口超时时间为5ms
uart.init(bits=8,
          parity=None,
          stop=1,
          tx=0,
          rx=1,
          timeout=5)

# 设置GPIO 25为LED输出引脚，下拉电阻使能
LED: Pin = Pin(25, Pin.OUT, Pin.PULL_DOWN)

# 创建一个轮询对象，用于监视串口的活动状态
uart_read_poll: select.poll = select.poll()
# 将串口对象注册到轮询对象中
# select.POLLIN 是一个常量，表示可读事件
# select.POLLERR 是一个常量，表示错误事件
# select.POLLHUP 是一个常量，表示设备断开事件
# 当串口设备上有数据可读时，uart_read_poll 对象会被触发
uart_read_poll.register(uart, select.POLLIN | select.POLLERR | select.POLLHUP)

# ========================================  主程序  ============================================

while True:

    # 点亮LED灯
    LED.on()

    # 使用Python的select模块中的poll方法来监听串口事件
    # poll方法会阻塞等待,直到串口上有数据可以接收
    # 设定超时时间为50ms，50ms内无事件发生则返回
    events = uart_read_poll.poll(50)

    # events是一个包含设备对象的列表，包括fd属性和events属性
    print('events =', events)

    # 如果有事件发生，处理这些事件
    if events:
        # 循环遍历事件列表
        for event in events:
            # 打印发生事件的设备和事件类型
            print(" fd attributes/the name of the device          : ",event[0])
            print(" events property/the trigger type of the event : ",event[1])

            # 若是uart对象有事件发生，则处理该事件
            if event[0] == uart:

                # 调用handle_uart_error方法处理串口错误事件
                handle_uart_error(event[1])

                # 调用handle_uart_disconnect方法处理串口断开事件
                handle_uart_disconnect(event[1])

                # 若是uart对象有可读数据，则读取并打印接收到的数据
                if event[1] & select.POLLIN:
                    # 尝试读取串口数据
                    try:
                        # 读取一个字节的串口数据
                        data = uart.read(1)
                        # 打印接收到的数据
                        print("uart recv data = ", data)
                        # 将数据添加到缓冲区
                        uart_buffer.append(data)
                        # 可以选择在此处处理或分析接收到的数据
                        # ...

                        # 若缓冲区数据满时，执行清空操作
                        if len(uart_buffer) >= uart_buffer_max_len:
                            # 清空缓冲区
                            uart_buffer.clear()

                        # 熄灭LED灯表示数据已读取完毕
                        LED.off()
                    except Exception as e:
                        print("uart read error: ", e)

events = [(UART(0, baudrate=9600, bits=8, parity=None, stop=1, tx=0, rx=1, txbuf=256, rxbuf=256, timeout=5, timeout_char=2, invert=None), 1)]
fd attributes/the name of the device          :  UART(0, baudrate=9600, bits=8, parity=None, stop=1, tx=0, rx=1, txbuf=256, rxbuf=256, timeout=5, timeout_char=2, invert=None)
events property/the trigger type of the event :  1
uart recv data =  b'a'

stream devices and block devices

The concepts of stream devices and block devices were primarily introduced in Linux systems, where devices are categorized into two main types: stream devices and block devices. Stream devices refer to devices that read and write data in the form of ByteFlow, such as tape drives, serial ports, and pipes. Block devices, on the other hand, are devices that read and write data in units of data blocks, such as hard disk drives and USB flash drives. In practical applications, we need to select the appropriate type of device based on specific requirements to achieve efficient data transmission.

Stream devices and block devices each have their own characteristics and application scenarios: