前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >Python串口通信案例

Python串口通信案例

作者头像
用户8949263
发布2022-11-07 12:57:45
3.9K0
发布2022-11-07 12:57:45
举报
文章被收录于专栏:Python数据分析实例

本次分享利用Python模拟串口通信案例。当前编程环境,并没有办法接好下位机平台,需要模拟出一个串口,不断发送、接收信息的过程。

串行通信端口(cluster communication port ),简称串口,即COM口。

pySerial 是 Python 中用于操作串口的第三方模块,它支持 Windows、Linux、OSX、BSD等多个平台。

1.win 下载虚拟串口

https://sourceforge.net/projects/com0com/files/latest/download 下载安装完成后,会在设备管理器中新增两个com。

这两个串口是互相接通的,一个com 可以当作上位机,一个com可以当作下位机。创建的一对虚拟串口。

或者利用虚拟串口的软件创建,VSPD(Virtual Serial Port Driver)

2、确定串口号

pyserial模块封装了python对串口的访问,为多平台的使用提供了统一的接口。

代码语言:javascript
复制
import serial
import serial.tools.list_ports
 
# 获取所有串口设备实例。

port_list = list(serial.tools.list_ports.comports())
def show_all_com():
    # 如果找到串口设备,则依次输出每个设备对应的串口号和描述信息。
    port_list_name = []
    if len(port_list) <= 0:
        print("the serial port can't find!")
    else:
        for itms in port_list:
            print(list(itms)[0], list(itms)[1])
            port_list_name.append(itms.device)

    return port_list_name
代码语言:javascript
复制
运行结果

COM1 通信端口 (COM1)
COM3 com0com - serial port emulator (COM3)
COM4 com0com - serial port emulator (COM4)
['COM1', 'COM3', 'COM4']

3、配置串口

代码语言:javascript
复制
import serial
 
# 打开 COM3,将波特率配置为115200,数据位为7,停止位为2,无校验位,读超时时间为0.5秒。
ser = serial.Serial(port="COM3",
                    baudrate=115200,
                    bytesize=serial.SEVENBITS,
                    parity=serial.PARITY_NONE,
                    stopbits=serial.STOPBITS_TWO,
                    timeout=0.5) 
代码语言:javascript
复制
常用的参数如下:

port - 串口设备名或 None。
baudrate - 波特率,可以是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。
bytesize - 数据位,可取值为:FIVEBITS, SIXBITS, SEVENBITS, EIGHTBITS。
parity - 校验位,可取值为:PARITY_NONE, PARITY_EVEN, PARITY_ODD, PARITY_MARK, PARITY_SPACE。
stopbits - 停止位,可取值为:STOPBITS_ONE, STOPBITS_ONE_POINT_FIVE, STOPBITS_TOW。
xonxoff - 软件流控,可取值为 True, False。
rtscts - 硬件(RTS/CTS)流控,可取值为 True, False。
dsr/dtr - 硬件(DSR/DTR)流控,可取值为 True, False。
timeout - 读超时时间,可取值为 None, 0 或者其他具体数值(支持小数)。当设置为 None 时,表示阻塞式读取,一直读到期望的所有数据才返回;当设置为 0 时,表示非阻塞式读取,无论读取到多少数据都立即返回;当设置为其他数值时,表示设置具体的超时时间(以秒为单位),如果在该时间内没有读取到所有数据,则直接返回。
write_timeout: 写超时时间,可取值为 None, 0 或者其他具体数值(支持小数)。参数值起到的效果参考 timeout 参数。

4、串口通信示例

python 实现向一个串口内写数据(发送端),另一个串口内接受数据(接收端)

  • 串口类工具

串口类功能函数组成:打开、关闭、发送数据、接受数据、解析保存数据功能

代码语言:javascript
复制
import serial
from time import sleep
import queue
import time
from datetime import datetime

q = queue.Queue()

