Repository: XKNX/xknx
Pull request #535 (2020-12-16 14:11:12)

Modify TelegramQueue.Callback filter from HA service
<!--
  You are awesome! Thanks for contributing to our project!
  Please, DO NOT DELETE ANY TEXT from this template!.
-->
## Description
<!--
Please include a summary of the change and which issue is fixed.
Please also include relevant motivation and context.
List any dependencies that are required for this change.
-->
- knx_event: renamed `fire_event_filter` to `event_filter` and deprecated `fire_event` config option. A callback is now always registered for HA to be able to modify its `group_addresses` filter from a service.
- added `knx.event_register` service allowing to add and remove group addresses to trigger knx_event without having to change configuration.
- TelegramQueue.Callback: add `group_addresses` attribute to store a list of GroupAddress triggering the callback (additionally to `address_filters`).

So now the yaml configured filters live in `Callback.address_filters` and the group_addresses added by knx.event_register live in `Callback.group_addresses`.

With this it is possible to build automations using knx_event on the fly without changing or even reloading configuration. 
So an HA automation could listen to automation_reloaded
```yaml
automation:
  trigger:
    platform: event
    event_type: automation_reloaded
  action:
    service: knx.event_register
    data:
      address: "1/2/3"
```
to setup a group address listener for itself and later on use the registered knx_event. I think this could come in handy with blueprints.


This fits nicely to #497 

## Type of change
<!--
Please tick the applicable options.
NOTE: Ticking multiple options most likely indicates
that your change is to big and it is suggested to split it into several smaller PRs.
-->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [x] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [x] This change requires a documentation update

## Checklist:

- [x] I have performed a self-review of my own code
- [ ] The documentation has been adjusted accordingly
- [x] The changes generate no new warnings
- [x] Tests have been added that prove the fix is effective or that the feature works
- [x] The changes are documented in the changelog
- [x] The Homeassistant plugin has been adjusted in case of new config options


