Merge branch 'develop' into felixdivo-patch-socketcan

Author: felixdivo

.travis.yml

@@ -105,6 +105,7 @@ jobs:
           can/bus.py
           can/interface.py
           can/interfaces/socketcan/**.py
+          can/interfaces/virtual.py
           can/listener.py
           can/logger.py
           can/message.py

can/bus.py

@@ -2,7 +2,7 @@
 Contains the ABC bus implementation and its documentation.
 """
 
-from typing import cast, Iterator, List, Optional, Sequence, Tuple, Union
+from typing import cast, Any, Iterator, List, Optional, Sequence, Tuple, Union
 
 import can.typechecking
 
@@ -43,7 +43,7 @@ class BusABC(metaclass=ABCMeta):
     @abstractmethod
     def __init__(
         self,
-        channel: can.typechecking.Channel,
+        channel: Any,
         can_filters: Optional[can.typechecking.CanFilters] = None,
         **kwargs: object
     ):

can/interfaces/ics_neovi/neovi_bus.py

@@ -306,6 +306,7 @@ def _ics_msg_to_message(self, ics_msg):
                 dlc=ics_msg.NumberBytesData,
                 is_extended_id=bool(ics_msg.StatusBitField & ics.SPY_STATUS_XTD_FRAME),
                 is_fd=is_fd,
+                is_rx=not bool(ics_msg.StatusBitField & ics.SPY_STATUS_TX_MSG),
                 is_remote_frame=bool(
                     ics_msg.StatusBitField & ics.SPY_STATUS_REMOTE_FRAME
                 ),
@@ -325,6 +326,7 @@ def _ics_msg_to_message(self, ics_msg):
                 dlc=ics_msg.NumberBytesData,
                 is_extended_id=bool(ics_msg.StatusBitField & ics.SPY_STATUS_XTD_FRAME),
                 is_fd=is_fd,
+                is_rx=not bool(ics_msg.StatusBitField & ics.SPY_STATUS_TX_MSG),
                 is_remote_frame=bool(
                     ics_msg.StatusBitField & ics.SPY_STATUS_REMOTE_FRAME
                 ),
@@ -361,11 +363,12 @@ def send(self, msg, timeout=None):
                 flag3 |= ics.SPY_STATUS3_CANFD_ESI
 
         message.ArbIDOrHeader = msg.arbitration_id
-        message.NumberBytesData = len(msg.data)
-        message.Data = tuple(msg.data[:8])
-        if msg.is_fd and len(msg.data) > 8:
+        msg_data = msg.data
+        message.NumberBytesData = len(msg_data)
+        message.Data = tuple(msg_data[:8])
+        if msg.is_fd and len(msg_data) > 8:
             message.ExtraDataPtrEnabled = 1
-            message.ExtraDataPtr = tuple(msg.data)
+            message.ExtraDataPtr = tuple(msg_data)
         message.StatusBitField = flag0
         message.StatusBitField2 = 0
         message.StatusBitField3 = flag3

can/interfaces/pcan/basic.py

@@ -14,7 +14,9 @@
 from string import *
 import platform
 import logging
-import winreg
+
+if platform.system() == "Windows":
+    import winreg
 
 
 logger = logging.getLogger("can.pcan")

can/interfaces/socketcan/socketcan.py

@@ -14,6 +14,7 @@
 import socket
 import struct
 import time
+import threading
 import errno
 
 log = logging.getLogger(__name__)
@@ -162,7 +163,7 @@ def build_can_frame(msg: Message) -> bytes:
         __u8    data[CANFD_MAX_DLEN] __attribute__((aligned(8)));
     };
     """
-    can_id = _add_flags_to_can_id(msg)
+    can_id = _compose_arbitration_id(msg)
     flags = 0
     if msg.bitrate_switch:
         flags |= CANFD_BRS
@@ -286,7 +287,7 @@ def send_bcm(bcm_socket: socket.socket, data: bytes) -> int:
             raise e
 
 
-def _add_flags_to_can_id(message: Message) -> int:
+def _compose_arbitration_id(message: Message) -> int:
     can_id = message.arbitration_id
     if message.is_extended_id:
         log.debug("sending an extended id type message")
@@ -297,7 +298,6 @@ def _add_flags_to_can_id(message: Message) -> int:
     if message.is_error_frame:
         log.debug("sending error frame")
         can_id |= CAN_ERR_FLAG
-
     return can_id
 
 
@@ -310,18 +310,22 @@ class CyclicSendTask(
         - setting of a task duration
         - modifying the data
         - stopping then subsequent restarting of the task
-
     """
 
     def __init__(
         self,
         bcm_socket: socket.socket,
+        task_id: int,
         messages: Union[Sequence[Message], Message],
         period: float,
         duration: Optional[float] = None,
     ):
-        """
+        """Construct and :meth:`~start` a task.
+
         :param bcm_socket: An open BCM socket on the desired CAN channel.
+        :param task_id:
+            The identifier used to uniquely reference particular cyclic send task
+            within Linux BCM.
         :param messages:
             The messages to be sent periodically.
         :param period:
@@ -336,12 +340,12 @@ def __init__(
         super().__init__(messages, period, duration)
 
         self.bcm_socket = bcm_socket
+        self.task_id = task_id
         self._tx_setup(self.messages)
 
     def _tx_setup(self, messages: Sequence[Message]) -> None:
         # Create a low level packed frame to pass to the kernel
         body = bytearray()
-        self.can_id_with_flags = _add_flags_to_can_id(messages[0])
         self.flags = CAN_FD_FRAME if messages[0].is_fd else 0
 
         if self.duration:
@@ -353,9 +357,19 @@ def _tx_setup(self, messages: Sequence[Message]) -> None:
             ival1 = 0.0
             ival2 = self.period
 
-        # First do a TX_READ before creating a new task, and check if we get
-        # EINVAL. If so, then we are referring to a CAN message with the same
-        # ID
+        self._check_bcm_task()
+
+        header = build_bcm_transmit_header(
+            self.task_id, count, ival1, ival2, self.flags, nframes=len(messages)
+        )
+        for message in messages:
+            body += build_can_frame(message)
+        log.debug("Sending BCM command")
+        send_bcm(self.bcm_socket, header + body)
+
+    def _check_bcm_task(self):
+        # Do a TX_READ on a task ID, and check if we get EINVAL. If so,
+        # then we are referring to a CAN message with the existing ID
         check_header = build_bcm_header(
             opcode=CAN_BCM_TX_READ,
             flags=0,
@@ -364,7 +378,7 @@ def _tx_setup(self, messages: Sequence[Message]) -> None:
             ival1_usec=0,
             ival2_seconds=0,
             ival2_usec=0,
-            can_id=self.can_id_with_flags,
+            can_id=self.task_id,
             nframes=0,
         )
         try:
@@ -374,45 +388,33 @@ def _tx_setup(self, messages: Sequence[Message]) -> None:
                 raise e
         else:
             raise ValueError(
-                "A periodic Task for Arbitration ID {} has already been created".format(
-                    messages[0].arbitration_id
+                "A periodic task for Task ID {} is already in progress by SocketCAN Linux layer".format(
+                    self.task_id
                 )
             )
 
-        header = build_bcm_transmit_header(
-            self.can_id_with_flags,
-            count,
-            ival1,
-            ival2,
-            self.flags,
-            nframes=len(messages),
-        )
-        for message in messages:
-            body += build_can_frame(message)
-        log.debug("Sending BCM command")
-        send_bcm(self.bcm_socket, header + body)
-
     def stop(self) -> None:
-        """Send a TX_DELETE message to cancel this task.
+        """Stop a task by sending TX_DELETE message to Linux kernel.
 
         This will delete the entry for the transmission of the CAN-message
-        with the specified can_id CAN identifier. The message length for the command
-        TX_DELETE is {[bcm_msg_head]} (only the header).
+        with the specified :attr:`~task_id` identifier. The message length
+        for the command TX_DELETE is {[bcm_msg_head]} (only the header).
         """
         log.debug("Stopping periodic task")
 
-        stopframe = build_bcm_tx_delete_header(self.can_id_with_flags, self.flags)
+        stopframe = build_bcm_tx_delete_header(self.task_id, self.flags)
         send_bcm(self.bcm_socket, stopframe)
 
     def modify_data(self, messages: Union[Sequence[Message], Message]) -> None:
-        """Update the contents of the periodically sent messages.
-
-        Note: The messages must all have the same
-        :attr:`~can.Message.arbitration_id` like the first message.
+        """Update the contents of the periodically sent CAN messages by
+        sending TX_SETUP message to Linux kernel.
 
-        Note: The number of new cyclic messages to be sent must be equal to the
+        The number of new cyclic messages to be sent must be equal to the
         original number of messages originally specified for this task.
 
+        .. note:: The messages must all have the same
+                  :attr:`~can.Message.arbitration_id` like the first message.
+
         :param messages:
             The messages with the new :attr:`can.Message.data`.
         """
@@ -423,14 +425,22 @@ def modify_data(self, messages: Union[Sequence[Message], Message]) -> None:
 
         body = bytearray()
         header = build_bcm_update_header(
-            can_id=self.can_id_with_flags, msg_flags=self.flags, nframes=len(messages)
+            can_id=self.task_id, msg_flags=self.flags, nframes=len(messages)
         )
         for message in messages:
             body += build_can_frame(message)
         log.debug("Sending BCM command")
         send_bcm(self.bcm_socket, header + body)
 
     def start(self) -> None:
+        """Start a periodic task by sending TX_SETUP message to Linux kernel.
+
+        It verifies presence of the particular BCM task through sending TX_READ
+        message to Linux kernel prior to scheduling.
+
+        :raises ValueError:
+            If the task referenced by :attr:`~task_id` is already running.
+        """
         self._tx_setup(self.messages)
 
 
@@ -443,16 +453,17 @@ class MultiRateCyclicSendTask(CyclicSendTask):
     def __init__(
         self,
         channel: socket.socket,
+        task_id: int,
         messages: Sequence[Message],
         count: int,
         initial_period: float,
         subsequent_period: float,
     ):
-        super().__init__(channel, messages, subsequent_period)
+        super().__init__(channel, task_id, messages, subsequent_period)
 
         # Create a low level packed frame to pass to the kernel
         header = build_bcm_transmit_header(
-            self.can_id_with_flags,
+            self.task_id,
             count,
             initial_period,
             subsequent_period,
@@ -509,10 +520,10 @@ def capture_message(
     # Fetching the Arb ID, DLC and Data
     try:
         if get_channel:
-            cf, addr = sock.recvfrom(CANFD_MTU)
+            cf, _, msg_flags, addr = sock.recvmsg(CANFD_MTU)
             channel = addr[0] if isinstance(addr, tuple) else addr
         else:
-            cf = sock.recv(CANFD_MTU)
+            cf, _, msg_flags, _ = sock.recvmsg(CANFD_MTU)
             channel = None
     except socket.error as exc:
         raise can.CanError("Error receiving: %s" % exc)
@@ -539,6 +550,9 @@ def capture_message(
     bitrate_switch = bool(flags & CANFD_BRS)
     error_state_indicator = bool(flags & CANFD_ESI)
 
+    # Section 4.7.1: MSG_DONTROUTE: set when the received frame was created on the local host.
+    is_rx = not bool(msg_flags & socket.MSG_DONTROUTE)
+
     if is_extended_frame_format:
         # log.debug("CAN: Extended")
         # TODO does this depend on SFF or EFF?
@@ -555,6 +569,7 @@ def capture_message(
         is_remote_frame=is_remote_transmission_request,
         is_error_frame=is_error_frame,
         is_fd=is_fd,
+        is_rx=is_rx,
         bitrate_switch=bitrate_switch,
         error_state_indicator=error_state_indicator,
         dlc=can_dlc,
@@ -567,8 +582,10 @@ def capture_message(
 
 
 class SocketcanBus(BusABC):
-    """
-    Implements :meth:`can.BusABC._detect_available_configs`.
+    """ A SocketCAN interface to CAN.
+
+    It implements :meth:`can.BusABC._detect_available_configs` to search for
+    available interfaces.
     """
 
     def __init__(
@@ -604,6 +621,8 @@ def __init__(
         self.channel_info = "socketcan channel '%s'" % channel
         self._bcm_sockets: Dict[str, socket.socket] = {}
         self._is_filtered = False
+        self._task_id = 0
+        self._task_id_guard = threading.Lock()
 
         # set the receive_own_messages parameter
         try:
@@ -718,18 +737,26 @@ def _send_periodic_internal(
     ) -> CyclicSendTask:
         """Start sending messages at a given period on this bus.
 
-        The kernel's Broadcast Manager SocketCAN API will be used.
+        The Linux kernel's Broadcast Manager SocketCAN API is used to schedule
+        periodic sending of CAN messages. The wrapping 32-bit counter (see
+        :meth:`~_get_next_task_id()`) designated to distinguish different
+        :class:`CyclicSendTask` within BCM provides flexibility to schedule
+        CAN messages sending with the same CAN ID, but different CAN data.
 
         :param messages:
-            The messages to be sent periodically
+            The message(s) to be sent periodically.
         :param period:
             The rate in seconds at which to send the messages.
         :param duration:
             Approximate duration in seconds to continue sending messages. If
             no duration is provided, the task will continue indefinitely.
 
+        :raises ValueError:
+            If task identifier passed to :class:`CyclicSendTask` can't be used
+            to schedule new task in Linux BCM.
+
         :return:
-            A started task instance. This can be used to modify the data,
+            A :class:`CyclicSendTask` task instance. This can be used to modify the data,
             pause/resume the transmission and to stop the transmission.
 
         .. note::
@@ -738,18 +765,20 @@ def _send_periodic_internal(
             be exactly the same as the duration specified by the user. In
             general the message will be sent at the given rate until at
             least *duration* seconds.
-
         """
         msgs = LimitedDurationCyclicSendTaskABC._check_and_convert_messages(msgs)
 
         msgs_channel = str(msgs[0].channel) if msgs[0].channel else None
         bcm_socket = self._get_bcm_socket(msgs_channel or self.channel)
-        # TODO: The SocketCAN BCM interface treats all cyclic tasks sharing an
-        # Arbitration ID as the same Cyclic group. We should probably warn the
-        # user instead of overwriting the old group?
-        task = CyclicSendTask(bcm_socket, msgs, period, duration)
+        task_id = self._get_next_task_id()
+        task = CyclicSendTask(bcm_socket, task_id, msgs, period, duration)
         return task
 
+    def _get_next_task_id(self) -> int:
+        with self._task_id_guard:
+            self._task_id = (self._task_id + 1) % (2 ** 32 - 1)
+            return self._task_id
+
     def _get_bcm_socket(self, channel: str) -> socket.socket:
         if channel not in self._bcm_sockets:
             self._bcm_sockets[channel] = create_bcm_socket(self.channel)