Python：简易的串口收发程序

Python：简易的串口收发程序

更新历史
20201204：

1. 首次发布

最近需要搞一个基于Python的串口程序，要求：

• 能够将字符串形式的十六进制数（例如：‘12 34 CD EF’）转化成对应的二进制数据后发送给一个串口设备
• 能够从串口设备接收二进制数据，转化成字符串形式的十六进制数后进行保存。

网上找了找，发现有现成可供参考的代码。于是拿来做了点修改，变成了我所需要的代码。感谢提供代码的网友，同时，我也将自己的代码贡献出来，以供参考：

#######################
#程序功能：串口通信的简易程序，基于Python3开发。
#by：永恒的止水（CSDN），https://blog.csdn.net/qq_38791897/article/details/110670718
#######################


import serial 
import serial.tools.list_ports
from binascii import b2a_hex, a2b_hex
import threading



#全局变量
GLOBAL_DATA_LIST = []
GLOBAL_NOTEND = True # 是否读取结束




#查找可用的串口设备
#输入：无
#输出：
#   port_list：列表，每个元素都是一个字符串，字符串为串口的名称
def get_serial_port_list():
    port_list = []

    port_list_temp = list(serial.tools.list_ports.comports())
    if len(port_list_temp) == 0:
        print("[Warning] 无可用串口！")
    else:
        print("[Successful] 存在可用的串口，如下：")
        for my_port in port_list_temp:
            print(my_port)
            port_list.append(str(my_port))

    return port_list







#打开一个串口，并为其创建一个读数据的线程
#输入：
#   portx：串口名称，例如，"COM5"
#   bps：波特率。标准值有：50, 75, 110, 134, 150, 200, 300, 600, 1200, 1800, 2400, 4800,
#                        9600, 19200, 38400, 57600, 115200, 230400, 460800, 500000,
#                        576000, 921600, 1000000, 1152000, 1500000, 2000000, 2500000,
#                        3000000, 3500000, 4000000
#   timeout：超时设置，None：永远等待操作；0：立即返回请求结果；其他：等待超时时间（单位为秒）
#   stopbits：停止位的长度，标准值有：1, 1.5, 2。
#   bytesize：有效载荷数据的位宽，标准值有：5, 6, 7, 8
#   parity：校验方式。可选的有：'None', 'Even', 'Odd', 'Mark', 'Space'
#输出：
#   ser：串口对象
#   successful：若为True，表示串口打开成功，ser有效；若为False，表示串口打开失败，ser无效。
def open_serial_port(portx, bps, timeout, stopbits, bytesize, parity):
    successful = False

    if parity == 'Even':
        my_parity = 'E'
    elif parity == 'Odd':
        my_parity = 'O'
    elif parity == 'Mark':
        my_parity = 'M'
    elif parity == 'Space':
        my_parity = 'S'
    else: #default: 'None'
        my_parity = 'N'

    try:
        # 打开串口，并得到串口对象
        ser = serial.Serial(portx, bps, timeout=timeout, stopbits=stopbits, bytesize=bytesize, parity=my_parity)
 
        # 判断是否成功打开
        if(ser.is_open):
            successful = True
            th = threading.Thread(target=read_from_serial_port, args=(ser,)) # 创建一个子线程去等待读数据
            th.start()
    except Exception as e:
        print("open_serial_port error!", e)
 
    return ser, successful



 
#从串口读数据的线程的本体，将读到的数据添加到一个global列表中
#输入：
#   ser：串口对象
#输出：无
def read_from_serial_port(ser):
    global GLOBAL_DATA_LIST, GLOBAL_NOTEND
 
    # 循环接收数据（此为死循环，可用线程实现）
    while GLOBAL_NOTEND:
        if ser.in_waiting:
            data = b2a_hex(ser.read(ser.in_waiting)).decode('utf-8') #16进制的字符串，例如：'4141'，'FF'
            print("\n[received] " + data)
            GLOBAL_DATA_LIST.append(data)



#关闭串口
#输入：
#   ser：串口对象
#输出：无
def close_serial_port(ser):
    global GLOBAL_NOTEND
    GLOBAL_NOTEND = False
    ser.close()
    print('串口已关闭！')




 
#向串口写数据
#输入：
#   ser：串口对象
#   text：16进制的字符串，必须要是偶数个字符，不能含有空格。例如：'4141'，'FF'，'12345678ABCDEF'
#输出：
#   byte_num_sent：向串口写入的字节数

def write_to_serial_port(ser, text):
    byte_num_sent = ser.write(a2b_hex(text))
    print("[Successful] 向串口写入了" + str(byte_num_sent) + '个字节。')
    return byte_num_sent
 




 
if __name__ == "__main__":
    #获取可用的串口的列表
    port_list = get_serial_port_list()

    #打开列表中的某个串口，进行数据收发
    if port_list:
        #选择串口
        while True:
            portx_in_port_list = False
            portx = input("请输入要打开的串口的名称(例如：COM5)：")
            for my_port in port_list:
                if portx == my_port[:4]:
                    portx_in_port_list = True
                    break
            if portx_in_port_list:
                break
        #设置其余参数
        bps = 9600
        timeout = 1
        stopbits = 1
        bytesize = 8
        parity = 'Odd'

        #打开串口
        ser, successful = open_serial_port(portx, bps, timeout, stopbits, bytesize, parity)

        #数据收发(数据接收是通过打开串口时启动的线程完成的，因而，此处的代码只需要进行数据的发送就可以了)
        if successful == True:
            while True:
                text = input("请输入'quit'或待发送的16进制字符串：\n>>")
                if text == "quit":
                    close_serial_port(ser)
                    break
                else:
                    #去除text中的空格，例如，将'12 12  3456'变成'12123456'
                    while ' ' in text:
                        text = text.replace(' ', '')
                    #将去除了空格的字符串发往串口
                    write_to_serial_port(ser, text)

            #程序结束，输出串口在程序运行过程中接收到的所有数据
            print('串口接收到的数据：')
            print(GLOBAL_DATA_LIST)