Repository: XKNX/xknx
Pull request #457 (2020-10-11 07:19:52)

StateUpdater: only request one group address at once
<!--
  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.
-->
- Request one group address at a time. StateUpdater now waits for a response (or timeout) and only then triggers the next GroupValueRead request.
- set timeout for GroupValueRead requests back to 1 sec (from 2)
- remove some debug logs

This PR does what #364 was intended for 🙃

## 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.
-->

- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

## Checklist:

- [x] I have performed a self-review of my own code
- [x] The documentation has been adjusted accordingly
- [ ] The changes generate no new warnings
- [ ] Tests have been added that prove the fix is effective or that the feature works
- [x] The changes are documented in the changelog
- [ ] The Homeassistant plugin has been adjusted in case of new config options


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

Coverage decreased (-0.02%) to 93.689% when pulling **8fece81b04d27b7625516623c51574bc452003c9 on farmio:state-updater-one-by-one** into **0ecf6ee88aefc742f6a44f7ed88683838a342b0c on XKNX:master**.


Patch:
diff --git a/changelog.md b/changelog.md
index 2d1038e8..cbbf8cb2 100644
--- a/changelog.md
+++ b/changelog.md
@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased changes
+
+### Internals
+
+- StateUpdater: Only request one GA at a time.
+
 ## 0.15.1 bugfix for binary sensors
 
 ### Devices
diff --git a/xknx/core/state_updater.py b/xknx/core/state_updater.py
index 65ff86be..268730df 100644
--- a/xknx/core/state_updater.py
+++ b/xknx/core/state_updater.py
@@ -1,14 +1,12 @@
 """Module for keeping the value of a RemoteValue from KNX bus up to date."""
 import asyncio
 from enum import Enum
-from functools import partial
 import logging
 
-from xknx.exceptions import ConversionError
 from xknx.remote_value import RemoteValue
-from xknx.telegram import GroupAddress
 
 DEFAULT_UPDATE_INTERVAL = 60
+MAX_UPDATE_INTERVAL = 1440
 
 logger = logging.getLogger("xknx.state_updater")
 
@@ -21,6 +19,7 @@ class StateUpdater:
         self.xknx = xknx
         self.started = False
         self._workers = {}
+        self._one_by_one = asyncio.Lock()
 
     def register_remote_value(self, remote_value: RemoteValue, tracker_options=True):
         """Register a RemoteValue to initialize its state and/or track for expiration."""
@@ -66,24 +65,41 @@ class StateUpdater:
                     tracker_type,
                     update_interval,
                 )
+            if update_interval > MAX_UPDATE_INTERVAL:
+                logger.warning(
+                    "StateUpdater interval of %s to long for %s. Using maximum of %s minutes (1 day)",
+                    tracker_options,
+                    remote_value,
+                    MAX_UPDATE_INTERVAL,
+                )
+                update_interval = MAX_UPDATE_INTERVAL
             return (tracker_type, update_interval)
 
-        tracker_type, update_inteval = parse_tracker_options(tracker_options)
+        async def read_state_mutex():
+            """Schedule to read the state from the KNX bus - one at a time."""
+            async with self._one_by_one:
+                # wait until there is nothing else to send to the bus
+                await self.xknx.telegram_queue.outgoing_queue.join()
+                logger.debug(
+                    "StateUpdater reading %s for %s - %s",
+                    remote_value.group_address_state,
+                    remote_value.device_name,
+                    remote_value.feature_name,
+                )
+                await remote_value.read_state(wait_for_result=True)
+
+        tracker_type, update_interval = parse_tracker_options(tracker_options)
         tracker = _StateTracker(
-            self.xknx,
-            device_name=remote_value.device_name,
-            feature_name=remote_value.feature_name,
-            group_address=remote_value.group_address_state,
             tracker_type=tracker_type,
-            interval_min=update_inteval,
-            read_state_awaitable=partial(remote_value.read_state, wait_for_result=True),
+            interval_min=update_interval,
+            read_state_awaitable=read_state_mutex,
         )
         self._workers[id(remote_value)] = tracker
 
         logger.debug(
             "StateUpdater registered %s %s for %s",
             tracker_type,
-            update_inteval,
+            update_interval,
             remote_value,
         )
         if self.started:
@@ -127,43 +143,26 @@ class _StateTracker:
     # pylint: disable=too-many-instance-attributes
     def __init__(
         self,
-        xknx,
-        device_name: str = None,
-        feature_name: str = None,
-        group_address: GroupAddress = None,
         tracker_type: StateTrackerType = StateTrackerType.EXPIRE,
         interval_min: int = 60,
         read_state_awaitable=None,
     ):
         """Initialize StateTracker class."""
         # pylint: disable=too-many-arguments
