Python Serial 串口基本操作（收發數據）

1、需要模塊以及測試工具

模塊名：pyserial

使用命令下載：python -m pip install pyserial

串口調試工具：sscom5.13.1.exe

2、導入模塊

import serial

3、打開串口

直接通過new一個Serial()的實例即可打開

返回實例

# encoding=utf-8
import serial

if __name__ == '__main__':
    com = serial.Serial('COM3', 115200)
    print com
    

運行結果

Serial<id=0x3518940, open=True>(port='COM3', baudrate=115200, bytesize=8, parity='N', stopbits=1, timeout=None, xonxoff=False, rtscts=False, dsrdtr=False)

4、發送數據

函數名write()

返回值為發送成功的字節數

# encoding=utf-8
import serial

if __name__ == '__main__':
    com = serial.Serial('COM3', 115200)
    success_bytes = com.write('This is data for test')
    print success_bytes

運行結果

21

串口工具界面

5、接收數據（接收固定長度數據）

函數名為read(size=1)

接收size單位的字符，是阻塞的，不接收到就一直等待接收，除非設置了超時時間（未設置該示例）

# encoding=utf-8
import serial

if __name__ == '__main__':
    com = serial.Serial('COM3', 115200)
    data = com.read(10)
    print data

運行結果

123456789a

串口工具界面

6、接收數據（超時時間內一直接收）

函數名為read(size=1)

參數為接收的長度，默認為1，一般傳入inWaiting(),它表示監測接收的字符串長度

配合While可以一直接收

# encoding=utf-8
import serial
import time

if __name__ == '__main__':
    com = serial.Serial('COM3', 115200)
    over_time = 30
    start_time = time.time()
    while True:
        end_time = time.time()
        if end_time - start_time < over_time:
            data = com.read(com.inWaiting())
            data = str(data)
            if data != '':
                print data

運行結果

111
222
aaa
bbb
1a2b3c4d

串口工具界面

7、封裝為類

# -*- encoding=utf-8 -*-
import serial
import time

import WriteLog


class COM:
    def __init__(self, port, baud):
        self.port = port
        self.baud = int(baud)
        self.open_com = None
        self.log = WriteLog.WriteLog('ITC_LOG.LOG')
        self.get_data_flag = True
        self.real_time_data = ''

    # return real time data form com
    def get_real_time_data(self):
        return self.real_time_data

    def clear_real_time_data(self):
        self.real_time_data = ''

    # set flag to receive data or not
    def set_get_data_flag(self, get_data_flag):
        self.get_data_flag = get_data_flag

    def open(self):
        try:
            self.open_com = serial.Serial(self.port, self.baud)
        except Exception as e:
            self.log.error('Open com fail:{}/{}'.format(self.port, self.baud))
            self.log.error('Exception:{}'.format(e))

    def close(self):
        if self.open_com is not None and self.open_com.isOpen:
            self.open_com.close()

    def send_data(self, data):
        if self.open_com is None:
            self.open()
        success_bytes = self.open_com.write(data.encode('UTF-8'))
        return success_bytes

    def get_data(self, over_time=30):
        all_data = ''
        if self.open_com is None:
            self.open()
        start_time = time.time()
        while True:
            end_time = time.time()
            if end_time - start_time < over_time and self.get_data_flag:
                data = self.open_com.read(self.open_com.inWaiting())
                # data = self.open_com.read() # read 1 size
                data = str(data)
                if data != '':
                    self.log.info('Get data is:{}'.format(data))
                    all_data = all_data + data
                    print data
                    self.real_time_data = all_data
            else:
                self.set_get_data_flag(True)
                break
        return all_data


if __name__ == '__main__':
    pass
    com = COM('com3', 115200)
    # com.open()
    print com.send_data('data')
    com.get_data(50)
    com.close()

8、對於一次函數說明（百度看到，並未測試）

readall()：讀取全部字符，是阻塞的，除非接收的字符串以EOF結尾或者超出緩沖區，否則函數不會返回。一般要結合超時設置，設置串口的timeout參數

Readline()：讀取一行，以/n結束，要是沒有/n就一直讀，阻塞。