Tweet

1. Fundamentals of RS232 Communication

2. Introduction to MAX3232 Chip

There are two types of DB9 connectors:

Male connector: with 9 protruding metal pins

Female connector: features 9 small holes to receive the pins of the male connector

3. USB to RS232 Adapter

4. Application Experiment

# 自定义RS232串口通信类，使用软件流控
class RS232:
    """
    自定义RS232串口通信类，支持软件流控（XON/XOFF协议）。

    该类实现了RS232串口的发送和接收功能，支持基于XON/XOFF的流控制机制。流控制用于在数据传输过程中，
    根据接收缓冲区的状态来控制数据的发送与暂停，防止接收缓冲区溢出。通过该类，可以进行串口数据的发送、接收和流控管理。

    Attributes:
        uart (UART): 串口实例，代表用于通信的硬件接口。
        tx_buffer (bytearray): 发送缓冲区，用于存储待发送的数据。
        rx_buffer (bytearray): 接收缓冲区，用于存储接收到的数据。
        rx_pos (int): 接收缓冲区的指针，标记接收数据的位置。
        rx_buf_size (int): 接收缓冲区的大小，表示接收缓冲区能够存储的最大数据量。
        tx_pos (int): 发送缓冲区的指针，标记已准备发送数据的位置。
        tx_buf_size (int): 发送缓冲区的大小，表示发送缓冲区能够存储的最大数据量。
        xon (bool): 软件流控状态，表示当前是否可以发送数据。`True`表示可以发送，`False`表示需要暂停。

    Methods:
        send(data: str) -> None:
            发送数据到串口。支持XON/XOFF流控，发送过程中如果流控为暂停状态，将等待直到XON信号恢复。

        receive() -> bytearray:
            从串口接收数据。接收过程中如果接收缓冲区已满，则会发送XOFF信号暂停数据传输。
    """

    # 软件流程相关类变量：两个 ASCII 控制字符
    # XON (CTRL-Q)
    XON = b'\x11'
    # XOFF (CTRL-S)
    XOFF = b'\x13'

    def __init__(self, uart, rx_buf_size=256, tx_buf_size=256):
        """
        初始化串口对象。

        Args:
            uart (UART): 使用的串口实例。
            rx_buf_size (int, optional): 接收缓冲区的大小，默认值为256。
            tx_buf_size (int, optional): 发送缓冲区的大小，默认值为256。

        Returns:
            None
        """

        # 将传入的 UART 对象保存
        self.uart = uart
        # 发送缓冲区，用于存储待发送的数据
        self.tx_buffer = bytearray(tx_buf_size)
        # 接收缓冲区，用于存储接收到的数据
        self.rx_buffer = bytearray(rx_buf_size)

        # 接收缓冲区的指针，跟踪数据位置
        self.rx_pos = 0
        # 接收缓冲区大小
        self.rx_buf_size = rx_buf_size

        # 发送缓冲区的指针，跟踪数据位置
        self.tx_pos = 0
        # 发送缓冲区大小
        self.tx_buf_size = tx_buf_size

        # XON 状态，表示可以发送数据
        self.xon = True

    def send(self, data : str) -> None:
        """
        发送数据。

        该函数将传入的数据发送到串口，遵循XON/XOFF流控机制。如果当前流控不允许
        发送数据，函数会等待直到流控解除。

        Args:
            data (str): 待发送的数据字符串。

        Returns:
            None
        """

        # 如果data不是字符串类型，进行数据转换
        if not isinstance(data, str):
            data = str(data)

        # 如果数据长度超过发送缓冲区大小，进行截断
        if len(data) > self.tx_buf_size:
            data = data[0:self.tx_buf_size]

        # 将待发送的数据加入到发送缓冲区中
        # 按照当前的发送缓冲区指针 self.tx_pos 插入数据
        self.tx_buffer[self.tx_pos:self.tx_pos+len(data)] = data.encode('utf-8')
        # 更新发送缓冲区指针
        self.tx_pos += len(data)

        # 遍历发送缓冲区中的数据
        for i in range(self.tx_pos):

            # 如果 XOFF 被激活，等待 XON 信号继续传输
            while not self.xon:
                time.sleep(0.01)  # 等待 10 毫秒

            # 发送单字节，确保为 bytes 类型
            self.uart.write(bytes([self.tx_buffer[i]]))

        # 发送完毕后，通过重置指针清空发送缓冲区
        self.tx_pos = 0

    def receive(self) -> bytearray:
        """
        接收数据。

        该函数检查串口接收缓冲区，读取接收到的数据并返回。若接收缓冲区已满，
        则通过XOFF信号通知暂停数据传输。

        Returns:
            bytearray: 接收到的数据，作为字节数组返回。
        """

        # 循环检查 UART 接收缓冲区是否有数据
        while uart.any():

            # 逐字节接收数据
            byte = self.uart.read(1)

            # 判断是否是 XON 或 XOFF 信号
            if byte == RS232.XON:
                # 收到 XON，允许继续传输
                self.xon = True
            elif byte == RS232.XOFF:
                # 收到 XOFF，暂停传输
                self.xon = False
            else:
                # 接收到的实际数据存储到接收缓冲区
                self.rx_buffer[self.rx_pos] = byte[0]
                # 移动缓冲区指针
                self.rx_pos += 1

                # 缓冲区已满，发送 XOFF 信号
                if self.rx_pos >= self.rx_buf_size:
                    self.uart.write(RS232.XOFF)
                    break
        # 截取数据
        data = self.rx_buffer[0:self.rx_pos]
        # 重置接收缓冲区指针
        self.rx_pos = 0
        # 返回接收的数据
        return data