-        if interval_min > 1440:
-            raise ConversionError(
-                "Interval_min to long. Maximum is 1440 minutes (1 day)",
-                interval_min=interval_min,
-                device_name=device_name,
-                feature_name=feature_name,
-            )
-        self.xknx = xknx
-        self.device_name = device_name
-        self.feature_name = feature_name
-        self.group_address = group_address
         self.tracker_type = tracker_type
         self.update_interval = interval_min * 60
-        self._read_state_awaitable = read_state_awaitable
+        self._read_state = read_state_awaitable
         self._task = None
 
     def start(self):
         """Start StateTracker - read state on call."""
-        logger.debug(
-            "StateUpdater initializing %s for %s - %s",
-            self.group_address,
-            self.device_name,
-            self.feature_name,
-        )
+        self._task = asyncio.create_task(self._start_init())
+
+    async def _start_init(self):
+        """Initialize state, start update loop if appropriate."""
+        await self._read_state()
         if self.tracker_type is not StateTrackerType.INIT:
             self._start_waiting()
-        self._read_state()
 
     def _start_waiting(self):
         """Start StateTracker - wait for value to expire."""
@@ -192,16 +191,4 @@ class _StateTracker:
         #   when no telegram was received it will try again endlessly
         while True:
             await asyncio.sleep(self.update_interval)
-            # wait until there is nothing else to send to the bus
-            await self.xknx.telegram_queue.outgoing_queue.join()
-            self._read_state()
-
-    def _read_state(self):
-        """Schedule to read the state from the KNX bus."""
-        logger.debug(
-            "StateUpdater scheduled reading %s for %s - %s",
-            self.group_address,
-            self.device_name,
-            self.feature_name,
-        )
-        asyncio.create_task(self._read_state_awaitable())
+            await self._read_state()
diff --git a/xknx/core/value_reader.py b/xknx/core/value_reader.py
index fddf7f02..5a2307d9 100644
--- a/xknx/core/value_reader.py
+++ b/xknx/core/value_reader.py
@@ -20,7 +20,7 @@ class ValueReader:
 
     # pylint: disable=too-many-instance-attributes
 
-    def __init__(self, xknx, group_address, timeout_in_seconds=2):
+    def __init__(self, xknx, group_address, timeout_in_seconds=1):
         """Initialize ValueReader class."""
         self.xknx = xknx
         self.group_address = group_address
diff --git a/xknx/remote_value/remote_value.py b/xknx/remote_value/remote_value.py
index f63cc03e..6951ce47 100644
--- a/xknx/remote_value/remote_value.py
+++ b/xknx/remote_value/remote_value.py
@@ -176,7 +176,6 @@ class RemoteValue:
     async def read_state(self, wait_for_result=False):
         """Send GroupValueRead telegram for state address to KNX bus."""
         if self.readable:
-            logger.debug("Sync %s - %s", self.device_name, self.feature_name)
             # pylint: disable=import-outside-toplevel
             # TODO: send a ReadRequset and start a timeout from here instead of ValueReader
             #       cancel timeout form process(); delete ValueReader


Test patch:
diff --git a/test/core_tests/state_updater_test.py b/test/core_tests/state_updater_test.py
index 3b05210d..67220a39 100644
--- a/test/core_tests/state_updater_test.py
+++ b/test/core_tests/state_updater_test.py
@@ -142,13 +142,18 @@ class TestStateUpdater(unittest.TestCase):
         self.assertEqual(_get_only_tracker().tracker_type, StateTrackerType.EXPIRE)
         self.assertEqual(_get_only_tracker().update_interval, 60 * 60)
         remote_value_float.__del__()
+        logging_warning_mock.reset_mock()
         # intervall too long
-        with (self.assertRaises(ConversionError)):
-            remote_value_long = RemoteValue(
-                xknx, sync_state=1441, group_address_state=GroupAddress("1/1/1")
-            )
-
-            remote_value_long.__del__()
+        remote_value_long = RemoteValue(
+            xknx, sync_state=1441, group_address_state=GroupAddress("1/1/1")
+        )
+        logging_warning_mock.assert_called_once_with(
+            "StateUpdater interval of %s to long for %s. Using maximum of %s minutes (1 day)",
+            1441,
+            remote_value_long,
+            1440,
+        )
+        remote_value_long.__del__()
 
     def test_state_updater_start_update_stop(self):
         """Test start, update_received and stop of StateUpdater."""