Added sdbus.utils.parse_properties_changed function

igo95862 · igo95862 · commit 2c4d0bffb017 · 2023-01-15T16:13:34.000+06:00
Parses the properties changed data in to a simple dictionary
there keys are translated to python names and invalidated
properties will have value of None.
diff --git a/docs/asyncio_api.rst b/docs/asyncio_api.rst
@@ -75,6 +75,9 @@ Classes
 
         Signal when one of the objects properties changes.
 
+        :py:func:`sdbus.utils.parse_properties_changed` can be used to transform
+        this signal data in to an easier to work with dictionary.
+
         Signal data is:
 
         Interface name : str
diff --git a/docs/index.rst b/docs/index.rst
@@ -38,6 +38,7 @@ If you are unfamiliar with D-Bus you might want to read following pages:
     asyncio_quick
     asyncio_api
     exceptions
+    utils
     examples
     proxies
     code_generator
diff --git a/docs/utils.rst b/docs/utils.rst
@@ -0,0 +1,22 @@
+Utilities
+=========
+
+Parsing utilities
++++++++++++++++++
+
+.. py:currentmodule:: sdbus.utils
+
+.. py:function:: parse_properties_changed(interface, properties_changed_data, on_unknown_member='error')
+
+    Parse data from :py:meth:`properties_changed <sdbus.DbusInterfaceCommonAsync.properties_changed>` signal.
+
+    Member names will be translated to python defined names.
+    Invalidated properties will have a value of None.
+
+    :param DbusInterfaceBaseAsync interface: Takes either D-Bus interface or interface class.
+    :param Tuple properties_changed_data: Tuple caught from signal.
+    :param str on_unknown_member: If an unknown D-Bus property was encountered
+            either raise an ``"error"`` (default), ``"ignore"`` the property
+            or ``"reuse"`` the D-Bus name for the member.
+    :returns: Dictionary of changed properties with keys translated to python
+            names. Invalidated properties will have value of None.
diff --git a/src/sdbus/dbus_common_funcs.py b/src/sdbus/dbus_common_funcs.py
@@ -22,7 +22,7 @@
 
 from asyncio import Future, get_running_loop
 from contextvars import ContextVar
