diff --git a/can/interfaces/ics_neovi/neovi_bus.py b/can/interfaces/ics_neovi/neovi_bus.py index 815ed6fa0..3dcb4320b 100644 --- a/can/interfaces/ics_neovi/neovi_bus.py +++ b/can/interfaces/ics_neovi/neovi_bus.py @@ -12,9 +12,8 @@ import logging import os import tempfile -from collections import Counter, defaultdict, deque +from collections import Counter, deque from datetime import datetime -from functools import partial from itertools import cycle from threading import Event from warnings import warn @@ -41,6 +40,19 @@ ics = None +def _build_ics_netid_lookup(ics_module): + if ics_module is None: + return {} + return { + name[6:]: getattr(ics_module, name) + for name in dir(ics_module) + if name.startswith("NETID_") + } + + +ICS_NETID_LOOKUP = _build_ics_netid_lookup(ics) + + try: from filelock import FileLock except ImportError as ie: @@ -217,7 +229,8 @@ def __init__(self, channel, can_filters=None, **kwargs): else: # Assume comma separated string of channels self.channels = [ch.strip() for ch in channel.split(",")] - self.channels = [NeoViBus.channel_to_netid(ch) for ch in self.channels] + self.channels = tuple(NeoViBus.channel_to_netid(ch) for ch in self.channels) + self._channel_set = set(self.channels) type_filter = kwargs.get("type_filter") serial = kwargs.get("serial") @@ -257,17 +270,15 @@ def __init__(self, channel, can_filters=None, **kwargs): logger.info(f"Using device: {self.channel_info}") self.rx_buffer = deque() - self.message_receipts = defaultdict(Event) + self.message_receipts = {} @staticmethod def channel_to_netid(channel_name_or_id): try: channel = int(channel_name_or_id) except ValueError: - netid = f"NETID_{channel_name_or_id.upper()}" - if hasattr(ics, netid): - channel = getattr(ics, netid) - else: + channel = ICS_NETID_LOOKUP.get(channel_name_or_id.upper()) + if channel is None: raise ValueError( "channel must be an integer or a valid ICS channel name" ) from None @@ -281,7 +292,7 @@ def get_serial_number(device): :return: ics device serial string :rtype: str """ - if int("0A0000", 36) < device.SerialNumber < int("ZZZZZZ", 36): + if ics.MIN_BASE36_SERIAL < device.SerialNumber < ics.MAX_SERIAL: return ics.base36enc(device.SerialNumber) else: return str(device.SerialNumber) @@ -344,24 +355,33 @@ def _process_msg_queue(self, timeout=0.1): messages, errors = ics.get_messages(self.dev, False, timeout) except ics.RuntimeError: return + + channel_set = self._channel_set + rx_append = self.rx_buffer.append + message_receipts = self.message_receipts + receive_own_messages = self._receive_own_messages + for ics_msg in messages: channel = ics_msg.NetworkID | (ics_msg.NetworkID2 << 8) - if channel not in self.channels: + if channel not in channel_set: continue - is_tx = bool(ics_msg.StatusBitField & ics.SPY_STATUS_TX_MSG) + status_bitfield = ics_msg.StatusBitField + is_tx = bool(status_bitfield & ics.SPY_STATUS_TX_MSG) if is_tx: - if bool(ics_msg.StatusBitField & ics.SPY_STATUS_GLOBAL_ERR): + if status_bitfield & ics.SPY_STATUS_GLOBAL_ERR: continue receipt_key = (ics_msg.ArbIDOrHeader, ics_msg.DescriptionID) - if ics_msg.DescriptionID and receipt_key in self.message_receipts: - self.message_receipts[receipt_key].set() - if not self._receive_own_messages: + if ics_msg.DescriptionID: + receipt_event = message_receipts.get(receipt_key) + if receipt_event is not None: + receipt_event.set() + if not receive_own_messages: continue - self.rx_buffer.append(ics_msg) + rx_append(ics_msg) if errors: logger.warning("%d error(s) found", errors) @@ -392,49 +412,59 @@ def _get_timestamp_for_msg(self, ics_msg): def _ics_msg_to_message(self, ics_msg): is_fd = ics_msg.Protocol == ics.SPY_PROTOCOL_CANFD - - message_from_ics = partial( - Message, - timestamp=self._get_timestamp_for_msg(ics_msg), - arbitration_id=ics_msg.ArbIDOrHeader, - is_extended_id=bool(ics_msg.StatusBitField & ics.SPY_STATUS_XTD_FRAME), - is_remote_frame=bool(ics_msg.StatusBitField & ics.SPY_STATUS_REMOTE_FRAME), - is_error_frame=bool(ics_msg.StatusBitField2 & ics.SPY_STATUS2_ERROR_FRAME), - channel=ics_msg.NetworkID | (ics_msg.NetworkID2 << 8), - dlc=ics_msg.NumberBytesData, - is_fd=is_fd, - is_rx=not bool(ics_msg.StatusBitField & ics.SPY_STATUS_TX_MSG), - ) + status_bitfield = ics_msg.StatusBitField + status_bitfield3 = ics_msg.StatusBitField3 + number_bytes = ics_msg.NumberBytesData + channel = ics_msg.NetworkID | (ics_msg.NetworkID2 << 8) + timestamp = self._get_timestamp_for_msg(ics_msg) + arbitration_id = ics_msg.ArbIDOrHeader + is_extended_id = bool(status_bitfield & ics.SPY_STATUS_XTD_FRAME) + is_remote_frame = bool(status_bitfield & ics.SPY_STATUS_REMOTE_FRAME) + is_error_frame = bool(ics_msg.StatusBitField2 & ics.SPY_STATUS2_ERROR_FRAME) + is_rx = not bool(status_bitfield & ics.SPY_STATUS_TX_MSG) if is_fd: if ics_msg.ExtraDataPtrEnabled: - data = ics_msg.ExtraDataPtr[: ics_msg.NumberBytesData] + data = ics_msg.ExtraDataPtr[:number_bytes] else: - data = ics_msg.Data[: ics_msg.NumberBytesData] - - return message_from_ics( - data=data, - error_state_indicator=bool( - ics_msg.StatusBitField3 & ics.SPY_STATUS3_CANFD_ESI - ), - bitrate_switch=bool( - ics_msg.StatusBitField3 & ics.SPY_STATUS3_CANFD_BRS - ), + data = ics_msg.Data[:number_bytes] + + return Message( + timestamp, + arbitration_id, + is_extended_id, + is_remote_frame, + is_error_frame, + channel, + number_bytes, + data, + is_fd, + is_rx, + bool(status_bitfield3 & ics.SPY_STATUS3_CANFD_BRS), + bool(status_bitfield3 & ics.SPY_STATUS3_CANFD_ESI), ) else: - return message_from_ics( - data=ics_msg.Data[: ics_msg.NumberBytesData], + return Message( + timestamp, + arbitration_id, + is_extended_id, + is_remote_frame, + is_error_frame, + channel, + number_bytes, + ics_msg.Data[:number_bytes], + is_fd, + is_rx, ) def _recv_internal(self, timeout=0.1): if not self.rx_buffer: self._process_msg_queue(timeout=timeout) - try: - ics_msg = self.rx_buffer.popleft() - msg = self._ics_msg_to_message(ics_msg) - except IndexError: + if not self.rx_buffer: return None, False - return msg, False + + ics_msg = self.rx_buffer.popleft() + return self._ics_msg_to_message(ics_msg), False @check_if_bus_open def send(self, msg, timeout=0): @@ -502,15 +532,17 @@ def send(self, msg, timeout=0): else: raise ValueError("msg.channel must be set when using multiple channels.") - message.NetworkID, message.NetworkID2 = int(network_id & 0xFF), int( - (network_id >> 8) & 0xFF + message.NetworkID, message.NetworkID2 = ( + network_id & 0xFF, + (network_id >> 8) & 0xFF, ) if timeout != 0: msg_desc_id = next(description_id) message.DescriptionID = msg_desc_id receipt_key = (msg.arbitration_id, msg_desc_id) - self.message_receipts[receipt_key].clear() + receipt_event = Event() + self.message_receipts[receipt_key] = receipt_event try: ics.transmit_messages(self.dev, message) @@ -521,8 +553,8 @@ def send(self, msg, timeout=0): # This requires a notifier for the bus or # some other thread calling recv periodically if timeout != 0: - got_receipt = self.message_receipts[receipt_key].wait(timeout) + got_receipt = receipt_event.wait(timeout) # We no longer need this receipt, so no point keeping it in memory - del self.message_receipts[receipt_key] + self.message_receipts.pop(receipt_key, None) if not got_receipt: raise CanTimeoutError("Transmit timeout") diff --git a/test/test_neovi.py b/test/test_neovi.py index cc6ddc297..c3825f279 100644 --- a/test/test_neovi.py +++ b/test/test_neovi.py @@ -4,8 +4,15 @@ import pickle import unittest +from collections import deque +from contextlib import ExitStack +from threading import Event +from types import SimpleNamespace +from unittest.mock import patch from can.interfaces.ics_neovi import ICSApiError +from can.interfaces.ics_neovi.neovi_bus import NeoViBus +from can.interfaces.ics_neovi import neovi_bus class ICSApiErrorTest(unittest.TestCase): @@ -22,5 +29,163 @@ def test_error_pickling(self): assert iae.__dict__ == un_pickled_iae.__dict__ +class NeoViBusBehaviorTest(unittest.TestCase): + def test_channel_to_netid_accepts_integer_and_named_channel(self): + fake_ics = SimpleNamespace(NETID_HSCAN=42) + + with ExitStack() as stack: + stack.enter_context(patch.object(neovi_bus, "ics", fake_ics)) + stack.enter_context( + patch.object(neovi_bus, "ICS_NETID_LOOKUP", {"HSCAN": 42}, create=True) + ) + + self.assertEqual(NeoViBus.channel_to_netid(7), 7) + self.assertEqual(NeoViBus.channel_to_netid("8"), 8) + self.assertEqual(NeoViBus.channel_to_netid("hscan"), 42) + + def test_channel_to_netid_rejects_unknown_channel_name(self): + fake_ics = SimpleNamespace(NETID_HSCAN=42) + + with ExitStack() as stack: + stack.enter_context(patch.object(neovi_bus, "ics", fake_ics)) + stack.enter_context( + patch.object(neovi_bus, "ICS_NETID_LOOKUP", {"HSCAN": 42}, create=True) + ) + + with self.assertRaises(ValueError): + NeoViBus.channel_to_netid("unknown") + + def test_ics_msg_to_message_converts_classic_frame(self): + fake_ics = SimpleNamespace( + SPY_PROTOCOL_CANFD=99, + SPY_STATUS_XTD_FRAME=0x01, + SPY_STATUS_REMOTE_FRAME=0x02, + SPY_STATUS_TX_MSG=0x04, + SPY_STATUS2_ERROR_FRAME=0x08, + SPY_STATUS3_CANFD_ESI=0x10, + SPY_STATUS3_CANFD_BRS=0x20, + ) + bus = NeoViBus.__new__(NeoViBus) + bus._use_system_timestamp = True + bus._is_shutdown = True + + ics_msg = SimpleNamespace( + Protocol=0, + StatusBitField=fake_ics.SPY_STATUS_XTD_FRAME, + StatusBitField2=0, + StatusBitField3=0, + NumberBytesData=4, + NetworkID=0x34, + NetworkID2=0x12, + ArbIDOrHeader=0x123, + ExtraDataPtrEnabled=0, + ExtraDataPtr=tuple(), + Data=(1, 2, 3, 4, 9, 9, 9, 9), + TimeSystem=12.5, + ) + + with patch.object(neovi_bus, "ics", fake_ics): + msg = bus._ics_msg_to_message(ics_msg) + + self.assertEqual(msg.timestamp, 12.5) + self.assertEqual(msg.arbitration_id, 0x123) + self.assertTrue(msg.is_extended_id) + self.assertFalse(msg.is_remote_frame) + self.assertFalse(msg.is_error_frame) + self.assertFalse(msg.is_fd) + self.assertTrue(msg.is_rx) + self.assertEqual(msg.channel, 0x1234) + self.assertEqual(msg.dlc, 4) + self.assertEqual(bytes(msg.data), b"\x01\x02\x03\x04") + + def test_ics_msg_to_message_converts_fd_frame(self): + fake_ics = SimpleNamespace( + SPY_PROTOCOL_CANFD=99, + SPY_STATUS_XTD_FRAME=0x01, + SPY_STATUS_REMOTE_FRAME=0x02, + SPY_STATUS_TX_MSG=0x04, + SPY_STATUS2_ERROR_FRAME=0x08, + SPY_STATUS3_CANFD_ESI=0x10, + SPY_STATUS3_CANFD_BRS=0x20, + ) + bus = NeoViBus.__new__(NeoViBus) + bus._use_system_timestamp = True + bus._is_shutdown = True + + ics_msg = SimpleNamespace( + Protocol=fake_ics.SPY_PROTOCOL_CANFD, + StatusBitField=0, + StatusBitField2=fake_ics.SPY_STATUS2_ERROR_FRAME, + StatusBitField3=fake_ics.SPY_STATUS3_CANFD_BRS + | fake_ics.SPY_STATUS3_CANFD_ESI, + NumberBytesData=12, + NetworkID=5, + NetworkID2=0, + ArbIDOrHeader=0x456, + ExtraDataPtrEnabled=1, + ExtraDataPtr=tuple(range(16)), + Data=tuple(range(8)), + TimeSystem=3.25, + ) + + with patch.object(neovi_bus, "ics", fake_ics): + msg = bus._ics_msg_to_message(ics_msg) + + self.assertEqual(msg.timestamp, 3.25) + self.assertEqual(msg.arbitration_id, 0x456) + self.assertFalse(msg.is_extended_id) + self.assertFalse(msg.is_remote_frame) + self.assertTrue(msg.is_error_frame) + self.assertTrue(msg.is_fd) + self.assertTrue(msg.is_rx) + self.assertTrue(msg.bitrate_switch) + self.assertTrue(msg.error_state_indicator) + self.assertEqual(msg.channel, 5) + self.assertEqual(msg.dlc, 12) + self.assertEqual(bytes(msg.data), bytes(range(12))) + + def test_recv_internal_returns_none_when_no_message_available(self): + bus = NeoViBus.__new__(NeoViBus) + bus._is_shutdown = True + bus.rx_buffer = deque() + bus._process_msg_queue = lambda timeout=0.1: None + + msg, already_filtered = bus._recv_internal(timeout=0) + + self.assertIsNone(msg) + self.assertFalse(already_filtered) + + def test_process_msg_queue_sets_receipt_without_echoing_transmit(self): + fake_ics = SimpleNamespace( + SPY_STATUS_TX_MSG=0x01, + SPY_STATUS_GLOBAL_ERR=0x02, + get_messages=lambda dev, include_errors, timeout: ((tx_msg,), 0), + ) + tx_msg = SimpleNamespace( + NetworkID=1, + NetworkID2=0, + StatusBitField=fake_ics.SPY_STATUS_TX_MSG, + ArbIDOrHeader=0x321, + DescriptionID=17, + ) + receipt_key = (tx_msg.ArbIDOrHeader, tx_msg.DescriptionID) + receipt_event = Event() + bus = NeoViBus.__new__(NeoViBus) + bus._is_shutdown = False + bus.dev = object() + bus.channels = [1] + bus._channel_set = {1} + bus.rx_buffer = deque() + bus.message_receipts = {receipt_key: receipt_event} + bus._receive_own_messages = False + + with patch.object(neovi_bus, "ics", fake_ics): + bus._process_msg_queue(timeout=0) + + self.assertTrue(receipt_event.is_set()) + self.assertEqual(len(bus.rx_buffer), 0) + bus._is_shutdown = True + + if __name__ == "__main__": unittest.main()