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

Author: MattWoodhead

.mergify.yml

@@ -1,7 +1,8 @@
 queue_rules:
   - name: default
     conditions:
-      - "status-success=test,format"  # "GitHub Actions works slightly differently [...], only the job name is used."
+      - "status-success=test"  # "GitHub Actions works slightly differently [...], only the job name is used."
+      - "status-success=format"
 
 pull_request_rules:
   - name: Automatic merge passing PR on up to date branch with approving CR
@@ -10,7 +11,8 @@ pull_request_rules:
       - "#approved-reviews-by>=1"
       - "#review-requested=0"
       - "#changes-requested-reviews-by=0"
-      - "status-success=test,format"
+      - "status-success=test"
+      - "status-success=format"
       - "label!=work-in-progress"
     actions:
       queue:

can/__init__.py

@@ -12,7 +12,7 @@
 
 log = logging.getLogger("can")
 
-rc: Dict[str, Any] = dict()
+rc: Dict[str, Any] = {}
 
 from .listener import Listener, BufferedReader, RedirectReader, AsyncBufferedReader
 

can/exceptions.py

@@ -15,7 +15,11 @@
 :class:`ValueError`. This should always be documented for the function at hand.
 """
 
+
+from contextlib import contextmanager
+
 from typing import Optional
+from typing import Type
 
 
 class CanError(Exception):
@@ -96,3 +100,18 @@ class CanTimeoutError(CanError, TimeoutError):
       - Some message could not be sent after the timeout elapsed
       - No message was read within the given time
     """
+
+
+@contextmanager
+def error_check(
+    error_message: Optional[str] = None,
+    exception_type: Type[CanError] = CanOperationError,
+) -> None:
+    """Catches any exceptions and turns them into the new type while preserving the stack trace."""
+    try:
+        yield
+    except Exception as error:
+        if error_message is None:
+            raise exception_type(str(error)) from error
+        else:
+            raise exception_type(error_message) from error

can/interfaces/canalystii.py

