編程學習網 > 編程語言 > Python > python中serial串口操作教程
2023
07-27

python中serial串口操作教程

python里面使用serial庫來操作串口,serial的使用流程跟平常的類似,也是打開、關閉、讀、寫。


1.打開串口
一般就是設置端口,波特率。

使用serial.Serial創建實體的時候會去打開串口,之后可以使用is_open開判斷下是否串口是否打開正常。

   def DOpenPort(portx, bps, timeout):
    try:
        # 打開串口,并得到串口對象
        ser = serial.Serial(portx, bps, timeout=timeout)
        # 判斷是否打開成功
        if(False == ser.is_open):
           ser = -1
    except Exception as e:
        print("---異常---:", e)

    return ser

2.關閉串口
使用ser.close即可關閉串口

def DColsePort(ser):
    uart.fdstate = -1
    ser.close()

3.寫數據
數據的寫使用ser.write接口,如果寫的是十六進制的數據使用bytearray來定義,如writebuf = bytearray([0x55, 0xaa, 0x00, 0x01, 0x00, 0x00])

def DWritePort(ser, data):
    result = ser.write(data)  # 寫數據
    logging.info(ser)
    logging.info("Write %s(%d)" % (data.hex(), result))
    return result

4.讀數據
讀數據使用ser.read接口,一般會先使用in_waiting來判斷下是否有數據,然后開始讀

# 十六進制顯示
def hexShow(argv):
    try:
        result = ''
        hLen = len(argv)
        for i in range(hLen):
            hvol = argv[i]
            hhex = '%02x' % hvol
            result += hhex+' '

        logging.info('Led Read:%s', result)
        return result
    except Exception as e:
        print("---異常---:", e)

def DReadPort(ser):
    # 循環接收數據,此為死循環,可用線程實現
    readstr = ""
    if ser.in_waiting:
        readbuf = ser.read(ser.in_waiting)
        if readbuf[0] == 0x55 and readbuf[1] == 0xaa:
            readstr = readbuf
        else:
            readstr = readstr + readbuf

        hexShow(readstr)

5.實例
下面舉一個例子,說明下我們在實際的使用情況。

一般會單獨創建一個進程來作為數據的接收,然后再配合上標記位或者信號量來處理邏輯

import serial  # 導入模塊
import threading
import time
import logging
import sys


class UartInfo(object):
    def __init__(self, fd, count, fail):
        self.fd = fd
        self.count = count  # 測試次數
        self.fail = fail  # 失敗次數

    response = False
    image_addr = 0x00
    image_crc = 0x00
    version = 0
    write_event = threading.Event()


uart = UartInfo(-1, 0, 0)


def logging_init():
    logging.basicConfig(#  filename="test.log", # 指定輸出的文件
        level=logging.DEBUG,
        format='%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s')

    return True


# 十六進制顯示
def hexShow(argv):
    try:
        result = ''
        hLen = len(argv)
        for i in range(hLen):
            hvol = argv[i]
            hhex = '%02x' % hvol
            result += hhex+' '

        logging.info('Led Read:%s', result)
        return result
    except Exception as e:
        print("---異常---:", e)


def crc_sum(data, data_len):
    crc = 0
    i = 0
    while(i < data_len):
        crc += data[i]
        i += 1
    return crc & 0x00FF


def crc_sum_u32(data, data_len):
    crc = 0
    i = 0
    while(i < data_len):
        crc += data[i]
        i += 1
    return crc

# 打開串口
def DOpenPort(portx, bps, timeout):
    try:
        # 打開串口,并得到串口對象
        ser = serial.Serial(portx, bps, timeout=timeout)
        # 判斷是否打開成功
        if(False == ser.is_open):
           ser = -1
    except Exception as e:
        print("---異常---:", e)

    return ser


# 關閉串口
def DColsePort(ser):
    uart.fdstate = -1
    ser.close()


# 寫數據
def DWritePort(ser, data):
    result = ser.write(data)  # 寫數據
    logging.info(ser)
    logging.info("Led Write %s(%d)" % (data.hex(), result))
    return result


# 讀數據
def ReadData_Thread(ser):
    # 循環接收數據,此為死循環,可用線程實現
    readstr = ""
    while(-1 != ser):
        if ser.in_waiting:
            try:  # 如果讀取的不是十六進制數據--
                readbuf = ser.read(ser.in_waiting)
                if readbuf[0] == 0x55 and readbuf[1] == 0xaa:
                    readstr = readbuf
                else:
                    readstr = readstr + readbuf

                hexShow(readstr)

                if (readstr[3] == 0x01) and (len(readstr) > 10):
                    uart.version = readstr[16]
                    uart.response = True
                    uart.write_event.set()
                elif (readstr[3] == 0x21) and (readstr[4] == 0x00) and (readstr[5] == 0x00):
                    uart.response = True
                    uart.write_event.set()
                elif (readstr[3] == 0x22) and (len(readstr) > 10):
                    uart.image_addr = (readstr[6] << 24 & 0xFF000000)
                    uart.image_addr += (readstr[7] << 16 & 0x00FF0000)
                    uart.image_addr += (readstr[8] << 8 & 0x0000FF00)
                    uart.image_addr += (readstr[9] << 0 & 0x000000FF)
                    uart.response = True
                    uart.write_event.set()
                elif (readstr[3] == 0x23) and (len(readstr) > 25):
                    uart.response = True
                    uart.write_event.set()

            except:  # --則將其作為字符串讀取
                readbuf = ser.read(ser.in_waiting)
                hexShow(readbuf)


def GetVersion(ser):
    print("GetVersion")
    writebuf = bytearray([0x55, 0xaa, 0x00, 0x01, 0x00, 0x00])
    # crc
    writebuf.append(crc_sum(writebuf, len(writebuf)))
    DWritePort(ser, writebuf)

    logging.info("take")
    uart.response = False
    uart.write_event.clear()
    uart.write_event.wait(timeout=3)
    uart.write_event.clear()
    logging.info("give")
    if uart.response == False:
        logging.info("fail")
        return False
    else:
        return True
        

# 測試任務
def Test_Thread(ser):
    while (-1 != ser):
        uart.response = False
        uart.image_addr = 0x00
        uart.image_crc = 0x00
        uart.version = 0
        logging.info("count:%d", uart.count)
        logging.info("fail:%d", uart.fail)
        print("count", uart.count)
        print("fail", uart.fail)

        if GetVersion(ser):
            logging.info(uart.version)
            if uart.version == 0x37:
                otafile = "./UartV108.bin"
            else:
                otafile = "./UartV107.bin"
        else:
            uart.fail += 1
            uart.count += 1
            continue

        print("ota:", otafile)
        logging.info("ota:%s", otafile)
        time.sleep(2)

        uart.count += 1
        time.sleep(5)


def TestStop(ser):
    DColsePort(uart.fd)  # 關閉串口


if __name__ == "__main__":
    if 2 != len(sys.argv):
        print("please enter COM")
        exit()
    else:
        uart.tty = sys.argv[1]

    logging_init()
    uart.fd = DOpenPort(uart.tty, 115200, None)

    if(uart.fd != -1):  # 判斷串口是否成功打開
        threading.Thread(target=Test_Thread, args=(uart.fd,)).start()

        threading.Thread(target=ReadData_Thread, args=(uart.fd,)).start()

以上就是python中serial串口操作教程的詳細內容,想要了解更多Python教程歡迎持續關注編程學習網。

掃碼二維碼 獲取免費視頻學習資料

Python編程學習

查 看2022高級編程視頻教程免費獲取