Skip to content

Commit c4752b2

Browse files
christiansandbergchristiansandberg
authored
Improved channel handling (hardbyte#332)
Make Notifier support multiple buses Add support for channels in more formats, interfaces, and loggers
1 parent b2c03b2 commit c4752b2

12 files changed

Lines changed: 174 additions & 52 deletions

File tree

‎can/interfaces/nican.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -154,6 +154,7 @@ def __init__(self, channel, can_filters=None, bitrate=None, log_errors=True, **k
154154
raise ImportError("The NI-CAN driver could not be loaded. "
155155
"Check that you are using 32-bit Python on Windows.")
156156

157+
self.channel = channel
157158
self.channel_info = "NI-CAN: " + channel
158159
if not isinstance(channel, bytes):
159160
channel = channel.encode()
@@ -242,6 +243,7 @@ def _recv_internal(self, timeout):
242243
arb_id &= 0x1FFFFFFF
243244
dlc = raw_msg.dlc
244245
msg = Message(timestamp=timestamp,
246+
channel=self.channel,
245247
is_remote_frame=is_remote_frame,
246248
is_error_frame=is_error_frame,
247249
extended_id=is_extended,

‎can/interfaces/vector/canlib.py‎

Lines changed: 17 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -90,17 +90,22 @@ def __init__(self, channel, can_filters=None, poll_interval=0.01,
9090
self.mask = 0
9191
self.fd = fd
9292
# Get channels masks
93+
self.channel_masks = {}
94+
self.index_to_channel = {}
9395
for channel in self.channels:
9496
hw_type = ctypes.c_uint(0)
9597
hw_index = ctypes.c_uint(0)
9698
hw_channel = ctypes.c_uint(0)
9799
vxlapi.xlGetApplConfig(self._app_name, channel, hw_type, hw_index,
98100
hw_channel, vxlapi.XL_BUS_TYPE_CAN)
99101
LOG.debug('Channel index %d found', channel)
100-
mask = vxlapi.xlGetChannelMask(hw_type.value, hw_index.value,
102+
idx = vxlapi.xlGetChannelIndex(hw_type.value, hw_index.value,
101103
hw_channel.value)
102-
LOG.debug('Channel %d, Type: %d, Mask: %d',
104+
mask = 1 << idx
105+
LOG.debug('Channel %d, Type: %d, Mask: 0x%X',
103106
hw_channel.value, hw_type.value, mask)
107+
self.channel_masks[channel] = mask
108+
self.index_to_channel[idx] = channel
104109
self.mask |= mask
105110

106111
permission_mask = vxlapi.XLaccess()
@@ -225,6 +230,7 @@ def _recv_internal(self, timeout):
225230
dlc = dlc2len(event.tagData.canRxOkMsg.dlc)
226231
flags = event.tagData.canRxOkMsg.msgFlags
227232
timestamp = event.timeStamp * 1e-9
233+
channel = self.index_to_channel.get(event.chanIndex)
228234
msg = Message(
229235
timestamp=timestamp + self._time_offset,
230236
arbitration_id=msg_id & 0x1FFFFFFF,
@@ -236,7 +242,7 @@ def _recv_internal(self, timeout):
236242
bitrate_switch=bool(flags & vxlapi.XL_CAN_RXMSG_FLAG_BRS),
237243
dlc=dlc,
238244
data=event.tagData.canRxOkMsg.data[:dlc],
239-
channel=event.chanIndex)
245+
channel=channel)
240246
return msg, self._is_filtered
241247
else:
242248
event_count.value = 1
@@ -251,6 +257,7 @@ def _recv_internal(self, timeout):
251257
dlc = event.tagData.msg.dlc
252258
flags = event.tagData.msg.flags
253259
timestamp = event.timeStamp * 1e-9
260+
channel = self.index_to_channel.get(event.chanIndex)
254261
msg = Message(
255262
timestamp=timestamp + self._time_offset,
256263
arbitration_id=msg_id & 0x1FFFFFFF,
@@ -260,7 +267,7 @@ def _recv_internal(self, timeout):
260267
is_fd=False,
261268
dlc=dlc,
262269
data=event.tagData.msg.data[:dlc],
263-
channel=event.chanIndex)
270+
channel=channel)
264271
return msg, self._is_filtered
265272

266273
if end_time is not None and time.time() > end_time:
@@ -286,6 +293,10 @@ def send(self, msg, timeout=None):
286293

287294
flags = 0
288295

296+
# If channel has been specified, try to send only to that one.
297+
# Otherwise send to all channels
298+
mask = self.channel_masks.get(msg.channel, self.mask)
299+
289300
if self.fd:
290301
if msg.is_fd:
291302
flags |= vxlapi.XL_CAN_TXMSG_FLAG_EDL
@@ -306,7 +317,7 @@ def send(self, msg, timeout=None):
306317
XLcanTxEvent.tagData.canMsg.dlc = len2dlc(msg.dlc)
307318
for idx, value in enumerate(msg.data):
308319
XLcanTxEvent.tagData.canMsg.data[idx] = value
309-
vxlapi.xlCanTransmitEx(self.port_handle, self.mask, message_count, MsgCntSent, XLcanTxEvent)
320+
vxlapi.xlCanTransmitEx(self.port_handle, mask, message_count, MsgCntSent, XLcanTxEvent)
310321

311322
else:
312323
if msg.is_remote_frame:
@@ -322,7 +333,7 @@ def send(self, msg, timeout=None):
322333
xl_event.tagData.msg.flags = flags
323334
for idx, value in enumerate(msg.data):
324335
xl_event.tagData.msg.data[idx] = value
325-
vxlapi.xlCanTransmit(self.port_handle, self.mask, message_count, xl_event)
336+
vxlapi.xlCanTransmit(self.port_handle, mask, message_count, xl_event)
326337

327338

328339
def flush_tx_buffer(self):

‎can/interfaces/virtual.py‎

Lines changed: 16 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
and reside in the same process will receive the same messages.
1010
"""
1111

12+
import copy
1213
import logging
1314
import time
1415
try:
@@ -42,7 +43,8 @@ class VirtualBus(BusABC):
4243
behaves here.
4344
"""
4445

45-
def __init__(self, channel=None, receive_own_messages=False, **config):
46+
def __init__(self, channel=None, receive_own_messages=False,
47+
rx_queue_size=0, **config):
4648
super(VirtualBus, self).__init__(channel=channel,
4749
receive_own_messages=receive_own_messages, **config)
4850

@@ -59,7 +61,7 @@ def __init__(self, channel=None, receive_own_messages=False, **config):
5961
channels[self.channel_id] = []
6062
self.channel = channels[self.channel_id]
6163

62-
self.queue = queue.Queue()
64+
self.queue = queue.Queue(rx_queue_size)
6365
self.channel.append(self.queue)
6466

6567
def _check_if_open(self):
@@ -81,11 +83,21 @@ def _recv_internal(self, timeout):
8183

8284
def send(self, msg, timeout=None):
8385
self._check_if_open()
84-
msg.timestamp = time.time()
86+
# Create a shallow copy for this channel
87+
msg_copy = copy.copy(msg)
88+
msg_copy.timestamp = time.time()
89+
msg_copy.data = bytearray(msg.data)
90+
msg_copy.channel = self.channel_id
91+
all_sent = True
8592
# Add message to all listening on this channel
8693
for bus_queue in self.channel:
8794
if bus_queue is not self.queue or self.receive_own_messages:
88-
bus_queue.put(msg)
95+
try:
96+
bus_queue.put(msg_copy, block=True, timeout=timeout)
97+
except queue.Full:
98+
all_sent = False
99+
if not all_sent:
100+
raise CanError('Could not send message to one or more recipients')
89101

90102
def shutdown(self):
91103
self._check_if_open()

‎can/io/asc.py‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313

1414
from can.listener import Listener
1515
from can.message import Message
16+
from can.util import channel2int
1617

1718
CAN_MSG_EXT = 0x80000000
1819
CAN_ID_MASK = 0x1FFFFFFF
@@ -197,8 +198,12 @@ def on_message_received(self, msg):
197198
if msg.is_extended_id:
198199
arb_id += 'x'
199200

200-
# Many interfaces start channel numbering at 0 which is invalid
201-
channel = msg.channel + 1 if isinstance(msg.channel, int) else self.channel
201+
channel = channel2int(msg.channel)
202+
if channel is None:
203+
channel = self.channel
204+
else:
205+
# Many interfaces start channel numbering at 0 which is invalid
206+
channel += 1
202207

203208
serialized = self.FORMAT_MESSAGE.format(channel=channel,
204209
id=arb_id,

‎can/io/blf.py‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@
2323

2424
from can.message import Message
2525
from can.listener import Listener
26-
from can.util import len2dlc, dlc2len
26+
from can.util import len2dlc, dlc2len, channel2int
2727

2828

2929
class BLFParseError(Exception):
@@ -275,8 +275,13 @@ def __init__(self, filename, channel=1):
275275
self.stop_timestamp = None
276276

277277
def on_message_received(self, msg):
278-
# Many interfaces start channel numbering at 0 which is invalid in BLF
279-
channel = msg.channel + 1 if isinstance(msg.channel, int) else self.channel
278+
channel = channel2int(msg.channel)
279+
if channel is None:
280+
channel = self.channel
281+
else:
282+
# Many interfaces start channel numbering at 0 which is invalid
283+
channel += 1
284+
280285
arb_id = msg.arbitration_id
281286
if msg.id_type:
282287
arb_id |= CAN_MSG_EXT

‎can/io/canutils.py‎

Lines changed: 11 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,6 @@
55
This module works with CAN data in ASCII log files (*.log).
66
It is is compatible with "candump -L" from the canutils program
77
(https://github.com/linux-can/can-utils).
8-
9-
TODO: "channel" is not uesed by CanutilsLogWriter. Is that supposed to be like that?
108
"""
119

1210
import time
@@ -44,9 +42,11 @@ def __iter__(self):
4442

4543
if temp:
4644

47-
(timestamp, bus, frame) = temp.split()
45+
(timestamp, channel, frame) = temp.split()
4846
timestamp = float(timestamp[1:-1])
4947
(canId, data) = frame.split('#')
48+
if channel.isdigit():
49+
channel = int(channel)
5050

5151
if len(canId) > 3:
5252
isExtended = True
@@ -73,7 +73,7 @@ def __iter__(self):
7373
else:
7474
msg = Message(timestamp=timestamp, arbitration_id=canId & 0x1FFFFFFF,
7575
extended_id=isExtended, is_remote_frame=isRemoteFrame,
76-
dlc=dlc, data=dataBin)
76+
dlc=dlc, data=dataBin, channel=channel)
7777
yield msg
7878

7979

@@ -113,20 +113,22 @@ def on_message_received(self, msg):
113113
timestamp = self.last_timestamp
114114
else:
115115
timestamp = msg.timestamp
116+
117+
channel = msg.channel if msg.channel is not None else self.channel
116118

117119
if msg.is_error_frame:
118-
self.log_file.write("(%f) vcan0 %08X#0000000000000000\n" % (timestamp, CAN_ERR_FLAG | CAN_ERR_BUSERROR))
120+
self.log_file.write("(%f) %s %08X#0000000000000000\n" % (timestamp, channel, CAN_ERR_FLAG | CAN_ERR_BUSERROR))
119121

120122
elif msg.is_remote_frame:
121123
data = []
122124
if msg.is_extended_id:
123-
self.log_file.write("(%f) vcan0 %08X#R\n" % (timestamp, msg.arbitration_id))
125+
self.log_file.write("(%f) %s %08X#R\n" % (timestamp, channel, msg.arbitration_id))
124126
else:
125-
self.log_file.write("(%f) vcan0 %03X#R\n" % (timestamp, msg.arbitration_id))
127+
self.log_file.write("(%f) %s %03X#R\n" % (timestamp, channel, msg.arbitration_id))
126128

127129
else:
128130
data = ["{:02X}".format(byte) for byte in msg.data]
129131
if msg.is_extended_id:
130-
self.log_file.write("(%f) vcan0 %08X#%s\n" % (timestamp, msg.arbitration_id, ''.join(data)))
132+
self.log_file.write("(%f) %s %08X#%s\n" % (timestamp, channel, msg.arbitration_id, ''.join(data)))
131133
else:
132-
self.log_file.write("(%f) vcan0 %03X#%s\n" % (timestamp, msg.arbitration_id, ''.join(data)))
134+
self.log_file.write("(%f) %s %03X#%s\n" % (timestamp, channel, msg.arbitration_id, ''.join(data)))

‎can/message.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -123,7 +123,7 @@ def __repr__(self):
123123
"dlc={}".format(self.dlc),
124124
"data=[{}]".format(", ".join(data))]
125125
if self.channel is not None:
126-
args.append("channel={}".format(self.channel))
126+
args.append("channel={!r}".format(self.channel))
127127
if self.is_fd:
128128
args.append("is_fd=True")
129129
args.append("bitrate_switch={}".format(self.bitrate_switch))

‎can/notifier.py‎

Lines changed: 40 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -7,63 +7,76 @@
77

88
import threading
99
import logging
10+
import time
1011

1112
logger = logging.getLogger('can.Notifier')
1213

1314

1415
class Notifier(object):
1516

16-
def __init__(self, bus, listeners, timeout=None):
17-
"""Manages the distribution of **Messages** from a given bus to a
17+
def __init__(self, bus, listeners, timeout=1):
18+
"""Manages the distribution of **Messages** from a given bus/buses to a
1819
list of listeners.
1920
20-
:param bus: The :ref:`bus` to listen too.
21-
:param listeners: An iterable of :class:`~can.Listener`s
22-
:param timeout: An optional maximum number of seconds to wait for any message.
21+
:param can.Bus bus: The :ref:`bus` or a list of buses to listen to.
22+
:param list listeners: An iterable of :class:`~can.Listener`s
23+
:param float timeout: An optional maximum number of seconds to wait for any message.
2324
"""
2425
self.listeners = listeners
2526
self.bus = bus
2627
self.timeout = timeout
2728

28-
# exception raised in thread
29+
#: Exception raised in thread
2930
self.exception = None
3031

31-
self._running = threading.Event()
32-
self._running.set()
32+
self._running = True
33+
self._lock = threading.Lock()
3334

34-
self._reader = threading.Thread(target=self._rx_thread,
35-
name='can.notifier for bus "{}"'.format(self.bus.channel_info))
36-
self._reader.daemon = True
37-
self._reader.start()
35+
self._readers = []
36+
buses = self.bus if isinstance(self.bus, list) else [self.bus]
37+
for bus in buses:
38+
reader = threading.Thread(target=self._rx_thread, args=(bus,),
39+
name='can.notifier for bus "{}"'.format(bus.channel_info))
40+
reader.daemon = True
41+
reader.start()
42+
self._readers.append(reader)
3843

39-
def stop(self):
44+
def stop(self, timeout=5):
4045
"""Stop notifying Listeners when new :class:`~can.Message` objects arrive
4146
and call :meth:`~can.Listener.stop` on each Listener.
42-
"""
43-
self._running.clear()
44-
if self.timeout is not None:
45-
self._reader.join(self.timeout + 0.1)
4647
47-
def _rx_thread(self):
48+
:param float timeout:
49+
Max time in seconds to wait for receive threads to finish.
50+
Should be longer than timeout given at instantiation.
51+
"""
52+
self._running = False
53+
end_time = time.time() + timeout
54+
for reader in self._readers:
55+
now = time.time()
56+
if now < end_time:
57+
reader.join(end_time - now)
58+
for listener in self.listeners:
59+
listener.stop()
60+
61+
def _rx_thread(self, bus):
62+
msg = None
4863
try:
49-
while self._running.is_set():
50-
msg = self.bus.recv(self.timeout)
64+
while self._running:
5165
if msg is not None:
52-
for callback in self.listeners:
53-
callback(msg)
66+
with self._lock:
67+
for callback in self.listeners:
68+
callback(msg)
69+
msg = bus.recv(self.timeout)
5470
except Exception as exc:
5571
self.exception = exc
5672
raise
57-
finally:
58-
for listener in self.listeners:
59-
listener.stop()
6073

6174
def add_listener(self, listener):
6275
"""Add new Listener to the notification list.
6376
If it is already present, it will be called two times
6477
each time a message arrives.
6578
66-
:param listener: a :class:`~can.Listener` object to be added to
79+
:param can.Listener listener: Listener to be added to
6780
the list to be notified
6881
"""
6982
self.listeners.append(listener)
@@ -73,7 +86,7 @@ def remove_listener(self, listener):
7386
trows an exception if the given listener is not part of the
7487
stored listeners.
7588
76-
:param listener: a :class:`~can.Listener` object to be removed from
89+
:param can.Listener listener: Listener to be removed from
7790
the list to be notified
7891
:raises ValueError: if `listener` was never added to this notifier
7992
"""

0 commit comments

Comments
 (0)