-from typing import Generator, Iterator
+from typing import Any, Dict, Generator, Iterator, Literal, Tuple
 from warnings import warn
 
 from .sd_bus_internals import (
@@ -159,3 +159,29 @@ def _check_sync_in_async_env() -> bool:
         return False
     except RuntimeError:
         return True
+
+
+def _parse_properties_vardict(
+        properties_name_map: Dict[str, str],
+        properties_vardict: Dict[str, Tuple[str, Any]],
+        on_unknown_member: Literal['error', 'ignore', 'reuse'],
+) -> Dict[str, Any]:
+
+    properties_translated: Dict[str, Any] = {}
+
+    for member_name, variant in properties_vardict.items():
+        try:
+            python_name = properties_name_map[member_name]
+        except KeyError:
+            if on_unknown_member == 'error':
+                raise
+            elif on_unknown_member == 'ignore':
+                continue
+            elif on_unknown_member == 'reuse':
+                python_name = member_name
+            else:
+                raise ValueError
+
+        properties_translated[python_name] = variant[1]
+
+    return properties_translated
diff --git a/src/sdbus/dbus_proxy_async_interfaces.py b/src/sdbus/dbus_proxy_async_interfaces.py
@@ -22,7 +22,7 @@
 from inspect import getmembers
 from typing import Any, Dict, List, Literal, Optional, Tuple
 
-from .dbus_common_funcs import get_default_bus
+from .dbus_common_funcs import _parse_properties_vardict, get_default_bus
 from .dbus_proxy_async_interface_base import DbusInterfaceBaseAsync
 from .dbus_proxy_async_method import dbus_method_async
 from .dbus_proxy_async_property import DbusPropertyAsyncBinded
@@ -94,20 +94,13 @@ async def properties_get_all_dict(
             dbus_properties_data = await self._properties_get_all(
                 interface_name)
 
-            for member_name, variant in dbus_properties_data.items():
-                try:
-                    python_name = self._dbus_to_python_name_map[member_name]
-                except KeyError:
-                    if on_unknown_member == 'error':
-                        raise
-                    elif on_unknown_member == 'ignore':
-                        continue
-                    elif on_unknown_member == 'reuse':
-                        python_name = member_name
-                    else:
-                        raise ValueError
-
-                properties[python_name] = variant[1]
+            properties.update(
+                _parse_properties_vardict(
+                    self._dbus_to_python_name_map,
+                    dbus_properties_data,
+                    on_unknown_member,
+                )
+            )
 
         return properties
 
diff --git a/src/sdbus/utils.py b/src/sdbus/utils.py
@@ -0,0 +1,48 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+
+# Copyright (C) 2023 igo95862
+
+# This file is part of python-sdbus
+
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
+from __future__ import annotations
+
+from typing import Any, Dict, Literal, Type, Union
+
+from .dbus_common_funcs import _parse_properties_vardict
+from .dbus_proxy_async_interface_base import DbusInterfaceBaseAsync
+from .dbus_proxy_async_interfaces import DBUS_PROPERTIES_CHANGED_TYPING
+
+
+def parse_properties_changed(
+        interface: Union[DbusInterfaceBaseAsync, Type[DbusInterfaceBaseAsync]],
+        properties_changed_data: DBUS_PROPERTIES_CHANGED_TYPING,
+        on_unknown_member: Literal['error', 'ignore', 'reuse'] = 'error',
+) -> Dict[str, Any]:
+    changed_properties_data = properties_changed_data[1]
+
+    for invalidated_property in properties_changed_data[2]:
+        changed_properties_data[invalidated_property] = ('0', None)
+
+    return _parse_properties_vardict(
+        interface._dbus_to_python_name_map,
+        properties_changed_data[1],
+        on_unknown_member,
+    )
+
+
+__all__ = (
+    'parse_properties_changed',
+)
diff --git a/test/test_sd_bus_async.py b/test/test_sd_bus_async.py
@@ -1,6 +1,6 @@
 # SPDX-License-Identifier: LGPL-2.1-or-later
 
-# Copyright (C) 2020, 2021 igo95862
+# Copyright (C) 2020-2023 igo95862
 
 # This file is part of python-sdbus
 
@@ -22,10 +22,11 @@
 
 from asyncio import Event, get_running_loop, sleep, wait_for
 from asyncio.subprocess import create_subprocess_exec
-from typing import Tuple
+from typing import Tuple, cast
 from unittest import SkipTest
 
 from sdbus.dbus_common_funcs import PROPERTY_FLAGS_MASK, count_bits
+from sdbus.dbus_proxy_async_interfaces import DBUS_PROPERTIES_CHANGED_TYPING
 from sdbus.exceptions import (
     DbusFailedError,
     DbusFileExistsError,
@@ -41,6 +42,7 @@
     is_interface_name_valid,
 )
 from sdbus.unittest import IsolatedDbusTestCase
+from sdbus.utils import parse_properties_changed
 
 from sdbus import (
     DbusInterfaceCommonAsync,
@@ -823,3 +825,42 @@ async def test_empty_signal(self) -> None:
         self.assertIsNone(await wait_for(aw_dbus, timeout=1))
 
         self.assertIsNone(await wait_for(q.get(), timeout=1))
+
+    async def test_properties_changed(self) -> None:
+        test_object, test_object_connection = initialize_object()
+
+        test_str = 'should_be_emited'
+
+        q = await test_object_connection.properties_changed._get_dbus_queue()
+
+        async def set_property() -> None:
+            await test_object_connection.test_property.set_async(test_str)
+
+        await set_property()
+
+        properties_changed_data = cast(
+            DBUS_PROPERTIES_CHANGED_TYPING,
+            (await q.get()).get_contents(),
+        )
+
+        parsed_dict_from_class = parse_properties_changed(
+            TestInterface, properties_changed_data)
+        self.assertEqual(
+            test_str,
+            parsed_dict_from_class['test_property'],
+        )
+
+        parsed_dict_from_object = parse_properties_changed(
+            test_object_connection, properties_changed_data)
+        self.assertEqual(
+            test_str,
+            parsed_dict_from_object['test_property'],
+        )
+
+        properties_changed_data[2].append('invalidated_property')
+        parsed_dict_with_invalidation = parse_properties_changed(
+            test_object, properties_changed_data,
+            on_unknown_member='reuse',
+        )
+        self.assertIsNone(
+            parsed_dict_with_invalidation['invalidated_property'])
