Source code for can.interfaces.ics_neovi.neovi_bus

Intrepid Control Systems (ICS) neoVI interface module.

python-ics is a Python wrapper around the API provided by Intrepid Control
Systems for communicating with their neoVI range of devices.

Implementation references:

import logging
import os
import tempfile
from collections import deque, defaultdict, Counter
from itertools import cycle
from threading import Event
from warnings import warn

from can import Message, BusABC
from ...exceptions import (

logger = logging.getLogger(__name__)

    import ics
except ImportError as ie:
        "You won't be able to use the ICS neoVI can backend without the "
        "python-ics module installed!: %s",
    ics = None

    from filelock import FileLock
except ImportError as ie:

        "Using ICS neoVI can backend without the "
        "filelock module installed may cause some issues!: %s",

    class FileLock:
        """Dummy file lock that does not actually do anything"""

        def __init__(self, lock_file, timeout=-1):
            self._lock_file = lock_file
            self.timeout = timeout

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc_val, exc_tb):
            return None

# Use inter-process mutex to prevent concurrent device open.
# When neoVI server is enabled, there is an issue with concurrent device open.
open_lock = FileLock(os.path.join(tempfile.gettempdir(), "neovi.lock"))
description_id = cycle(range(1, 0x8000))

[docs]class ICSApiError(CanError): """ Indicates an error with the ICS API. """ # A critical error which affects operation or accuracy. ICS_SPY_ERR_CRITICAL = 0x10 # An error which is not understood. ICS_SPY_ERR_QUESTION = 0x20 # An important error which may be critical depending on the application ICS_SPY_ERR_EXCLAMATION = 0x30 # An error which probably does not need attention. ICS_SPY_ERR_INFORMATION = 0x40 def __init__( self, error_code: int, description_short: str, description_long: str, severity: int, restart_needed: int, ): super().__init__(f"{description_short}. {description_long}", error_code) self.description_short = description_short self.description_long = description_long self.severity = severity self.restart_needed = restart_needed == 1 def __reduce__(self): return type(self), ( self.error_code, self.description_short, self.description_long, self.severity, self.restart_needed, ) @property def error_number(self) -> int: """Deprecated. Renamed to :attr:`can.CanError.error_code`.""" warn( "ICSApiError::error_number has been renamed to error_code defined by CanError", DeprecationWarning, ) return self.error_code @property def is_critical(self) -> bool: return self.severity == self.ICS_SPY_ERR_CRITICAL
[docs]class ICSInitializationError(ICSApiError, CanInitializationError): pass
[docs]class ICSOperationError(ICSApiError, CanOperationError): pass
[docs]class NeoViBus(BusABC): """ The CAN Bus implemented for the python_ics interface """ def __init__(self, channel, can_filters=None, **kwargs): """ :param channel: The channel ids to create this bus with. Can also be a single integer, netid name or a comma separated string. :type channel: int or str or list(int) or list(str) :param list can_filters: See :meth:`can.BusABC.set_filters` for details. :param bool receive_own_messages: If transmitted messages should also be received by this bus. :param bool use_system_timestamp: Use system timestamp for can messages instead of the hardware time stamp :param str serial: Serial to connect (optional, will use the first found if not supplied) :param int bitrate: Channel bitrate in bit/s. (optional, will enable the auto bitrate feature if not supplied) :param bool fd: If CAN-FD frames should be supported. :param int data_bitrate: Which bitrate to use for data phase in CAN FD. Defaults to arbitration bitrate. :param override_library_name: Absolute path or relative path to the library including filename. :raise ImportError: If *python-ics* is not available :raise CanInitializationError: If the bus could not be set up. May or may not be a :class:`~ICSInitializationError`. """ if ics is None: raise ImportError("Please install python-ics") super().__init__(channel=channel, can_filters=can_filters, **kwargs)"CAN Filters: {}".format(can_filters))"Got configuration of: {}".format(kwargs)) if "override_library_name" in kwargs: ics.override_library_name(kwargs.get("override_library_name")) if isinstance(channel, (list, tuple)): self.channels = channel elif isinstance(channel, int): self.channels = [channel] 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] type_filter = kwargs.get("type_filter") serial = kwargs.get("serial") = self._find_device(type_filter, serial) with open_lock: ics.open_device( try: if "bitrate" in kwargs: for channel in self.channels: ics.set_bit_rate(, kwargs.get("bitrate"), channel) if kwargs.get("fd", False): if "data_bitrate" in kwargs: for channel in self.channels: ics.set_fd_bit_rate(, kwargs.get("data_bitrate"), channel ) except ics.RuntimeError as re: logger.error(re) err = ICSInitializationError(*ics.get_last_api_error( try: self.shutdown() finally: raise err self._use_system_timestamp = bool(kwargs.get("use_system_timestamp", False)) self._receive_own_messages = kwargs.get("receive_own_messages", True) self.channel_info = "%s %s CH:%s" % (, self.get_serial_number(, self.channels, )"Using device: {}".format(self.channel_info)) self.rx_buffer = deque() self.message_receipts = defaultdict(Event) @staticmethod def channel_to_netid(channel_name_or_id): try: channel = int(channel_name_or_id) except ValueError: netid = "NETID_{}".format(channel_name_or_id.upper()) if hasattr(ics, netid): channel = getattr(ics, netid) else: raise ValueError( "channel must be an integer or a valid ICS channel name" ) from None return channel @staticmethod def get_serial_number(device): """Decode (if needed) and return the ICS device serial string :param device: ics device :return: ics device serial string :rtype: str """ if int("AA0000", 36) < device.SerialNumber < int("ZZZZZZ", 36): return ics.base36enc(device.SerialNumber) else: return str(device.SerialNumber) def shutdown(self): super().shutdown() ics.close_device( @staticmethod def _detect_available_configs(): """Detect all configurations/channels that this interface could currently connect with. :rtype: Iterator[dict] :return: an iterable of dicts, each being a configuration suitable for usage in the interface's bus constructor. """ if ics is None: return [] try: devices = ics.find_devices() except Exception as e: logger.debug("Failed to detect configs: %s", e) return [] # TODO: add the channel(s) return [ {"interface": "neovi", "serial": NeoViBus.get_serial_number(device)} for device in devices ] def _find_device(self, type_filter=None, serial=None): """Returns the first matching device or raises an error. :raise CanInitializationError: If not matching device could be found """ if type_filter is not None: devices = ics.find_devices(type_filter) else: devices = ics.find_devices() for device in devices: if serial is None or self.get_serial_number(device) == str(serial): return device else: msg = ["No device"] if type_filter is not None: msg.append("with type {}".format(type_filter)) if serial is not None: msg.append("with serial {}".format(serial)) msg.append("found.") raise CanInitializationError(" ".join(msg)) def _process_msg_queue(self, timeout=0.1): try: messages, errors = ics.get_messages(, False, timeout) except ics.RuntimeError: return for ics_msg in messages: if ics_msg.NetworkID not in self.channels: continue is_tx = bool(ics_msg.StatusBitField & ics.SPY_STATUS_TX_MSG) if is_tx: if bool(ics_msg.StatusBitField & 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: continue self.rx_buffer.append(ics_msg) if errors: logger.warning("%d error(s) found", errors) for msg, count in Counter(ics.get_error_messages( error = ICSApiError(*msg) if count > 1: logger.warning(f"{error} (Repeated {count} times)") else: logger.warning(error) def _get_timestamp_for_msg(self, ics_msg): if self._use_system_timestamp: # This is the system time stamp. # TimeSystem is loaded with the value received from the timeGetTime # call in the WIN32 multimedia API. # # The timeGetTime accuracy is up to 1 millisecond. See the WIN32 # API documentation for more information. # # This timestamp is useful for time comparing with other system # events or data which is not synced with the neoVI timestamp. # # Currently, TimeSystem2 is not used. return ics_msg.TimeSystem else: # This is the hardware time stamp. return ics.get_timestamp_for_msg(, ics_msg) def _ics_msg_to_message(self, ics_msg): is_fd = ics_msg.Protocol == ics.SPY_PROTOCOL_CANFD if is_fd: if ics_msg.ExtraDataPtrEnabled: data = ics_msg.ExtraDataPtr[: ics_msg.NumberBytesData] else: data = ics_msg.Data[: ics_msg.NumberBytesData] return Message( timestamp=self._get_timestamp_for_msg(ics_msg), arbitration_id=ics_msg.ArbIDOrHeader, data=data, dlc=ics_msg.NumberBytesData, is_extended_id=bool(ics_msg.StatusBitField & ics.SPY_STATUS_XTD_FRAME), is_fd=is_fd, is_rx=not bool(ics_msg.StatusBitField & ics.SPY_STATUS_TX_MSG), is_remote_frame=bool( ics_msg.StatusBitField & ics.SPY_STATUS_REMOTE_FRAME ), is_error_frame=bool( ics_msg.StatusBitField2 & ics.SPY_STATUS2_ERROR_FRAME ), error_state_indicator=bool( ics_msg.StatusBitField3 & ics.SPY_STATUS3_CANFD_ESI ), bitrate_switch=bool( ics_msg.StatusBitField3 & ics.SPY_STATUS3_CANFD_BRS ), channel=ics_msg.NetworkID, ) else: return Message( timestamp=self._get_timestamp_for_msg(ics_msg), arbitration_id=ics_msg.ArbIDOrHeader, data=ics_msg.Data[: ics_msg.NumberBytesData], dlc=ics_msg.NumberBytesData, is_extended_id=bool(ics_msg.StatusBitField & ics.SPY_STATUS_XTD_FRAME), is_fd=is_fd, is_rx=not bool(ics_msg.StatusBitField & ics.SPY_STATUS_TX_MSG), 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, ) 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: return None, False return msg, False def send(self, msg, timeout=0): """Transmit a message to the CAN bus. :param Message msg: A message object. :param float timeout: If > 0, wait up to this many seconds for message to be ACK'ed. If timeout is exceeded, an exception will be raised. None blocks indefinitely. :raises ValueError: if the message is invalid :raises can.CanTimeoutError: if sending timed out :raises CanOperationError: If the bus is closed or the message could otherwise not be sent. May or may not be a :class:`~ICSOperationError`. """ if not ics.validate_hobject( raise CanOperationError("bus not open") # Check for valid DLC to avoid passing extra long data to the driver if msg.is_fd: if msg.dlc > 64: raise ValueError( f"DLC was {msg.dlc} but it should be <= 64 for CAN FD frames" ) elif msg.dlc > 8: raise ValueError( f"DLC was {msg.dlc} but it should be <= 8 for normal CAN frames" ) message = ics.SpyMessage() flag0 = 0 if msg.is_extended_id: flag0 |= ics.SPY_STATUS_XTD_FRAME if msg.is_remote_frame: flag0 |= ics.SPY_STATUS_REMOTE_FRAME flag3 = 0 if msg.is_fd: message.Protocol = ics.SPY_PROTOCOL_CANFD if msg.bitrate_switch: flag3 |= ics.SPY_STATUS3_CANFD_BRS if msg.error_state_indicator: flag3 |= ics.SPY_STATUS3_CANFD_ESI message.ArbIDOrHeader = msg.arbitration_id msg_data =[: msg.dlc] message.NumberBytesData = msg.dlc message.Data = tuple(msg_data[:8]) if msg.is_fd and len(msg_data) > 8: message.ExtraDataPtrEnabled = 1 message.ExtraDataPtr = tuple(msg_data) message.StatusBitField = flag0 message.StatusBitField2 = 0 message.StatusBitField3 = flag3 if is not None: message.NetworkID = elif len(self.channels) == 1: message.NetworkID = self.channels[0] else: raise ValueError(" must be set when using multiple channels.") 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() try: ics.transmit_messages(, message) except ics.RuntimeError: raise ICSOperationError(*ics.get_last_api_error( from None # If timeout is set, wait for ACK # 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) # We no longer need this receipt, so no point keeping it in memory del self.message_receipts[receipt_key] if not got_receipt: raise CanTimeoutError("Transmit timeout")