Poll all status data in Vera (#35703)

* Vera now polls for all status data, no only incremental.
Vera polling is not handled using hass event loops with proper backoffs.

* Using long polling.

* Addressing PR feedback.

* Addressing PR feedback.
Adding controller stop on config unload.

homeassistant/components/vera/__init__.py

@@ -24,7 +24,7 @@ from homeassistant.helpers.entity import Entity
 from homeassistant.util import convert, slugify
 from homeassistant.util.dt import utc_from_timestamp
 
-from .common import ControllerData, get_configured_platforms
+from .common import ControllerData, SubscriptionRegistry, get_configured_platforms
 from .config_flow import fix_device_id_list, new_options
 from .const import (
     ATTR_CURRENT_ENERGY_KWH,
@@ -95,12 +95,11 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
         )
 
     # Initialize the Vera controller.
-    controller = veraApi.VeraController(base_url)
-    controller.start()
+    subscription_registry = SubscriptionRegistry(hass)
+    controller = veraApi.VeraController(base_url, subscription_registry)
+    await hass.async_add_executor_job(controller.start)
 
-    hass.bus.async_listen_once(
-        EVENT_HOMEASSISTANT_STOP, lambda event: controller.stop()
-    )
+    hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, controller.stop)
 
     try:
         all_devices = await hass.async_add_executor_job(controller.get_devices)
@@ -143,12 +142,13 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
 
 async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
     """Unload Withings config entry."""
-    controller_data = hass.data[DOMAIN]
+    controller_data: ControllerData = hass.data[DOMAIN]
 
     tasks = [
         hass.config_entries.async_forward_entry_unload(config_entry, platform)
         for platform in get_configured_platforms(controller_data)
     ]
+    tasks.append(hass.async_add_executor_job(controller_data.controller.stop))
     await asyncio.gather(*tasks)
 
     return True

homeassistant/components/vera/common.py

@@ -5,6 +5,8 @@ from typing import DefaultDict, List, NamedTuple, Set
 import pyvera as pv
 
 from homeassistant.components.scene import DOMAIN as SCENE_DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.event import call_later
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -27,3 +29,38 @@ def get_configured_platforms(controller_data: ControllerData) -> Set[str]:
         platforms.append(SCENE_DOMAIN)
 
     return set(platforms)
+
+
+class SubscriptionRegistry(pv.AbstractSubscriptionRegistry):
+    """Manages polling for data from vera."""
+
+    def __init__(self, hass: HomeAssistant) -> None:
+        """Initialize the object."""
+        super().__init__()
+        self._hass = hass
+        self._cancel_poll = None
+
+    def start(self) -> None:
+        """Start polling for data."""
+        self.stop()
+        self._schedule_poll(1)
+
+    def stop(self) -> None:
+        """Stop polling for data."""
+        if self._cancel_poll:
+            self._cancel_poll()
+            self._cancel_poll = None
+
+    def _schedule_poll(self, delay: float) -> None:
+        self._cancel_poll = call_later(self._hass, delay, self._run_poll_server)
+
+    def _run_poll_server(self, now) -> None:
+        delay = 1
+
+        # Long poll for changes. The downstream API instructs the endpoint to wait a
+        # a minimum of 200ms before returning data and a maximum of 9s before timing out.
+        if not self.poll_server_once():
+            # If an error was encountered, wait a bit longer before trying again.
+            delay = 60
+
+        self._schedule_poll(delay)

homeassistant/components/vera/manifest.json

@@ -3,6 +3,6 @@
   "name": "Vera",
   "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/vera",
-  "requirements": ["pyvera==0.3.7"],
+  "requirements": ["pyvera==0.3.9"],
   "codeowners": ["@vangorra"]
 }

requirements_all.txt

@@ -1799,7 +1799,7 @@ pyuptimerobot==0.0.5
 # pyuserinput==0.1.11
 
 # homeassistant.components.vera
-pyvera==0.3.7
+pyvera==0.3.9
 
 # homeassistant.components.versasense
 pyversasense==0.0.6

requirements_test_all.txt

@@ -756,7 +756,7 @@ pytraccar==0.9.0
 pytradfri[async]==6.4.0
 
 # homeassistant.components.vera
-pyvera==0.3.7
+pyvera==0.3.9
 
 # homeassistant.components.vesync
 pyvesync==1.1.0

tests/components/vera/test_common.py

@@ -0,0 +1,50 @@
+"""Tests for common vera code."""
+from datetime import timedelta
+
+from homeassistant.components.vera import SubscriptionRegistry
+from homeassistant.core import HomeAssistant
+from homeassistant.util.dt import utcnow
+
+from tests.async_mock import MagicMock
+from tests.common import async_fire_time_changed
+
+
+async def test_subscription_registry(hass: HomeAssistant) -> None:
+    """Test subscription registry polling."""
+    subscription_registry = SubscriptionRegistry(hass)
+    # pylint: disable=protected-access
+    subscription_registry.poll_server_once = poll_server_once_mock = MagicMock()
+
+    poll_server_once_mock.return_value = True
+    await hass.async_add_executor_job(subscription_registry.start)
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=1))
+    await hass.async_block_till_done()
+    poll_server_once_mock.assert_called_once()
+
+    # Last poll was successful and already scheduled the next poll for 1s in the future.
+    # This will ensure that future poll will fail.
+    poll_server_once_mock.return_value = False
+
+    # Asserting future poll runs.
+    poll_server_once_mock.reset_mock()
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=2))
+    await hass.async_block_till_done()
+    poll_server_once_mock.assert_called_once()
+
+    # Asserting a future poll is delayed due to the failure set above.
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=2))
+    poll_server_once_mock.reset_mock()
+    poll_server_once_mock.assert_not_called()
+
+    poll_server_once_mock.reset_mock()
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=60))
+    await hass.async_block_till_done()
+    poll_server_once_mock.assert_called_once()
+
+    poll_server_once_mock.reset_mock()
+    await hass.async_add_executor_job(subscription_registry.stop)
+
+    # Assert no further polling is performed.
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=65))
+    await hass.async_block_till_done()
+    poll_server_once_mock.assert_not_called()