Python實作序列槽通信(pyserial)
pyserial子產品封裝了對序列槽的通路,相容各種平台。
安裝
pip insatll pyserial
初始化
簡單初始化示例
import serial
ser = 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 created
baudrate=9600, # baud rate
bytesize=EIGHTBITS, # number of databits
parity=PARITY_NONE, # enable parity checking
stopbits=STOPBITS_ONE, # number of stopbits
timeout=None, # set a timeout value, None for waiting forever
xonxoff=0, # enable software flow control
rtscts=0, # enable RTS/CTS flow control
interCharTimeout=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 serial
import serial.tools.list_ports
class 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)#轉為十進制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()#方式二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