coveralls: 
[![Coverage Status](https://coveralls.io/builds/35763179/badge)](https://coveralls.io/builds/35763179)

Coverage increased (+0.005%) to 94.778% when pulling **697bf4ff883b86f7f25afc0d7509d01d03643640 on farmio:dynamic_knx_event** into **339f526fc82f646414b1bc8cddbb8896d761a31b on XKNX:master**.

farmio: 🤔 It strikes me that it is not possible to use multiple trigger-action blocks in one automation. Who knew? 
Well, maybe this will get changed in the future, I think the service can be useful nevertheless - even if only for testing purposes.

Patch:
diff --git a/changelog.md b/changelog.md
index f51c72ef..65c96329 100644
--- a/changelog.md
+++ b/changelog.md
@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased changes
+
+### Internals
+
+- remove DPTComparator: DPTBinary and DPTArray are not equal, even if their .value is, and are never equal to `None`.
+
 ## 0.16.0 APCI possibilities considerably increased 2021-01-01
 
 ### Devices
@@ -9,7 +15,9 @@
 
 ### HA integration
 
+- knx_event: renamed `fire_event_filter` to `event_filter` and deprecated `fire_event` config option. A callback is now always registered for HA to be able to modify its `group_addresses` filter from a service.
 - knx_event: renamed `address` to `destination` and added `source`, `telegramtype`, `direction` attributes.
+- added `knx.event_register` service allowing to add and remove group addresses to trigger knx_event without having to change configuration.
 
 ### Internals
 
@@ -26,6 +34,7 @@
 - Farewell Travis CI; Welcome Github Actions!
 - StateUpdater allow float values for `register_remote_value(tracker_options)` attribute.
 - Handle exceptions from received unsupported or not implemented KNXIP Service Type identifiers
+- TelegramQueue.Callback: add `group_addresses` attribute to store a list of GroupAddress triggering the callback (additionally to `address_filters`).
 
 ## 0.15.6 Bugfix for StateUpater 2020-11-26
 
diff --git a/home-assistant-plugin/custom_components/xknx/__init__.py b/home-assistant-plugin/custom_components/xknx/__init__.py
index 431662ca..12b06c11 100644
--- a/home-assistant-plugin/custom_components/xknx/__init__.py
+++ b/home-assistant-plugin/custom_components/xknx/__init__.py
@@ -4,6 +4,7 @@ import logging
 
 import voluptuous as vol
 from xknx import XKNX
+from xknx.core.telegram_queue import TelegramQueue
 from xknx.devices import DateTime, ExposeSensor
 from xknx.dpt import DPTArray, DPTBase, DPTBinary
 from xknx.exceptions import XKNXException
@@ -59,7 +60,7 @@ CONF_XKNX_CONFIG = "config_file"
 CONF_XKNX_ROUTING = "routing"
 CONF_XKNX_TUNNELING = "tunneling"
 CONF_XKNX_FIRE_EVENT = "fire_event"
-CONF_XKNX_FIRE_EVENT_FILTER = "fire_event_filter"
+CONF_XKNX_EVENT_FILTER = "event_filter"
 CONF_XKNX_INDIVIDUAL_ADDRESS = "individual_address"
 CONF_XKNX_MCAST_GRP = "multicast_group"
 CONF_XKNX_MCAST_PORT = "multicast_port"
@@ -71,62 +72,72 @@ SERVICE_XKNX_SEND = "send"
 SERVICE_XKNX_ATTR_ADDRESS = "address"
 SERVICE_XKNX_ATTR_PAYLOAD = "payload"
 SERVICE_XKNX_ATTR_TYPE = "type"
+SERVICE_XKNX_ATTR_REMOVE = "remove"
+SERVICE_XKNX_EVENT_REGISTER = "event_register"
 
 CONFIG_SCHEMA = vol.Schema(
     {
-        DOMAIN: vol.Schema(
-            {
-                vol.Optional(CONF_XKNX_CONFIG): cv.string,
-                vol.Exclusive(
-                    CONF_XKNX_ROUTING, "connection_type"
-                ): ConnectionSchema.ROUTING_SCHEMA,
-                vol.Exclusive(
-                    CONF_XKNX_TUNNELING, "connection_type"
-                ): ConnectionSchema.TUNNELING_SCHEMA,
-                vol.Inclusive(CONF_XKNX_FIRE_EVENT, "fire_ev"): cv.boolean,
-                vol.Inclusive(CONF_XKNX_FIRE_EVENT_FILTER, "fire_ev"): vol.All(
-                    cv.ensure_list, [cv.string]
-                ),
-                vol.Optional(
-                    CONF_XKNX_INDIVIDUAL_ADDRESS, default=XKNX.DEFAULT_ADDRESS
-                ): cv.string,
-                vol.Optional(CONF_XKNX_MCAST_GRP, default=DEFAULT_MCAST_GRP): cv.string,
-                vol.Optional(CONF_XKNX_MCAST_PORT, default=DEFAULT_MCAST_PORT): cv.port,
-                vol.Optional(CONF_XKNX_STATE_UPDATER, default=True): cv.boolean,
-                vol.Optional(CONF_XKNX_RATE_LIMIT, default=20): vol.All(
-                    vol.Coerce(int), vol.Range(min=1, max=100)
-                ),
-                vol.Optional(CONF_XKNX_EXPOSE): vol.All(
-                    cv.ensure_list, [ExposeSchema.SCHEMA]
-                ),
-                vol.Optional(SupportedPlatforms.cover.value): vol.All(
-                    cv.ensure_list, [CoverSchema.SCHEMA]
-                ),
-                vol.Optional(SupportedPlatforms.binary_sensor.value): vol.All(
-                    cv.ensure_list, [BinarySensorSchema.SCHEMA]
-                ),
-                vol.Optional(SupportedPlatforms.light.value): vol.All(
-                    cv.ensure_list, [LightSchema.SCHEMA]
-                ),
-                vol.Optional(SupportedPlatforms.climate.value): vol.All(
-                    cv.ensure_list, [ClimateSchema.SCHEMA]
-                ),
-                vol.Optional(SupportedPlatforms.notify.value): vol.All(
-                    cv.ensure_list, [NotifySchema.SCHEMA]
-                ),
-                vol.Optional(SupportedPlatforms.switch.value): vol.All(
-                    cv.ensure_list, [SwitchSchema.SCHEMA]
-                ),
-                vol.Optional(SupportedPlatforms.sensor.value): vol.All(
-                    cv.ensure_list, [SensorSchema.SCHEMA]
-                ),
-                vol.Optional(SupportedPlatforms.scene.value): vol.All(
-                    cv.ensure_list, [SceneSchema.SCHEMA]
-                ),
-                vol.Optional(SupportedPlatforms.weather.value): vol.All(
-                    cv.ensure_list, [WeatherSchema.SCHEMA]
-                ),
-            }
+        DOMAIN: vol.All(
+            cv.deprecated(CONF_XKNX_FIRE_EVENT, invalidation_version="2021.12"),
+            cv.deprecated("fire_event_filter", replacement_key=CONF_XKNX_EVENT_FILTER),
+            vol.Schema(
+                {
+                    vol.Optional(CONF_XKNX_CONFIG): cv.string,
+                    vol.Exclusive(
+                        CONF_XKNX_ROUTING, "connection_type"
+                    ): ConnectionSchema.ROUTING_SCHEMA,
+                    vol.Exclusive(
+                        CONF_XKNX_TUNNELING, "connection_type"
+                    ): ConnectionSchema.TUNNELING_SCHEMA,
+                    vol.Optional(CONF_XKNX_FIRE_EVENT): cv.boolean,
+                    vol.Optional(CONF_XKNX_EVENT_FILTER, default=[]): vol.All(
+                        cv.ensure_list, [cv.string]
+                    ),
+                    vol.Optional(
+                        CONF_XKNX_INDIVIDUAL_ADDRESS, default=XKNX.DEFAULT_ADDRESS
+                    ): cv.string,
+                    vol.Optional(
+                        CONF_XKNX_MCAST_GRP, default=DEFAULT_MCAST_GRP
+                    ): cv.string,
+                    vol.Optional(
+                        CONF_XKNX_MCAST_PORT, default=DEFAULT_MCAST_PORT
+                    ): cv.port,
+                    vol.Optional(CONF_XKNX_STATE_UPDATER, default=True): cv.boolean,
+                    vol.Optional(CONF_XKNX_RATE_LIMIT, default=20): vol.All(
+                        vol.Coerce(int), vol.Range(min=1, max=100)
+                    ),
+                    vol.Optional(CONF_XKNX_EXPOSE): vol.All(
+                        cv.ensure_list, [ExposeSchema.SCHEMA]
+                    ),
+                    vol.Optional(SupportedPlatforms.cover.value): vol.All(
+                        cv.ensure_list, [CoverSchema.SCHEMA]
+                    ),
+                    vol.Optional(SupportedPlatforms.binary_sensor.value): vol.All(
+                        cv.ensure_list, [BinarySensorSchema.SCHEMA]
+                    ),
+                    vol.Optional(SupportedPlatforms.light.value): vol.All(
+                        cv.ensure_list, [LightSchema.SCHEMA]
+                    ),
+                    vol.Optional(SupportedPlatforms.climate.value): vol.All(
+                        cv.ensure_list, [ClimateSchema.SCHEMA]
+                    ),
+                    vol.Optional(SupportedPlatforms.notify.value): vol.All(
+                        cv.ensure_list, [NotifySchema.SCHEMA]
+                    ),
+                    vol.Optional(SupportedPlatforms.switch.value): vol.All(
+                        cv.ensure_list, [SwitchSchema.SCHEMA]
+                    ),
+                    vol.Optional(SupportedPlatforms.sensor.value): vol.All(
+                        cv.ensure_list, [SensorSchema.SCHEMA]
+                    ),
+                    vol.Optional(SupportedPlatforms.scene.value): vol.All(
+                        cv.ensure_list, [SceneSchema.SCHEMA]
+                    ),
+                    vol.Optional(SupportedPlatforms.weather.value): vol.All(
+                        cv.ensure_list, [WeatherSchema.SCHEMA]
+                    ),
+                }
+            ),
         )
     },
     extra=vol.ALLOW_EXTRA,
@@ -151,6 +162,13 @@ SERVICE_XKNX_SEND_SCHEMA = vol.Any(
     ),
 )
 
+SERVICE_XKNX_EVENT_REGISTER_SCHEMA = vol.Schema(
+    {
+        vol.Required(SERVICE_XKNX_ATTR_ADDRESS): cv.string,
+        vol.Optional(SERVICE_XKNX_ATTR_REMOVE, default=False): cv.boolean,
+    }
+)
+
 
 async def async_setup(hass, config):
     """Set up the KNX component."""
@@ -188,6 +206,14 @@ async def async_setup(hass, config):
         schema=SERVICE_XKNX_SEND_SCHEMA,
     )
 
+    async_register_admin_service(
+        hass,
+        DOMAIN,
+        SERVICE_XKNX_EVENT_REGISTER,
+        hass.data[DOMAIN].service_event_register_modify,
+        schema=SERVICE_XKNX_EVENT_REGISTER_SCHEMA,
+    )
+
     async def reload_service_handler(service_call: ServiceCallType) -> None:
         """Remove all KNX components and load new ones from config."""
 
@@ -221,10 +247,11 @@ class KNXModule:
         self.hass = hass
         self.config = config
         self.connected = False
-        self.init_xknx()
-        self.register_callbacks()
         self.exposures = []
 
+        self.init_xknx()
+        self._knx_event_callback: TelegramQueue.Callback = self.register_callback()
+
     def init_xknx(self):
         """Initialize of KNX object."""
         self.xknx = XKNX(
@@ -289,19 +316,6 @@ class KNXModule:
             auto_reconnect=True,
         )
 
-    def register_callbacks(self):
-        """Register callbacks within XKNX object."""
-        if (
-            CONF_XKNX_FIRE_EVENT in self.config[DOMAIN]
-            and self.config[DOMAIN][CONF_XKNX_FIRE_EVENT]
-        ):
-            address_filters = list(
-                map(AddressFilter, self.config[DOMAIN][CONF_XKNX_FIRE_EVENT_FILTER])
-            )
-            self.xknx.telegram_queue.register_telegram_received_cb(
-                self.telegram_received_cb, address_filters
-            )
-
     @callback
     def async_create_exposures(self):
         """Create exposures."""
@@ -349,6 +363,25 @@ class KNXModule:
             },
         )
 
