Python 调用串口通信DLL

Posted 2021-08-27 一身转战三千里

#!D:/ide/Python/Python37/
# -*- coding: utf-8 -*-
"""     加载系统相关包    """
import sys
import os
import ctypes
from ctypes import *
from binascii import unhexlify, b2a_hex
from crcmod import mkCrcFun
dll = cdll.LoadLibrary("./UartDll/UartDll.dll")

"""     QT 相关包    """
from PyQt5.QtWidgets import QApplication
from PyQt5.QtCore import QThread, QMutex, QObject, pyqtSignal, QMutexLocker

"""
# @文  件: UartCommuClass.py
# @作  者: 徐生海
# @日  期: 2021/8/26 0026
# @说  明:
"""


class UartSerial(QObject):
    # 打开串口信号
    sign_OpenSerial = pyqtSignal(object, object, object, object, object)
    # 关闭串口信号
    sign_CloseSerial = pyqtSignal()
    # 串口状态信号：用于对外发送
    sign_SerialIsOpen = pyqtSignal(object)
    # 发送串口对象给设备对象
    sign_SendSerialObject = pyqtSignal(QObject)
    # 发送和接收延时
    ReadWriteDelay = int(180)
    # 读取数据延时
    ReadDataDelay = int(5)
    # 当前串口句柄
    CurrentHandle = int(-1)
    # 创建句柄函数
    InitUart = dll.Create
    # 打开串口函数
    OpenUart = dll.Open
    # 关闭串口函数
    CloseUart = dll.Close
    # 发送数据函数
    SendByte = dll.Send
    # 接收数据函数
    ReceiveData = dll.Reveice
    # 发送数据后延时接收函数
    SerSend = dll.SerSend
    # 注消句柄函数
    UnInitUart = dll.UnInit
    # 串口读写锁
    UartMutex: QMutex

    # 串口打开状态
    isOpen = False
    def __init__(self, parent=None):
        super(UartSerial, self).__init__(parent)
        self.isOpen = False
        self.CurrentHandle = -1
        self.UartMutex = QMutex()
        self.InitHandle()
        self.setReadWriteDelay()

    def __del__(self):
        self.UnInitHandle()

    def getHandle(self) -> int:
        return self.CurrentHandle
    # 初始化串口句柄
    def InitHandle(self) -> int:
        self.CurrentHandle = -1
        recH = c_int(-1)
        self.InitUart.argtype = [POINTER(c_int)]
        self.InitUart.restype = c_int
        recD = self.InitUart(byref(recH))
        if recD is 1:
            self.CurrentHandle = recH.value
        return self.CurrentHandle
    # 注消句柄
    def UnInitHandle(self) -> bool:
        self.CloseUart.argtype = [c_int]
        self.CloseUart.restype = c_int
        recD = self.CloseUart(c_int(self.CurrentHandle))
        if recD == 1:
            print("注消句柄成功")
            self.isOpen = False
            self.CurrentHandle = -1
            return True
        else:
            print("注消句柄失败")
            return False
    # 打开串口
    def OpenHandle(self, portName: str, baudRate: int, dataBits=int(8), flowControl=int(0), parity=int(0), stopBits=int(1)) -> bool:
        info = portName + ","
        info += str(baudRate) + ","
        info += str(dataBits) + ","
        info += str(flowControl) + ","
        info += str(parity) + ","
        info += str(stopBits) + ","
        if self.CurrentHandle < 0:
            self.InitHandle()
        byteInfo = info.encode("utf-8")

        self.OpenUart.argtype = [c_int, c_void_p]
        self.OpenUart.restype = c_int
        recD = self.OpenUart(c_int(self.CurrentHandle), byteInfo)
        if recD == 1:
            self.isOpen = True
        else:
            self.isOpen = False
        self.sign_SerialIsOpen.emit(self.isOpen)
        self.sign_SendSerialObject.emit(self)
        return self.isOpen

    # 关闭串口
    def CloseHandle(self) -> bool:
        self.UnInitHandle()
        self.sign_SerialIsOpen.emit(self.isOpen)
        return self.isOpen

    # 发送数据
    def SendHandle(self, Data: str) -> bool:
        Locker = QMutexLocker(self.UartMutex)
        SendBuffer = self.crc16Add(Data)
        SendByteData = bytes.fromhex(SendBuffer)
        if self.CurrentHandle < 0:
            return False
        if not self.isOpen:
            return False
        self.SendByte.argtype = [c_int, c_void_p, c_int]
        self.SendByte.restype = c_int
        recD = self.SendByte(c_int(self.CurrentHandle), SendByteData, c_int(len(SendByteData)))
        if recD != 1:
            return False
        return True
        pass
    # 发送数据延时接收
    def serSend(self, Data: str) -> str:
        Locker = QMutexLocker(self.UartMutex)
        SendBuffer = self.crc16Add(Data)
        SendByteData = bytes.fromhex(SendBuffer)
        if self.CurrentHandle < 0:
            return False
        if not self.isOpen:
            return False
        self.SerSend.argtype = []
        self.SerSend.restype = c_int
        readPtr = create_string_buffer(128)
        readSize = c_int(0)
        self.SerSend.argtype = [c_int, c_void_p, c_int, c_char_p, POINTER(c_int)]
        self.SerSend.restype = c_int
        recD = self.SerSend(c_int(self.CurrentHandle), SendByteData, c_int(len(SendByteData)),
                            pointer(readPtr), byref(readSize))
        if recD != 1:
            return str("")
        return str(readPtr.value, \'utf-8\')

        pass
    # 接收数据
    def Receive(self) -> str:
        Locker = QMutexLocker(self.UartMutex)
        readPtr = create_string_buffer(1024)
        readSize = c_int(0)
        resultStr = str(\'\')
        if self.CurrentHandle < 0:
            return resultStr
        if not self.isOpen:
            return resultStr
        self.ReceiveData.argtype = [c_int, c_void_p, POINTER(c_int)]
        self.ReceiveData.restype = c_int
        recD = self.ReceiveData(c_int(self.CurrentHandle), pointer(readPtr), byref(readSize))
        if recD != 1:
            return resultStr
        resultStr = str(readPtr.value, \'utf-8\')
        return resultStr
        pass

    # 设置延时时长
    def setReadWriteDelay(self, WriteDelay=180, ReadDelay=5) -> bool:
        self.ReadWriteDelay = WriteDelay
        self.ReadDataDelay = ReadDelay
        if self.CurrentHandle < 0:
            return False
        dll.SetDelay.argtype = [c_int, c_int, c_int]
        dll.SetDelay.restype = c_int
        recD = dll.SetDelay(c_int(self.CurrentHandle), c_int(self.ReadWriteDelay), c_int(self.ReadDataDelay))
        if recD != 1:
            return False
        return True
    # CRC16-ModBus
    @staticmethod
    def crc16Add(read) -> str:
        crc16 = mkCrcFun(0x18005, rev=True, initCrc=0xFFFF, xorOut=0x0000)
        data = read.replace(" ", "")
        readout = hex(crc16(unhexlify(data))).upper()
        str_list = list(readout)
        if len(str_list) < 6:
            str_list.insert(2, \'0\' * (6 - len(str_list)))  # 位数不足补0
        crc_data = "".join(str_list)
        read = read.strip() + crc_data[4:] + crc_data[2:4]
        return read

if __name__ == \'__main__\':
    app = QApplication(sys.argv)
    Uart = UartSerial()
    print("初始化后的句柄:", Uart.CurrentHandle)

    result = Uart.OpenHandle("COM1", 115200, 8, 0, 0, 1)
    if result:
        print("打开串口成功")

    for i in range(0, 200):
        result = Uart.SendHandle(str("010300030001"))
        if result:
            recData = Uart.Receive()
            print("发送数据状态:", str("010300030001"), "收到数据:", recData)