Add hooks for CAN errors

Author: tomasjakubik

pycyphal/transport/can/_can.py

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 import copy
+import enum
 import typing
 import asyncio
 import logging
@@ -15,6 +16,7 @@
 from ._session import CANInputSession, CANOutputSession, SendTransaction
 from ._session import BroadcastCANOutputSession, UnicastCANOutputSession
 from ._frame import CyphalFrame, TRANSFER_ID_MODULO
+from .media import Media
 from ._identifier import CANID, generate_filter_configurations
 from ._input_dispatch_table import InputDispatchTable
 from ._tracer import CANTracer, CANCapture
@@ -84,6 +86,14 @@ class CANTransport(pycyphal.transport.Transport):
 
     TRANSFER_ID_MODULO = TRANSFER_ID_MODULO
 
+    class Error(enum.Enum):
+        """Transport-specific error codes."""
+
+        SEND_TIMEOUT = enum.auto()  # Did not send within the specified deadline
+
+    ErrorHandler = typing.Callable[[Timestamp, Error | Media.Error], None]
+    """The error handler is non-blocking and non-yielding; returns immediately."""
+
     def __init__(
         self,
         media: Media,
@@ -115,6 +125,8 @@ def __init__(
 
         self._frame_stats = CANTransportStatistics()
 
+        self._error_hooks: typing.List[CANTransport.ErrorHandler] = []
+
         if self._local_node_id is not None and not 0 <= self._local_node_id <= CANID.NODE_ID_MASK:
             raise ValueError(f"Invalid node ID for CAN: {self._local_node_id}")
 
@@ -130,7 +142,16 @@ def __init__(
                 f"The number of acceptance filters is too low: {media.number_of_acceptance_filters}"
             )
 
-        media.start(self._on_frames_received, no_automatic_retransmission=self._local_node_id is None)
+        media.start(self._on_frames_received, self._on_error, no_automatic_retransmission=self._local_node_id is None)
+
+    def add_error_hook(self, hook: CANTransport.ErrorHandler) -> None:
+        """Register an error hook. Called on transport or media error."""
+        self._error_hooks.append(hook)
+
+    def _on_error(self, timestamp: Timestamp, error: CANTransport.Error | Media.Error) -> None:
+        """Call all registered hooks on error in media or transport layer."""
+        for hook in self._error_hooks:
+            hook(timestamp, error)
 
     @property
     def protocol_parameters(self) -> pycyphal.transport.ProtocolParameters:
@@ -157,6 +178,8 @@ def output_sessions(self) -> typing.Sequence[CANOutputSession]:
         return list(self._output_registry.values())
 
     def close(self) -> None:
+        self._error_hooks.clear()
+
         for s in (*self.input_sessions, *self.output_sessions):
             try:
                 s.close()
@@ -373,6 +396,7 @@ async def _do_send(self, t: SendTransaction) -> bool:
                 len(t.frames),
                 can_id_int,
             )
+            self._on_error(Timestamp.now(), CANTransport.Error.SEND_TIMEOUT)
 
         return not unsent_frames
 

pycyphal/transport/can/media/_media.py

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 import abc
+import enum
 import typing
 import asyncio
 import warnings
@@ -28,6 +29,21 @@ class Media(abc.ABC):
     The timestamp is provided individually per frame.
     """
 
+    class Error(enum.Enum):
+        """Media-specific error codes."""
+
+        CAN_TX_TIMEOUT = enum.auto()  # A transmission request timed out
+        CAN_BUS_OFF = enum.auto()  # The CAN controller entered the bus-off state
+        CAN_RX_OVERFLOW = enum.auto()  # Overflow in the CAN controller
+        CAN_TX_OVERFLOW = enum.auto()  # Overflow in the CAN controller
+        CAN_RX_WARNING = enum.auto()  # The CAN controller issued a warning
+        CAN_TX_WARNING = enum.auto()  # The CAN controller issued a warning
+        CAN_TX_PASSIVE = enum.auto()  # The CAN controller entered the error passive state
+        CAN_RX_PASSIVE = enum.auto()  # The CAN controller entered the error passive state
+
+    ErrorHandler = typing.Callable[[Timestamp, Error], None]
+    """The error handler is non-blocking and non-yielding; returns immediately."""
+
     VALID_MTU_SET = {8, 12, 16, 20, 24, 32, 48, 64}
     """Valid MTU values for Classic CAN and CAN FD."""
 
@@ -73,7 +89,12 @@ def number_of_acceptance_filters(self) -> int:
         raise NotImplementedError
 
     @abc.abstractmethod
-    def start(self, handler: ReceivedFramesHandler, no_automatic_retransmission: bool) -> None:
+    def start(
+        self,
+        handler: ReceivedFramesHandler,
+        error_handler: typing.Optional[ErrorHandler],
+        no_automatic_retransmission: bool,
+    ) -> None:
         """
         Every received frame shall be timestamped. Both monotonic and system timestamps are required.
         There are no timestamping accuracy requirements. An empty set of frames should never be reported.
@@ -97,6 +118,9 @@ def start(self, handler: ReceivedFramesHandler, no_automatic_retransmission: boo
 
         :param handler: Behold my transformation. You are empowered to do as you please.
 
+        :param error_handler: Informs about media errors. This feature is optional in both directions.
+            Ignore if not implemented. Set to None if error reporting is not needed by the transport.
+
         :param no_automatic_retransmission: If True, the CAN controller should be configured to abort transmission
             of CAN frames after first error or arbitration loss (time-triggered transmission mode).
             This mode is used by Cyphal to facilitate the PnP node-ID allocation process on the client side.

pycyphal/transport/can/media/candump/_candump.py

@@ -117,7 +117,12 @@ def mtu(self) -> int:
     def number_of_acceptance_filters(self) -> int:
         return 1
 
-    def start(self, handler: Media.ReceivedFramesHandler, no_automatic_retransmission: bool) -> None:
+    def start(
+        self,
+        handler: Media.ReceivedFramesHandler,
+        error_handler: None | Media.ErrorHandler,
+        no_automatic_retransmission: bool,
+    ) -> None:
         _ = no_automatic_retransmission
         if self._thread is not None:
             raise RuntimeError(f"{self!r}: Already started")

pycyphal/transport/can/media/pythoncan/_pythoncan.py

@@ -245,7 +245,12 @@ def is_fd(self) -> bool:
         """
         return self._is_fd
 
-    def start(self, handler: Media.ReceivedFramesHandler, no_automatic_retransmission: bool) -> None:
+    def start(
+        self,
+        handler: Media.ReceivedFramesHandler,
+        error_handler: typing.Optional[Media.ErrorHandler],
+        no_automatic_retransmission: bool,
+    ) -> None:
         self._tx_thread.start()
         if self._maybe_thread is None:
             self._rx_handler = handler

pycyphal/transport/can/media/socketcan/_socketcan.py

@@ -3,6 +3,7 @@
 # Author: Pavel Kirienko <pavel@opencyphal.org>
 # pylint: disable=duplicate-code
 
+from dataclasses import dataclass
 import enum
 import time
 import errno
@@ -29,6 +30,14 @@
 _logger = logging.getLogger(__name__)
 
 
+@dataclass
+class _TimestampedErrorList:
+    """Collection of media errors with a single timestamp. Used as a helper for typing."""
+
+    timestamp: Timestamp
+    errors: typing.List[Media.Error]
+
+
 class SocketCANMedia(Media):
     """
     This media implementation provides a simple interface for the standard Linux SocketCAN media layer.
@@ -125,17 +134,29 @@ def number_of_acceptance_filters(self) -> int:
         """
         return 512
 
-    def start(self, handler: Media.ReceivedFramesHandler, no_automatic_retransmission: bool) -> None:
+    def start(
+        self,
+        handler: Media.ReceivedFramesHandler,
+        error_handler: typing.Optional[Media.ErrorHandler],
+        no_automatic_retransmission: bool,
+    ) -> None:
         if self._maybe_thread is None:
             self._maybe_thread = threading.Thread(
-                target=self._thread_function, name=str(self), args=(handler, asyncio.get_event_loop()), daemon=True
+                target=self._thread_function,
+                name=str(self),
+                args=(handler, error_handler, asyncio.get_event_loop()),
+                daemon=True,
             )
             self._maybe_thread.start()
             if no_automatic_retransmission:
                 _logger.info("%s non-automatic retransmission is not supported", self)
         else:
             raise RuntimeError("The RX frame handler is already set up")
 
+        if error_handler is not None:
+            err_mask = _CAN_ERR_TX_TIMEOUT | _CAN_ERR_CRTL | _CAN_ERR_BUSOFF
+            self._sock.setsockopt(socket.SOL_CAN_RAW, _CAN_RAW_ERR_FILTER, err_mask)
+
     def configure_acceptance_filters(self, configuration: typing.Sequence[FilterConfiguration]) -> None:
         if self._closed:
             raise pycyphal.transport.ResourceClosedError(repr(self))
@@ -194,14 +215,30 @@ def close(self) -> None:
             self._ctl_worker.close()
             self._ctl_main.close()
 
-    def _thread_function(self, handler: Media.ReceivedFramesHandler, loop: asyncio.AbstractEventLoop) -> None:
+    def _thread_function(
+        self,
+        handler: Media.ReceivedFramesHandler,
+        error_handler: typing.Optional[Media.ErrorHandler],
+        loop: asyncio.AbstractEventLoop,
+    ) -> None:
         def handler_wrapper(frs: typing.Sequence[typing.Tuple[Timestamp, Envelope]]) -> None:
             try:
                 if not self._closed:  # Don't call after closure to prevent race conditions and use-after-close.
                     handler(frs)
             except Exception as exc:
                 _logger.exception("%s: Unhandled exception in the receive handler: %s; lost frames: %s", self, exc, frs)
 
+        def error_handler_wrapper(errors: _TimestampedErrorList) -> None:
+            try:
+                # Check if we are not closed and the handler exists
+                if not self._closed and error_handler is not None:
+                    for error in errors.errors:
+                        error_handler(errors.timestamp, error)
+            except Exception as exc:
+                _logger.exception(
+                    "%s: Unhandled exception in the receive error handler: %s; lost error: %s", self, exc, errors
+                )
+
         while not self._closed and not loop.is_closed():
             try:
                 (
@@ -213,14 +250,22 @@ def handler_wrapper(frs: typing.Sequence[typing.Tuple[Timestamp, Envelope]]) ->
 
                 if self._sock in read_ready:
                     frames: typing.List[typing.Tuple[Timestamp, Envelope]] = []
+                    errors: typing.Optional[_TimestampedErrorList] = None
                     try:
                         while True:
-                            frames.append(self._read_frame(ts_mono_ns))
+                            out = self._read_frame(ts_mono_ns)
+                            if isinstance(out, _TimestampedErrorList):
+                                errors = out
+                                break  # Report previously received frames first
+                            else:
+                                frames.append(out)
                     except OSError as ex:
                         if ex.errno != errno.EAGAIN:
                             raise
                     try:
                         loop.call_soon_threadsafe(handler_wrapper, frames)
+                        if errors:
+                            loop.call_soon_threadsafe(error_handler_wrapper, errors)
                     except RuntimeError as ex:
                         _logger.debug("%s: Event loop is closed, exiting: %r", self, ex)
                         break
@@ -241,7 +286,7 @@ def handler_wrapper(frs: typing.Sequence[typing.Tuple[Timestamp, Envelope]]) ->
         self._closed = True
         _logger.debug("%s thread is about to exit", self)
 
-    def _read_frame(self, ts_mono_ns: int) -> typing.Tuple[Timestamp, Envelope]:
+    def _read_frame(self, ts_mono_ns: int) -> typing.Tuple[Timestamp, Envelope] | _TimestampedErrorList:
         while True:
             data, ancdata, msg_flags, _addr = self._sock.recvmsg(  # type: ignore
                 self._native_frame_size, self._ancillary_data_buffer_size
@@ -266,9 +311,13 @@ def _read_frame(self, ts_mono_ns: int) -> typing.Tuple[Timestamp, Envelope]:
 
             assert ts_system_ns > 0, "Missing the timestamp; does the driver support timestamping?"
             timestamp = Timestamp(system_ns=ts_system_ns, monotonic_ns=ts_mono_ns)
-            out = SocketCANMedia._parse_native_frame(data)
-            if out is not None:
+            out = self._parse_native_frame(data)
+            if isinstance(out, DataFrame):
                 return timestamp, Envelope(out, loopback=loopback)
+            elif isinstance(out, list):
+                return _TimestampedErrorList(timestamp, out)
+            else:
+                assert False, "Unreachable"
 
     def _compile_native_frame(self, source: DataFrame) -> bytes:
         flags = _CANFD_BRS if (self._is_fd and not self._disable_brs) else 0
@@ -278,13 +327,51 @@ def _compile_native_frame(self, source: DataFrame) -> bytes:
         assert len(out) == self._native_frame_size
         return out
 
-    @staticmethod
-    def _parse_native_frame(source: bytes) -> typing.Optional[DataFrame]:
+    def _parse_native_frame(self, source: bytes) -> None | DataFrame | typing.List[Media.Error]:
         header_size = _FRAME_HEADER_STRUCT.size
         ident_raw, data_length, _flags = _FRAME_HEADER_STRUCT.unpack(source[:header_size])
-        if (ident_raw & _CAN_RTR_FLAG) or (ident_raw & _CAN_ERR_FLAG):  # Unsupported format, ignore silently
+        if ident_raw & _CAN_RTR_FLAG:  # Unsupported format, ignore silently
             _logger.debug("Unsupported CAN frame dropped; raw SocketCAN ID is %08x", ident_raw)
             return None
+
+        if ident_raw & _CAN_ERR_FLAG:
+            out_error = []
+            if ident_raw & _CAN_ERR_TX_TIMEOUT:
+                _logger.error("Error Tx Timeout on %s", self._iface_name)
+                out_error.append(Media.Error.CAN_TX_TIMEOUT)
+            if ident_raw & _CAN_ERR_CRTL:  # Controller problem, details are in data[1]
+                error_byte = source[header_size + 1]
+                if error_byte & _CAN_ERR_CRTL_RX_OVERFLOW:
+                    _logger.error("Error Rx Overflow State on %s", self._iface_name)
+                    out_error.append(Media.Error.CAN_RX_OVERFLOW)
+                if error_byte & _CAN_ERR_CRTL_TX_OVERFLOW:
+                    _logger.error("Error Tx Overflow State on %s", self._iface_name)
+                    out_error.append(Media.Error.CAN_TX_OVERFLOW)
+                if error_byte & _CAN_ERR_CRTL_RX_WARNING:
+                    _logger.warning("Error Rx Warning State on %s", self._iface_name)
+                    out_error.append(Media.Error.CAN_RX_WARNING)
+                if error_byte & _CAN_ERR_CRTL_TX_WARNING:
+                    _logger.warning("Error Tx Warning State on %s", self._iface_name)
+                    out_error.append(Media.Error.CAN_TX_WARNING)
+                if error_byte & _CAN_ERR_CRTL_TX_PASSIVE:
+                    _logger.error("Error Tx Passive State on %s", self._iface_name)
+                    out_error.append(Media.Error.CAN_TX_PASSIVE)
+                if error_byte & _CAN_ERR_CRTL_RX_PASSIVE:
+                    _logger.error("Error Rx Passive State on %s", self._iface_name)
+                    out_error.append(Media.Error.CAN_RX_PASSIVE)
+            if ident_raw & _CAN_ERR_BUSOFF:
+                _logger.error("CAN Bus Off on %s", self._iface_name)
+                out_error.append(Media.Error.CAN_BUS_OFF)
+
+            if len(out_error) > 0:
+                return out_error
+            else:
+                _logger.debug(
+                    "Unsupported CAN error frame dropped; raw SocketCAN ID is %08x",
+                    ident_raw,
+                )
+                return None
+
         frame_format = FrameFormat.EXTENDED if ident_raw & _CAN_EFF_FLAG else FrameFormat.BASE
         data = source[header_size : header_size + data_length]
         assert len(data) == data_length
@@ -351,6 +438,48 @@ class _NativeFrameDataCapacity(enum.IntEnum):
 _CAN_RTR_FLAG = 0x40000000
 _CAN_ERR_FLAG = 0x20000000
 
+# From the Linux kernel (linux/include/uapi/linux/can/error.h); not exposed via the Python's socket module
+_CAN_ERR_TX_TIMEOUT = 0x00000001
+"""TX timeout (by netdevice driver)"""
+_CAN_ERR_LOSTARB = 0x00000002
+"""lost arbitration    / data[0]"""
+_CAN_ERR_CRTL = 0x00000004
+"""controller problems / data[1]"""
+_CAN_ERR_PROT = 0x00000008
+"""protocol violations / data[2..3]"""
+_CAN_ERR_TRX = 0x00000010
+"""transceiver status  / data[4]"""
+_CAN_ERR_ACK = 0x00000020
+"""received no ACK on transmission"""
+_CAN_ERR_BUSOFF = 0x00000040
+"""bus off"""
+_CAN_ERR_BUSERROR = 0x00000080
+"""bus error (may flood!)"""
+_CAN_ERR_RESTARTED = 0x00000100
+"""controller restarted"""
+_CAN_ERR_CNT = 0x00000200
+"""TX error counter / data[6], RX error counter / data[7]"""
+
+_CAN_ERR_CRTL_UNSPEC = 0x00
+""" unspecified"""
+_CAN_ERR_CRTL_RX_OVERFLOW = 0x01
+""" RX buffer overflow"""
+_CAN_ERR_CRTL_TX_OVERFLOW = 0x02
+""" TX buffer overflow"""
+_CAN_ERR_CRTL_RX_WARNING = 0x04
+""" reached warning level for RX errors"""
+_CAN_ERR_CRTL_TX_WARNING = 0x08
+""" reached warning level for TX errors"""
+_CAN_ERR_CRTL_RX_PASSIVE = 0x10
+""" reached error passive status RX"""
+_CAN_ERR_CRTL_TX_PASSIVE = 0x20
+""" reached error passive status TX (at least one error counter exceeds the protocol-defined level of 127)"""
+_CAN_ERR_CRTL_ACTIVE = 0x40
+""" recovered to error active state"""
+
+# From the Linux kernel (linux/include/uapi/linux/can/raw.h); not exposed via the Python's socket module
+_CAN_RAW_ERR_FILTER = 2
+
 _CAN_EFF_MASK = 0x1FFFFFFF
 
 # approximate sk_buffer kernel struct overhead.