Software filtering for all Buses

.appveyor.yml

@@ -33,4 +33,5 @@ test_script:
   # Note that you must use the environment variable %PYTHON% to refer to
   # the interpreter you're using - Appveyor does not do anything special
   # to put the Python version you want to use on PATH.
-  - "%PYTHON%\\python.exe setup.py test -v"
+  - "%PYTHON%\\python.exe setup.py test -v" # --timeout=300 -> TODO:
+  # need to switch to pytest like on Unix first, but that's for another PR

.travis.yml

@@ -51,4 +51,4 @@ install:
   - travis_retry pip install .[test]
 
 script:
-  - py.test -v
+  - py.test -v --timeout=300

can/bus.py

@@ -2,79 +2,137 @@
 # coding: utf-8
 
 """
-Contains the ABC bus implementation.
+Contains the ABC bus implementation and its documentation.
 """
 
 from __future__ import print_function, absolute_import
 
 from abc import ABCMeta, abstractmethod
 import logging
 import threading
+from time import time
 from collections import namedtuple
 
-from can.broadcastmanager import ThreadBasedCyclicSendTask
-
-logger = logging.getLogger(__name__)
+from .broadcastmanager import ThreadBasedCyclicSendTask
 
 
 BusState = namedtuple('BusState', 'ACTIVE, PASSIVE, ERROR')
 
 
 class BusABC(object):
-    """CAN Bus Abstract Base Class
-
-    Concrete implementations must implement the following methods:
-        * send
-        * recv
-
-    As well as setting the `channel_info` attribute to a string describing the
-    interface.
-
-    They may implement :meth:`~can.BusABC._detect_available_configs` to allow
-    the interface to report which configurations are currently available for
-    new connections.
-
+    """The CAN Bus Abstract Base Class that serves as the basis
+    for all concrete interfaces.
     """
 
-    #: a string describing the underlying bus channel
+    #: a string describing the underlying bus and/or channel
     channel_info = 'unknown'
 
     @abstractmethod
     def __init__(self, channel=None, can_filters=None, **config):
-        """
+        """Construct and open a CAN bus instance of the specified type.
+
+        Subclasses should call though this method with all given parameters
+        as it handles generic tasks like applying filters.
+
         :param channel:
             The can interface identifier. Expected type is backend dependent.
 
         :param list can_filters:
-            A list of dictionaries each containing a "can_id", a "can_mask",
-            and an "extended" key.
-
-            >>> [{"can_id": 0x11, "can_mask": 0x21, "extended": False}]
-
-            A filter matches, when ``<received_can_id> & can_mask == can_id & can_mask``
+            See :meth:`~can.BusABC.set_filters` for details.
 
         :param dict config:
             Any backend dependent configurations are passed in this dictionary
         """
-        pass
+        self.set_filters(can_filters)
 
     def __str__(self):
         return self.channel_info
 
-    @abstractmethod
     def recv(self, timeout=None):
         """Block waiting for a message from the Bus.
 
-        :param float timeout: Seconds to wait for a message.
+        :param float timeout:
+            seconds to wait for a message or None to wait indefinitely
 
+        :rtype: can.Message or None
         :return:
             None on timeout or a :class:`can.Message` object.
+        :raises can.CanError:
+            if an error occurred while reading
+        """
+        start = time()
+        time_left = timeout
+
+        while True:
+
+            # try to get a message
+            msg, already_filtered = self._recv_internal(timeout=time_left)
+
+            # return it, if it matches
+            if msg and (already_filtered or self._matches_filters(msg)):
+                return msg
+
+            # if not, and timeout is None, try indefinitely
+            elif timeout is None:
+                continue
+
+            # try next one only if there still is time, and with reduced timeout
+            else:
+
+                time_left = timeout - (time() - start)
+
+                if time_left > 0:
+                    continue
+                else:
+                    return None
+
+    def _recv_internal(self, timeout):
+        """
+        Read a message from the bus and tell whether it was filtered.
+        This methods may be called by :meth:`~can.BusABC.recv`
+        to read a message multiple times if the filters set by
+        :meth:`~can.BusABC.set_filters` do not match and the call has
+        not yet timed out.
+
+        New implementations should always override this method instead of
+        :meth:`~can.BusABC.recv`, to be able to take advantage of the
+        software based filtering provided by :meth:`~can.BusABC.recv`
+        as a fallback. This method should never be called directly.
+
+        .. note::
+
+            This method is not an `@abstractmethod` (for now) to allow older
+            external implementations to continue using their existing
+            :meth:`~can.BusABC.recv` implementation.
+
+        .. note::
+
+            The second return value (whether filtering was already done) may change
+            over time for some interfaces, like for example in the Kvaser interface.
+            Thus it cannot be simplified to a constant value.
+
+        :param float timeout: seconds to wait for a message,
+                              see :meth:`can.BusABC.send`
+
+        :rtype: tuple[can.Message, bool] or tuple[None, bool]
+        :return:
+            1.  a message that was read or None on timeout
+            2.  a bool that is True if message filtering has already
+                been done and else False
+
+        :raises can.CanError:
+            if an error occurred while reading
+        :raises NotImplementedError:
+            if the bus provides it's own :meth:`~can.BusABC.recv`
+            implementation (legacy implementation)
+
         """
         raise NotImplementedError("Trying to read from a write only bus?")
 
     @abstractmethod
     def send(self, msg, timeout=None):
