Update Hue data fetching (#31338)

* Refactor Hue Lights to use DataCoordinator

* Redo how Hue updates data

* Address comments

* Inherit from Entity and remove pylint disable

* Add tests for debounce

homeassistant/components/hue/__init__.py

@@ -122,7 +122,7 @@ async def async_setup_entry(
     if not await bridge.async_setup():
         return False
 
-    hass.data[DOMAIN][host] = bridge
+    hass.data[DOMAIN][entry.entry_id] = bridge
     config = bridge.api.config
 
     # For backwards compat
@@ -151,5 +151,5 @@ async def async_setup_entry(
 
 async def async_unload_entry(hass, entry):
     """Unload a config entry."""
-    bridge = hass.data[DOMAIN].pop(entry.data["host"])
+    bridge = hass.data[DOMAIN].pop(entry.entry_id)
     return await bridge.async_reset()

homeassistant/components/hue/binary_sensor.py

@@ -6,27 +6,18 @@ from homeassistant.components.binary_sensor import (
     DEVICE_CLASS_MOTION,
     BinarySensorDevice,
 )
-from homeassistant.components.hue.sensor_base import (
+
-    GenericZLLSensor,
+from .const import DOMAIN as HUE_DOMAIN
-    SensorManager,
+from .sensor_base import SENSOR_CONFIG_MAP, GenericZLLSensor
-    async_setup_entry as shared_async_setup_entry,
-)
 
 PRESENCE_NAME_FORMAT = "{} motion"
 
 
 async def async_setup_entry(hass, config_entry, async_add_entities):
     """Defer binary sensor setup to the shared sensor module."""
-    SensorManager.sensor_config_map.update(
+    await hass.data[HUE_DOMAIN][
-        {
+        config_entry.entry_id
-            TYPE_ZLL_PRESENCE: {
+    ].sensor_manager.async_register_component(True, async_add_entities)
-                "binary": True,
-                "name_format": PRESENCE_NAME_FORMAT,
-                "class": HuePresence,
-            }
-        }
-    )
-    await shared_async_setup_entry(hass, config_entry, async_add_entities, binary=True)
 
 
 class HuePresence(GenericZLLSensor, BinarySensorDevice):
@@ -34,9 +25,6 @@ class HuePresence(GenericZLLSensor, BinarySensorDevice):
 
     device_class = DEVICE_CLASS_MOTION
 
-    async def _async_update_ha_state(self, *args, **kwargs):
-        await self.async_update_ha_state(self, *args, **kwargs)
-
     @property
     def is_on(self):
         """Return true if the binary sensor is on."""
@@ -51,3 +39,14 @@ class HuePresence(GenericZLLSensor, BinarySensorDevice):
         if "sensitivitymax" in self.sensor.config:
             attributes["sensitivity_max"] = self.sensor.config["sensitivitymax"]
         return attributes
+
+
+SENSOR_CONFIG_MAP.update(
+    {
+        TYPE_ZLL_PRESENCE: {
+            "binary": True,
+            "name_format": PRESENCE_NAME_FORMAT,
+            "class": HuePresence,
+        }
+    }
+)

homeassistant/components/hue/bridge.py

@@ -13,6 +13,7 @@ from homeassistant.helpers import aiohttp_client, config_validation as cv
 from .const import DOMAIN, LOGGER
 from .errors import AuthenticationRequired, CannotConnect
 from .helpers import create_config_flow
+from .sensor_base import SensorManager
 
 SERVICE_HUE_SCENE = "hue_activate_scene"
 ATTR_GROUP_NAME = "group_name"
@@ -35,6 +36,9 @@ class HueBridge:
         self.authorized = False
         self.api = None
         self.parallel_updates_semaphore = None
+        # Jobs to be executed when API is reset.
+        self.reset_jobs = []
+        self.sensor_manager = None
 
     @property
     def host(self):
@@ -72,6 +76,7 @@ class HueBridge:
             return False
 
         self.api = bridge
+        self.sensor_manager = SensorManager(self)
 
         hass.async_create_task(
             hass.config_entries.async_forward_entry_setup(self.config_entry, "light")
@@ -118,6 +123,9 @@ class HueBridge:
 
         self.hass.services.async_remove(DOMAIN, SERVICE_HUE_SCENE)
 
+        while self.reset_jobs:
+            self.reset_jobs.pop()()
+
         # If setup was successful, we set api variable, forwarded entry and
         # register service
         results = await asyncio.gather(
@@ -131,6 +139,7 @@ class HueBridge:
                 self.config_entry, "sensor"
             ),
         )
+
         # None and True are OK
         return False not in results
 

homeassistant/components/hue/const.py

@@ -4,3 +4,7 @@ import logging
 LOGGER = logging.getLogger(__package__)
 DOMAIN = "hue"
 API_NUPNP = "https://www.meethue.com/api/nupnp"
+
+# How long to wait to actually do the refresh after requesting it.
+# We wait some time so if we control multiple lights, we batch requests.
+REQUEST_REFRESH_DELAY = 0.3

homeassistant/components/hue/helpers.py

@@ -6,7 +6,7 @@ from homeassistant.helpers.entity_registry import async_get_registry as get_ent_
 from .const import DOMAIN
 
 
-async def remove_devices(hass, config_entry, api_ids, current):
+async def remove_devices(bridge, api_ids, current):
     """Get items that are removed from api."""
     removed_items = []
 
@@ -18,16 +18,16 @@ async def remove_devices(hass, config_entry, api_ids, current):
         entity = current[item_id]
         removed_items.append(item_id)
         await entity.async_remove()
-        ent_registry = await get_ent_reg(hass)
+        ent_registry = await get_ent_reg(bridge.hass)
         if entity.entity_id in ent_registry.entities:
             ent_registry.async_remove(entity.entity_id)
-        dev_registry = await get_dev_reg(hass)
+        dev_registry = await get_dev_reg(bridge.hass)
         device = dev_registry.async_get_device(
             identifiers={(DOMAIN, entity.device_id)}, connections=set()
         )
         if device is not None:
             dev_registry.async_update_device(
-                device.id, remove_config_entry_id=config_entry.entry_id
+                device.id, remove_config_entry_id=bridge.config_entry.entry_id
             )
 
     for item_id in removed_items:

homeassistant/components/hue/light.py

@@ -1,14 +1,13 @@
 """Support for the Philips Hue lights."""
 import asyncio
 from datetime import timedelta
+from functools import partial
 import logging
 import random
-from time import monotonic
 
 import aiohue
 import async_timeout
 
-from homeassistant.components import hue
 from homeassistant.components.light import (
     ATTR_BRIGHTNESS,
     ATTR_COLOR_TEMP,
@@ -28,8 +27,13 @@ from homeassistant.components.light import (
     SUPPORT_TRANSITION,
     Light,
 )
+from homeassistant.core import callback
+from homeassistant.exceptions import PlatformNotReady
+from homeassistant.helpers.debounce import Debouncer
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 from homeassistant.util import color
 
+from .const import DOMAIN as HUE_DOMAIN, REQUEST_REFRESH_DELAY
 from .helpers import remove_devices
 
 SCAN_INTERVAL = timedelta(seconds=5)
@@ -70,9 +74,40 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
 
 async def async_setup_entry(hass, config_entry, async_add_entities):
     """Set up the Hue lights from a config entry."""
-    bridge = hass.data[hue.DOMAIN][config_entry.data["host"]]
+    bridge = hass.data[HUE_DOMAIN][config_entry.entry_id]
-    cur_lights = {}
+
-    cur_groups = {}
+    light_coordinator = DataUpdateCoordinator(
+        hass,
+        _LOGGER,
+        "light",
+        partial(async_safe_fetch, bridge, bridge.api.lights.update),
+        SCAN_INTERVAL,
+        Debouncer(bridge.hass, _LOGGER, REQUEST_REFRESH_DELAY, True),
+    )
+
+    # First do a refresh to see if we can reach the hub.
+    # Otherwise we will declare not ready.
+    await light_coordinator.async_refresh()
+
+    if light_coordinator.failed_last_update:
+        raise PlatformNotReady
+
+    update_lights = partial(
+        async_update_items,
+        bridge,
+        bridge.api.lights,
+        {},
+        async_add_entities,
+        partial(HueLight, light_coordinator, bridge, False),
+    )
+
+    # We add a listener after fetching the data, so manually trigger listener
+    light_coordinator.async_add_listener(update_lights)
+    update_lights()
+
+    bridge.reset_jobs.append(
+        lambda: light_coordinator.async_remove_listener(update_lights)
+    )
 
     api_version = tuple(int(v) for v in bridge.api.config.apiversion.split("."))
 
@@ -81,168 +116,60 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
         _LOGGER.warning("Please update your Hue bridge to support groups")
         allow_groups = False
 
-    # Hue updates all lights via a single API call.
+    if not allow_groups:
-    #
-    # If we call a service to update 2 lights, we only want the API to be
-    # called once.
-    #
-    # The throttle decorator will return right away if a call is currently
-    # in progress. This means that if we are updating 2 lights, the first one
-    # is in the update method, the second one will skip it and assume the
-    # update went through and updates it's data, not good!
-    #
-    # The current mechanism will make sure that all lights will wait till
-    # the update call is done before writing their data to the state machine.
-    #
-    # An alternative approach would be to disable automatic polling by Home
-    # Assistant and take control ourselves. This works great for polling as now
-    # we trigger from 1 time update an update to all entities. However it gets
-    # tricky from inside async_turn_on and async_turn_off.
-    #
-    # If automatic polling is enabled, Home Assistant will call the entity
-    # update method after it is done calling all the services. This means that
-    # when we update, we know all commands have been processed. If we trigger
-    # the update from inside async_turn_on, the update will not capture the
-    # changes to the second entity until the next polling update because the
-    # throttle decorator will prevent the call.
-
-    progress = None
-    light_progress = set()
-    group_progress = set()
-
-    async def request_update(is_group, object_id):
-        """Request an update.
-
-        We will only make 1 request to the server for updating at a time. If a
-        request is in progress, we will join the request that is in progress.
-
-        This approach is possible because should_poll=True. That means that
-        Home Assistant will ask lights for updates during a polling cycle or
-        after it has called a service.
-
-        We keep track of the lights that are waiting for the request to finish.
-        When new data comes in, we'll trigger an update for all non-waiting
-        lights. This covers the case where a service is called to enable 2
-        lights but in the meanwhile some other light has changed too.
-        """
-        nonlocal progress
-
-        progress_set = group_progress if is_group else light_progress
-        progress_set.add(object_id)
-
-        if progress is not None:
-            return await progress
-
-        progress = asyncio.ensure_future(update_bridge())
-        result = await progress
-        progress = None
-        light_progress.clear()
-        group_progress.clear()
-        return result
-
-    async def update_bridge():
-        """Update the values of the bridge.
-
-        Will update lights and, if enabled, groups from the bridge.
-        """
-        tasks = []
-        tasks.append(
-            async_update_items(
-                hass,
-                config_entry,
-                bridge,
-                async_add_entities,
-                request_update,
-                False,
-                cur_lights,
-                light_progress,
-            )
-        )
-
-        if allow_groups:
-            tasks.append(
-                async_update_items(
-                    hass,
-                    config_entry,
-                    bridge,
-                    async_add_entities,
-                    request_update,
-                    True,
-                    cur_groups,
-                    group_progress,
-                )
-            )
-
-        await asyncio.wait(tasks)
-
-    await update_bridge()
-
-
-async def async_update_items(
-    hass,
-    config_entry,
-    bridge,
-    async_add_entities,
-    request_bridge_update,
-    is_group,
-    current,
-    progress_waiting,
-):
-    """Update either groups or lights from the bridge."""
-    if not bridge.authorized:
         return
 
-    if is_group:
+    group_coordinator = DataUpdateCoordinator(
-        api_type = "group"
+        hass,
-        api = bridge.api.groups
+        _LOGGER,
-    else:
+        "group",
-        api_type = "light"
+        partial(async_safe_fetch, bridge, bridge.api.groups.update),
-        api = bridge.api.lights
+        SCAN_INTERVAL,
+        Debouncer(bridge.hass, _LOGGER, REQUEST_REFRESH_DELAY, True),
+    )
 
+    update_groups = partial(
+        async_update_items,
+        bridge,
+        bridge.api.groups,
+        {},
+        async_add_entities,
+        partial(HueLight, group_coordinator, bridge, True),
+    )
+
+    group_coordinator.async_add_listener(update_groups)
+    await group_coordinator.async_refresh()
+
+    bridge.reset_jobs.append(
+        lambda: group_coordinator.async_remove_listener(update_groups)
+    )
+
+
+async def async_safe_fetch(bridge, fetch_method):
+    """Safely fetch data."""
     try:
-        start = monotonic()
         with async_timeout.timeout(4):
-            await bridge.async_request_call(api.update())
+            return await bridge.async_request_call(fetch_method())
     except aiohue.Unauthorized:
         await bridge.handle_unauthorized_error()
-        return
+        raise UpdateFailed
-    except (asyncio.TimeoutError, aiohue.AiohueException) as err:
+    except (asyncio.TimeoutError, aiohue.AiohueException):
-        _LOGGER.debug("Failed to fetch %s: %s", api_type, err)
+        raise UpdateFailed
 
-        if not bridge.available:
-            return
-
-        _LOGGER.error("Unable to reach bridge %s (%s)", bridge.host, err)
-        bridge.available = False
-
-        for item_id, item in current.items():
-            if item_id not in progress_waiting:
-                item.async_schedule_update_ha_state()
-
-        return
-
-    finally:
-        _LOGGER.debug(
-            "Finished %s request in %.3f seconds", api_type, monotonic() - start
-        )
-
-    if not bridge.available:
-        _LOGGER.info("Reconnected to bridge %s", bridge.host)
-        bridge.available = True
 
+@callback
+def async_update_items(bridge, api, current, async_add_entities, create_item):
+    """Update items."""
     new_items = []
 
     for item_id in api:
-        if item_id not in current:
+        if item_id in current:
-            current[item_id] = HueLight(
+            continue
-                api[item_id], request_bridge_update, bridge, is_group
-            )
 
-            new_items.append(current[item_id])
+        current[item_id] = create_item(api[item_id])
-        elif item_id not in progress_waiting:
+        new_items.append(current[item_id])
-            current[item_id].async_schedule_update_ha_state()
 
-    await remove_devices(hass, config_entry, api, current)
+    bridge.hass.async_create_task(remove_devices(bridge, api, current))
 
     if new_items:
         async_add_entities(new_items)
@@ -251,10 +178,10 @@ async def async_update_items(
 class HueLight(Light):
     """Representation of a Hue light."""
 
-    def __init__(self, light, request_bridge_update, bridge, is_group=False):
+    def __init__(self, coordinator, bridge, is_group, light):
         """Initialize the light."""
         self.light = light
-        self.async_request_bridge_update = request_bridge_update
+        self.coordinator = coordinator
         self.bridge = bridge
         self.is_group = is_group
 
@@ -289,6 +216,11 @@ class HueLight(Light):
         """Return the unique ID of this Hue light."""
         return self.light.uniqueid
 
+    @property
+    def should_poll(self):
+        """No polling required."""
+        return False
+
     @property
     def device_id(self):
         """Return the ID of this Hue light."""
@@ -345,14 +277,10 @@ class HueLight(Light):
     @property
     def available(self):
         """Return if light is available."""
-        return (
+        return not self.coordinator.failed_last_update and (
-            self.bridge.available
+            self.is_group
-            and self.bridge.authorized
+            or self.bridge.allow_unreachable
-            and (
+            or self.light.state["reachable"]
-                self.is_group
-                or self.bridge.allow_unreachable
-                or self.light.state["reachable"]
-            )
         )
 
     @property
@@ -379,7 +307,7 @@ class HueLight(Light):
             return None
 
         return {
-            "identifiers": {(hue.DOMAIN, self.device_id)},
+            "identifiers": {(HUE_DOMAIN, self.device_id)},
             "name": self.name,
             "manufacturer": self.light.manufacturername,
             # productname added in Hue Bridge API 1.24
@@ -387,9 +315,17 @@ class HueLight(Light):
             "model": self.light.productname or self.light.modelid,
             # Not yet exposed as properties in aiohue
             "sw_version": self.light.raw["swversion"],
-            "via_device": (hue.DOMAIN, self.bridge.api.config.bridgeid),
+            "via_device": (HUE_DOMAIN, self.bridge.api.config.bridgeid),
         }
 
+    async def async_added_to_hass(self):
+        """When entity is added to hass."""
+        self.coordinator.async_add_listener(self.async_write_ha_state)
+
+    async def async_will_remove_from_hass(self):
+        """When entity will be removed from hass."""
+        self.coordinator.async_remove_listener(self.async_write_ha_state)
+
     async def async_turn_on(self, **kwargs):
         """Turn the specified or all lights on."""
         command = {"on": True}
@@ -440,6 +376,8 @@ class HueLight(Light):
         else:
             await self.bridge.async_request_call(self.light.set_state(**command))
 
+        await self.coordinator.async_request_refresh()
+
     async def async_turn_off(self, **kwargs):
         """Turn the specified or all lights off."""
         command = {"on": False}
@@ -463,9 +401,14 @@ class HueLight(Light):
         else:
             await self.bridge.async_request_call(self.light.set_state(**command))
 
+        await self.coordinator.async_request_refresh()
+
     async def async_update(self):
-        """Synchronize state with bridge."""
+        """Update the entity.
-        await self.async_request_bridge_update(self.is_group, self.light.id)
+
+        Only used by the generic entity update service.
+        """
+        await self.coordinator.async_request_refresh()
 
     @property
     def device_state_attributes(self):

homeassistant/components/hue/sensor.py

@@ -1,11 +1,6 @@
 """Hue sensor entities."""
 from aiohue.sensors import TYPE_ZLL_LIGHTLEVEL, TYPE_ZLL_TEMPERATURE
 
-from homeassistant.components.hue.sensor_base import (
-    GenericZLLSensor,
-    SensorManager,
-    async_setup_entry as shared_async_setup_entry,
-)
 from homeassistant.const import (
     DEVICE_CLASS_ILLUMINANCE,
     DEVICE_CLASS_TEMPERATURE,
@@ -13,27 +8,18 @@ from homeassistant.const import (
 )
 from homeassistant.helpers.entity import Entity
 
+from .const import DOMAIN as HUE_DOMAIN
+from .sensor_base import SENSOR_CONFIG_MAP, GenericZLLSensor
+
 LIGHT_LEVEL_NAME_FORMAT = "{} light level"
 TEMPERATURE_NAME_FORMAT = "{} temperature"
 
 
 async def async_setup_entry(hass, config_entry, async_add_entities):
     """Defer sensor setup to the shared sensor module."""
-    SensorManager.sensor_config_map.update(
+    await hass.data[HUE_DOMAIN][
-        {
+        config_entry.entry_id
-            TYPE_ZLL_LIGHTLEVEL: {
+    ].sensor_manager.async_register_component(False, async_add_entities)
-                "binary": False,
-                "name_format": LIGHT_LEVEL_NAME_FORMAT,
-                "class": HueLightLevel,
-            },
-            TYPE_ZLL_TEMPERATURE: {
-                "binary": False,
-                "name_format": TEMPERATURE_NAME_FORMAT,
-                "class": HueTemperature,
-            },
-        }
-    )
-    await shared_async_setup_entry(hass, config_entry, async_add_entities, binary=False)
 
 
 class GenericHueGaugeSensorEntity(GenericZLLSensor, Entity):
@@ -91,3 +77,19 @@ class HueTemperature(GenericHueGaugeSensorEntity):
             return None
 
         return self.sensor.temperature / 100
+
+
+SENSOR_CONFIG_MAP.update(
+    {
+        TYPE_ZLL_LIGHTLEVEL: {
+            "binary": False,
+            "name_format": LIGHT_LEVEL_NAME_FORMAT,
+            "class": HueLightLevel,
+        },
+        TYPE_ZLL_TEMPERATURE: {
+            "binary": False,
+            "name_format": TEMPERATURE_NAME_FORMAT,
+            "class": HueTemperature,
+        },
+    }
+)

homeassistant/components/hue/sensor_base.py

@@ -2,22 +2,19 @@
 import asyncio
 from datetime import timedelta
 import logging
-from time import monotonic
 
 from aiohue import AiohueException, Unauthorized
 from aiohue.sensors import TYPE_ZLL_PRESENCE
 import async_timeout
 
-from homeassistant.components import hue
+from homeassistant.core import callback
-from homeassistant.exceptions import NoEntitySpecifiedError
+from homeassistant.helpers import debounce, entity
-from homeassistant.helpers.event import async_track_point_in_utc_time
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
-from homeassistant.util.dt import utcnow
 
+from .const import DOMAIN as HUE_DOMAIN, REQUEST_REFRESH_DELAY
 from .helpers import remove_devices
 
-CURRENT_SENSORS_FORMAT = "{}_current_sensors"
+SENSOR_CONFIG_MAP = {}
-SENSOR_MANAGER_FORMAT = "{}_sensor_manager"
-
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -29,22 +26,6 @@ def _device_id(aiohue_sensor):
     return device_id
 
 
-async def async_setup_entry(hass, config_entry, async_add_entities, binary=False):
-    """Set up the Hue sensors from a config entry."""
-    sensor_key = CURRENT_SENSORS_FORMAT.format(config_entry.data["host"])
-    bridge = hass.data[hue.DOMAIN][config_entry.data["host"]]
-    hass.data[hue.DOMAIN].setdefault(sensor_key, {})
-
-    sm_key = SENSOR_MANAGER_FORMAT.format(config_entry.data["host"])
-    manager = hass.data[hue.DOMAIN].get(sm_key)
-    if manager is None:
-        manager = SensorManager(hass, bridge, config_entry)
-        hass.data[hue.DOMAIN][sm_key] = manager
-
-    manager.register_component(binary, async_add_entities)
-    await manager.start()
-
-
 class SensorManager:
     """Class that handles registering and updating Hue sensor entities.
 
@@ -52,84 +33,60 @@ class SensorManager:
     """
 
     SCAN_INTERVAL = timedelta(seconds=5)
-    sensor_config_map = {}
 
-    def __init__(self, hass, bridge, config_entry):
+    def __init__(self, bridge):
         """Initialize the sensor manager."""
-        self.hass = hass
         self.bridge = bridge
-        self.config_entry = config_entry
         self._component_add_entities = {}
-        self._started = False
+        self.current = {}
+        self.coordinator = DataUpdateCoordinator(
+            bridge.hass,
+            _LOGGER,
+            "sensor",
+            self.async_update_data,
+            self.SCAN_INTERVAL,
+            debounce.Debouncer(bridge.hass, _LOGGER, REQUEST_REFRESH_DELAY, True),
+        )
 
-    def register_component(self, binary, async_add_entities):
+    async def async_update_data(self):
+        """Update sensor data."""
+        try:
+            with async_timeout.timeout(4):
+                return await self.bridge.async_request_call(
+                    self.bridge.api.sensors.update()
+                )
+        except Unauthorized:
+            await self.bridge.handle_unauthorized_error()
+            raise UpdateFailed
+        except (asyncio.TimeoutError, AiohueException):
+            raise UpdateFailed
+
+    async def async_register_component(self, binary, async_add_entities):
         """Register async_add_entities methods for components."""
         self._component_add_entities[binary] = async_add_entities
 
-    async def start(self):
+        if len(self._component_add_entities) < 2:
-        """Start updating sensors from the bridge on a schedule."""
-        # but only if it's not already started, and when we've got both
-        # async_add_entities methods
-        if self._started or len(self._component_add_entities) < 2:
             return
 
-        self._started = True
+        # We have all components available, start the updating.
-        _LOGGER.info(
+        self.coordinator.async_add_listener(self.async_update_items)
-            "Starting sensor polling loop with %s second interval",
+        self.bridge.reset_jobs.append(
-            self.SCAN_INTERVAL.total_seconds(),
+            lambda: self.coordinator.async_remove_listener(self.async_update_items)
         )
+        await self.coordinator.async_refresh()
 
-        async def async_update_bridge(now):
+    @callback
-            """Will update sensors from the bridge."""
+    def async_update_items(self):
-
-            # don't update when we are not authorized
-            if not self.bridge.authorized:
-                return
-
-            await self.async_update_items()
-
-            async_track_point_in_utc_time(
-                self.hass, async_update_bridge, utcnow() + self.SCAN_INTERVAL
-            )
-
-        await async_update_bridge(None)
-
-    async def async_update_items(self):
         """Update sensors from the bridge."""
         api = self.bridge.api.sensors
 
-        try:
+        if len(self._component_add_entities) < 2:
-            start = monotonic()
-            with async_timeout.timeout(4):
-                await self.bridge.async_request_call(api.update())
-        except Unauthorized:
-            await self.bridge.handle_unauthorized_error()
             return
-        except (asyncio.TimeoutError, AiohueException) as err:
-            _LOGGER.debug("Failed to fetch sensor: %s", err)
-
-            if not self.bridge.available:
-                return
-
-            _LOGGER.error("Unable to reach bridge %s (%s)", self.bridge.host, err)
-            self.bridge.available = False
-
-            return
-
-        finally:
-            _LOGGER.debug(
-                "Finished sensor request in %.3f seconds", monotonic() - start
-            )
-
-        if not self.bridge.available:
-            _LOGGER.info("Reconnected to bridge %s", self.bridge.host)
-            self.bridge.available = True
 
         new_sensors = []
         new_binary_sensors = []
         primary_sensor_devices = {}
-        sensor_key = CURRENT_SENSORS_FORMAT.format(self.config_entry.data["host"])
+        current = self.current
-        current = self.hass.data[hue.DOMAIN][sensor_key]
 
         # Physical Hue motion sensors present as three sensors in the API: a
         # presence sensor, a temperature sensor, and a light level sensor. Of
@@ -155,11 +112,10 @@ class SensorManager:
         for item_id in api:
             existing = current.get(api[item_id].uniqueid)
             if existing is not None:
-                self.hass.async_create_task(existing.async_maybe_update_ha_state())
                 continue
 
             primary_sensor = None
-            sensor_config = self.sensor_config_map.get(api[item_id].type)
+            sensor_config = SENSOR_CONFIG_MAP.get(api[item_id].type)
             if sensor_config is None:
                 continue
 
@@ -177,22 +133,19 @@ class SensorManager:
             else:
                 new_sensors.append(current[api[item_id].uniqueid])
 
-        await remove_devices(
+        self.bridge.hass.async_create_task(
-            self.hass,
+            remove_devices(
-            self.config_entry,
+                self.bridge, [value.uniqueid for value in api.values()], current,
-            [value.uniqueid for value in api.values()],
+            )
-            current,
         )
 
-        async_add_sensor_entities = self._component_add_entities.get(False)
+        if new_sensors:
-        async_add_binary_entities = self._component_add_entities.get(True)
+            self._component_add_entities[False](new_sensors)
-        if new_sensors and async_add_sensor_entities:
+        if new_binary_sensors:
-            async_add_sensor_entities(new_sensors)
+            self._component_add_entities[True](new_binary_sensors)
-        if new_binary_sensors and async_add_binary_entities:
-            async_add_binary_entities(new_binary_sensors)
 
 
-class GenericHueSensor:
+class GenericHueSensor(entity.Entity):
     """Representation of a Hue sensor."""
 
     should_poll = False
@@ -230,10 +183,8 @@ class GenericHueSensor:
     @property
     def available(self):
         """Return if sensor is available."""
-        return (
+        return not self.bridge.sensor_manager.coordinator.failed_last_update and (
-            self.bridge.available
+            self.bridge.allow_unreachable or self.sensor.config["reachable"]
-            and self.bridge.authorized
-            and (self.bridge.allow_unreachable or self.sensor.config["reachable"])
         )
 
     @property
@@ -241,15 +192,24 @@ class GenericHueSensor:
         """Return detail of available software updates for this device."""
         return self.primary_sensor.raw.get("swupdate", {}).get("state")
 
-    async def async_maybe_update_ha_state(self):
+    async def async_added_to_hass(self):
-        """Try to update Home Assistant with current state of entity.
+        """When entity is added to hass."""
+        self.bridge.sensor_manager.coordinator.async_add_listener(
+            self.async_write_ha_state
+        )
 
-        But if it's not been added to hass yet, then don't throw an error.
+    async def async_will_remove_from_hass(self):
+        """When entity will be removed from hass."""
+        self.bridge.sensor_manager.coordinator.async_remove_listener(
+            self.async_write_ha_state
+        )
+
+    async def async_update(self):
+        """Update the entity.
+
+        Only used by the generic entity update service.
         """
-        try:
+        await self.bridge.sensor_manager.coordinator.coordinator.async_request_refresh()
-            await self._async_update_ha_state()
-        except (RuntimeError, NoEntitySpecifiedError):
-            _LOGGER.debug("Hue sensor update requested before it has been added.")
 
     @property
     def device_info(self):
@@ -258,12 +218,12 @@ class GenericHueSensor:
         Links individual entities together in the hass device registry.
         """
         return {
-            "identifiers": {(hue.DOMAIN, self.device_id)},
+            "identifiers": {(HUE_DOMAIN, self.device_id)},
             "name": self.primary_sensor.name,
             "manufacturer": self.primary_sensor.manufacturername,
             "model": (self.primary_sensor.productname or self.primary_sensor.modelid),
             "sw_version": self.primary_sensor.swversion,
-            "via_device": (hue.DOMAIN, self.bridge.api.config.bridgeid),
+            "via_device": (HUE_DOMAIN, self.bridge.api.config.bridgeid),
         }
 
 

homeassistant/helpers/debounce.py

@@ -0,0 +1,77 @@
+"""Debounce helper."""
+import asyncio
+from logging import Logger
+from typing import Any, Awaitable, Callable, Optional
+
+from homeassistant.core import HomeAssistant, callback
+
+
+class Debouncer:
+    """Class to rate limit calls to a specific command."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        logger: Logger,
+        cooldown: float,
+        immediate: bool,
+        function: Optional[Callable[..., Awaitable[Any]]] = None,
+    ):
+        """Initialize debounce.
+
+        immediate: indicate if the function needs to be called right away and
+                   wait 0.3s until executing next invocation.
+        function: optional and can be instantiated later.
+        """
+        self.hass = hass
+        self.logger = logger
+        self.function = function
+        self.cooldown = cooldown
+        self.immediate = immediate
+        self._timer_task: Optional[asyncio.TimerHandle] = None
+        self._execute_at_end_of_timer: bool = False
+
+    async def async_call(self) -> None:
+        """Call the function."""
+        assert self.function is not None
+
+        if self._timer_task:
+            if not self._execute_at_end_of_timer:
+                self._execute_at_end_of_timer = True
+
+            return
+
+        if self.immediate:
+            await self.hass.async_add_job(self.function)  # type: ignore
+        else:
+            self._execute_at_end_of_timer = True
+
+        self._timer_task = self.hass.loop.call_later(
+            self.cooldown,
+            lambda: self.hass.async_create_task(self._handle_timer_finish()),
+        )
+
+    async def _handle_timer_finish(self) -> None:
+        """Handle a finished timer."""
+        assert self.function is not None
+
+        self._timer_task = None
+
+        if not self._execute_at_end_of_timer:
+            return
+
+        self._execute_at_end_of_timer = False
+
+        try:
+            await self.hass.async_add_job(self.function)  # type: ignore
+        except Exception:  # pylint: disable=broad-except
+            self.logger.exception("Unexpected exception from %s", self.function)
+
+    @callback
+    def async_cancel(self) -> None:
+        """Cancel any scheduled call."""
+        if self._timer_task:
+            self._timer_task.cancel()
+            self._timer_task = None
+
+        self._execute_at_end_of_timer = False

homeassistant/helpers/event.py

@@ -225,7 +225,7 @@ track_point_in_time = threaded_listener_factory(async_track_point_in_time)
 @callback
 @bind_hass
 def async_track_point_in_utc_time(
-    hass: HomeAssistant, action: Callable[..., None], point_in_time: datetime
+    hass: HomeAssistant, action: Callable[..., Any], point_in_time: datetime
 ) -> CALLBACK_TYPE:
     """Add a listener that fires once after a specific point in UTC time."""
     # Ensure point_in_time is UTC

homeassistant/helpers/update_coordinator.py

@@ -0,0 +1,135 @@
+"""Helpers to help coordinate updates."""
+import asyncio
+from datetime import datetime, timedelta
+import logging
+from time import monotonic
+from typing import Any, Awaitable, Callable, List, Optional
+
+from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
+from homeassistant.helpers.event import async_track_point_in_utc_time
+from homeassistant.util.dt import utcnow
+
+from .debounce import Debouncer
+
+
+class UpdateFailed(Exception):
+    """Raised when an update has failed."""
+
+
+class DataUpdateCoordinator:
+    """Class to manage fetching data from single endpoint."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        logger: logging.Logger,
+        name: str,
+        update_method: Callable[[], Awaitable],
+        update_interval: timedelta,
+        request_refresh_debouncer: Debouncer,
+    ):
+        """Initialize global data updater."""
+        self.hass = hass
+        self.logger = logger
+        self.name = name
+        self.update_method = update_method
+        self.update_interval = update_interval
+
+        self.data: Optional[Any] = None
+
+        self._listeners: List[CALLBACK_TYPE] = []
+        self._unsub_refresh: Optional[CALLBACK_TYPE] = None
+        self._request_refresh_task: Optional[asyncio.TimerHandle] = None
+        self.failed_last_update = False
+        self._debounced_refresh = request_refresh_debouncer
+        request_refresh_debouncer.function = self._async_do_refresh
+
+    @callback
+    def async_add_listener(self, update_callback: CALLBACK_TYPE) -> None:
+        """Listen for data updates."""
+        schedule_refresh = not self._listeners
+
+        self._listeners.append(update_callback)
+
+        # This is the first listener, set up interval.
+        if schedule_refresh:
+            self._schedule_refresh()
+
+    @callback
+    def async_remove_listener(self, update_callback: CALLBACK_TYPE) -> None:
+        """Remove data update."""
+        self._listeners.remove(update_callback)
+
+        if not self._listeners and self._unsub_refresh:
+            self._unsub_refresh()
+            self._unsub_refresh = None
+
+    async def async_refresh(self) -> None:
+        """Refresh the data."""
+        if self._unsub_refresh:
+            self._unsub_refresh()
+            self._unsub_refresh = None
+
+        await self._async_do_refresh()
+
+    @callback
+    def _schedule_refresh(self) -> None:
+        """Schedule a refresh."""
+        if self._unsub_refresh:
+            self._unsub_refresh()
+            self._unsub_refresh = None
+
+        self._unsub_refresh = async_track_point_in_utc_time(
+            self.hass, self._handle_refresh_interval, utcnow() + self.update_interval
+        )
+
+    async def _handle_refresh_interval(self, _now: datetime) -> None:
+        """Handle a refresh interval occurrence."""
+        self._unsub_refresh = None
+        await self._async_do_refresh()
+
+    async def async_request_refresh(self) -> None:
+        """Request a refresh.
+
+        Refresh will wait a bit to see if it can batch them.
+        """
+        await self._debounced_refresh.async_call()
+
+    async def _async_do_refresh(self) -> None:
+        """Time to update."""
+        if self._unsub_refresh:
+            self._unsub_refresh()
+            self._unsub_refresh = None
+
+        self._debounced_refresh.async_cancel()
+
+        try:
+            start = monotonic()
+            self.data = await self.update_method()
+
+        except UpdateFailed as err:
+            if not self.failed_last_update:
+                self.logger.error("Error fetching %s data: %s", self.name, err)
+                self.failed_last_update = True
+
+        except Exception as err:  # pylint: disable=broad-except
+            self.failed_last_update = True
+            self.logger.exception(
+                "Unexpected error fetching %s data: %s", self.name, err
+            )
+
+        else:
+            if self.failed_last_update:
+                self.failed_last_update = False
+                self.logger.info("Fetching %s data recovered")
+
+        finally:
+            self.logger.debug(
+                "Finished fetching %s data in %.3f seconds",
+                self.name,
+                monotonic() - start,
+            )
+            self._schedule_refresh()
+
+        for update_callback in self._listeners:
+            update_callback()

tests/components/hue/conftest.py

@@ -0,0 +1,11 @@
+"""Test helpers for Hue."""
+from unittest.mock import patch
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def no_request_delay():
+    """Make the request refresh delay 0 for instant tests."""
+    with patch("homeassistant.components.hue.light.REQUEST_REFRESH_DELAY", 0):
+        yield

tests/components/hue/test_light.py

@@ -179,11 +179,13 @@ LIGHT_GAMUT_TYPE = "A"
 def mock_bridge(hass):
     """Mock a Hue bridge."""
     bridge = Mock(
+        hass=hass,
         available=True,
         authorized=True,
         allow_unreachable=False,
         allow_groups=False,
         api=Mock(),
+        reset_jobs=[],
         spec=hue.HueBridge,
     )
     bridge.mock_requests = []
@@ -218,7 +220,6 @@ def mock_bridge(hass):
 async def setup_bridge(hass, mock_bridge):
     """Load the Hue light platform with the provided bridge."""
     hass.config.components.add(hue.DOMAIN)
-    hass.data[hue.DOMAIN] = {"mock-host": mock_bridge}
     config_entry = config_entries.ConfigEntry(
         1,
         hue.DOMAIN,
@@ -228,6 +229,8 @@ async def setup_bridge(hass, mock_bridge):
         config_entries.CONN_CLASS_LOCAL_POLL,
         system_options={},
     )
+    mock_bridge.config_entry = config_entry
+    hass.data[hue.DOMAIN] = {config_entry.entry_id: mock_bridge}
     await hass.config_entries.async_forward_entry_setup(config_entry, "light")
     # To flush out the service call to update the group
     await hass.async_block_till_done()
@@ -363,8 +366,8 @@ async def test_new_group_discovered(hass, mock_bridge):
     await hass.services.async_call(
         "light", "turn_on", {"entity_id": "light.group_1"}, blocking=True
     )
-    # 2x group update, 2x light update, 1 turn on request
+    # 2x group update, 1x light update, 1 turn on request
-    assert len(mock_bridge.mock_requests) == 5
+    assert len(mock_bridge.mock_requests) == 4
     assert len(hass.states.async_all()) == 3
 
     new_group = hass.states.get("light.group_3")
@@ -443,8 +446,8 @@ async def test_group_removed(hass, mock_bridge):
         "light", "turn_on", {"entity_id": "light.group_1"}, blocking=True
     )
 
-    # 2x group update, 2x light update, 1 turn on request
+    # 2x group update, 1x light update, 1 turn on request
-    assert len(mock_bridge.mock_requests) == 5
+    assert len(mock_bridge.mock_requests) == 4
     assert len(hass.states.async_all()) == 1
 
     group = hass.states.get("light.group_1")
@@ -524,8 +527,8 @@ async def test_other_group_update(hass, mock_bridge):
     await hass.services.async_call(
         "light", "turn_on", {"entity_id": "light.group_1"}, blocking=True
     )
-    # 2x group update, 2x light update, 1 turn on request
+    # 2x group update, 1x light update, 1 turn on request
-    assert len(mock_bridge.mock_requests) == 5
+    assert len(mock_bridge.mock_requests) == 4
     assert len(hass.states.async_all()) == 2
 
     group_2 = hass.states.get("light.group_2")
@@ -599,7 +602,6 @@ async def test_update_timeout(hass, mock_bridge):
     await setup_bridge(hass, mock_bridge)
     assert len(mock_bridge.mock_requests) == 0
     assert len(hass.states.async_all()) == 0
-    assert mock_bridge.available is False
 
 
 async def test_update_unauthorized(hass, mock_bridge):
@@ -701,7 +703,7 @@ def test_available():
             colorgamuttype=LIGHT_GAMUT_TYPE,
             colorgamut=LIGHT_GAMUT,
         ),
-        request_bridge_update=None,
+        coordinator=Mock(failed_last_update=False),
         bridge=Mock(allow_unreachable=False),
         is_group=False,
     )
@@ -715,7 +717,7 @@ def test_available():
             colorgamuttype=LIGHT_GAMUT_TYPE,
             colorgamut=LIGHT_GAMUT,
         ),
-        request_bridge_update=None,
+        coordinator=Mock(failed_last_update=False),
         bridge=Mock(allow_unreachable=True),
         is_group=False,
     )
@@ -729,7 +731,7 @@ def test_available():
             colorgamuttype=LIGHT_GAMUT_TYPE,
             colorgamut=LIGHT_GAMUT,
         ),
-        request_bridge_update=None,
+        coordinator=Mock(failed_last_update=False),
         bridge=Mock(allow_unreachable=False),
         is_group=True,
     )
@@ -746,7 +748,7 @@ def test_hs_color():
             colorgamuttype=LIGHT_GAMUT_TYPE,
             colorgamut=LIGHT_GAMUT,
         ),
-        request_bridge_update=None,
+        coordinator=Mock(failed_last_update=False),
         bridge=Mock(),
         is_group=False,
     )
