#!/usr/bin/env python # -*- encoding: utf-8 -*- ''' @文件 :canConnect.py @时间 :2021/12/09 13:47:58 @作者 :None @版本 :1.0 @说明 :CAN连接驱动 ''' from ctypes.util import find_library from sys import platform import ctypes from ctypes import c_ubyte, c_uint from utils.log_signal import LogSignal from utils.resource import resource_path, resource_path_extend PYINSTALLER = 1 if PYINSTALLER: if platform == "win64": _CanDLLName = resource_path("config\\ControlCANx64\\ControlCAN.dll") ZLGCAN = ctypes.windll.LoadLibrary(_CanDLLName) elif platform == "win32": _CanDLLName = resource_path("config\\ControlCANx86\\ControlCAN.dll") ZLGCAN = ctypes.windll.LoadLibrary(_CanDLLName) elif platform == "linux": _CanDLLName = resource_path("config/linux/libusbcan.so") ZLGCAN = ctypes.cdll.LoadLibrary(_CanDLLName) else: ZLGCAN = ctypes.cdll.LoadLibrary(find_library("libPCBUSB.dylib")) else: if platform == "win64": resource_path_extend("config") ZLGCAN = ctypes.windll.LoadLibrary("ControlCAN.dll") elif platform == "win32": resource_path_extend("config") ZLGCAN = ctypes.windll.LoadLibrary("ControlCAN.dll") else: ZLGCAN = "" ubyte_array = c_ubyte * 8 ubyte_3array = c_ubyte * 3 # can type CANTYPE = { 'USBCAN-I': 3, 'USBCAN-II': 4, } # can mode NORMAL_MODE = 0 LISTEN_MODE = 1 # filter type SINGLE_FILTER = 0 DOUBLE_FILTER = 1 # status STATUS_OK = 1 # sendtype SEND_NORMAL = 0 SEND_SINGLE = 1 SELF_SEND_RECV = 2 SELF_SEND_RECV_SINGLE = 3 class VCI_INIT_CONFIG(ctypes.Structure): _fields_ = [("AccCode", c_uint), # 验收码。SJA1000的帧过滤验收码。对经过屏蔽码过滤为“有关位”进行匹配，全部匹配成功后，此帧可以被接收。 # 屏蔽码。SJA1000的帧过滤屏蔽码。对接收的CAN帧ID进行过滤，对应位为0的是“有关位”，对应位为1的是“无关位”。屏蔽码推荐设置为0xFFFFFFFF，即全部接收。 ("AccMask", c_uint), # 保留 ("Reserved", c_uint), # 滤波方式 ("Filter", c_ubyte), # 波特率定时器 0 ("Timing0", c_ubyte), # 波特率定时器 1 ("Timing1", c_ubyte), # 模式。=0表示正常模式（相当于正常节点），=1表示只听模式（只接收，不影响总线），=2表示自发自收模式（环回模式）。 ("Mode", c_ubyte) ] # VCI_CAN_OBJ结构体是CAN帧结构体，即1个结构体表示一个帧的数据结构。在发送函数VCI_Transmit和接收函数VCI_Receive中，被用来传送CAN信息帧。 class VCI_CAN_OBJ(ctypes.Structure): _fields_ = [("ID", c_uint), # 帧ID。32位变量，数据格式为靠右对齐 # 设备接收到某一帧的时间标识。时间标示从CAN卡上电开始计时，计时单位为0.1ms。 ("TimeStamp", c_uint), # 是否使用时间标识，为1时TimeStamp有效，TimeFlag和TimeStamp只在此帧为接收帧时有意义。 ("TimeFlag", c_ubyte), # 发送帧类型。=0时为正常发送（发送失败会自动重发，重发时间为4秒，4秒内没有发出则取消）；=1时为单次发送（只发送一次，发送失败不会自动重发，总线只产生一帧数据）；其它值无效。 ("SendType", c_ubyte), # 是否是远程帧。=0时为为数据帧，=1时为远程帧（数据段空）。 ("RemoteFlag", c_ubyte), # 是否是扩展帧。=0时为标准帧（11位ID），=1时为扩展帧（29位ID）。 ("ExternFlag", c_ubyte), # 数据长度 DLC (<=8)，即CAN帧Data有几个字节。约束了后面Data[8]中的有效字节 ("DataLen", c_ubyte), ("Data", c_ubyte * 8), # CAN帧的数据。由于CAN规定了最大是8个字节，所以这里预留了8个字节的空间，受DataLen约束。如DataLen定义为3，即Data[0]、Data[1]、Data[2]是有效的 ("Reserved", c_ubyte * 3) # 系统保留 ] class PVCI_ERR_INFO(ctypes.Structure): _fields_ = [("ErrorCode", c_uint), ("PassiveErrData", c_ubyte * 3), ("ArLostErrData", c_ubyte) ] baudRateConfig = { '5Kbps': {'time0': 0xBF, 'time1': 0xFF}, '10Kbps': {'time0': 0x31, 'time1': 0x1C}, '20Kbps': {'time0': 0x18, 'time1': 0x1C}, '40Kbps': {'time0': 0x87, 'time1': 0xFF}, '50Kbps': {'time0': 0x09, 'time1': 0x1C}, '80Kbps': {'time0': 0x83, 'time1': 0xFF}, '100Kbps': {'time0': 0x04, 'time1': 0x1C}, '125Kbps': {'time0': 0x03, 'time1': 0x1C}, '200Kbps': {'time0': 0x81, 'time1': 0xFA}, '250Kbps': {'time0': 0x01, 'time1': 0x1C}, '400Kbps': {'time0': 0x80, 'time1': 0xFA}, '500Kbps': {'time0': 0x00, 'time1': 0x1C}, '666Kbps': {'time0': 0x80, 'time1': 0xB6}, '800Kbps': {'time0': 0x00, 'time1': 0x16}, '1000Kbps': {'time0': 0x00, 'time1': 0x14}, } class MessageDeal: def __init__(self): self.canType = CANTYPE['USBCAN-II'] def set_can_board(self, canIndex, canChannel, canBaudrate): self.canIndex = canIndex self.canChannel = canChannel self.canBaudrate = canBaudrate def open_device(self): ret = ZLGCAN.VCI_OpenDevice(self.canType, self.canChannel, self.canChannel) if ret != STATUS_OK: # LogSignal.print_log_signal().log_emit('打开CAN卡失败: {}'.format(str(self.canChannel)) if SD.SYSTEM_LANGUAGE == 0 else 'CAN Device Error: {}'.format(str(self.canChannel))) return False return True def init_can(self, accCode, accMask): # 初始化通道 _vci_initconfig = VCI_INIT_CONFIG(accCode, accMask, 0, DOUBLE_FILTER, baudRateConfig[self.canBaudrate]['time0'], baudRateConfig[self.canBaudrate]['time1'], NORMAL_MODE) ret = ZLGCAN.VCI_InitCAN(self.canType, self.canIndex, self.canChannel, ctypes.byref(_vci_initconfig)) if ret != STATUS_OK: # LogSignal.print_log_signal().log_emit('初始化CAN卡失败: {}'.format(str(self.canChannel)) if SD.SYSTEM_LANGUAGE == 0 else 'CAN Open Error: {}'.format(str(self.canChannel))) return False ret = ZLGCAN.VCI_StartCAN(self.canType, self.canIndex, self.canChannel) if ret != STATUS_OK: # LogSignal.print_log_signal().log_emit('启动CAN卡失败: {}'.format(str(self.canChannel))if SD.SYSTEM_LANGUAGE == 0 else 'CAN Start Error: {}'.format(str(self.canChannel))) return False return True def get_undeal_number(self): return ZLGCAN.VCI_GetReceiveNum(self.canType, self.canIndex, self.canChannel) def receive(self, number=1): objs = (VCI_CAN_OBJ * number)() ret = ZLGCAN.VCI_Receive(self.canType, self.canIndex, self.canChannel, ctypes.byref(objs), number, 10) if ret == 0xFFFFFFFF: return None else: return objs[:ret] def send(self, ID, data, remote_flag=False, extern_flag=False, data_len=8): vci_can_obj = VCI_CAN_OBJ() vci_can_obj.ID = ID vci_can_obj.SendType = SEND_NORMAL if remote_flag: vci_can_obj.RemoteFlag = 1 else: vci_can_obj.RemoteFlag = 0 if extern_flag: vci_can_obj.ExternFlag = 1 else: vci_can_obj.ExternFlag = 0 vci_can_obj.DataLen = data_len if len(data) < 8: data += (8 - len(data)) * [0] vci_can_obj.Data = (c_ubyte * 8)(data[0], data[1], data[2], data[3], data[4], data[5], data[6], data[7]) ret = ZLGCAN.VCI_Transmit(self.canType, self.canIndex, self.canChannel, ctypes.byref(vci_can_obj), 1) if ret != STATUS_OK: LogSignal.print_log_signal().log_emit("CAN Send Error！") return False else: return True def read_err_info(self): errInfo = PVCI_ERR_INFO(0, ubyte_3array(0, 0, 0), 0) ZLGCAN.VCI_ReadErrInfo(self.canType, self.canIndex, self.canChannel, ctypes.byref(errInfo)) LogSignal.print_log_signal().log_emit(errInfo.ErrorCode, errInfo.PassiveErrData[0], errInfo.PassiveErrData[1], errInfo.PassiveErrData[2], errInfo.ArLostErrData) def clear_buffer(self): return ZLGCAN.VCI_ClearBuffer(self.canType, self.canIndex, self.canChannel) def close_can(self): return ZLGCAN.VCI_CloseDevice(self.canType, self.canIndex)