1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
|
from ctypes import *
import logging
import platform
from can import BusABC, Message
logger = logging.getLogger(__name__)
class VCI_INIT_CONFIG(Structure):
_fields_ = [("AccCode", c_int32),
("AccMask", c_int32),
("Reserved", c_int32),
("Filter", c_ubyte),
("Timing0", c_ubyte),
("Timing1", c_ubyte),
("Mode", c_ubyte)]
class VCI_CAN_OBJ(Structure):
_fields_ = [("ID", c_uint),
("TimeStamp", c_int),
("TimeFlag", c_byte),
("SendType", c_byte),
("RemoteFlag", c_byte),
("ExternFlag", c_byte),
("DataLen", c_byte),
("Data", c_ubyte * 8),
("Reserved", c_byte * 3)]
VCI_USBCAN2 = 4
STATUS_OK = 0x01
STATUS_ERR = 0x00
TIMING_DICT = {
5000: (0xBF, 0xFF),
10000: (0x31, 0x1C),
20000: (0x18, 0x1C),
33330: (0x09, 0x6F),
40000: (0x87, 0xFF),
50000: (0x09, 0x1C),
66660: (0x04, 0x6F),
80000: (0x83, 0xFF),
83330: (0x03, 0x6F),
100000: (0x04, 0x1C),
125000: (0x03, 0x1C),
200000: (0x81, 0xFA),
250000: (0x01, 0x1C),
400000: (0x80, 0xFA),
500000: (0x00, 0x1C),
666000: (0x80, 0xB6),
800000: (0x00, 0x16),
1000000: (0x00, 0x14),
}
try:
if platform.system() == "Windows":
CANalystII = WinDLL("./ControlCAN.dll")
else:
CANalystII = CDLL("./libcontrolcan.so")
logger.info("Loaded CANalystII library")
except OSError as e:
CANalystII = None
logger.info("Cannot load CANalystII library")
class CANalystIIBus(BusABC):
def __init__(self, channel, device=0, baud=None, Timing0=None, Timing1=None, can_filters=None):
"""
:param channel: channel number
:param device: device number
:param baud: baud rate
:param Timing0: customize the timing register if baudrate is not specified
:param Timing1:
:param can_filters: filters for packet
"""
super(CANalystIIBus, self).__init__(channel, can_filters)
if isinstance(channel, (list, tuple)):
self.channels = channel
elif isinstance(channel, int):
self.channels = [channel]
else:
# Assume comma separated string of channels
self.channels = [int(ch.strip()) for ch in channel.split(',')]
self.device = device
self.channel_info = "CANalyst-II: device {}, channels {}".format(self.device, self.channels)
if baud is not None:
try:
Timing0, Timing1 = TIMING_DICT[baud]
except KeyError:
raise ValueError("Baudrate is not supported")
if Timing0 is None or Timing1 is None:
raise ValueError("Timing registers are not set")
self.init_config = VCI_INIT_CONFIG(0, 0xFFFFFFFF, 0, 1, Timing0, Timing1, 0)
if CANalystII.VCI_OpenDevice(VCI_USBCAN2, self.device, 0) == STATUS_ERR:
logger.error("VCI_OpenDevice Error")
for channel in self.channels:
if CANalystII.VCI_InitCAN(VCI_USBCAN2, self.device, channel, byref(self.init_config)) == STATUS_ERR:
logger.error("VCI_InitCAN Error")
self.shutdown()
return
if CANalystII.VCI_StartCAN(VCI_USBCAN2, self.device, channel) == STATUS_ERR:
logger.error("VCI_StartCAN Error")
self.shutdown()
return
def send(self, msg, timeout=None):
"""
:param msg: message to send
:param timeout: timeout is not used here
:return:
"""
extern_flag = 1 if msg.is_extended_id else 0
raw_message = VCI_CAN_OBJ(msg.arbitration_id, 0, 0, 1, msg.is_remote_frame, extern_flag, msg.dlc, (c_ubyte * 8)(*msg.data), (c_byte * 3)(*[0, 0, 0]))
if msg.channel is not None:
channel = msg.channel
elif len(self.channels) == 1:
channel = self.channels[0]
else:
raise ValueError(
"msg.channel must be set when using multiple channels.")
CANalystII.VCI_Transmit(VCI_USBCAN2, self.device, channel, byref(raw_message), 1)
def _recv_internal(self, timeout=None):
"""
:param timeout: float in seconds
:return:
"""
raw_message = VCI_CAN_OBJ()
timeout = -1 if timeout is None else int(timeout * 1000)
if CANalystII.VCI_Receive(VCI_USBCAN2, self.device, self.channels[0], byref(raw_message), 1, timeout) <= STATUS_ERR:
return None, False
else:
return (
Message(
timestamp=raw_message.TimeStamp if raw_message.TimeFlag else 0.0,
arbitration_id=raw_message.ID,
is_remote_frame=raw_message.RemoteFlag,
channel=0,
dlc=raw_message.DataLen,
data=raw_message.Data,
),
False,
)
def flush_tx_buffer(self):
for channel in self.channels:
CANalystII.VCI_ClearBuffer(VCI_USBCAN2, self.device, channel)
def shutdown(self):
CANalystII.VCI_CloseDevice(VCI_USBCAN2, self.device)
|