# Python env   : MicroPython v1.23.0
# -*- coding: utf-8 -*-        
# @Time    : 2024/9/22 下午11:23   
# @Author  : 李清水            
# @File    : main.py       
# @Description : UART类实验，RS232串口通信实现软件流控功能

# ======================================== 导入相关模块 =========================================

# 硬件相关的模块
from machine import UART, Pin, Timer
# 时间相关的模块
import time

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

# 串口发送计数变量
send_count: int = 0

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

# 定时器回调函数，用于定时接收数据
def receive_data(timer: Timer) -> None:
    """
    定时器回调函数，用于定时接收数据。

    该函数会检查串口接收的数据，并根据流控状态控制LED灯的开关。
    如果流控信号允许发送数据，LED会熄灭；如果流控信号禁止发送数据，LED会亮起。

    Args:
        timer (Timer): 定时器实例。

    Returns:
        None
    """
    global rs232,LED

    # 接收数据
    data: bytearray = rs232.receive()

    # 判断是否接收到数据
    if len(data) > 0:
        # 将接收到的数据转换为十六进制并打印
        hex_data = [hex(byte) for byte in data]
        print("Received data:", hex_data)

    # 判断rs232的流控状态
    if rs232.xon:
        # 如果可以发送数据，熄灭 LED
        LED.value(0)
    else:
        # 如果不能发送数据，点亮 LED
        LED.value(1)

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