+    def register_callback(self) -> TelegramQueue.Callback:
+        """Register callback within XKNX TelegramQueue."""
+        address_filters = list(
+            map(AddressFilter, self.config[DOMAIN][CONF_XKNX_EVENT_FILTER])
+        )
+        return self.xknx.telegram_queue.register_telegram_received_cb(
+            self.telegram_received_cb,
+            address_filters=address_filters,
+            group_addresses=[],
+        )
+
+    async def service_event_register_modify(self, call):
+        """Service for adding or removing a GroupAddress to the knx_event filter."""
+        group_address = GroupAddress(call.data.get(SERVICE_XKNX_ATTR_ADDRESS))
+        if call.data.get(SERVICE_XKNX_ATTR_REMOVE):
+            self._knx_event_callback.group_addresses.remove(group_address)
+        elif group_address not in self._knx_event_callback.group_addresses:
+            self._knx_event_callback.group_addresses.append(group_address)
+
     async def service_send_to_knx_bus(self, call):
         """Service for sending an arbitrary KNX message to the KNX bus."""
         attr_payload = call.data.get(SERVICE_XKNX_ATTR_PAYLOAD)
diff --git a/home-assistant-plugin/custom_components/xknx/services.yaml b/home-assistant-plugin/custom_components/xknx/services.yaml
index 03d4e69b..cab8c100 100644
--- a/home-assistant-plugin/custom_components/xknx/services.yaml
+++ b/home-assistant-plugin/custom_components/xknx/services.yaml
@@ -10,3 +10,11 @@ send:
     type:
       description: "Optional. If set, the payload will not be sent as raw bytes, but encoded as given DPT. Knx sensor types are valid values (see https://www.home-assistant.io/integrations/sensor.knx)."
       example: "temperature"
