Source code for can.interfaces.ics_neovi.neovi_bus

# coding: utf-8

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
from collections import deque

from can import Message, CanError, BusABC

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", ie
    ics = None

class ICSApiError(CanError):
    Indicates an error with the ICS API.

    # A critical error which affects operation or accuracy.
    # An error which is not understood.
    # An important error which may be critical depending on the application
    # An error which probably does not need attention.

    def __init__(
            self, error_number, description_short, description_long,
            severity, restart_needed
        super(ICSApiError, self).__init__(description_short)
        self.error_number = error_number
        self.description_short = description_short
        self.description_long = description_long
        self.severity = severity
        self.restart_needed = restart_needed == 1

    def __str__(self):
        return "{} {}".format(self.description_short, self.description_long)

    def is_critical(self):
        return self.severity == self.ICS_SPY_ERR_CRITICAL

[docs]class NeoViBus(BusABC): """ The CAN Bus implemented for the python_ics interface """ def __init__(self, channel, can_filters=None, **config): """ :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 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. """ if ics is None: raise ImportError('Please install python-ics') super(NeoViBus, self).__init__( channel=channel, can_filters=can_filters, **config)"CAN Filters: {}".format(can_filters))"Got configuration of: {}".format(config)) 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 = config.get('type_filter') serial = config.get('serial') = self._find_device(type_filter, serial) ics.open_device( if 'bitrate' in config: for channel in self.channels: ics.set_bit_rate(, config.get('bitrate'), channel) fd = config.get('fd', False) if fd: if 'data_bitrate' in config: for channel in self.channels: ics.set_fd_bit_rate(, config.get('data_bitrate'), channel) self._use_system_timestamp = bool( config.get('use_system_timestamp', False) ) self.channel_info = '%s %s CH:%s' % (, self.get_serial_number(, self.channels )"Using device: {}".format(self.channel_info)) self.rx_buffer = deque() @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' ) return channel
[docs] @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 """ a0000 = 604661760 if device.SerialNumber >= a0000: return ics.base36enc(device.SerialNumber) return str(device.SerialNumber)
[docs] def shutdown(self): super(NeoViBus, self).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): 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): dev = device break 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 Exception(' '.join(msg)) return dev 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 self.rx_buffer.append(ics_msg) if errors: logger.warning("%d error(s) found" % errors) for msg in ics.get_error_messages( error = ICSApiError(*msg) 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, extended_id=bool( ics_msg.StatusBitField & ics.SPY_STATUS_XTD_FRAME ), is_fd=is_fd, is_remote_frame=bool( ics_msg.StatusBitField & ics.SPY_STATUS_REMOTE_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, extended_id=bool( ics_msg.StatusBitField & ics.SPY_STATUS_XTD_FRAME ), is_fd=is_fd, is_remote_frame=bool( ics_msg.StatusBitField & ics.SPY_STATUS_REMOTE_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
[docs] def send(self, msg, timeout=None): if not ics.validate_hobject( raise CanError("bus not open") 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 message.NumberBytesData = len( message.Data = tuple([:8]) if msg.is_fd and len( > 8: message.ExtraDataPtrEnabled = 1 message.ExtraDataPtr = tuple( 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.") try: ics.transmit_messages(, message) except ics.RuntimeError: raise ICSApiError(*ics.get_last_api_error(