代码收藏家技术教程 2022-07-22

python–串口通信

Python之串口通信

串口通信是嵌入式工程师的必备技能，如果能自己写一个简单的上位机的话，肯定会加分不少。
本系列是把相关的工作做个记录，毕竟Python不是经常用，老是忘记很多东西。用的时候总是重头再学一遍太麻烦了。毕竟3个月以后，自己写的程序也如同天书一般。

开发环境：
Python版本是最新的3.10.1，
编译和调试用的是PyCharm2021.4社区版，
需要用到的库：串口通信库pyserial，命令行：pip3 install pyserial回车即可安装。
串口设备：自己的STM32板子
各种版本的软件，工具，我都喜欢用新版本，主要工具软件大概是每年更新一次。
因为之前吃过亏，Protel99用了好多年，后面一下用到AD2013，感觉全变了，极不适应。还有UG也是这样。每年升级的话，软件更新不会太多，接受起来比较容易。

pyserial 使用

首先导入pyserial库和定义全局变量

import serial
import serial.tools.list_ports
import threading

com_rx_buf = ''				# 接收缓冲区
com_tx_buf = ''				# 发送缓冲区
COMM = serial.Serial()		# 定义串口对象
port_list: list				# 可用串口列表
port_select: list			# 选择好的串口

port_list = serial.tools.list_ports.comports() # 返回值为可用的串口列表，其__len__属性为可用串口的数量。本机的port_list.len()为2。

post_list.len()，为2
post_list[0].device字段为串口编号，为COM99
post_list[1].device字段为串口编号，为COM97
串口的编号，在设备管理器中是可以设置的，我一般是将指定设备安排一个比较大的串口号，列表的最后一项就是我想要的设备了。

打开串口

    COMM = serial.Serial(serial_port, 115200, timeout=0.01)
    if COMM.isOpen():
        print(serial_port, "open success")
        return 0
    else:
        print("open failed")
        return 255

关闭串口

对应的是关闭串口，程序结束时必须关闭串口，否则，如果程序异常退出后，再次运行程序时，串口会因为已经处于打开状态而出错。

    COMM.close()

发送数据

buf = [0,3,55,32] # buf的格式
COMM.write(buf)

接收数据

buf = COMM.readall() 可以读取出串口接收的所有数据。可以开一个接收线程，里面是一个死循环，每经过一定的延迟时间，执行一次本指令即可。
需要注意的是，如果对方发送的是一长串数据，又恰好在数据发送了一部分的时候延时时间到，这样收到的数据就不完全，此时可以再经过一个延迟，再执行一次readall()，即可读取全部数据了。这个工作放在接收线程中完成即可。

全部程序如下，多出了好多函数，文件是从现有工程中摘出来的，捡有用的看吧。

# 安装：pip3 install pyserial   //python3
import serial
import serial.tools.list_ports
import time
import threading


com_rx_buf = ''				# 接收缓冲区
com_tx_buf = ''				# 发送缓冲区
COMM = serial.Serial()		# 定义串口对象
port_list: list				# 可用串口列表
port_select: list			# 选择好的串口


# 无串口返回0，
# 返回可用的串口列表
def get_com_list():
    global port_list
    # a = serial.tools.list_ports.comports()
    # print(a)
    # port_list = list(serial.tools.list_ports.comports())
    port_list = serial.tools.list_ports.comports()
    return port_list


def set_com_port(n=0):
    global port_list
    global port_select
    port_select = port_list[n]
    return port_select.device


# 打开串口
def serial_open(n=0):
    global COMM
    serial_port = set_com_port(n)
    COMM = serial.Serial(serial_port, 115200, timeout=0.01)
    if COMM.isOpen():
        print(serial_port, "open success")
        return 0
    else:
        print("open failed")
        return 255


# 关闭串口
def serial_close():
    global COMM
    COMM.close()
    print(COMM.name + "closed.")


def set_com_rx_buf(buf=''):
    global com_rx_buf
    com_rx_buf = buf


def set_com_tx_buf(buf=''):
    global com_tx_buf
    com_tx_buf = buf


def get_com_rx_buf():
    global com_rx_buf
    return com_rx_buf


def get_com_tx_buf():
    global com_tx_buf
    return com_tx_buf


def thread_com_receive():
    while True:
        try:
            rx_buf = ''
            rx_buf = COMM.read()  # 转化为整型数字
            if rx_buf != b'':
                time.sleep(0.01)
                rx_buf = rx_buf + COMM.read_all()
                print("串口收到消息:", rx_buf)
            time.sleep(0.01)
        except:
            pass
    pass


# def serial_encode(addr=0, command=0, param1=0, param0=0):
#     buf = [addr, command, param1, param0, 0, 0, 0, 0]
#     print(buf)
#     return buf


def serial_send_command(addr=0, command=0, param1=0, param0=0, data3=0, data2=0, data1=0, data0=0):
    buf = [addr, command, param1, param0, data3, data2, data1, data0]
    COMM.write(buf)
    pass


def serial_init():
    buf = "AT+CG\r\n"
    COMM.write(buf)
    time.sleep(0.05)
    buf = COMM.read_all()
    if buf != "OK\r\n":
        return 254  # 进入调试模式失败

    buf = "AT+CAN_MODE=0\r\n"
    COMM.write(buf)
    time.sleep(0.05)
    buf = COMM.read_all()
    if buf != "OK\r\n":
        return 253          # 进入正常模式失败，模块处于1状态，即环回模式中

    buf = "AT+CAN_BAUD=500000\r\n"
    COMM.write(buf)
    time.sleep(0.05)
    buf = COMM.read_all()
    if buf != "OK\r\n":
        return 253          # 波特率设置失败

    buf = "AT+FRAMEFORMAT=1,0,\r\n"
    COMM.write(buf)
    time.sleep(0.05)
    buf = COMM.read_all()
    if buf != "OK\r\n":
        return 253          # 波特率设置失败

    buf = "AT+ET\r\n"       # 进入透传模式
    COMM.write(buf)
    time.sleep(0.05)
    buf = COMM.read_all()
    if buf != "OK\r\n":
        return 255  # 不是CAN模块


if __name__ == '__main__':
    get_com_list()
    len = port_list.__len__()
    device = port_list[0].device
    print(len, device)
    serial_open()
    thread1 = threading.Thread(target=thread_com_receive)
    thread1.start()
 
    buf = [33, 2, 0x55, 1, 4]
    COMM.write(buf)

    # serial_close()