Reduce the number of polling tasks (#21)

* Reduce the number of polling tasks

* missing looptime and test coverage

* coverage

tests/test_device.py

@@ -120,7 +120,7 @@ async def ota_zha_device(
 async def _send_time_changed(zha_gateway: Gateway, seconds: int):
     """Send a time changed event."""
     await asyncio.sleep(seconds)
-    await zha_gateway.async_block_till_done()
+    await zha_gateway.async_block_till_done(wait_background_tasks=True)
 
 
 @patch(
@@ -132,11 +132,15 @@ async def test_check_available_success(
     zha_gateway: Gateway,
     device_with_basic_cluster_handler: ZigpyDevice,  # pylint: disable=redefined-outer-name
     device_joined: Callable[[ZigpyDevice], Awaitable[Device]],
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     """Check device availability success on 1st try."""
     zha_device = await device_joined(device_with_basic_cluster_handler)
     basic_ch = device_with_basic_cluster_handler.endpoints[3].basic
 
+    assert not zha_device.is_coordinator
+    assert not zha_device.is_active_coordinator
+
     basic_ch.read_attributes.reset_mock()
     device_with_basic_cluster_handler.last_seen = None
     assert zha_device.available is True
@@ -156,22 +160,46 @@ def _update_last_seen(*args, **kwargs):  # pylint: disable=unused-argument
     basic_ch.read_attributes.side_effect = _update_last_seen
 
     # successfully ping zigpy device, but zha_device is not yet available
-    await _send_time_changed(zha_gateway, zha_device.__polling_interval + 1)
+    await _send_time_changed(
+        zha_gateway, zha_gateway._device_availability_checker.__polling_interval + 1
+    )
     assert basic_ch.read_attributes.await_count == 1
     assert basic_ch.read_attributes.await_args[0][0] == ["manufacturer"]
     assert zha_device.available is False
 
     # There was traffic from the device: pings, but not yet available
-    await _send_time_changed(zha_gateway, zha_device.__polling_interval + 1)
+    await _send_time_changed(
+        zha_gateway, zha_gateway._device_availability_checker.__polling_interval + 1
+    )
     assert basic_ch.read_attributes.await_count == 2
     assert basic_ch.read_attributes.await_args[0][0] == ["manufacturer"]
     assert zha_device.available is False
 
     # There was traffic from the device: don't try to ping, marked as available
-    await _send_time_changed(zha_gateway, zha_device.__polling_interval + 1)
+    await _send_time_changed(
+        zha_gateway, zha_gateway._device_availability_checker.__polling_interval + 1
+    )
     assert basic_ch.read_attributes.await_count == 2
     assert basic_ch.read_attributes.await_args[0][0] == ["manufacturer"]
     assert zha_device.available is True
+    assert zha_device.on_network is True
+
+    assert "Device is not on the network, marking unavailable" not in caplog.text
+    zha_device.on_network = False
+
+    assert zha_device.available is False
+    assert zha_device.on_network is False
+
+    sleep_time = max(
+        zha_gateway.global_updater.__polling_interval,
+        zha_gateway._device_availability_checker.__polling_interval,
+    )
+    sleep_time += 2
+
+    await asyncio.sleep(sleep_time)
+    await zha_gateway.async_block_till_done(wait_background_tasks=True)
+
+    assert "Device is not on the network, marking unavailable" in caplog.text
 
 
 @patch(
@@ -197,21 +225,27 @@ async def test_check_available_unsuccessful(
     )
 
     # unsuccessfully ping zigpy device, but zha_device is still available
-    await _send_time_changed(zha_gateway, zha_device.__polling_interval + 1)
+    await _send_time_changed(
+        zha_gateway, zha_gateway._device_availability_checker.__polling_interval + 1
+    )
 
     assert basic_ch.read_attributes.await_count == 1
     assert basic_ch.read_attributes.await_args[0][0] == ["manufacturer"]
     assert zha_device.available is True
 
     # still no traffic, but zha_device is still available
-    await _send_time_changed(zha_gateway, zha_device.__polling_interval + 1)
+    await _send_time_changed(
+        zha_gateway, zha_gateway._device_availability_checker.__polling_interval + 1
+    )
 
     assert basic_ch.read_attributes.await_count == 2
     assert basic_ch.read_attributes.await_args[0][0] == ["manufacturer"]
     assert zha_device.available is True
 
     # not even trying to update, device is unavailable
-    await _send_time_changed(zha_gateway, zha_device.__polling_interval + 1)
+    await _send_time_changed(
+        zha_gateway, zha_gateway._device_availability_checker.__polling_interval + 1
+    )
 
     assert basic_ch.read_attributes.await_count == 2
     assert basic_ch.read_attributes.await_args[0][0] == ["manufacturer"]
@@ -241,7 +275,9 @@ async def test_check_available_no_basic_cluster_handler(
     )
 
     assert "does not have a mandatory basic cluster" not in caplog.text
-    await _send_time_changed(zha_gateway, zha_device.__polling_interval + 1)
+    await _send_time_changed(
+        zha_gateway, zha_gateway._device_availability_checker.__polling_interval + 1
+    )
 
     assert zha_device.available is False
     assert "does not have a mandatory basic cluster" in caplog.text

tests/test_gateway.py

@@ -173,6 +173,7 @@ async def test_device_left(
     zha_gateway.device_left(zigpy_dev_basic)
     await zha_gateway.async_block_till_done()
     assert zha_dev_basic.available is False
+    assert zha_dev_basic.on_network is False
 
 
 async def test_gateway_group_methods(
@@ -390,6 +391,7 @@ async def test_gateway_force_multi_pan_channel(
 
 
 @pytest.mark.parametrize("radio_concurrency", [1, 2, 8])
+@pytest.mark.looptime
 async def test_startup_concurrency_limit(
     radio_concurrency: int,
     zigpy_app_controller: ControllerApplication,
@@ -567,3 +569,57 @@ def test_gateway_raw_device_initialized(
             event="raw_device_initialized",
         ),
     )
+
+
+@pytest.mark.looptime
+async def test_pollers_skip(
+    zha_gateway: Gateway,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test pollers skip when they should."""
+
+    assert "Global updater interval skipped" not in caplog.text
+    assert "Device availability checker interval skipped" not in caplog.text
+
+    assert zha_gateway.config.allow_polling is True
+    zha_gateway.config.allow_polling = False
+    assert zha_gateway.config.allow_polling is False
+
+    sleep_time = max(
+        zha_gateway.global_updater.__polling_interval,
+        zha_gateway._device_availability_checker.__polling_interval,
+    )
+    sleep_time += 2
+
+    await asyncio.sleep(sleep_time)
+    await zha_gateway.async_block_till_done(wait_background_tasks=True)
+
+    assert "Global updater interval skipped" in caplog.text
+    assert "Device availability checker interval skipped" in caplog.text
+
+
+async def test_gateway_handle_message(
+    zha_gateway: Gateway,
+    zha_dev_basic: Device,  # pylint: disable=redefined-outer-name
+) -> None:
+    """Test handle message."""
+
+    assert zha_dev_basic.available is True
+    assert zha_dev_basic.on_network is True
+
+    zha_dev_basic.on_network = False
+
+    assert zha_dev_basic.available is False
+    assert zha_dev_basic.on_network is False
+
+    zha_gateway.handle_message(
+        zha_dev_basic.device,
+        zha.PROFILE_ID,
+        general.Basic.cluster_id,
+        1,
+        1,
+        b"",
+    )
+
+    assert zha_dev_basic.available is True
+    assert zha_dev_basic.on_network is True

tests/test_sensor.py

@@ -643,15 +643,16 @@ async def test_electrical_measurement_init(
     assert cluster_handler.ac_power_multiplier == 1
     assert entity.state["state"] == 4.0
 
-    zha_device.available = False
+    zha_device.on_network = False
 
-    await asyncio.sleep(70)
+    await asyncio.sleep(entity.__polling_interval + 1)
+    await zha_gateway.async_block_till_done(wait_background_tasks=True)
     assert (
         "1-2820: skipping polling for updated state, available: False, allow polled requests: True"
         in caplog.text
     )
 
-    zha_device.available = True
+    zha_device.on_network = True
 
     await send_attributes_report(
         zha_gateway,
@@ -1173,13 +1174,14 @@ async def test_device_counter_sensors(
         "counter_1"
     ].increment()
 
-    await entity.async_update()
-    await zha_gateway.async_block_till_done()
+    await asyncio.sleep(zha_gateway.global_updater.__polling_interval + 2)
+    await zha_gateway.async_block_till_done(wait_background_tasks=True)
 
     assert entity.state["state"] == 2
 
     coordinator.available = False
     await asyncio.sleep(120)
+    await zha_gateway.async_block_till_done(wait_background_tasks=True)
 
     assert (
         "counter_1: skipping polling for updated state, available: False, allow polled requests: True"

zha/application/gateway.py

@@ -42,7 +42,7 @@
     ZHA_GW_MSG_RAW_INIT,
     RadioType,
 )
-from zha.application.helpers import ZHAData
+from zha.application.helpers import DeviceAvailabilityChecker, GlobalUpdater, ZHAData
 from zha.async_ import (
     AsyncUtilMixin,
     create_eager_task,
@@ -162,6 +162,10 @@ def __init__(self, config: ZHAData) -> None:
             setup_quirks(
                 custom_quirks_path=config.yaml_config.get(CONF_CUSTOM_QUIRKS_PATH)
             )
+        self.global_updater: GlobalUpdater = GlobalUpdater(self)
+        self._device_availability_checker: DeviceAvailabilityChecker = (
+            DeviceAvailabilityChecker(self)
+        )
         self.config.gateway = self
 
     def get_application_controller_data(self) -> tuple[ControllerApplication, dict]:
@@ -224,6 +228,8 @@ async def async_initialize(self) -> None:
 
         self.application_controller.add_listener(self)
         self.application_controller.groups.add_listener(self)
+        self.global_updater.start()
+        self._device_availability_checker.start()
 
     def connection_lost(self, exc: Exception) -> None:
         """Handle connection lost event."""
@@ -408,7 +414,10 @@ def device_initialized(self, device: zigpy.device.Device) -> None:
 
     def device_left(self, device: zigpy.device.Device) -> None:
         """Handle device leaving the network."""
-        self.async_update_device(device, False)
+        zha_device: Device = self._devices.get(device.ieee)
+        if zha_device is not None:
+            zha_device.on_network = False
+        self.async_update_device(device, available=False)
 
     def group_member_removed(
         self, zigpy_group: zigpy.group.Group, endpoint: zigpy.endpoint.Endpoint
@@ -522,7 +531,9 @@ def get_or_create_group(self, zigpy_group: zigpy.group.Group) -> Group:
         return zha_group
 
     def async_update_device(
-        self, sender: zigpy.device.Device, available: bool = True
+        self,
+        sender: zigpy.device.Device,
+        available: bool = True,
     ) -> None:
         """Update device that has just become available."""
         if sender.ieee in self.devices:
@@ -569,6 +580,7 @@ async def async_device_initialized(self, device: zigpy.device.Device) -> None:
 
     async def _async_device_joined(self, zha_device: Device) -> None:
         zha_device.available = True
+        zha_device.on_network = True
         await zha_device.async_configure()
         device_info = ExtendedDeviceInfoWithPairingStatus(
             pairing_status=DevicePairingStatus.CONFIGURED.name,
@@ -600,7 +612,7 @@ async def _async_device_rejoined(self, zha_device: Device) -> None:
         )
         # force async_initialize() to fire so don't explicitly call it
         zha_device.available = False
-        zha_device.update_available(True)
+        zha_device.on_network = True
 
     async def async_create_zigpy_group(
         self,
@@ -660,6 +672,9 @@ async def shutdown(self) -> None:
             _LOGGER.debug("Ignoring duplicate shutdown event")
             return
 
+        self.global_updater.stop()
+        self._device_availability_checker.stop()
+
         async def _cancel_tasks(tasks_to_cancel: Iterable) -> None:
             tasks = [t for t in tasks_to_cancel if not (t.done() or t.cancelled())]
             for task in tasks:
@@ -696,4 +711,5 @@ def handle_message(  # pylint: disable=unused-argument
     ) -> None:
         """Handle message from a device Event handler."""
         if sender.ieee in self.devices and not self.devices[sender.ieee].available:
+            self.devices[sender.ieee].on_network = True
             self.async_update_device(sender, available=True)