首页 > 编程 > Python > 正文

Python实现串口通信(pyserial)过程解析

2019-11-25 11:40:06
字体:
来源:转载
供稿:网友

pyserial模块封装了对串口的访问,兼容各种平台。

安装

pip insatll pyserial

初始化

简单初始化示例

import serialser = serial.Serial('com1', 9600, timeout=1)

所有参数

ser = serial.Serial(port=None,       # number of device, numbering starts at# zero. if everything fails, the user# can specify a device string, note# that this isn't portable anymore# if no port is specified an unconfigured# an closed serial port object is createdbaudrate=9600,     # baud ratebytesize=EIGHTBITS,   # number of databitsparity=PARITY_NONE,   # enable parity checkingstopbits=STOPBITS_ONE, # number of stopbitstimeout=None,      # set a timeout value, None for waiting foreverxonxoff=0,       # enable software flow controlrtscts=0,        # enable RTS/CTS flow controlinterCharTimeout=None  # Inter-character timeout, None to disable)

不同平台下初始化

ser=serial.Serial("/dev/ttyUSB0",9600,timeout=0.5) #使用USB连接串行口ser=serial.Serial("/dev/ttyAMA0",9600,timeout=0.5) #使用树莓派的GPIO口连接串行口ser=serial.Serial(1,9600,timeout=0.5)#winsows系统使用com1口连接串行口ser=serial.Serial("com1",9600,timeout=0.5)#winsows系统使用com1口连接串行口ser=serial.Serial("/dev/ttyS1",9600,timeout=0.5)#Linux系统使用com1口连接串行口

serial.Serial类(另外初始化的方法)

class serial.Serial(){  def __init__(port=None, baudrate=9600, bytesize=EIGHTBITS,parity=PARITY_NONE, stopbits=STOPBITS_ONE, timeout=None, xonxoff=False, rtscts=False, writeTimeout=None, dsrdtr=False, interCharTimeout=None)}

ser对象属性

name:设备名字port:读或者写端口baudrate:波特率bytesize:字节大小parity:校验位stopbits:停止位timeout:读超时设置writeTimeout:写超时xonxoff:软件流控rtscts:硬件流控dsrdtr:硬件流控interCharTimeout:字符间隔超时

ser对象常用方法

ser.isOpen():查看端口是否被打开。ser.open() :打开端口‘。ser.close():关闭端口。ser.read():从端口读字节数据。默认1个字节。ser.read_all():从端口接收全部数据。ser.write("hello"):向端口写数据。ser.readline():读一行数据。ser.readlines():读多行数据。in_waiting():返回接收缓存中的字节数。flush():等待所有数据写出。flushInput():丢弃接收缓存中的所有数据。flushOutput():终止当前写操作,并丢弃发送缓存中的数据。

封装参考

import serialimport serial.tools.list_portsclass Communication():  #初始化  def __init__(self,com,bps,timeout):    self.port = com    self.bps = bps    self.timeout =timeout    global Ret    try:      # 打开串口,并得到串口对象       self.main_engine= serial.Serial(self.port,self.bps,timeout=self.timeout)      # 判断是否打开成功       if (self.main_engine.is_open):        Ret = True    except Exception as e:      print("---异常---:", e)  # 打印设备基本信息  def Print_Name(self):    print(self.main_engine.name) #设备名字    print(self.main_engine.port)#读或者写端口    print(self.main_engine.baudrate)#波特率    print(self.main_engine.bytesize)#字节大小    print(self.main_engine.parity)#校验位    print(self.main_engine.stopbits)#停止位    print(self.main_engine.timeout)#读超时设置    print(self.main_engine.writeTimeout)#写超时    print(self.main_engine.xonxoff)#软件流控    print(self.main_engine.rtscts)#软件流控    print(self.main_engine.dsrdtr)#硬件流控    print(self.main_engine.interCharTimeout)#字符间隔超时  #打开串口  def Open_Engine(self):    self.main_engine.open()  #关闭串口  def Close_Engine(self):    self.main_engine.close()    print(self.main_engine.is_open) # 检验串口是否打开  # 打印可用串口列表  @staticmethod  def Print_Used_Com():    port_list = list(serial.tools.list_ports.comports())    print(port_list)  #接收指定大小的数据  #从串口读size个字节。如果指定超时,则可能在超时后返回较少的字节;如果没有指定超时,则会一直等到收完指定的字节数。  def Read_Size(self,size):    return self.main_engine.read(size=size)  #接收一行数据  # 使用readline()时应该注意:打开串口时应该指定超时,否则如果串口没有收到新行,则会一直等待。  # 如果没有超时,readline会报异常。  def Read_Line(self):    return self.main_engine.readline()  #发数据  def Send_data(self,data):    self.main_engine.write(data)  #更多示例  # self.main_engine.write(chr(0x06).encode("utf-8")) # 十六制发送一个数据  # print(self.main_engine.read().hex()) # # 十六进制的读取读一个字节  # print(self.main_engine.read())#读一个字节  # print(self.main_engine.read(10).decode("gbk"))#读十个字节  # print(self.main_engine.readline().decode("gbk"))#读一行  # print(self.main_engine.readlines())#读取多行,返回列表,必须匹配超时(timeout)使用  # print(self.main_engine.in_waiting)#获取输入缓冲区的剩余字节数  # print(self.main_engine.out_waiting)#获取输出缓冲区的字节数  # print(self.main_engine.readall())#读取全部字符。  #接收数据  #一个整型数据占两个字节  #一个字符占一个字节  def Recive_data(self,way):    # 循环接收数据,此为死循环,可用线程实现    print("开始接收数据:")    while True:      try:        # 一个字节一个字节的接收        if self.main_engine.in_waiting:          if(way == 0):            for i in range(self.main_engine.in_waiting):              print("接收ascii数据:"+str(self.Read_Size(1)))              data1 = self.Read_Size(1).hex()#转为十六进制              data2 = int(data1,16)#转为十进制              if (data2 == "exit"): # 退出标志                break              else:                 print("收到数据十六进制:"+data1+" 收到数据十进制:"+str(data2))          if(way == 1):            #整体接收            # data = self.main_engine.read(self.main_engine.in_waiting).decode("utf-8")#方式一            data = self.main_engine.read_all()#方式二            if (data == "exit"): # 退出标志              break            else:               print("接收ascii数据:", data)      except Exception as e:        print("异常报错:",e)Communication.Print_Used_Com()Ret =False #是否创建成功标志Engine1 = Communication("com12",115200,0.5)if (Ret):  Engine1.Recive_data(0)
while(1)  {   //发送测试   uint8_t a = 61;   delayms(300);   printf("%c", a);}
开始接收数据:接收ascii数据:b'='收到数据十六进制:3d 收到数据十进制:61

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持武林网。

发表评论 共有条评论
用户名: 密码:
验证码: 匿名发表