Extend ruff linter configuration (#1724)

* update ruff configuration

* skip test_send_periodic_duration on PyPy

* remove format-code.yml

Author: zariiii9003

.github/workflows/format-code.yml

@@ -281,7 +281,7 @@ def wrapped_stop_method(remove_task: bool = True) -> None:
                     pass  # allow the task to be already removed
             original_stop_method()
 
-        task.stop = wrapped_stop_method  # type: ignore
+        task.stop = wrapped_stop_method  # type: ignore[method-assign]
 
         if store_task:
             self._periodic_tasks.append(task)
@@ -401,7 +401,8 @@ def set_filters(
             messages based only on the arbitration ID and mask.
         """
         self._filters = filters or None
-        self._apply_filters(self._filters)
+        with contextlib.suppress(NotImplementedError):
+            self._apply_filters(self._filters)
 
     def _apply_filters(self, filters: Optional[can.typechecking.CanFilters]) -> None:
         """
@@ -411,6 +412,7 @@ def _apply_filters(self, filters: Optional[can.typechecking.CanFilters]) -> None
         :param filters:
             See :meth:`~can.BusABC.set_filters` for details.
         """
+        raise NotImplementedError
 
     def _matches_filters(self, msg: Message) -> bool:
         """Checks whether the given message matches at least one of the
@@ -450,6 +452,7 @@ def _matches_filters(self, msg: Message) -> bool:
 
     def flush_tx_buffer(self) -> None:
         """Discard every message that may be queued in the output buffer(s)."""
+        raise NotImplementedError
 
     def shutdown(self) -> None:
         """

can/bus.py

@@ -1,4 +1,3 @@
-# type: ignore
 """
 This module contains common `ctypes` utils.
 """
@@ -11,11 +10,10 @@
 
 __all__ = ["CLibrary", "HANDLE", "PHANDLE", "HRESULT"]
 
-
-try:
+if sys.platform == "win32":
     _LibBase = ctypes.WinDLL
     _FUNCTION_TYPE = ctypes.WINFUNCTYPE
-except AttributeError:
+else:
     _LibBase = ctypes.CDLL
     _FUNCTION_TYPE = ctypes.CFUNCTYPE
 
@@ -60,7 +58,7 @@ def map_symbol(
                 f'Could not map function "{func_name}" from library {self._name}'
             ) from None
 
-        func._name = func_name  # pylint: disable=protected-access
+        func._name = func_name  # type: ignore[attr-defined] # pylint: disable=protected-access
         log.debug(
             'Wrapped function "%s", result type: %s, error_check %s',
             func_name,

can/ctypesutil.py

@@ -61,7 +61,7 @@ def _get_class_for_interface(interface: str) -> Type[BusABC]:
     bustype="interface",
     context="config_context",
 )
-def Bus(
+def Bus(  # noqa: N802
     channel: Optional[Channel] = None,
     interface: Optional[str] = None,
     config_context: Optional[str] = None,

can/interface.py

@@ -229,10 +229,10 @@ def _apply_filters(self, filters: Optional[can.typechecking.CanFilters]) -> None
 
         self._oci_filters = (ctypes.POINTER(OCI_CANRxFilterEx) * len(filters))()
 
-        for i, filter in enumerate(filters):
+        for i, filter_ in enumerate(filters):
             f = OCI_CANRxFilterEx()
-            f.frameIDValue = filter["can_id"]
-            f.frameIDMask = filter["can_mask"]
+            f.frameIDValue = filter_["can_id"]
+            f.frameIDMask = filter_["can_mask"]
             f.tag = 0
             f.flagsValue = 0
             if self.receive_own_messages:
@@ -241,7 +241,7 @@ def _apply_filters(self, filters: Optional[can.typechecking.CanFilters]) -> None
             else:
                 # enable the SR bit in the mask. since the bit is 0 in flagsValue -> do not self-receive
                 f.flagsMask = OCI_CAN_MSG_FLAG_SELFRECEPTION
-            if filter.get("extended"):
+            if filter_.get("extended"):
                 f.flagsValue |= OCI_CAN_MSG_FLAG_EXTENDED
                 f.flagsMask |= OCI_CAN_MSG_FLAG_EXTENDED
             self._oci_filters[i].contents = f

can/interfaces/etas/__init__.py

@@ -95,8 +95,8 @@ def send(self, msg: can.Message, timeout: Optional[float] = None):
 
         try:
             self.gs_usb.send(frame)
-        except usb.core.USBError:
-            raise CanOperationError("The message could not be sent")
+        except usb.core.USBError as exc:
+            raise CanOperationError("The message could not be sent") from exc
 
     def _recv_internal(
         self, timeout: Optional[float]

can/interfaces/gs_usb.py

@@ -109,8 +109,10 @@ def __reduce__(self):
     def error_number(self) -> int:
         """Deprecated. Renamed to :attr:`can.CanError.error_code`."""
         warn(
-            "ICSApiError::error_number has been renamed to error_code defined by CanError",
+            "ICSApiError::error_number has been replaced by ICSApiError.error_code in python-can 4.0"
+            "and will be remove in version 5.0.",
             DeprecationWarning,
+            stacklevel=2,
         )
         return self.error_code
 
@@ -223,10 +225,8 @@ def __init__(self, channel, can_filters=None, **kwargs):
         self._use_system_timestamp = bool(kwargs.get("use_system_timestamp", False))
         self._receive_own_messages = kwargs.get("receive_own_messages", True)
 
-        self.channel_info = "{} {} CH:{}".format(
-            self.dev.Name,
-            self.get_serial_number(self.dev),
-            self.channels,
+        self.channel_info = (
+            f"{self.dev.Name} {self.get_serial_number(self.dev)} CH:{self.channels}"
         )
         logger.info(f"Using device: {self.channel_info}")
 

can/interfaces/ics_neovi/neovi_bus.py

@@ -79,8 +79,8 @@ def __vciFormatErrorExtended(
         Formatted string
     """
     # TODO: make sure we don't generate another exception
-    return "{} - arguments were {}".format(
-        __vciFormatError(library_instance, function, vret), args
+    return (
+        f"{__vciFormatError(library_instance, function, vret)} - arguments were {args}"
     )
 
 
@@ -512,13 +512,11 @@ def __init__(
                 if unique_hardware_id is None:
                     raise VCIDeviceNotFoundError(
                         "No IXXAT device(s) connected or device(s) in use by other process(es)."
-                    )
+                    ) from None
                 else:
                     raise VCIDeviceNotFoundError(
-                        "Unique HW ID {} not connected or not available.".format(
-                            unique_hardware_id
-                        )
-                    )
+                        f"Unique HW ID {unique_hardware_id} not connected or not available."
+                    ) from None
             else:
                 if (unique_hardware_id is None) or (
                     self._device_info.UniqueHardwareId.AsChar
@@ -815,7 +813,8 @@ def _send_periodic_internal(
         # fallback to thread based cyclic task
         warnings.warn(
             f"{self.__class__.__name__} falls back to a thread-based cyclic task, "
-            "when the `modifier_callback` argument is given."
+            "when the `modifier_callback` argument is given.",
+            stacklevel=3,
         )
         return BusABC._send_periodic_internal(
             self,

can/interfaces/ixxat/canlib_vcinpl.py

@@ -77,8 +77,8 @@ def __vciFormatErrorExtended(
         Formatted string
     """
     # TODO: make sure we don't generate another exception
-    return "{} - arguments were {}".format(
-        __vciFormatError(library_instance, function, vret), args
+    return (
+        f"{__vciFormatError(library_instance, function, vret)} - arguments were {args}"
     )
 
 
@@ -553,13 +553,11 @@ def __init__(
                 if unique_hardware_id is None:
                     raise VCIDeviceNotFoundError(
                         "No IXXAT device(s) connected or device(s) in use by other process(es)."
-                    )
+                    ) from None
                 else:
                     raise VCIDeviceNotFoundError(
-                        "Unique HW ID {} not connected or not available.".format(
-                            unique_hardware_id
-                        )
-                    )
+                        f"Unique HW ID {unique_hardware_id} not connected or not available."
+                    ) from None
             else:
                 if (unique_hardware_id is None) or (
                     self._device_info.UniqueHardwareId.AsChar
@@ -579,7 +577,9 @@ def __init__(
                 ctypes.byref(self._device_handle),
             )
         except Exception as exception:
-            raise CanInitializationError(f"Could not open device: {exception}")
+            raise CanInitializationError(
+                f"Could not open device: {exception}"
+            ) from exception
 
         log.info("Using unique HW ID %s", self._device_info.UniqueHardwareId.AsChar)
 
@@ -600,7 +600,7 @@ def __init__(
         except Exception as exception:
             raise CanInitializationError(
                 f"Could not open and initialize channel: {exception}"
-            )
+            ) from exception
 
         # Signal TX/RX events when at least one frame has been handled
         _canlib.canChannelInitialize(
@@ -959,7 +959,8 @@ def _send_periodic_internal(
         # fallback to thread based cyclic task
         warnings.warn(
             f"{self.__class__.__name__} falls back to a thread-based cyclic task, "
-            "when the `modifier_callback` argument is given."
+            "when the `modifier_callback` argument is given.",
+            stacklevel=3,
         )
         return BusABC._send_periodic_internal(
             self,

can/interfaces/ixxat/canlib_vcinpl2.py

@@ -458,7 +458,7 @@ def __init__(
         try:
             channel = int(channel)
         except ValueError:
-            raise ValueError("channel must be an integer")
+            raise ValueError("channel must be an integer") from None
 
         self.channel = channel
         self.single_handle = single_handle

can/interfaces/kvaser/canlib.py

@@ -192,10 +192,12 @@ def __init__(
 
     @property
     def fd(self) -> bool:
+        class_name = self.__class__.__name__
         warnings.warn(
-            "The NiXNETcanBus.fd property is deprecated and superseded by "
-            "BusABC.protocol. It is scheduled for removal in version 5.0.",
+            f"The {class_name}.fd property is deprecated and superseded by "
+            f"{class_name}.protocol. It is scheduled for removal in python-can version 5.0.",
             DeprecationWarning,
+            stacklevel=2,
         )
         return self._can_protocol is CanProtocol.CAN_FD
 

can/interfaces/nixnet.py

@@ -694,9 +694,9 @@ def Initialize(
         self,
         Channel,
         Btr0Btr1,
-        HwType=TPCANType(0),
-        IOPort=c_uint(0),
-        Interrupt=c_ushort(0),
+        HwType=TPCANType(0),  # noqa: B008
+        IOPort=c_uint(0),  # noqa: B008
+        Interrupt=c_ushort(0),  # noqa: B008
     ):
         """Initializes a PCAN Channel
 

can/interfaces/pcan/basic.py

@@ -657,10 +657,12 @@ def shutdown(self):
 
     @property
     def fd(self) -> bool:
+        class_name = self.__class__.__name__
         warnings.warn(
-            "The PcanBus.fd property is deprecated and superseded by BusABC.protocol. "
-            "It is scheduled for removal in version 5.0.",
+            f"The {class_name}.fd property is deprecated and superseded by {class_name}.protocol. "
+            "It is scheduled for removal in python-can version 5.0.",
             DeprecationWarning,
+            stacklevel=2,
         )
         return self._can_protocol is CanProtocol.CAN_FD
 

can/interfaces/pcan/pcan.py

@@ -376,7 +376,7 @@ def fileno(self):
         except io.UnsupportedOperation:
             raise NotImplementedError(
                 "fileno is not implemented using current CAN bus on this platform"
-            )
+            ) from None
         except Exception as exception:
             raise CanOperationError("Cannot fetch fileno") from exception
 

can/interfaces/robotell.py

@@ -63,7 +63,6 @@ def __init__(
         frame_type="STD",
         operation_mode="normal",
         bitrate=500000,
-        *args,
         **kwargs,
     ):
         """
@@ -115,7 +114,7 @@ def __init__(
                 "could not create the serial device"
             ) from error
 
-        super().__init__(channel=channel, *args, **kwargs)
+        super().__init__(channel=channel, **kwargs)
         self.init_frame()
 
     def shutdown(self):

can/interfaces/seeedstudio/seeedstudio.py

@@ -131,13 +131,15 @@ def send(self, msg: Message, timeout: Optional[float] = None) -> None:
         try:
             timestamp = struct.pack("<I", int(msg.timestamp * 1000))
         except struct.error:
-            raise ValueError("Timestamp is out of range")
+            raise ValueError(f"Timestamp is out of range: {msg.timestamp}") from None
 
         # Pack arbitration ID
         try:
             arbitration_id = struct.pack("<I", msg.arbitration_id)
         except struct.error:
-            raise ValueError("Arbitration ID is out of range")
+            raise ValueError(
+                f"Arbitration ID is out of range: {msg.arbitration_id}"
+            ) from None
 
         # Assemble message
         byte_msg = bytearray()
@@ -222,7 +224,7 @@ def fileno(self) -> int:
         except io.UnsupportedOperation:
             raise NotImplementedError(
                 "fileno is not implemented using current CAN bus on this platform"
-            )
+            ) from None
         except Exception as exception:
             raise CanOperationError("Cannot fetch fileno") from exception
 

can/interfaces/serial/serial_can.py

@@ -268,7 +268,7 @@ def fileno(self) -> int:
         except io.UnsupportedOperation:
             raise NotImplementedError(
                 "fileno is not implemented using current CAN bus on this platform"
-            )
+            ) from None
         except Exception as exception:
             raise CanOperationError("Cannot fetch fileno") from exception