首页编程正文

Python应用开发之实现串口通信_python_

2023-05-26 07:25:12 32

前言

在嵌入式开发中我们经常会用到串口，串口通信简单，使用起来方便，且适用场景多，因此串口常常用来输出调试Log或者跟其他外设进行通讯，也可以用作上位机和下位机之间的通讯。那这一讲我就简单介绍一下如何用Python编写一个串口收发程序。

1.环境搭建

Python的环境这里就不具体介绍了，网上教程也很多，我主要介绍一下我们这一讲用到的类库以及当前的版本。

package	version
time	注：python标准库，无需安装
pyserial	v3.5

注：不懂如何安装库的同学请自行查阅资料。

特别说明：本文举的例子皆是基于windows系统下python3来实现的，其他平台不一定完全适配。

2.硬件准备

我们要通过电脑的USB接口发送串口数据，在硬件上首先得保证有可用的串口设备接入，并且正确的安装了驱动。

我这里用的是一个CH340的串口转TTL的模块来测试的。

可以用两个这种模块相互收发，也可以单个模块自发自收，或者接入其他嵌入式设备也行。

硬件接入之后我们可以在设备管理器看到设备的端口号，如下图所示：

3.代码编写与测试

3.1 简单测一下串口收发

示例代码如下：

import serial #导入模块

try:
    # 端口号，根据自己实际情况输入，可以在设备管理器查看
    port = "COM6"
    # 串口波特率，根据自己实际情况输入
    bps = 9600
    # 超时时间,None：永远等待操作，0为立即返回请求结果，其他值为等待超时时间(单位为秒）
    time = 5
    # 打开串口，并返回串口对象
    uart = serial.Serial(port, bps, timeout = time)

    # 串口发送一个字符串
    len = uart.write("hello world".encode('utf-8')) 
    print("send len: ", len)

    # 串口接收一个字符串
    str = ''
    for i in range(len):
        str += uart.read().decode("utf-8")
    print("receive data: ", str)

    # 关闭串口
    uart.close()

except Exception as result:
    print("******error******：", result)

运行测试：

我这里是自发自收，如果运行没有出错并且成功发送和接收到正确的数据，说明串口通信是走通了，那么接下来就可以继续完善这个流程。
如果出现其他问题，则需要检查硬件和软件，确保所有配置都没问题才能继续进行下一步。

3.2 补充细节

1、串口发送各种不同类型的数据。

串口数据常用字符串和十六进制(hex)表示。下面列举了一些例子，可以作为一种参考。

示例如下：

data1 = "hello world"        # 字符串
data2 = b"hello world"       # bytes
data3 = "你好"               # 中文字符串
data4 = 0x0A                 # 整形(以16进制表示)
data5 = [0x10, 0x11, 0x12]   # 列表/数组(以16进制表示)

len = uart.write(data1.encode('utf-8'))         # 发送字符串"hello world"
len = uart.write(data2)                         # 发送字符串"hello world"
len = uart.write(data3.encode('utf-8'))         # 以utf-8编码方式发送字符串"你好"（6字节）
len = uart.write(data3.encode('gbk'))           # 以gbk编码方式发送字符串"你好"（4字节）
len = uart.write(chr(data4.encode("utf-8"))     # 发送16进制数据0x0A（1字节）
for x in data5:                                 # 遍历列表/数组的所有元素并依次发送
    len = uart.write(chr(x).encode("utf-8"))

2、完善串口接收流程

串口接收跟发送一样，接收数据也是常用字符串和十六进制(hex)表示。

可以根据以下两种方式接收数据:

str = uart.read(uart.in_waiting).decode("utf-8")   # 以字符串接收
str = uart.read().hex()                            # 以16进制(hex)接收

接收的时间很短，大部分时间其实是在等待接收，所以我们需要加一个死循环或者循环线程来确保串口一直在等待接收。

示例如下：

while True:
    if uart.in_waiting:
        # str = uart.read(uart.in_waiting).decode("utf-8")   # 以字符串接收
        str = uart.read().hex()                            # 以16进制(hex)接收
        print(str)                                         # 打印数据

3、扫描端口

在不知道串口端口号的情况下可以先扫描一下可用的端口。

示例如下：

import serial 
import serial.tools.list_ports

port_list = list(serial.tools.list_ports.comports())
print(port_list)
if len(port_list) == 0:
   print('无可用串口')
else:
    for i in range(0,len(port_list)):
        print(port_list[i])

3.3 完善整个收发流程

根据上面举的一些例子，我们把整个流程补充完整。

示例如下：

# 导入模块
import threading
import time
import serial
import serial.tools.list_ports

# 自定义变量
port = "COM6"  # 端口号，根据自己实际情况输入，可以在设备管理器查看
bps = 9600     # 串口波特率，根据自己实际情况输入
timeout = 5       # 超时时间,None：永远等待操作，0为立即返回请求结果，其他值为等待超时时间(单位为秒）
rxdata = ''    # 接收的数据

# 扫描端口
def check_uart_port():
    port_list = list(serial.tools.list_ports.comports())
    # print(port_list)
    if len(port_list) == 0:
        print('can not fine uart port')
        return False
    else:
        for i in range(0,len(port_list)):
            print(port_list[i])
    return True

# 打开串口
def open_uart(port, bps, timeout):
    try:
        # 打开串口，并返回串口对象
        uart = serial.Serial(port, bps, timeout=timeout)
        return uart
    except Exception as result:
        print("can not open uart")
        print(result)
        return False

# 发送数据
def uart_send_data(uart, txbuf):
    len = uart.write(txbuf.encode('utf-8'))  # 写数据
    return len

# 接收数据
def uart_receive_data(uart):
    if uart.in_waiting:
        rxdata = uart.read(uart.in_waiting).decode("utf-8")   # 以字符串接收
        # rxdata = uart.read().hex()  # 以16进制(hex)接收
        print(rxdata)  # 打印数据

# 关闭串口
def close_uart(uart):
    uart.close()

# 创建一个线程用来等待串口接收数据
class myThread (threading.Thread):   # 继承父类threading.Thread
    def __init__(self, uart):
        threading.Thread.__init__(self)
        self.uart = uart
    def run(self):                   # 把要执行的代码写到run函数里面 线程在创建后会直接运行run函数
        while True:
            # print("thread_uart_receive")
            uart_receive_data(self.uart)  # 接收数据
            # time.sleep(0.01)

# 主函数
def main():
    # 扫描端口
    result = check_uart_port()
    if(result == False):
        return

    # 打开串口
    result = open_uart(port, bps, timeout)
    if (result == False):
        return
    else:
        uart1 = result

    # 创建一个线程用来接收串口数据
    thread_uart = myThread(uart1)
    thread_uart.start()

    while True:
        # 定时发送数据
        txbuf = "hello world"
        len = uart_send_data(uart1, txbuf)
        print("send len: ", len)
        time.sleep(1)

# 启动主函数
main()

运行测试：