@@ -53,7 +53,7 @@ def __init__(
         elif isinstance(channel, int):
             self.channels = [channel]
         else:  # Sequence[int]
-            self.channels = [c for c in channel]
+            self.channels = list(channel)
 
         self.rx_queue = collections.deque(
             maxlen=rx_queue_size

can/interfaces/cantact.py

@@ -7,14 +7,20 @@
 from unittest.mock import Mock
 
 from can import BusABC, Message
+from ..exceptions import (
+    CanInitializationError,
+    CanInterfaceNotImplementedError,
+    error_check,
+)
 
 logger = logging.getLogger(__name__)
 
 try:
     import cantact
 except ImportError:
+    cantact = None
     logger.warning(
-        "The CANtact module is not installed. Install it using `python3 -m pip install cantact`"
+        "The CANtact module is not installed. Install it using `python -m pip install cantact`"
     )
 
 
@@ -25,13 +31,15 @@ class CantactBus(BusABC):
     def _detect_available_configs():
         try:
             interface = cantact.Interface()
-        except (NameError, SystemError):
-            # couldn't import cantact, so no configurations are available
+        except (NameError, SystemError, AttributeError):
+            logger.debug(
+                "Could not import or instantiate cantact, so no configurations are available"
+            )
             return []
 
         channels = []
         for i in range(0, interface.channel_count()):
-            channels.append({"interface": "cantact", "channel": "ch:%d" % i})
+            channels.append({"interface": "cantact", "channel": f"ch:{i}"})
         return channels
 
     def __init__(
@@ -42,7 +50,7 @@ def __init__(
         monitor=False,
         bit_timing=None,
         _testing=False,
-        **kwargs
+        **kwargs,
     ):
         """
         :param int channel:
@@ -58,36 +66,45 @@ def __init__(
         if _testing:
             self.interface = MockInterface()
         else:
-            self.interface = cantact.Interface()
+            if cantact is None:
+                raise CanInterfaceNotImplementedError(
+                    "The CANtact module is not installed. Install it using `python -m pip install cantact`"
+                )
+            with error_check(
+                "Cannot create the cantact.Interface", CanInitializationError
+            ):
+                self.interface = cantact.Interface()
 
         self.channel = int(channel)
-        self.channel_info = "CANtact: ch:%s" % channel
-
-        # configure the interface
-        if bit_timing is None:
-            # use bitrate
-            self.interface.set_bitrate(int(channel), int(bitrate))
-        else:
-            # use custom bit timing
-            self.interface.set_bit_timing(
-                int(channel),
-                int(bit_timing.brp),
-                int(bit_timing.tseg1),
-                int(bit_timing.tseg2),
-                int(bit_timing.sjw),
-            )
-        self.interface.set_enabled(int(channel), True)
-        self.interface.set_monitor(int(channel), monitor)
-        self.interface.start()
+        self.channel_info = f"CANtact: ch:{channel}"
+
+        # Configure the interface
+        with error_check("Cannot setup the cantact.Interface", CanInitializationError):
+            if bit_timing is None:
+                # use bitrate
+                self.interface.set_bitrate(int(channel), int(bitrate))
+            else:
+                # use custom bit timing
+                self.interface.set_bit_timing(
+                    int(channel),
+                    int(bit_timing.brp),
+                    int(bit_timing.tseg1),
+                    int(bit_timing.tseg2),
+                    int(bit_timing.sjw),
+                )
+            self.interface.set_enabled(int(channel), True)
+            self.interface.set_monitor(int(channel), monitor)
+            self.interface.start()
 
         super().__init__(
             channel=channel, bitrate=bitrate, poll_interval=poll_interval, **kwargs
         )
 
     def _recv_internal(self, timeout):
-        frame = self.interface.recv(int(timeout * 1000))
+        with error_check("Cannot receive message"):
+            frame = self.interface.recv(int(timeout * 1000))
         if frame is None:
-            # timeout occured
+            # timeout occurred
             return None, False
 
         msg = Message(
@@ -103,31 +120,33 @@ def _recv_internal(self, timeout):
         return msg, False
 
     def send(self, msg, timeout=None):
-        self.interface.send(
-            self.channel,
-            msg.arbitration_id,
-            bool(msg.is_extended_id),
-            bool(msg.is_remote_frame),
-            msg.dlc,
-            msg.data,
-        )
+        with error_check("Cannot send message"):
+            self.interface.send(
+                self.channel,
+                msg.arbitration_id,
+                bool(msg.is_extended_id),
+                bool(msg.is_remote_frame),
+                msg.dlc,
+                msg.data,
+            )
 
     def shutdown(self):
-        self.interface.stop()
+        with error_check("Cannot shutdown interface"):
+            self.interface.stop()
 
 
 def mock_recv(timeout):
     if timeout > 0:
-        frame = {}
-        frame["id"] = 0x123
-        frame["extended"] = False
-        frame["timestamp"] = time.time()
-        frame["loopback"] = False
-        frame["rtr"] = False
-        frame["dlc"] = 8
-        frame["data"] = [1, 2, 3, 4, 5, 6, 7, 8]
-        frame["channel"] = 0
-        return frame
+        return {
+            "id": 0x123,
+            "extended": False,
+            "timestamp": time.time(),
+            "loopback": False,
+            "rtr": False,
+            "dlc": 8,
+            "data": [1, 2, 3, 4, 5, 6, 7, 8],
+            "channel": 0,
+        }
     else:
         # simulate timeout when timeout = 0
         return None
@@ -144,7 +163,6 @@ class MockInterface:
     set_bit_timing = Mock()
     set_enabled = Mock()
     set_monitor = Mock()
-    start = Mock()
     stop = Mock()
     send = Mock()
     channel_count = Mock(return_value=1)

can/interfaces/ics_neovi/neovi_bus.py

@@ -226,8 +226,8 @@ def channel_to_netid(channel_name_or_id):
                 channel = getattr(ics, netid)
             else:
                 raise ValueError(
-                    "channel must be an integer or " "a valid ICS channel name"
-                )
+                    "channel must be an integer or a valid ICS channel name"
+                ) from None
         return channel
 
     @staticmethod

can/interfaces/ixxat/canlib.py

@@ -604,12 +604,12 @@ def _inWaiting(self):
             return 1
 
     def flush_tx_buffer(self):
-        """ Flushes the transmit buffer on the IXXAT """
+        """Flushes the transmit buffer on the IXXAT"""
         # TODO #64: no timeout?
         _canlib.canChannelWaitTxEvent(self._channel_handle, constants.INFINITE)
 
     def _recv_internal(self, timeout):
-        """ Read a message from IXXAT device. """
+        """Read a message from IXXAT device."""
 
         # TODO: handling CAN error messages?
         data_received = False

can/interfaces/ixxat/exceptions.py

@@ -21,15 +21,15 @@
 
 
 class VCITimeout(CanTimeoutError):
-    """ Wraps the VCI_E_TIMEOUT error """
+    """Wraps the VCI_E_TIMEOUT error"""
 
 
 class VCIError(CanOperationError):
-    """ Try to display errors that occur within the wrapped C library nicely. """
+    """Try to display errors that occur within the wrapped C library nicely."""
 
 
 class VCIRxQueueEmptyError(VCIError):
-    """ Wraps the VCI_E_RXQUEUE_EMPTY error """
+    """Wraps the VCI_E_RXQUEUE_EMPTY error"""
 
     def __init__(self):
         super().__init__("Receive queue is empty")

can/interfaces/kvaser/canlib.py

@@ -713,11 +713,7 @@ def get_channel_info(channel):
         ctypes.sizeof(number),
     )
 
-    return "%s, S/N %d (#%d)" % (
-        name.value.decode("ascii"),
-        serial.value,
-        number.value + 1,
-    )
+    return f"{name.value.decode('ascii')}, S/N {serial.value} (#{number.value + 1})"
 
 
 init_kvaser_library()

can/interfaces/neousys/neousys.py

@@ -47,7 +47,7 @@
 
 
 class NeousysCanSetup(Structure):
-    """ C CAN Setup struct """
+    """C CAN Setup struct"""
 
     _fields_ = [
         ("bitRate", c_uint),
@@ -58,7 +58,7 @@ class NeousysCanSetup(Structure):
 
 
 class NeousysCanMsg(Structure):
-    """ C CAN Message struct """
+    """C CAN Message struct"""
 
     _fields_ = [
         ("id", c_uint),
@@ -75,7 +75,7 @@ class NeousysCanMsg(Structure):
 # 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 """
+    """C CAN BIT Clock struct"""
 
     _fields_ = [
         ("syncPropPhase1Seg", c_ushort),

can/interfaces/nixnet.py

@@ -30,7 +30,7 @@
         )
     except ImportError:
         logger.error("Error, NIXNET python module cannot be loaded.")
-        raise ImportError()
+        raise
 else:
     logger.error("NI-XNET interface is only available on Windows systems")
     raise NotImplementedError("NiXNET is not supported on not Win32 platforms")

can/interfaces/seeedstudio/seeedstudio.py

@@ -113,7 +113,7 @@ def __init__(
                 "could not create the serial device"
             ) from error
 
-        super(SeeedBus, self).__init__(channel=channel, *args, **kwargs)
+        super().__init__(channel=channel, *args, **kwargs)
         self.init_frame()
 
     def shutdown(self):