@@ -760,7 +762,7 @@ def test_hs_color():
             colorgamuttype=LIGHT_GAMUT_TYPE,
             colorgamut=LIGHT_GAMUT,
         ),
-        request_bridge_update=None,
+        coordinator=Mock(failed_last_update=False),
         bridge=Mock(),
         is_group=False,
     )
@@ -774,7 +776,7 @@ def test_hs_color():
             colorgamuttype=LIGHT_GAMUT_TYPE,
             colorgamut=LIGHT_GAMUT,
         ),
-        request_bridge_update=None,
+        coordinator=Mock(failed_last_update=False),
         bridge=Mock(),
         is_group=False,
     )

tests/components/hue/test_sensor_base.py

@@ -1,7 +1,6 @@
 """Philips Hue sensors platform tests."""
 import asyncio
 from collections import deque
-import datetime
 import logging
 from unittest.mock import Mock
 
@@ -252,16 +251,19 @@ SENSOR_RESPONSE = {
 }
 
 
-def create_mock_bridge():
+def create_mock_bridge(hass):
     """Create a mock Hue bridge."""
     bridge = Mock(
+        hass=hass,
         available=True,
         authorized=True,
         allow_unreachable=False,
         allow_groups=False,
         api=Mock(),
+        reset_jobs=[],
         spec=hue.HueBridge,
     )