class SerialPort:
    def __init__(self, port, buand):
        self.port = serial.Serial(port, buand)
        self.port.close()
        if not self.port.isOpen():
            self.port.open()

    def port_open(self):
        if not self.port.isOpen():
            self.port.open()

    def port_close(self):
        self.port.close()

    #发送数据
    def send_data(self):
        """
        #write() 方法只能发送 bytes 类型的数据,所以需要对字符串进行 encode 编码。
        # write() 方法执行完成后,会将发送的字节数作为返回值。
        """
        while True:   # 循环发送数据
            date = datetime.now().strftime('%H:%M:%S.%f')[:-3]
            print("send:", date)
            self.port.write(date.encode())
            time.sleep()

    # 读取数据
    def read_data(self):
        """
        #  read()方法默认一次读取一个字节,可以通过传入参数指定每次读取的字节数;
        read() 方法会将读取的内容作为返回值,类型为 bytes。
        """
        while True:
            count = self.port.inWaiting()
            if count != :
                recv = self.port.read(count)
                self.port.flushInput()
                q.put(recv.decode())
                print("receive:",recv.decode())
            sleep(0.5)

    #解析保存数据
    def save_json(self):
        while True:
            if q.empty():
                pass
            else:
                print("队列长度",q.qsize())
                data = q.get()
                print(data)
            sleep(0.1)
  • 发送端

开启一个线程定时发送数据

代码语言:javascript
复制
from tools.LoRaSerial import SerialPort
import threading
from time import sleep

if __name__=='__main__':
    baunRate = 
    serialPort_w = "COM3"
    mSerial_w = SerialPort(serialPort_w, baunRate)
    t1 = threading.Thread(target=mSerial_w.send_data)
    t1.daemon = True
    t1.start()

    while True:
        sleep()
  • 接收端

接收端开启两个线程接收数据:读取数据和解析存储数据。

代码语言:javascript
复制
from tools.LoRaSerial import SerialPort
import threading
from time import sleep

if __name__=='__main__':
    threads=[]
    baunRate = 

    serialPort_r = "COM4"
    mSerial_r = SerialPort(serialPort_r, baunRate)

    t2 = threading.Thread(target=mSerial_r.read_data)
    threads.append(t2)

    t3 = threading.Thread(target=mSerial_r.save_json)
    threads.append(t3)

    for t in threads:
        t.start()

    while True:
        sleep()

运行结果

从一个线程向另一个线程发送数据最安全的方式可能就是使用 queue 库中的队列了。创建一个被多个线程共享的 Queue 对象,这些线程通过使用put() 和 get() 操作来向队列中添加或者删除元素。

当终端发送数据量较大时,其数据解析也是比较耗时,可以用消息队列缓存接收到的发送端数据,当接收线程进行IO监听时,负责解析线程去消息队列取消息并解析。如下图,将解析过程时间延迟由0.1至2s。

为了方便模拟传递数据变化效果,这里用时间函数代替的,实际上设备传递具体消息可解析为具体的json数据格式

消息类别

Ori_ID发送端设备ID

des_ID接收端设备ID

CMD消息类别

VALUE数组

消息方向

上线消息

1

123

Online

[]

终端—网关

心跳消息

1

123

Heart

[{“battery”:”60”}]

终端—网关

环境信息

1

123

Env

[{“temp”:”35”},{“hum”:”66”}]

终端—网关

控制指令

1

123

Alarm

[{“light”:”on”},{“pump”:”Stop”}]

网关—终端

报警消息

123

1

Control

[]

终端—网关

备注:

1、 上线消息-------终端启动时给网关发送上线的消息,网关返回该终端的各种数据及运行状态

2、 心跳消息--------终端定期发送心跳消息给网关

3、 环境信息--------终端定期上报各种环境参数

4、 控制指令-------网关发送指令给终端,实现对终端的控制,如开灯、关灯等

5、 报警消息-------设备监测到报警信息,发送报警信息给网关

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2022-08-30,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 Python数据分析实例 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 1.win 下载虚拟串口
  • python 实现向一个串口内写数据(发送端),另一个串口内接受数据(接收端)
  • 运行结果
相关产品与服务
消息队列 CMQ 版
消息队列 CMQ 版(TDMQ for CMQ,简称 TDMQ CMQ 版)是一款分布式高可用的消息队列服务,它能够提供可靠的,基于消息的异步通信机制,能够将分布式部署的不同应用(或同一应用的不同组件)中的信息传递,存储在可靠有效的 CMQ 队列中,防止消息丢失。TDMQ CMQ 版支持多进程同时读写,收发互不干扰,无需各应用或组件始终处于运行状态。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档