Clean up socketcan

can/interfaces/socketcan/socketcan.py

@@ -34,6 +34,7 @@
     RestartableCyclicTaskABC,
     LimitedDurationCyclicSendTaskABC,
 )
+from can.typechecking import CanFilters
 from can.interfaces.socketcan.constants import *  # CAN_RAW, CAN_*_FLAG
 from can.interfaces.socketcan.utils import pack_filters, find_available_interfaces
 
@@ -593,20 +594,28 @@ def __init__(
         channel: str = "",
         receive_own_messages: bool = False,
         fd: bool = False,
+        can_filters: Optional[CanFilters] = None,
         **kwargs,
     ) -> None:
-        """
+        """Creates a new socketcan bus.
+
+        If setting some socket options fails, an error will be printed but no exception will be thrown.
+        This includes enabling:
+         - that own messages should be received,
+         - CAN-FD frames and
+         - error frames.
+
         :param channel:
-            The can interface name with which to create this bus. An example channel
-            would be 'vcan0' or 'can0'.
+            The can interface name with which to create this bus.
+            An example channel would be 'vcan0' or 'can0'.
             An empty string '' will receive messages from all channels.
             In that case any sent messages must be explicitly addressed to a
             channel using :attr:`can.Message.channel`.
         :param receive_own_messages:
             If transmitted messages should also be received by this bus.
         :param fd:
             If CAN-FD frames should be supported.
-        :param list can_filters:
+        :param can_filters:
             See :meth:`can.BusABC.set_filters`.
         """
         self.socket = create_socket()
@@ -622,53 +631,56 @@ def __init__(
             self.socket.setsockopt(
                 SOL_CAN_RAW, CAN_RAW_RECV_OWN_MSGS, 1 if receive_own_messages else 0
             )
-        except socket.error as e:
-            log.error("Could not receive own messages (%s)", e)
+        except socket.error as error:
+            log.error("Could not receive own messages (%s)", error)
 
+        # enable CAN-FD frames
         if fd:
-            # TODO handle errors
-            self.socket.setsockopt(SOL_CAN_RAW, CAN_RAW_FD_FRAMES, 1)
+            try:
+                self.socket.setsockopt(SOL_CAN_RAW, CAN_RAW_FD_FRAMES, 1)
+            except socket.error as error:
+                log.error("Could not enable CAN-FD frames (%s)", error)
 
-        # Enable error frames
-        self.socket.setsockopt(SOL_CAN_RAW, CAN_RAW_ERR_FILTER, 0x1FFFFFFF)
+        # enable error frames
+        try:
+            self.socket.setsockopt(SOL_CAN_RAW, CAN_RAW_ERR_FILTER, 0x1FFFFFFF)
+        except socket.error as error:
+            log.error("Could not enable error frames (%s)", error)
 
         bind_socket(self.socket, channel)
         kwargs.update({"receive_own_messages": receive_own_messages, "fd": fd})
-        super().__init__(channel=channel, **kwargs)
+        super().__init__(channel=channel, can_filters=can_filters, **kwargs)
 
     def shutdown(self) -> None:
         """Stops all active periodic tasks and closes the socket."""
         self.stop_all_periodic_tasks()
-        for channel in self._bcm_sockets:
-            log.debug("Closing bcm socket for channel {}".format(channel))
-            bcm_socket = self._bcm_sockets[channel]
+        for channel, bcm_socket in self._bcm_sockets.items():
+            log.debug("Closing bcm socket for channel %s", channel)
             bcm_socket.close()
         log.debug("Closing raw can socket")
         self.socket.close()
 
     def _recv_internal(
         self, timeout: Optional[float]
     ) -> Tuple[Optional[Message], bool]:
-        # get all sockets that are ready (can be a list with a single value
-        # being self.socket or an empty list if self.socket is not ready)
         try:
             # get all sockets that are ready (can be a list with a single value
             # being self.socket or an empty list if self.socket is not ready)
             ready_receive_sockets, _, _ = select.select([self.socket], [], [], timeout)
         except socket.error as exc:
             # something bad happened (e.g. the interface went down)
-            raise can.CanError("Failed to receive: %s" % exc)
+            raise can.CanError(f"Failed to receive: {exc}")
 
-        if ready_receive_sockets:  # not empty or True
+        if ready_receive_sockets:  # not empty
             get_channel = self.channel == ""
             msg = capture_message(self.socket, get_channel)
             if msg and not msg.channel and self.channel:
                 # Default to our own channel
                 msg.channel = self.channel
             return msg, self._is_filtered
-        else:
-            # socket wasn't readable or timeout occurred
-            return None, self._is_filtered
+
+        # socket wasn't readable or timeout occurred
+        return None, self._is_filtered
 
     def send(self, msg: Message, timeout: Optional[float] = None) -> None:
         """Transmit a message to the CAN bus.