+    bridge.sensor_manager = hue_sensor_base.SensorManager(bridge)
     bridge.mock_requests = []
     # We're using a deque so we can schedule multiple responses
     # and also means that `popleft()` will blow up if we get more updates
@@ -289,13 +291,7 @@ def create_mock_bridge():
 @pytest.fixture
 def mock_bridge(hass):
     """Mock a Hue bridge."""
-    return create_mock_bridge()
+    return create_mock_bridge(hass)
-
-
-@pytest.fixture
-def increase_scan_interval(hass):
-    """Increase the SCAN_INTERVAL to prevent unexpected scans during tests."""
-    hue_sensor_base.SensorManager.SCAN_INTERVAL = datetime.timedelta(days=365)
 
 
 async def setup_bridge(hass, mock_bridge, hostname=None):
@@ -303,7 +299,6 @@ async def setup_bridge(hass, mock_bridge, hostname=None):
     if hostname is None:
         hostname = "mock-host"
     hass.config.components.add(hue.DOMAIN)
-    hass.data[hue.DOMAIN] = {hostname: mock_bridge}
     config_entry = config_entries.ConfigEntry(
         1,
         hue.DOMAIN,
@@ -313,6 +308,8 @@ async def setup_bridge(hass, mock_bridge, hostname=None):
         config_entries.CONN_CLASS_LOCAL_POLL,
         system_options={},
     )
