Merge remote-tracking branch 'upstream/develop' into etas

Author: Felix Nieuwenhuizen

can/interfaces/nixnet.py

@@ -13,7 +13,9 @@
 import time
 import struct
 
-from can import CanError, BusABC, Message
+from can import BusABC, Message
+from ..exceptions import CanInitializationError, CanOperationError
+
 
 logger = logging.getLogger(__name__)
 
@@ -28,18 +30,15 @@
             database,
             XnetError,
         )
-    except ImportError:
-        logger.error("Error, NIXNET python module cannot be loaded.")
-        raise
+    except ImportError as error:
+        raise ImportError("NIXNET python module cannot be loaded") from error
 else:
-    logger.error("NI-XNET interface is only available on Windows systems")
-    raise NotImplementedError("NiXNET is not supported on not Win32 platforms")
+    raise NotImplementedError("NiXNET only supported on Win32 platforms")
 
 
 class NiXNETcanBus(BusABC):
     """
     The CAN Bus implemented for the NI-XNET interface.
-
     """
 
     def __init__(
@@ -52,7 +51,7 @@ def __init__(
         brs=False,
         can_termination=False,
         log_errors=True,
-        **kwargs
+        **kwargs,
     ):
         """
         :param str channel:
@@ -69,9 +68,8 @@ def __init__(
             ``is_error_frame`` set to True and ``arbitration_id`` will identify
             the error (default True)
 
-        :raises can.interfaces.nixnet.NiXNETError:
+        :raises can.exceptions.CanInitializationError:
             If starting communication fails
-
         """
         self._rx_queue = []
         self.channel = channel
@@ -120,16 +118,18 @@ def __init__(
             self.__session_send.start()
             self.__session_receive.start()
 
-        except errors.XnetError as err:
-            raise NiXNETError(function="__init__", error_message=err.args[0]) from None
+        except errors.XnetError as error:
+            raise CanInitializationError(
+                f"{error.args[0]} ({error.error_type})", error.error_code
+            ) from None
 
         self._is_filtered = False
         super(NiXNETcanBus, self).__init__(
             channel=channel,
             can_filters=can_filters,
             bitrate=bitrate,
             log_errors=log_errors,
-            **kwargs
+            **kwargs,
         )
 
     def _recv_internal(self, timeout):
@@ -160,8 +160,7 @@ def _recv_internal(self, timeout):
             )
 
             return msg, self._filters is None
-        except Exception as e:
-            # print('Error: ', e)
+        except Exception:
             return None, self._filters is None
 
     def send(self, msg, timeout=None):
@@ -174,7 +173,7 @@ def send(self, msg, timeout=None):
         :param float timeout:
             Max time to wait for the device to be ready in seconds, None if time is infinite
 
-        :raises can.interfaces.nixnet.NiXNETError:
+        :raises can.exceptions.CanOperationError:
             If writing to transmit buffer fails.
             It does not wait for message to be ACKed currently.
         """
@@ -201,8 +200,10 @@ def send(self, msg, timeout=None):
 
         try:
             self.__session_send.frames.write([can_frame], timeout)
-        except errors.XnetError as err:
-            raise NiXNETError(function="send", error_message=err.args[0]) from None
+        except errors.XnetError as error:
+            raise CanOperationError(
+                f"{error.args[0]} ({error.error_type})", error.error_code
+            ) from None
 
     def reset(self):
         """
@@ -254,18 +255,3 @@ def _detect_available_configs():
             logger.debug("An error occured while searching for configs: %s", str(error))
 
         return configs
-
-
-# To-Do review error management, I don't like this implementation
-class NiXNETError(CanError):
-    """Error from NI-XNET driver."""
-
-    def __init__(self, function="", error_message=""):
-        super(NiXNETError, self).__init__()
-        #: Function that failed
-        self.function = function
-        #: Arguments passed to function
-        self.error_message = error_message
-
-    def __str__(self):
-        return "Function %s failed:\n%s" % (self.function, self.error_message)

can/interfaces/pcan/basic.py

@@ -814,7 +814,7 @@ def Read(self, Channel):
         """Reads a CAN message from the receive queue of a PCAN Channel
 
         Remarks:
-          The return value of this method is a 3-touple, where
+          The return value of this method is a 3-tuple, where
           the first value is the result (TPCANStatus) of the method.
           The order of the values are:
           [0]: A TPCANStatus error code
@@ -843,7 +843,7 @@ def ReadFD(self, Channel):
         """Reads a CAN message from the receive queue of a FD capable PCAN Channel
 
         Remarks:
-          The return value of this method is a 3-touple, where
+          The return value of this method is a 3-tuple, where
           the first value is the result (TPCANStatus) of the method.
           The order of the values are:
           [0]: A TPCANStatus error code
@@ -943,7 +943,7 @@ def GetValue(self, Channel, Parameter):
           of Hardware (PCAN Channel) being used. If a parameter is not available,
           a PCAN_ERROR_ILLPARAMTYPE error will be returned.
 
-          The return value of this method is a 2-touple, where
+          The return value of this method is a 2-tuple, where
           the first value is the result (TPCANStatus) of the method and
           the second one, the asked value
 
@@ -1038,7 +1038,7 @@ def GetErrorText(self, Error, Language=0):
           Neutral (0x00), German (0x07), English (0x09), Spanish (0x0A),
           Italian (0x10) and French (0x0C)
 
-          The return value of this method is a 2-touple, where
+          The return value of this method is a 2-tuple, where
           the first value is the result (TPCANStatus) of the method and
           the second one, the error text
 
@@ -1063,7 +1063,7 @@ def LookUpChannel(self, Parameters):
 
         Remarks:
 
-          The return value of this method is a 2-touple, where
+          The return value of this method is a 2-tuple, where
           the first value is the result (TPCANStatus) of the method and
           the second one a TPCANHandle value
 

can/interfaces/robotell.py

@@ -6,6 +6,7 @@
 import logging
 
 from can import BusABC, Message
+from ..exceptions import CanInterfaceNotImplementedError
 
 logger = logging.getLogger(__name__)
 
@@ -56,15 +57,17 @@ def __init__(
     ):
         """
         :param str channel:
-            port of underlying serial or usb device (e.g. /dev/ttyUSB0, COM8, ...)
-            Must not be empty.
+            port of underlying serial or usb device (e.g. ``/dev/ttyUSB0``, ``COM8``, ...)
+            Must not be empty. Can also end with ``@115200`` (or similarly) to specify the baudrate.
         :param int ttyBaudrate:
-            baudrate of underlying serial or usb device
+            baudrate of underlying serial or usb device (Ignored if set via the ``channel`` parameter)
         :param int bitrate:
             CAN Bitrate in bit/s. Value is stored in the adapter and will be used as default if no bitrate is specified
         :param bool rtscts:
             turn hardware handshake (RTS/CTS) on and off
         """
+        if serial is None:
+            raise CanInterfaceNotImplementedError("The serial module is not installed")
 
         if not channel:  # if None or empty
             raise TypeError("Must specify a serial port.")
@@ -74,7 +77,7 @@ def __init__(
             channel, baudrate=ttyBaudrate, rtscts=rtscts
         )
 
-        ## Disable flushing queued config ACKs on lookup channel (for unit tests)
+        # Disable flushing queued config ACKs on lookup channel (for unit tests)
         self._loopback_test = channel == "loop://"
 
         self._rxbuffer = bytearray()  # raw bytes from the serial port
@@ -86,11 +89,10 @@ def __init__(
         if bitrate is not None:
             self.set_bitrate(bitrate)
 
-        self.channel_info = "Robotell USB-CAN s/n %s on %s" % (
-            self.get_serial_number(1),
-            channel,
+        self.channel_info = (
+            f"Robotell USB-CAN s/n {self.get_serial_number(1)} on {channel}"
         )
-        logger.info("Using device: {}".format(self.channel_info))
+        logger.info("Using device: %s", self.channel_info)
 
         super().__init__(channel=channel, **kwargs)
 
@@ -103,9 +105,7 @@ def set_bitrate(self, bitrate):
         if bitrate <= self._MAX_CAN_BAUD:
             self._writeconfig(self._CAN_BAUD_ID, bitrate)
         else:
-            raise ValueError(
-                "Invalid bitrate, must be less than " + str(self._MAX_CAN_BAUD)
-            )
+            raise ValueError(f"Invalid bitrate, must be less than {self._MAX_CAN_BAUD}")
 
     def set_auto_retransmit(self, retrans_flag):
         """
@@ -119,8 +119,8 @@ def set_auto_bus_management(self, auto_man):
         :param bool auto_man:
             Enable/disable automatic bus management
         """
-        ## Not sure what "automatic bus managemenet" does. Does not seem to control
-        ## automatic ACK of CAN frames (listen only mode)
+        # Not sure what "automatic bus management" does. Does not seem to control
+        # automatic ACK of CAN frames (listen only mode)
         self._writeconfig(self._CAN_ABOM_ID, 1 if auto_man else 0)
 
     def set_serial_rate(self, serial_bps):
@@ -161,7 +161,7 @@ def _getconfigsize(self, configid):
             return 4
         if configid == self._CAN_READ_SERIAL1 or configid <= self._CAN_READ_SERIAL2:
             return 8
-        if configid >= self._CAN_FILTER_BASE_ID and configid <= self._CAN_FILTER_MAX_ID:
+        if self._CAN_FILTER_BASE_ID <= configid <= self._CAN_FILTER_MAX_ID:
             return 8
         return 0
 
@@ -178,9 +178,7 @@ def _readconfig(self, configid, timeout):
         newmsg = self._readmessage(not self._loopback_test, True, timeout)
         if newmsg is None:
             logger.warning(
-                "Timeout waiting for response when reading config value {:04X}.".format(
-                    configid
-                )
+                f"Timeout waiting for response when reading config value {configid:04X}."
             )
             return None
         return newmsg[4:12]
@@ -211,8 +209,7 @@ def _writeconfig(self, configid, value, value2=0):
         newmsg = self._readmessage(not self._loopback_test, True, 1)
         if newmsg is None:
             logger.warning(
-                "Timeout waiting for response when writing config value "
-                + str(configid)
+                "Timeout waiting for response when writing config value %d", configid
             )
 
     def _readmessage(self, flushold, cfgchannel, timeout):
@@ -261,17 +258,16 @@ def _readmessage(self, flushold, cfgchannel, timeout):
                         cs = (cs + newmsg[idx]) & 0xFF
                     if newmsg[16] == cs:
                         # OK, valid message - place it in the correct queue
-                        if newmsg[13] == 0xFF:  ## Check for config channel
+                        if newmsg[13] == 0xFF:  # Check for config channel
                             self._configmsg.append(newmsg)
                         else:
                             self._rxmsg.append(newmsg)
                     else:
                         logger.warning("Incorrect message checksum, discarded message")
                 else:
                     logger.warning(
-                        "Invalid message structure length "
-                        + str(len(newmsg))
-                        + ", ignoring message"
+                        "Invalid message structure length %d, ignoring message",
+                        len(newmsg),
                     )
 
             # Check if we have a message in the desired queue - if so copy and return