# 自定义RS232串口通信类，使用软件流控
class RS232:
    """
    自定义RS232串口通信类，支持软件流控（XON/XOFF协议）。

    该类实现了RS232串口的发送和接收功能，支持基于XON/XOFF的流控制机制。流控制用于在数据传输过程中，
    根据接收缓冲区的状态来控制数据的发送与暂停，防止接收缓冲区溢出。通过该类，可以进行串口数据的发送、接收和流控管理。

    Attributes:
        uart (UART): 串口实例，代表用于通信的硬件接口。
        tx_buffer (bytearray): 发送缓冲区，用于存储待发送的数据。
        rx_buffer (bytearray): 接收缓冲区，用于存储接收到的数据。
        rx_pos (int): 接收缓冲区的指针，标记接收数据的位置。
        rx_buf_size (int): 接收缓冲区的大小，表示接收缓冲区能够存储的最大数据量。
        tx_pos (int): 发送缓冲区的指针，标记已准备发送数据的位置。
        tx_buf_size (int): 发送缓冲区的大小，表示发送缓冲区能够存储的最大数据量。
        xon (bool): 软件流控状态，表示当前是否可以发送数据。`True`表示可以发送，`False`表示需要暂停。

    Methods:
        send(data: str) -> None:
            发送数据到串口。支持XON/XOFF流控，发送过程中如果流控为暂停状态，将等待直到XON信号恢复。

        receive() -> bytearray:
            从串口接收数据。接收过程中如果接收缓冲区已满，则会发送XOFF信号暂停数据传输。
    """

    # 软件流程相关类变量：两个 ASCII 控制字符
    # XON (CTRL-Q)
    XON = b'\x11'
    # XOFF (CTRL-S)
    XOFF = b'\x13'

    def __init__(self, uart, rx_buf_size=256, tx_buf_size=256):
        """
        初始化串口对象。

        Args:
            uart (UART): 使用的串口实例。
            rx_buf_size (int, optional): 接收缓冲区的大小，默认值为256。
            tx_buf_size (int, optional): 发送缓冲区的大小，默认值为256。

        Returns:
            None
        """

        # 将传入的 UART 对象保存
        self.uart = uart
        # 发送缓冲区，用于存储待发送的数据
        self.tx_buffer = bytearray(tx_buf_size)
        # 接收缓冲区，用于存储接收到的数据
        self.rx_buffer = bytearray(rx_buf_size)

        # 接收缓冲区的指针，跟踪数据位置
        self.rx_pos = 0
        # 接收缓冲区大小
        self.rx_buf_size = rx_buf_size

        # 发送缓冲区的指针，跟踪数据位置
        self.tx_pos = 0
        # 发送缓冲区大小
        self.tx_buf_size = tx_buf_size

        # XON 状态，表示可以发送数据
        self.xon = True

    def send(self, data : str) -> None:
        """
        发送数据。

        该函数将传入的数据发送到串口，遵循XON/XOFF流控机制。如果当前流控不允许
        发送数据，函数会等待直到流控解除。

        Args:
            data (str): 待发送的数据字符串。

        Returns:
            None
        """

        # 如果data不是字符串类型，进行数据转换
        if not isinstance(data, str):
            data = str(data)

        # 如果数据长度超过发送缓冲区大小，进行截断
        if len(data) > self.tx_buf_size:
            data = data[0:self.tx_buf_size]

        # 将待发送的数据加入到发送缓冲区中
        # 按照当前的发送缓冲区指针 self.tx_pos 插入数据
        self.tx_buffer[self.tx_pos:self.tx_pos+len(data)] = data.encode('utf-8')
        # 更新发送缓冲区指针
        self.tx_pos += len(data)

        # 遍历发送缓冲区中的数据
        for i in range(self.tx_pos):

            # 如果 XOFF 被激活，等待 XON 信号继续传输
            while not self.xon:
                time.sleep(0.01)  # 等待 10 毫秒

            # 发送单字节，确保为 bytes 类型
            self.uart.write(bytes([self.tx_buffer[i]]))

        # 发送完毕后，通过重置指针清空发送缓冲区
        self.tx_pos = 0

    def receive(self) -> bytearray:
        """
        接收数据。

        该函数检查串口接收缓冲区，读取接收到的数据并返回。若接收缓冲区已满，
        则通过XOFF信号通知暂停数据传输。

        Returns:
            bytearray: 接收到的数据，作为字节数组返回。
        """

        # 循环检查 UART 接收缓冲区是否有数据
        while uart.any():

            # 逐字节接收数据
            byte = self.uart.read(1)

            # 判断是否是 XON 或 XOFF 信号
            if byte == RS232.XON:
                # 收到 XON，允许继续传输
                self.xon = True
            elif byte == RS232.XOFF:
                # 收到 XOFF，暂停传输
                self.xon = False
            else:
                # 接收到的实际数据存储到接收缓冲区
                self.rx_buffer[self.rx_pos] = byte[0]
                # 移动缓冲区指针
                self.rx_pos += 1

                # 缓冲区已满，发送 XOFF 信号
                if self.rx_pos >= self.rx_buf_size:
                    self.uart.write(RS232.XOFF)
                    break
        # 截取数据
        data = self.rx_buffer[0:self.rx_pos]
        # 重置接收缓冲区指针
        self.rx_pos = 0
        # 返回接收的数据
        return data

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

# 上电延时3s
time.sleep(3)
# 打印调试消息
print("FreakStudio: RS232 Software Flow Control Demo")

# 创建串口对象，设置波特率为115200
uart: UART = UART(0, 115200)
# 初始化uart对象，波特率为115200，数据位为8，无校验位，停止位为1
# 设置接收引脚为GPIO17，发送引脚为GPIO16
# 设置串口超时时间为100ms
uart.init(baudrate  = 115200,
          bits      = 8,
          parity    = None,
          stop      = 1,
          tx        = 16,
          rx        = 17,
          timeout   = 100)

# 初始化RS232对象,设置接收缓冲区大小为32，发送缓冲区大小为32
rs232: RS232 = RS232(uart, 32, 32)

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

# 创建定时器对象，每隔10ms接收数据
timer: Timer = Timer(-1)
timer.init(period=10, mode=Timer.PERIODIC, callback=receive_data)

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

# 循环执行
while True:
    # 发送固定数据，每次发送完成都换行
    rs232.send("RS232 Send Data :" + str(send_count)+'\r\n')
    # 发送计数变量递增
    send_count = send_count + 1
    # 延时1s
    time.sleep(1)

11 22 33 44 55 66 77 88 99 00 11 22 33 44 55 66 77 88 99 00 11 22 33 44 55 66 77 88 99 00 11 22 33 44 55 66 77 88 99 00 11 22 33 44 55 66 77 88 99 00 11 22 33 44 55 66 77 88 99 00