+    mock_bridge.config_entry = config_entry
+    hass.data[hue.DOMAIN] = {config_entry.entry_id: mock_bridge}
     await hass.config_entries.async_forward_entry_setup(config_entry, "binary_sensor")
     await hass.config_entries.async_forward_entry_setup(config_entry, "sensor")
     # and make sure it completes before going further
@@ -330,7 +327,7 @@ async def test_no_sensors(hass, mock_bridge):
 
 async def test_sensors_with_multiple_bridges(hass, mock_bridge):
     """Test the update_items function with some sensors."""
-    mock_bridge_2 = create_mock_bridge()
+    mock_bridge_2 = create_mock_bridge(hass)
     mock_bridge_2.mock_sensor_responses.append(
         {
             "1": PRESENCE_SENSOR_3_PRESENT,
@@ -412,11 +409,7 @@ async def test_new_sensor_discovered(hass, mock_bridge):
     mock_bridge.mock_sensor_responses.append(new_sensor_response)
 
     # Force updates to run again
-    sm_key = hue_sensor_base.SENSOR_MANAGER_FORMAT.format("mock-host")
+    await mock_bridge.sensor_manager.coordinator.async_refresh()
-    sm = hass.data[hue.DOMAIN][sm_key]
-    await sm.async_update_items()
-
-    # To flush out the service call to update the group
     await hass.async_block_till_done()
 
     assert len(mock_bridge.mock_requests) == 2
@@ -443,9 +436,7 @@ async def test_sensor_removed(hass, mock_bridge):
     mock_bridge.mock_sensor_responses.append({k: SENSOR_RESPONSE[k] for k in keys})
 
     # Force updates to run again
-    sm_key = hue_sensor_base.SENSOR_MANAGER_FORMAT.format("mock-host")
+    await mock_bridge.sensor_manager.coordinator.async_refresh()
-    sm = hass.data[hue.DOMAIN][sm_key]
-    await sm.async_update_items()
 
     # To flush out the service call to update the group
     await hass.async_block_till_done()
@@ -466,7 +457,6 @@ async def test_update_timeout(hass, mock_bridge):
     await setup_bridge(hass, mock_bridge)
     assert len(mock_bridge.mock_requests) == 0
     assert len(hass.states.async_all()) == 0
-    assert mock_bridge.available is False
 
 
 async def test_update_unauthorized(hass, mock_bridge):

tests/helpers/test_debounce.py

@@ -0,0 +1,62 @@
+"""Tests for debounce."""
+from asynctest import CoroutineMock
+
+from homeassistant.helpers import debounce
+
+
+async def test_immediate_works(hass):
+    """Test immediate works."""
+    calls = []
+    debouncer = debounce.Debouncer(
+        hass, None, 0.01, True, CoroutineMock(side_effect=lambda: calls.append(None))
+    )
+
+    await debouncer.async_call()
+    assert len(calls) == 1
+    assert debouncer._timer_task is not None
+    assert debouncer._execute_at_end_of_timer is False
+
+    await debouncer.async_call()
+    assert len(calls) == 1
+    assert debouncer._timer_task is not None
+    assert debouncer._execute_at_end_of_timer is True
+
+    debouncer.async_cancel()
+    assert debouncer._timer_task is None
+    assert debouncer._execute_at_end_of_timer is False
+
+    await debouncer.async_call()
+    assert len(calls) == 2
+    await debouncer._handle_timer_finish()
+    assert len(calls) == 2
+    assert debouncer._timer_task is None
+    assert debouncer._execute_at_end_of_timer is False
+
+
+async def test_not_immediate_works(hass):
+    """Test immediate works."""
+    calls = []
+    debouncer = debounce.Debouncer(
+        hass, None, 0.01, False, CoroutineMock(side_effect=lambda: calls.append(None))
+    )
+
+    await debouncer.async_call()
+    assert len(calls) == 0
+    assert debouncer._timer_task is not None
+    assert debouncer._execute_at_end_of_timer is True
+
+    await debouncer.async_call()
+    assert len(calls) == 0
+    assert debouncer._timer_task is not None
+    assert debouncer._execute_at_end_of_timer is True
+
+    debouncer.async_cancel()
+    assert debouncer._timer_task is None
+    assert debouncer._execute_at_end_of_timer is False
+
+    await debouncer.async_call()
+    assert len(calls) == 0
+    await debouncer._handle_timer_finish()
+    assert len(calls) == 1
+    assert debouncer._timer_task is None
+    assert debouncer._execute_at_end_of_timer is False