+event_register:
+  description: "Add or remove single group address to knx_event filter for triggering `knx_event`s. Only addresses added with this service can be removed."
+  fields:
+    address:
+      description: "Group address that shall be added or removed."
+      example: "1/1/0"
+    remove:
+      description: "Optional. If `True` the group address will be removed. Defaults to `False`."
diff --git a/xknx/core/telegram_queue.py b/xknx/core/telegram_queue.py
index 9c6cbc36..c793ebc4 100644
--- a/xknx/core/telegram_queue.py
+++ b/xknx/core/telegram_queue.py
@@ -36,20 +36,25 @@ class TelegramQueue:
             self,
             callback: "AsyncTelegramCallback",
             address_filters: Optional[List[AddressFilter]] = None,
+            group_addresses: Optional[List[GroupAddress]] = None,
         ):
             """Initialize Callback class."""
             self.callback = callback
-            self.address_filters = address_filters
+            self._match_all = address_filters is None and group_addresses is None
+            self.address_filters = [] if address_filters is None else address_filters
+            self.group_addresses = [] if group_addresses is None else group_addresses
 
         def is_within_filter(self, telegram: Telegram) -> bool:
             """Test if callback is filtering for group address."""
-            if self.address_filters is None:
+            if self._match_all:
                 return True
