usb2canabstractionlayer.py

"""
This wrapper is for windows or direct access via CANAL API.
Socket CAN is recommended under Unix/Linux systems.
"""

from ctypes import *
from enum import IntEnum
import logging

import can
from ...exceptions import error_check

log = logging.getLogger("can.usb2can")

# type definitions
flags_t = c_ulong
pConfigureStr = c_char_p
handle_t = c_long
timeout_t = c_ulong
filter_t = c_ulong

# flags mappings
IS_ERROR_FRAME = 4
IS_REMOTE_FRAME = 2
IS_ID_TYPE = 1


class CanalError(IntEnum):
    SUCCESS = 0
    BAUDRATE = 1
    BUS_OFF = 2
    BUS_PASSIVE = 3
    BUS_WARNING = 4
    CAN_ID = 5
    CAN_MESSAGE = 6
    CHANNEL = 7
    FIFO_EMPTY = 8
    FIFO_FULL = 9
    FIFO_SIZE = 10
    FIFO_WAIT = 11
    GENERIC = 12
    HARDWARE = 13
    INIT_FAIL = 14
    INIT_MISSING = 15
    INIT_READY = 16
    NOT_SUPPORTED = 17
    OVERRUN = 18
    RCV_EMPTY = 19
    REGISTER = 20
    TRM_FULL = 21
    ERRFRM_STUFF = 22
    ERRFRM_FORM = 23
    ERRFRM_ACK = 24
    ERRFRM_BIT1 = 25
    ERRFRM_BIT0 = 26
    ERRFRM_CRC = 27
    LIBRARY = 28
    PROCADDRESS = 29
    ONLY_ONE_INSTANCE = 30
    SUB_DRIVER = 31
    TIMEOUT = 32
    NOT_OPEN = 33
    PARAMETER = 34
    MEMORY = 35
    INTERNAL = 36
    COMMUNICATION = 37


class CanalStatistics(Structure):
    _fields_ = [
        ("ReceiveFrams", c_ulong),
        ("TransmistFrams", c_ulong),
        ("ReceiveData", c_ulong),
        ("TransmitData", c_ulong),
        ("Overruns", c_ulong),
        ("BusWarnings", c_ulong),
        ("BusOff", c_ulong),
    ]


stat = CanalStatistics


class CanalStatus(Structure):
    _fields_ = [
        ("channel_status", c_ulong),
        ("lasterrorcode", c_ulong),
        ("lasterrorsubcode", c_ulong),
        ("lasterrorstr", c_byte * 80),
    ]


# data type for the CAN Message
class CanalMsg(Structure):
    _fields_ = [
        ("flags", c_ulong),
        ("obid", c_ulong),
        ("id", c_ulong),
        ("sizeData", c_ubyte),
        ("data", c_ubyte * 8),
        ("timestamp", c_ulong),
    ]


class Usb2CanAbstractionLayer:
    """A low level wrapper around the usb2can library.

    Documentation: http://www.8devices.com/media/products/usb2can/downloads/CANAL_API.pdf
    """

    def __init__(self, dll="usb2can.dll"):
        """
        :type dll: str or path-like
        :param dll (optional): the path to the usb2can DLL to load

        :raises can.CanInterfaceNotImplementedError: if the DLL could not be loaded
        """
        try:
            self.__m_dllBasic = windll.LoadLibrary(dll)
            if self.__m_dllBasic is None:
                raise Exception("__m_dllBasic is None")

        except Exception as error:
            message = f"DLL failed to load at path: {dll}"
            raise can.CanInterfaceNotImplementedError(message) from error

    def open(self, configuration: str, flags: int):
        """
        Opens a CAN connection using `CanalOpen()`.

        :param configuration: the configuration: "device_id; baudrate"
        :param flags: the flags to be set

        :raises can.CanInitializationError: if any error occurred
        :returns: Valid handle for CANAL API functions on success
        """
        try:
            # we need to convert this into bytes, since the underlying DLL cannot
            # handle non-ASCII configuration strings
            config_ascii = configuration.encode("ascii", "ignore")
            result = self.__m_dllBasic.CanalOpen(config_ascii, flags)
        except Exception as ex:
            # catch any errors thrown by this call and re-raise
            raise can.CanInitializationError(
                f'CanalOpen() failed, configuration: "{configuration}", error: {ex}'
            )
        else:
            # any greater-than-zero return value indicates a success
            # (see https://grodansparadis.gitbooks.io/the-vscp-daemon/canal_interface_specification.html)
            # raise an error if the return code is <= 0
            if result <= 0:
                raise can.CanInitializationError(
                    f'CanalOpen() failed, configuration: "{configuration}"',
                    error_code=result,
                )
            else:
                return result

    def close(self, handle) -> CanalError:
        with error_check("Failed to close"):
            return CanalError(self.__m_dllBasic.CanalClose(handle))

    def send(self, handle, msg) -> CanalError:
        with error_check("Failed to transmit frame"):
            return CanalError(self.__m_dllBasic.CanalSend(handle, msg))

    def receive(self, handle, msg) -> CanalError:
        with error_check("Receive error"):
            return CanalError(self.__m_dllBasic.CanalReceive(handle, msg))

    def blocking_send(self, handle, msg, timeout) -> CanalError:
        with error_check("Blocking send error"):
            return CanalError(self.__m_dllBasic.CanalBlockingSend(handle, msg, timeout))

    def blocking_receive(self, handle, msg, timeout) -> CanalError:
        with error_check("Blocking Receive Failed"):
            return CanalError(
                self.__m_dllBasic.CanalBlockingReceive(handle, msg, timeout)
            )

    def get_status(self, handle, status) -> CanalError:
        with error_check("Get status failed"):
            return CanalError(self.__m_dllBasic.CanalGetStatus(handle, status))

    def get_statistics(self, handle, statistics) -> CanalError:
        with error_check("Get Statistics failed"):
            return CanalError(self.__m_dllBasic.CanalGetStatistics(handle, statistics))

    def get_version(self):
        with error_check("Failed to get version info"):
            return self.__m_dllBasic.CanalGetVersion()

    def get_library_version(self):
        with error_check("Failed to get DLL version"):
            return self.__m_dllBasic.CanalGetDllVersion()

    def get_vendor_string(self):
        with error_check("Failed to get vendor string"):
            vstr = self.__m_dllBasic.CanalGetVendorString
            vstr.restype = c_char_p
            return vstr()

    def set_filters(self, handle, len, filters_arr):
        with error_check("Failed to set filters"):
            return CanalError(self.__m_dllBasic.CanalSetFilters(handle, len, filters_arr))