Hide navigation sidebar
Hide table of contents sidebar
Back to top
Toggle table of contents sidebar

# Source code for can.interfaces.neousys.neousys

    """ Neousys CAN bus driver """

    #
    # This kind of interface can be found for example on Neousys POC-551VTC
    # One needs to have correct drivers and DLL (Share object for Linux) from Neousys
    #
    # https://www.neousys-tech.com/en/support-service/resources/category/299-poc-551vtc-driver
    #
    # Beware this is only tested on Linux kernel higher than v5.3. This should be drop in
    # with Windows but you have to replace with correct named DLL
    #

    # pylint: disable=too-few-public-methods
    # pylint: disable=too-many-instance-attributes
    # pylint: disable=wrong-import-position

    import logging
    import platform
    import queue
    from ctypes import (
        CFUNCTYPE,
        POINTER,
        Structure,
        byref,
        c_ubyte,
        c_uint,
        c_ushort,
        sizeof,
    )
    from time import time

    try:
        from ctypes import WinDLL
    except ImportError:
        from ctypes import CDLL

    from can import (
        BusABC,
        CanInitializationError,
        CanInterfaceNotImplementedError,
        CanOperationError,
        CanProtocol,
        Message,
    )

    logger = logging.getLogger(__name__)

    class NeousysCanSetup(Structure):
        """C CAN Setup struct"""

        _fields_ = [
            ("bitRate", c_uint),
            ("recvConfig", c_uint),
            ("recvId", c_uint),
            ("recvMask", c_uint),
        ]

    class NeousysCanMsg(Structure):
        """C CAN Message struct"""

        _fields_ = [
            ("id", c_uint),
            ("flags", c_ushort),
            ("extra", c_ubyte),
            ("len", c_ubyte),
            ("data", c_ubyte * 8),
        ]

    # valid:2~16, sum of the Synchronization, Propagation, and
    #             Phase Buffer 1 segments, measured in time quanta.
    # valid:1~8, the Phase Buffer 2 segment in time quanta.
    # valid:1~4, Resynchronization Jump Width in time quanta
    # valid:1~1023, CAN_CLK divider used to determine time quanta
    class NeousysCanBitClk(Structure):
        """C CAN BIT Clock struct"""

        _fields_ = [
            ("syncPropPhase1Seg", c_ushort),
            ("phase2Seg", c_ushort),
            ("jumpWidth", c_ushort),
            ("quantumPrescaler", c_ushort),
        ]

    NEOUSYS_CAN_MSG_CALLBACK = CFUNCTYPE(None, POINTER(NeousysCanMsg), c_uint)
    NEOUSYS_CAN_STATUS_CALLBACK = CFUNCTYPE(None, c_uint)

    NEOUSYS_CAN_MSG_EXTENDED_ID = 0x0004
    NEOUSYS_CAN_MSG_REMOTE_FRAME = 0x0040
    NEOUSYS_CAN_MSG_DATA_NEW = 0x0080
    NEOUSYS_CAN_MSG_DATA_LOST = 0x0100

    NEOUSYS_CAN_MSG_USE_ID_FILTER = 0x00000008
    NEOUSYS_CAN_MSG_USE_DIR_FILTER = (
        0x00000010 | NEOUSYS_CAN_MSG_USE_ID_FILTER
    )  # only accept the direction specified in the message type
    NEOUSYS_CAN_MSG_USE_EXT_FILTER = (
        0x00000020 | NEOUSYS_CAN_MSG_USE_ID_FILTER
    )  # filters on only extended identifiers

    NEOUSYS_CAN_STATUS_BUS_OFF = 0x00000080
    NEOUSYS_CAN_STATUS_EWARN = (
        0x00000040  # can controller error level has reached warning level.
    )
    NEOUSYS_CAN_STATUS_EPASS = (
        0x00000020  # can controller error level has reached error passive level.
    )
    NEOUSYS_CAN_STATUS_LEC_STUFF = 0x00000001  # a bit stuffing error has occurred.
    NEOUSYS_CAN_STATUS_LEC_FORM = 0x00000002  # a formatting error has occurred.
    NEOUSYS_CAN_STATUS_LEC_ACK = 0x00000003  # an acknowledge error has occurred.
    NEOUSYS_CAN_STATUS_LEC_BIT1 = (
        0x00000004  # the bus remained a bit level of 1 for longer than is allowed.
    )
    NEOUSYS_CAN_STATUS_LEC_BIT0 = (
        0x00000005  # the bus remained a bit level of 0 for longer than is allowed.
    )
    NEOUSYS_CAN_STATUS_LEC_CRC = 0x00000006  # a crc error has occurred.
    NEOUSYS_CAN_STATUS_LEC_MASK = (
        0x00000007  # this is the mask for the can last error code (lec).
    )

    NEOUSYS_CANLIB = None

    try:
        if platform.system() == "Windows":
            NEOUSYS_CANLIB = WinDLL("./WDT_DIO.dll")
        else:
            NEOUSYS_CANLIB = CDLL("libwdt_dio.so")
        logger.info("Loaded Neousys WDT_DIO Can driver")
    except OSError as error:
        logger.info("Cannot load Neousys CAN bus dll or shared object: %s", error)

    class NeousysBus(BusABC):
        """Neousys CAN bus Class"""

        def __init__(self, channel, device=0, bitrate=500000, **kwargs):
            """
            :param channel: channel number
            :param device: device number
            :param bitrate: bit rate.
            """
            super().__init__(channel, **kwargs)

            if NEOUSYS_CANLIB is None:
                raise CanInterfaceNotImplementedError("Neousys WDT_DIO Can driver missing")

            self.channel = channel
            self.device = device
            self.channel_info = f"Neousys Can: device {self.device}, channel {self.channel}"
            self._can_protocol = CanProtocol.CAN_20

            self.queue = queue.Queue()

            # Init with accept all and wanted bitrate
            self.init_config = NeousysCanSetup(bitrate, NEOUSYS_CAN_MSG_USE_ID_FILTER, 0, 0)

            self._neousys_recv_cb = NEOUSYS_CAN_MSG_CALLBACK(self._neousys_recv_cb)
            self._neousys_status_cb = NEOUSYS_CAN_STATUS_CALLBACK(self._neousys_status_cb)

            if NEOUSYS_CANLIB.CAN_RegisterReceived(0, self._neousys_recv_cb) == 0:
                raise CanInitializationError("Neousys CAN bus Setup receive callback")

            if NEOUSYS_CANLIB.CAN_RegisterStatus(0, self._neousys_status_cb) == 0:
                raise CanInitializationError("Neousys CAN bus Setup status callback")

            if (
                NEOUSYS_CANLIB.CAN_Setup(
                    channel, byref(self.init_config), sizeof(self.init_config)
                )
                == 0
            ):
                raise CanInitializationError("Neousys CAN bus Setup Error")

            if NEOUSYS_CANLIB.CAN_Start(channel) == 0:
                raise CanInitializationError("Neousys CAN bus Start Error")

        def send(self, msg, timeout=None) -> None:
            """
            :param msg: message to send
            :param timeout: timeout is not used here
            """

            tx_msg = NeousysCanMsg(
                msg.arbitration_id, 0, 0, msg.dlc, (c_ubyte * 8)(*msg.data)
            )

            if NEOUSYS_CANLIB.CAN_Send(self.channel, byref(tx_msg), sizeof(tx_msg)) == 0:
                raise CanOperationError("Neousys Can can't send message")

        def _recv_internal(self, timeout):
            try:
                return self.queue.get(block=True, timeout=timeout), False
            except queue.Empty:
                return None, False

        def _neousys_recv_cb(self, msg, sizeof_msg) -> None:
            """
            :param msg: struct CAN_MSG
            :param sizeof_msg: message number
            """
            msg_bytes = bytearray(msg.contents.data)
            remote_frame = bool(msg.contents.flags & NEOUSYS_CAN_MSG_REMOTE_FRAME)
            extended_frame = bool(msg.contents.flags & NEOUSYS_CAN_MSG_EXTENDED_ID)

            if msg.contents.flags & NEOUSYS_CAN_MSG_DATA_LOST:
                logger.error("_neousys_recv_cb flag CAN_MSG_DATA_LOST")

            msg = Message(
                timestamp=time(),
                arbitration_id=msg.contents.id,
                is_remote_frame=remote_frame,
                is_extended_id=extended_frame,
                channel=self.channel,
                dlc=msg.contents.len,
                data=msg_bytes[: msg.contents.len],
            )

            # Reading happens in Callback function and
            # with Python-CAN it happens polling
            # so cache stuff in array to for poll
            try:
                self.queue.put(msg)
            except queue.Full:
                raise CanOperationError("Neousys message Queue is full") from None

        def _neousys_status_cb(self, status) -> None:
            """
            :param status: BUS Status
            """
            logger.info("%s _neousys_status_cb: %d", self.init_config, status)

        def shutdown(self):
            super().shutdown()
            NEOUSYS_CANLIB.CAN_Stop(self.channel)

        @staticmethod
        def _detect_available_configs():
            # There is only one channel
            return [{"interface": "neousys", "channel": 0}]