-            for address_filter in self.address_filters:
-                if isinstance(
-                    telegram.destination_address, GroupAddress
-                ) and address_filter.match(telegram.destination_address):
-                    return True
+            if isinstance(telegram.destination_address, GroupAddress):
+                for address_filter in self.address_filters:
+                    if address_filter.match(telegram.destination_address):
+                        return True
+                for group_address in self.group_addresses:
+                    if telegram.destination_address == group_address:
+                        return True
             return False
 
     def __init__(self, xknx: "XKNX"):
@@ -63,9 +68,14 @@ class TelegramQueue:
         self,
         telegram_received_cb: "AsyncTelegramCallback",
         address_filters: Optional[List[AddressFilter]] = None,
+        group_addresses: Optional[List[GroupAddress]] = None,
     ) -> Callback:
         """Register callback for a telegram beeing received from KNX bus."""
-        callback = TelegramQueue.Callback(telegram_received_cb, address_filters)
+        callback = TelegramQueue.Callback(
+            telegram_received_cb,
+            address_filters=address_filters,
+            group_addresses=group_addresses,
+        )
         self.telegram_received_cbs.append(callback)
         return callback
 
diff --git a/xknx/dpt/__init__.py b/xknx/dpt/__init__.py
index 13fb10e3..aa0c3182 100644
--- a/xknx/dpt/__init__.py
+++ b/xknx/dpt/__init__.py
@@ -5,7 +5,7 @@ Module for encoding and decoding KNX datatypes.
 * Derived KNX Values like Scaling, Temperature
 """
 # flake8: noqa
-from .dpt import DPTArray, DPTBase, DPTBinary, DPTComparator
+from .dpt import DPTArray, DPTBase, DPTBinary
 from .dpt_1byte_signed import DPTPercentV8, DPTSignedRelativeValue, DPTValue1Count
 from .dpt_1byte_uint import (
     DPTDecimalFactor,
@@ -208,7 +208,6 @@ __all__ = [
     "DPTChargeDensityVolume",
     "DPTColorTemperature",
     "DPTCommonTemperature",
-    "DPTComparator",
     "DPTCompressibility",
     "DPTConductance",
     "DPTControllerStatus",
diff --git a/xknx/dpt/dpt.py b/xknx/dpt/dpt.py
index 4cc3c2b9..f1363dc7 100644
--- a/xknx/dpt/dpt.py
+++ b/xknx/dpt/dpt.py
@@ -128,7 +128,7 @@ class DPTBinary:
     APCI_BITMASK = 0x3F
     APCI_MAX_VALUE = APCI_BITMASK
 
-    def __init__(self, value) -> None:
+    def __init__(self, value: int) -> None:
         """Initialize DPTBinary class."""
         if not isinstance(value, int):
             raise TypeError()
@@ -137,9 +137,11 @@ class DPTBinary:
 
         self.value = value
 
-    def __eq__(self, other):
+    def __eq__(self, other: object) -> bool:
         """Equal operator."""
-        return DPTComparator.compare(self, other)
+        if isinstance(other, DPTBinary):
+            return self.value == other.value
+        return False
 
     def __str__(self) -> str:
         """Return object as readable string."""
@@ -163,49 +165,12 @@ class DPTArray:
         else:
             raise TypeError()
 
-    def __eq__(self, other):
+    def __eq__(self, other: object) -> bool:
         """Equal operator."""
-        return DPTComparator.compare(self, other)
+        if isinstance(other, DPTArray):
+            return self.value == other.value
+        return False
 
     def __str__(self) -> str:
         """Return object as readable string."""
         return '<DPTArray value="[{}]" />'.format(",".join(hex(b) for b in self.value))
-
-
-class DPTComparator:
-    """Helper class to compare different types of DPT objects."""
-
-    # pylint: disable=too-few-public-methods
-
-    @staticmethod
-    def compare(a, b):
-        """Test if 'a' and 'b' are the same."""
-        # pylint: disable=invalid-name,too-many-return-statements,len-as-condition
-        if a is None and b is None:
-            return True
-
-        if a is None:
-            if isinstance(b, DPTBinary):
-                return b.value == 0
-            if isinstance(b, DPTArray):
-                return len(b.value) == 0
-
-        if b is None:
-            if isinstance(a, DPTBinary):
-                return a.value == 0
-            if isinstance(a, DPTArray):
-                return len(a.value) == 0
-
-        if isinstance(a, DPTArray) and isinstance(b, DPTArray):
-            return a.value == b.value
-
-        if isinstance(a, DPTBinary) and isinstance(b, DPTBinary):
-            return a.value == b.value
-
-        if isinstance(a, DPTBinary) and isinstance(b, DPTArray):
-            return a.value == 0 and len(b.value) == 0
-
-        if isinstance(a, DPTArray) and isinstance(b, DPTBinary):
-            return len(a.value) == 0 and b.value == 0
-
-        raise TypeError()


Test patch:
diff --git a/test/core_tests/telegram_queue_test.py b/test/core_tests/telegram_queue_test.py
index 0c27b794..9543f66b 100644
--- a/test/core_tests/telegram_queue_test.py
+++ b/test/core_tests/telegram_queue_test.py
@@ -107,12 +107,7 @@ class TestTelegramQueue(unittest.TestCase):
         """Test telegram_received_callback after state of switch was changed."""
         # pylint: disable=no-self-use
         xknx = XKNX()
-
-        telegram_received_callback = Mock()
-
-        async def async_telegram_received_cb(device):
-            """Async callback."""
-            telegram_received_callback(device)
+        async_telegram_received_cb = AsyncMock()
 
         xknx.telegram_queue.register_telegram_received_cb(async_telegram_received_cb)
 
@@ -124,7 +119,7 @@ class TestTelegramQueue(unittest.TestCase):
         self.loop.run_until_complete(
             xknx.telegram_queue.process_telegram_incoming(telegram)
         )
-        telegram_received_callback.assert_called_once_with(telegram)
+        async_telegram_received_cb.assert_called_once_with(telegram)
 
     #
     # TEST UNREGISTER
@@ -259,20 +254,14 @@ class TestTelegramQueue(unittest.TestCase):
             ),
         )
 
-    @patch("xknx.core.TelegramQueue.process_telegram_outgoing")
-    @patch("xknx.core.TelegramQueue.process_telegram_incoming")
+    @patch("xknx.core.TelegramQueue.process_telegram_outgoing", new_callable=AsyncMock)
+    @patch("xknx.core.TelegramQueue.process_telegram_incoming", new_callable=AsyncMock)
     def test_process_all_telegrams(
         self, process_telegram_incoming_mock, process_telegram_outgoing_mock
     ):
         """Test _process_all_telegrams for clearing the queue."""
-        # pylint: disable=no-self-use
         xknx = XKNX()
 
-        async_process_mock = asyncio.Future()
-        async_process_mock.set_result(None)
-        process_telegram_incoming_mock.return_value = async_process_mock
-        process_telegram_outgoing_mock.return_value = async_process_mock
-
         telegram_in = Telegram(
             destination_address=GroupAddress("1/2/3"),
             direction=TelegramDirection.INCOMING,
@@ -289,24 +278,20 @@ class TestTelegramQueue(unittest.TestCase):
         res = self.loop.run_until_complete(xknx.telegram_queue._process_all_telegrams())
 
         self.assertIsNone(res)
-        self.assertEqual(process_telegram_incoming_mock.call_count, 1)
-        self.assertEqual(process_telegram_outgoing_mock.call_count, 1)
+        process_telegram_incoming_mock.assert_called_once()
+        process_telegram_outgoing_mock.assert_called_once()
 
     #
     # TEST NO FILTERS
     #
-    def test_no_filters(self):
+    def test_callback_no_filters(self):
         """Test telegram_received_callback after state of switch was changed."""
-        # pylint: disable=no-self-use
         xknx = XKNX()
+        async_telegram_received_callback = AsyncMock()
 
-        telegram_received_callback = Mock()
-
-        async def async_telegram_received_cb(device):
-            """Async callback."""
-            telegram_received_callback(device)
-
-        xknx.telegram_queue.register_telegram_received_cb(async_telegram_received_cb)
+        xknx.telegram_queue.register_telegram_received_cb(
+            async_telegram_received_callback
+        )
 
         telegram = Telegram(
             destination_address=GroupAddress("1/2/3"),
@@ -316,25 +301,19 @@ class TestTelegramQueue(unittest.TestCase):
         xknx.telegrams.put_nowait(telegram)
         self.loop.run_until_complete(xknx.telegram_queue._process_all_telegrams())
 
-        telegram_received_callback.assert_called_with(telegram)
+        async_telegram_received_callback.assert_called_with(telegram)
 
     #
     # TEST POSITIVE FILTERS
     #
-    def test_positive_filters(self):
+    def test_callback_positive_address_filters(self):
         """Test telegram_received_callback after state of switch was changed."""
-        # pylint: disable=no-self-use
         xknx = XKNX()
-
-        telegram_received_callback = Mock()
-
-        async def async_telegram_received_cb(device):
-            """Async callback."""
-            telegram_received_callback(device)
+        async_telegram_received_callback = AsyncMock()
 
         xknx.telegram_queue.register_telegram_received_cb(
-            async_telegram_received_cb,
-            [AddressFilter("2/4-8/*"), AddressFilter("1/2/-8")],
+            async_telegram_received_callback,
+            address_filters=[AddressFilter("2/4-8/*"), AddressFilter("1/2/-8")],
         )
 
         telegram = Telegram(
@@ -345,25 +324,19 @@ class TestTelegramQueue(unittest.TestCase):
         xknx.telegrams.put_nowait(telegram)
         self.loop.run_until_complete(xknx.telegram_queue._process_all_telegrams())
 
-        telegram_received_callback.assert_called_with(telegram)
+        async_telegram_received_callback.assert_called_with(telegram)
 
     #
     # TEST NEGATIVE FILTERS
     #
-    def test_negative_filters(self):
+    def test_callback_negative_address_filters(self):
         """Test telegram_received_callback after state of switch was changed."""
-        # pylint: disable=no-self-use
         xknx = XKNX()
-
-        telegram_received_callback = Mock()
-
-        async def async_telegram_received_cb(device):
-            """Async callback."""
-            telegram_received_callback(device)
+        async_telegram_received_callback = AsyncMock()
 
         xknx.telegram_queue.register_telegram_received_cb(
-            async_telegram_received_cb,
-            [AddressFilter("2/4-8/*"), AddressFilter("1/2/8-")],
+            async_telegram_received_callback,
+            address_filters=[AddressFilter("2/4-8/*"), AddressFilter("1/2/8-")],
         )
 
         telegram = Telegram(
@@ -374,4 +347,40 @@ class TestTelegramQueue(unittest.TestCase):
         xknx.telegrams.put_nowait(telegram)
         self.loop.run_until_complete(xknx.telegram_queue._process_all_telegrams())
 
-        telegram_received_callback.assert_not_called()
+        async_telegram_received_callback.assert_not_called()
+
+    def test_callback_group_addresses(self):
+        """Test telegram_received_callback after state of switch was changed."""
+        xknx = XKNX()
+        async_telegram_received_cb_one = AsyncMock()
+        async_telegram_received_cb_two = AsyncMock()
+
+        callback_one = xknx.telegram_queue.register_telegram_received_cb(
+            async_telegram_received_cb_one,
+            address_filters=[],
+            group_addresses=[GroupAddress("1/2/3")],
+        )
+        callback_two = xknx.telegram_queue.register_telegram_received_cb(
+            async_telegram_received_cb_two, address_filters=[], group_addresses=[]
+        )
+
+        telegram = Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            direction=TelegramDirection.INCOMING,
+            payload=GroupValueWrite(DPTBinary(1)),
+        )
+        self.loop.run_until_complete(
+            xknx.telegram_queue.process_telegram_incoming(telegram)
+        )
+        async_telegram_received_cb_one.assert_called_once_with(telegram)
+        async_telegram_received_cb_two.assert_not_called()
+
+        async_telegram_received_cb_one.reset_mock()
+        # modify the filters - add/remove a GroupAddress
+        callback_one.group_addresses.remove(GroupAddress("1/2/3"))
+        callback_two.group_addresses.append(GroupAddress("1/2/3"))
+        self.loop.run_until_complete(
+            xknx.telegram_queue.process_telegram_incoming(telegram)
+        )
+        async_telegram_received_cb_one.assert_not_called()
+        async_telegram_received_cb_two.assert_called_once_with(telegram)
diff --git a/test/dpt_tests/dpt_test.py b/test/dpt_tests/dpt_test.py
index 95a3575c..873356f8 100644
--- a/test/dpt_tests/dpt_test.py
+++ b/test/dpt_tests/dpt_test.py
@@ -1,7 +1,7 @@
 """Unit test for KNX binary/integer objects."""
 import unittest
 
-from xknx.dpt import DPTArray, DPTBase, DPTBinary, DPTComparator
+from xknx.dpt import DPTArray, DPTBase, DPTBinary
 from xknx.exceptions import ConversionError
 
 
@@ -13,6 +13,8 @@ class TestDPT(unittest.TestCase):
     def test_compare_binary(self):
         """Test comparison of DPTBinary objects."""
         self.assertEqual(DPTBinary(0), DPTBinary(0))
+        self.assertEqual(DPTBinary(0), DPTBinary(False))
+        self.assertEqual(DPTBinary(1), DPTBinary(True))
         self.assertEqual(DPTBinary(2), DPTBinary(2))
         self.assertNotEqual(DPTBinary(1), DPTBinary(4))
         self.assertNotEqual(DPTBinary(2), DPTBinary(0))
@@ -30,11 +32,11 @@ class TestDPT(unittest.TestCase):
         self.assertNotEqual(DPTArray((1, 2, 3)), DPTArray([1, 2, 4]))
 
     def test_compare_none(self):
-        """Test comparison of empty DPTArray objects with None."""
-        self.assertEqual(DPTArray(()), None)
-        self.assertEqual(None, DPTArray(()))
-        self.assertEqual(DPTBinary(0), None)
-        self.assertEqual(None, DPTBinary(0))
+        """Test comparison DPTArray objects with None."""
+        self.assertNotEqual(DPTArray(()), None)
+        self.assertNotEqual(None, DPTArray(()))
+        self.assertNotEqual(DPTBinary(0), None)
+        self.assertNotEqual(None, DPTBinary(0))
         self.assertNotEqual(DPTArray((1, 2, 3)), None)
         self.assertNotEqual(None, DPTArray((1, 2, 3)))
         self.assertNotEqual(DPTBinary(1), None)
@@ -42,8 +44,10 @@ class TestDPT(unittest.TestCase):
 
     def test_compare_array_binary(self):
         """Test comparison of empty DPTArray objects with DPTBinary objects."""
-        self.assertEqual(DPTArray(()), DPTBinary(0))
-        self.assertEqual(DPTBinary(0), DPTArray(()))
+        self.assertNotEqual(DPTArray(()), DPTBinary(0))
+        self.assertNotEqual(DPTBinary(0), DPTArray(()))
+        self.assertNotEqual(DPTBinary(0), DPTArray(0))
+        self.assertNotEqual(DPTBinary(1), DPTArray(1))
         self.assertNotEqual(DPTArray((1, 2, 3)), DPTBinary(2))
         self.assertNotEqual(DPTBinary(2), DPTArray((1, 2, 3)))
         self.assertNotEqual(DPTArray((2,)), DPTBinary(2))
@@ -68,17 +72,6 @@ class TestDPT(unittest.TestCase):
         with self.assertRaises(TypeError):
             DPTArray("bla")
 
-    def test_dpt_comparator_none_with_none(self):
-        """Test comperator for DPTBinary and DPTBinary - missing cases."""
-        self.assertTrue(DPTComparator.compare(None, None))
-        self.assertTrue(DPTComparator.compare(None, DPTBinary(0)))
-        self.assertTrue(DPTComparator.compare(None, DPTArray(())))
-
-    def test_dpt_comparator_none_with_wrong_parameter(self):
-        """Test comperator for DPTBinary and DPTBinary - wrong parameter."""
-        with self.assertRaises(TypeError):
-            DPTComparator.compare("bla", DPTBinary(0))
-
 
 class TestDPTBase(unittest.TestCase):
     """Test class for transcoder base object."""
diff --git a/test/knxip_tests/cemi_frame_test.py b/test/knxip_tests/cemi_frame_test.py
index 0dbdc19c..c3fd05ff 100644
--- a/test/knxip_tests/cemi_frame_test.py
+++ b/test/knxip_tests/cemi_frame_test.py
@@ -3,7 +3,7 @@
 from unittest.mock import MagicMock
 
 from pytest import fixture, raises
-from xknx.dpt import DPTBinary, DPTComparator
+from xknx.dpt import DPTBinary
 from xknx.exceptions import ConversionError, CouldNotParseKNXIP, UnsupportedCEMIMessage
 from xknx.knxip.cemi_frame import CEMIFrame
 from xknx.knxip.knxip_enum import CEMIFlags, CEMIMessageCode
@@ -92,7 +92,7 @@ def test_invalid_payload(frame):
     frame.code = CEMIMessageCode.L_DATA_IND
     frame.flags = 0
     frame.mpdu_len = 1
-    frame.payload = DPTComparator()
+    frame.payload = DPTBinary(1)
     frame.src_addr = IndividualAddress(0)
     frame.dst_addr = IndividualAddress(0)