add utility function create_filter

Author: zariiii9003

can/__init__.py

@@ -31,7 +31,7 @@
 from .io import CSVWriter, CSVReader
 from .io import SqliteWriter, SqliteReader
 
-from .util import set_logging_level
+from .util import set_logging_level, create_filter
 
 from .message import Message
 from .bus import BusABC, BusState

can/bus.py

@@ -344,7 +344,7 @@ def set_filters(
         filters to ``None``.
 
         :param filters:
-            A iterable of dictionaries each containing a "can_id",
+            An iterable of dictionaries each containing a "can_id",
             a "can_mask", and an optional "extended" key.
 
             >>> [{"can_id": 0x11, "can_mask": 0x21, "extended": False}]
@@ -354,6 +354,9 @@ def set_filters(
             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.
+
+            The utility function :func:`~can.create_filter` is provided
+            to generate suitable filters.
         """
         self._filters = filters or None
         self._apply_filters(self._filters)

can/typechecking.py

@@ -8,9 +8,13 @@
 
 import typing_extensions
 
-CanFilter = typing_extensions.TypedDict("CanFilter", {"can_id": int, "can_mask": int})
+CanFilter = typing_extensions.TypedDict(
+    "CanFilter",
+    {"can_id": int, "can_mask": int, "extended": typing_extensions.Literal[False]},
+)
 CanFilterExtended = typing_extensions.TypedDict(
-    "CanFilterExtended", {"can_id": int, "can_mask": int, "extended": bool}
+    "CanFilterExtended",
+    {"can_id": int, "can_mask": int, "extended": typing_extensions.Literal[True]},
 )
 CanFilters = typing.Sequence[typing.Union[CanFilter, CanFilterExtended]]
 

can/util.py

@@ -371,6 +371,49 @@ def time_perfcounter_correlation() -> Tuple[float, float]:
     return t1, performance_counter
 
 
+def create_filter(
+    accepted_std_ids: Iterable[int], accepted_ext_ids: Iterable[int]
+) -> typechecking.CanFilters:
+    """Calculate CAN filter settings for given arbitration IDs.
+
+    :param accepted_std_ids:
+        An iterable of standard arbitration IDs (11bit) to be accepted.
+    :param accepted_ext_ids:
+        An iterable of extended arbitration IDs (29bit) to be accepted.
+    :returns:
+        CAN filters to be used with :meth:`~can.BusABC.set_filters`.
+    """
+    std_code = 0x7FF
+    std_mask = 0x7FF
+    _std_mem = 0x7FF
+
+    ext_code = 0x1FFFFFFF
+    ext_mask = 0x1FFFFFFF
+    _ext_mem = 0x1FFFFFFF
+
+    for can_id in accepted_std_ids:
+        if not (0 <= can_id <= 0x7FF):
+            raise ValueError(f"Invalid CAN ID {can_id}")
+
+        if std_code != 0x7FF:
+            _std_mem = ~(std_code ^ can_id) & 0x7FF
+        std_code &= can_id
+        std_mask &= _std_mem
+
+    for can_id in accepted_ext_ids:
+        if not (0 <= can_id <= 0x1FFFFFFF):
+            raise ValueError(f"Invalid Extended CAN ID {can_id}")
+        if ext_code != 0x1FFFFFFF:
+            _ext_mem = ~(ext_code ^ can_id) & 0x1FFFFFFF
+        ext_code &= can_id
+        ext_mask &= _ext_mem
+
+    return [
+        {"can_id": std_code, "can_mask": std_mask, "extended": False},
+        {"can_id": ext_code, "can_mask": ext_mask, "extended": True},
+    ]
+
+
 if __name__ == "__main__":
     print("Searching for configuration named:")
     print("\n".join(CONFIG_FILES))

doc/api.rst

@@ -27,6 +27,8 @@ Utilities
 
 .. autofunction:: can.detect_available_configs
 
+.. autofunction:: can.create_filter
+
 
 .. _notifier:
 

test/test_util.py

@@ -1,7 +1,9 @@
 import unittest
 import warnings
 
-from can.util import _rename_kwargs
+import pytest
+
+from can.util import _rename_kwargs, create_filter
 
 
 class RenameKwargsTest(unittest.TestCase):
@@ -47,3 +49,115 @@ def test_with_new_and_alias_present(self):
         aliases = {"old_a": "a", "old_b": "b", "z": None}
         with self.assertRaises(TypeError):
             self._test(kwargs, aliases)
+
+
+def test_create_filter():
+    # test exceptions
+    with pytest.raises(ValueError):
+        create_filter(
+            accepted_std_ids=[-1],
+            accepted_ext_ids=[100],
+        )
+    with pytest.raises(ValueError):
+        create_filter(
+            accepted_std_ids=[100],
+            accepted_ext_ids=[-1],
+        )
+    with pytest.raises(ValueError):
+        create_filter(
+            accepted_std_ids=[0x7FF + 1],
+            accepted_ext_ids=[100],
+        )
+    with pytest.raises(ValueError):
+        create_filter(
+            accepted_std_ids=[100],
+            accepted_ext_ids=[0x1FFFFFFF + 1],
+        )
+
+    accepted_std_range = range(200, 205)
+    accepted_ext_range = range(105_000, 105_005)
+
+    can_filters = create_filter(
+        accepted_std_ids=accepted_std_range,
+        accepted_ext_ids=accepted_ext_range,
+    )
+    std_code = can_filters[0]["can_id"]
+    std_mask = can_filters[0]["can_mask"]
+    ext_code = can_filters[1]["can_id"]
+    ext_mask = can_filters[1]["can_mask"]
+
+    """
+    Standard ID proof:
+    b00011001000 | id:    200
+    b00011001001 | id:    201
+    b00011001010 | id:    202
+    b00011001011 | id:    203
+    b00011001100 | id:    204
+    -------------------------
+    b00011001000 | code:  200
+    b11111111000 | mask: 2040
+    """
+    assert std_code == 200
+    assert std_mask == 2040
+
+    """
+    Extended ID proof:
+    b00000000000011001101000101000 | id:      105000
+    b00000000000011001101000101001 | id:      105001
+    b00000000000011001101000101010 | id:      105002
+    b00000000000011001101000101011 | id:      105003
+    b00000000000011001101000101100 | id:      105004
+    -------------------------
+    b00000000000011001101000101000 | code:    105000
+    b11111111111111111111111111000 | mask: 536870904
+    """
+    assert ext_code == 105000
+    assert ext_mask == 536870904
+
+    # test that standard IDs in given range are accepted
+    for can_id in accepted_std_range:
+        assert 0 == (can_id ^ std_code) & std_mask
+
+    # test that extended IDs in given range are accepted
+    for can_id in accepted_ext_range:
+        assert 0 == (can_id ^ ext_code) & ext_mask
+
+    # test non-consecutive IDs
+    can_filters = create_filter(
+        accepted_std_ids=[333, 444],
+        accepted_ext_ids=[33333, 44444],
+    )
+    std_code = can_filters[0]["can_id"]
+    std_mask = can_filters[0]["can_mask"]
+    ext_code = can_filters[1]["can_id"]
+    ext_mask = can_filters[1]["can_mask"]
+
+    """
+    Standard ID proof (non-consecutive):
+    b00101001101 | id:    333
+    b00110111100 | id:    444
+    -------------------------
+    b00100001100 | code:  268
+    b11100001110 | mask: 1806
+    """
+    assert std_code == 268
+    assert std_mask == 1806
+
+    """
+    Extended ID proof (non-consecutive):
+    b00000000000001000001000110101 | id:       33333
+    b00000000000001010110110011100 | id:       44444
+    -------------------------
+    b00000000000001000000000010100 | code:     32788
+    b11111111111111101000001010110 | mask: 536858710
+    """
+    assert ext_code == 32788
+    assert ext_mask == 536858710
+
+    # test that standard IDs in given sequence are accepted
+    for can_id in [333, 444]:
+        assert 0 == (can_id ^ std_code) & std_mask
+
+    # test that extended IDs in given sequence are accepted
+    for can_id in [33333, 44444]:
+        assert 0 == (can_id ^ ext_code) & ext_mask