@@ -777,13 +789,12 @@ def _get_bcm_socket(self, channel: str) -> socket.socket:
     def _apply_filters(self, filters: Optional[can.typechecking.CanFilters]) -> None:
         try:
             self.socket.setsockopt(SOL_CAN_RAW, CAN_RAW_FILTER, pack_filters(filters))
-        except socket.error as err:
+        except socket.error as error:
             # fall back to "software filtering" (= not in kernel)
             self._is_filtered = False
-            # TODO Is this serious enough to raise a CanError exception?
             log.error(
                 "Setting filters failed; falling back to software filtering (not in kernel): %s",
-                err,
+                error,
             )
         else:
             self._is_filtered = True

can/interfaces/socketcan/utils.py

@@ -2,17 +2,16 @@
 Defines common socketcan functions.
 """
 
-from typing import cast, Iterable, Optional
-import can.typechecking as typechecking
-
+import errno
 import logging
 import os
-import errno
+import re
 import struct
 import subprocess
-import re
+from typing import cast, Iterable, Optional
 
 from can.interfaces.socketcan.constants import CAN_EFF_FLAG
+import can.typechecking as typechecking
 
 log = logging.getLogger(__name__)
 
@@ -46,46 +45,38 @@ def find_available_interfaces() -> Iterable[str]:
     """Returns the names of all open can/vcan interfaces using
     the ``ip link list`` command. If the lookup fails, an error
     is logged to the console and an empty list is returned.
-
-    :rtype: an iterable of :class:`str`
     """
 
     try:
-        # it might be good to add "type vcan", but that might (?) exclude physical can devices
+        # adding "type vcan" would exclude physical can devices
         command = ["ip", "-o", "link", "list", "up"]
         output = subprocess.check_output(command, universal_newlines=True)
 
-    except Exception as e:  # subprocess.CalledProcessError was too specific
+    except Exception as e:  # subprocess.CalledProcessError is too specific
         log.error("failed to fetch opened can devices: %s", e)
         return []
 
     else:
         # log.debug("find_available_interfaces(): output=\n%s", output)
         # output contains some lines like "1: vcan42: <NOARP,UP,LOWER_UP> ..."
         # extract the "vcan42" of each line
-        interface_names = [line.split(": ", 3)[1] for line in output.splitlines()]
-        log.debug("find_available_interfaces(): detected: %s", interface_names)
-        return filter(_PATTERN_CAN_INTERFACE.match, interface_names)
+        interfaces = [line.split(": ", 3)[1] for line in output.splitlines()]
+        log.debug(
+            "find_available_interfaces(): detected these interfaces (before filtering): %s",
+            interfaces,
+        )
+        return filter(_PATTERN_CAN_INTERFACE.match, interfaces)
 
 
-def error_code_to_str(code: int) -> str:
+def error_code_to_str(code: Optional[int]) -> str:
     """
     Converts a given error code (errno) to a useful and human readable string.
 
-    :param int code: a possibly invalid/unknown error code
-    :rtype: str
+    :param code: a possibly invalid/unknown error code
     :returns: a string explaining and containing the given error code, or a string
               explaining that the errorcode is unknown if that is the case
     """
+    name = errno.errorcode.get(code, "UNKNOWN")  # type: ignore
+    description = os.strerror(code) if code is not None else "NO DESCRIPTION AVAILABLE"
 
-    try:
-        name = errno.errorcode[code]
-    except KeyError:
-        name = "UNKNOWN"
-
-    try:
-        description = os.strerror(code)
-    except ValueError:
-        description = "no description available"
-
-    return "{} (errno {}): {}".format(name, code, description)
+    return f"{name} (errno {code}): {description}"

can/io/blf.py

@@ -286,9 +286,17 @@ def _parse_data(self, data):
                 )
             elif obj_type == CAN_FD_MESSAGE:
                 members = unpack_can_fd_msg(data, pos)
-                channel, flags, dlc, can_id, _, _, fd_flags, valid_bytes, can_data = (
-                    members
-                )
+                (
+                    channel,
+                    flags,
+                    dlc,
+                    can_id,
+                    _,
+                    _,
+                    fd_flags,
+                    valid_bytes,
+                    can_data,
+                ) = members
                 yield Message(
                     timestamp=timestamp,
                     arbitration_id=can_id & 0x1FFFFFFF,

test/test_socketcan_helpers.py

@@ -9,7 +9,7 @@
 
 from can.interfaces.socketcan.utils import find_available_interfaces, error_code_to_str
 
-from .config import *
+from .config import IS_LINUX, TEST_INTERFACE_SOCKETCAN
 
 
 class TestSocketCanHelpers(unittest.TestCase):
@@ -21,7 +21,7 @@ def test_error_code_to_str(self):
         """
 
         # all possible & also some invalid error codes
-        test_data = list(range(0, 256)) + [-1, 256, 5235, 346264]
+        test_data = list(range(0, 256)) + [-1, 256, 5235, 346264, None]
 
         for error_code in test_data:
             string = error_code_to_str(error_code)