-        """Transmit a message to CAN bus.
+        """Transmit a message to the CAN bus.
+
         Override this method to enable the transmit path.
 
         :param can.Message msg: A message object.
@@ -84,7 +142,7 @@ def send(self, msg, timeout=None):
             If timeout is exceeded, an exception will be raised.
             Might not be supported by all interfaces.
 
-        :raise: :class:`can.CanError`
+        :raises can.CanError:
             if the message could not be written.
         """
         raise NotImplementedError("Trying to write to a readonly bus?")
@@ -103,14 +161,16 @@ def send_periodic(self, msg, period, duration=None):
         :return: A started task instance
         :rtype: can.CyclicSendTaskABC
 
+        .. note::
+
             Note the duration before the message stops being sent may not
             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.
 
         """
         if not hasattr(self, "_lock"):
-            # Create send lock for this bus
+            # Create a send lock for this bus
             self._lock = threading.Lock()
         return ThreadBasedCyclicSendTask(self, self._lock, msg, period, duration)
 
@@ -121,27 +181,91 @@ def __iter__(self):
             ...     print(msg)
 
 
-        :yields: :class:`can.Message` msg objects.
+        :yields:
+            :class:`can.Message` msg objects.
         """
         while True:
             msg = self.recv(timeout=1.0)
             if msg is not None:
                 yield msg
 
-    def set_filters(self, can_filters=None):
+    @property
+    def filters(self):
+        return self._filters
+
+    @filters.setter
+    def filters(self, filters):
+        self.set_filters(filters)
+
+    def set_filters(self, filters=None):
         """Apply filtering to all messages received by this Bus.
 
-        Calling without passing any filters will reset the applied filters.
+        All messages that match at least one filter are returned.
+        If `filters` is `None`, all messages are matched.
+        If it is a zero size interable, no messages are matched.
 
-        :param list can_filters:
-            A list of dictionaries each containing a "can_id" and a "can_mask".
+        Calling without passing any filters will reset the applied
+        filters to `None`.
 
-            >>> [{"can_id": 0x11, "can_mask": 0x21}]
+        :param Iterator[dict] filters:
+            A iterable of dictionaries each containing a "can_id", a "can_mask",
+            and an optional "extended" key.
 
-            A filter matches, when ``<received_can_id> & can_mask == can_id & can_mask``
+            >>> [{"can_id": 0x11, "can_mask": 0x21, "extended": False}]
+
+            A filter matches, when ``<received_can_id> & can_mask == can_id & can_mask``.
+            If ``extended`` is set as well, it only matches messages where
+            ``<received_is_extended> == extended``. Else it matches every messages based
+            only on the arbitration ID and mask.
+
+        """
+        self._filters = filters
+        self._apply_filters(self._filters)
 
+    def _apply_filters(self, filters):
         """
-        raise NotImplementedError("Trying to set_filters on unsupported bus")
+        Hook for applying the filters to the underlying kernel or
+        hardware if supported/implemented by the interface.
+
+        :param Iterator[dict] filters:
+            See :meth:`~can.BusABC.set_filters` for details.
+        """
+        pass
+
+    def _matches_filters(self, msg):
+        """Checks whether the given message matches at least one of the
+        current filters. See :meth:`~can.BusABC.set_filters` for details
+        on how the filters work.
+
+        This method should not be overridden.
+
+        :param can.Message msg:
+            the message to check if matching
+        :rtype: bool
+        :return: whether the given message matches at least one filter
+        """
+
+        # if no filters are set, all messages are matched
+        if self._filters is None:
+            return True
+
+        for filter in self._filters:
+            # check if this filter even applies to the message
+            if 'extended' in filter and \
+                filter['extended'] != msg.is_extended_id:
+                continue
+
+            # then check for the mask and id
+            can_id = filter['can_id']
+            can_mask = filter['can_mask']
+
+            # basically, we compute `msg.arbitration_id & can_mask == can_id & can_mask`
+            # by using the shorter, but equivalent from below:
+            if (can_id ^ msg.arbitration_id) & can_mask == 0:
+                return True
+
+        # nothing matched
+        return False
 
     def flush_tx_buffer(self):
         """Discard every message that may be queued in the output buffer(s).
@@ -153,7 +277,7 @@ def shutdown(self):
         Called to carry out any interface specific cleanup required
         in shutting down a bus.
         """
-        self.flush_tx_buffer()
+        pass
 
     @property
     def state(self):

can/interface.py

@@ -22,7 +22,6 @@
 if sys.version_info.major > 2:
     basestring = str
 
-
 log = logging.getLogger('can.interface')
 log_autodetect = log.getChild('detect_available_configs')