Centralize august activity updates (#32197)

* Reduce August doorbell detail updates

* Doorbell images now get updates from the activity feed

* Tests for activity updates

* py-august now provides bridge_is_online for available state

* py-august now provides is_standby for available state

* py-august now provides get_doorbell_image (eliminate requests)

* remove debug

* black after merge conflict

* Centralize august activity updates

* Updates appear significantly more responsive

* Should address the community complaints about "lag"

* Reduce detail updates (device end points) to one hour interval

* Signal entities to update via dispatcher when new activity arrives

* Resolves out of sync state (skipped test is now unskipped)

* pylint

* fix merge conflict

* review comments

* Remove stray

* Address review items that can be done without refactor

homeassistant/components/august/__init__.py

@@ -1,7 +1,6 @@
 """Support for August devices."""
 import asyncio
-from datetime import timedelta
+import itertools
-from functools import partial
 import logging
 
 from august.api import AugustApiHTTPError
@@ -16,10 +15,13 @@ from homeassistant.const import CONF_PASSWORD, CONF_TIMEOUT, CONF_USERNAME
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.dispatcher import async_dispatcher_send
 from homeassistant.util import Throttle
 
+from .activity import ActivityStream
 from .const import (
     AUGUST_COMPONENTS,
+    AUGUST_DEVICE_UPDATE,
     CONF_ACCESS_TOKEN_CACHE_FILE,
     CONF_INSTALL_ID,
     CONF_LOGIN_METHOD,
@@ -29,7 +31,6 @@ from .const import (
     DEFAULT_TIMEOUT,
     DOMAIN,
     LOGIN_METHODS,
-    MIN_TIME_BETWEEN_ACTIVITY_UPDATES,
     MIN_TIME_BETWEEN_DETAIL_UPDATES,
     VERIFICATION_CODE_KEY,
 )
@@ -40,7 +41,7 @@ _LOGGER = logging.getLogger(__name__)
 
 TWO_FA_REVALIDATE = "verify_configurator"
 
-DEFAULT_SCAN_INTERVAL = timedelta(seconds=10)
+DEFAULT_SCAN_INTERVAL = MIN_TIME_BETWEEN_DETAIL_UPDATES
 
 CONFIG_SCHEMA = vol.Schema(
     {
@@ -133,6 +134,7 @@ async def async_setup_august(hass, config_entry, august_gateway):
     hass.data[DOMAIN][entry_id][DATA_AUGUST] = await hass.async_add_executor_job(
         AugustData, hass, august_gateway
     )
+    await hass.data[DOMAIN][entry_id][DATA_AUGUST].activity_stream.async_start()
 
     for component in AUGUST_COMPONENTS:
         hass.async_create_task(
@@ -178,6 +180,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
 
 async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
     """Unload a config entry."""
+    hass.data[DOMAIN][entry.entry_id][DATA_AUGUST].activity_stream.async_stop()
+
     unload_ok = all(
         await asyncio.gather(
             *[
@@ -196,8 +200,6 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
 class AugustData:
     """August data object."""
 
-    DEFAULT_ACTIVITY_FETCH_LIMIT = 10
-
     def __init__(self, hass, august_gateway):
         """Init August data object."""
         self._hass = hass
@@ -211,12 +213,11 @@ class AugustData:
             self._api.get_operable_locks(self._august_gateway.access_token) or []
         )
         self._house_ids = set()
-        for device in self._doorbells + self._locks:
+        for device in itertools.chain(self._doorbells, self._locks):
             self._house_ids.add(device.house_id)
 
         self._doorbell_detail_by_id = {}
         self._lock_detail_by_id = {}
-        self._activities_by_id = {}
 
         # We check the locks right away so we can
         # remove inoperative ones
@@ -224,6 +225,10 @@ class AugustData:
         self._update_doorbells_detail()
         self._filter_inoperative_locks()
 
+        self.activity_stream = ActivityStream(
+            hass, self._api, self._august_gateway, self._house_ids
+        )
+
     @property
     def house_ids(self):
         """Return a list of house_ids."""
@@ -239,49 +244,6 @@ class AugustData:
         """Return a list of locks."""
         return self._locks
 
-    async def async_get_device_activities(self, device_id, *activity_types):
-        """Return a list of activities."""
-        _LOGGER.debug("Getting device activities for %s", device_id)
-        await self._async_update_device_activities()
-
-        activities = self._activities_by_id.get(device_id, [])
-        if activity_types:
-            return [a for a in activities if a.activity_type in activity_types]
-        return activities
-
-    async def async_get_latest_device_activity(self, device_id, *activity_types):
-        """Return latest activity."""
-        activities = await self.async_get_device_activities(device_id, *activity_types)
-        return next(iter(activities or []), None)
-
-    @Throttle(MIN_TIME_BETWEEN_ACTIVITY_UPDATES)
-    async def _async_update_device_activities(self, limit=DEFAULT_ACTIVITY_FETCH_LIMIT):
-        """Update data object with latest from August API."""
-
-        # This is the only place we refresh the api token
-        await self._august_gateway.async_refresh_access_token_if_needed()
-
-        return await self._hass.async_add_executor_job(
-            partial(self._update_device_activities, limit=limit)
-        )
-
-    def _update_device_activities(self, limit=DEFAULT_ACTIVITY_FETCH_LIMIT):
-        _LOGGER.debug("Start retrieving device activities")
-        for house_id in self.house_ids:
-            _LOGGER.debug("Updating device activity for house id %s", house_id)
-
-            activities = self._api.get_house_activities(
-                self._august_gateway.access_token, house_id, limit=limit
-            )
-
-            device_ids = {a.device_id for a in activities}
-            for device_id in device_ids:
-                self._activities_by_id[device_id] = [
-                    a for a in activities if a.device_id == device_id
-                ]
-
-        _LOGGER.debug("Completed retrieving device activities")
-
     async def async_get_device_detail(self, device):
         """Return the detail for a device."""
         if isinstance(device, Lock):
@@ -317,11 +279,11 @@ class AugustData:
         await self._async_update_locks_detail()
         return self._lock_detail_by_id[device_id]
 
-    def get_lock_name(self, device_id):
+    def get_device_name(self, device_id):
-        """Return lock name as August has it stored."""
+        """Return doorbell or lock name as August has it stored."""
-        for lock in self._locks:
+        for device in itertools.chain(self._locks, self._doorbells):
-            if lock.device_id == device_id:
+            if device.device_id == device_id:
-                return lock.device_name
+                return device.device_name
 
     @Throttle(MIN_TIME_BETWEEN_DETAIL_UPDATES)
     async def _async_update_locks_detail(self):
@@ -354,11 +316,17 @@ class AugustData:
         _LOGGER.debug("Completed retrieving %s detail", device_type)
         return detail_by_id
 
+    async def async_signal_operation_changed_device_state(self, device_id):
+        """Signal a device update when an operation changes state."""
+        _LOGGER.debug(
+            "async_dispatcher_send (from operation): AUGUST_DEVICE_UPDATE-%s", device_id
+        )
+        async_dispatcher_send(self._hass, f"{AUGUST_DEVICE_UPDATE}-{device_id}")
+
     def lock(self, device_id):
         """Lock the device."""
-        return _call_api_operation_that_requires_bridge(
+        return self._call_api_op_requires_bridge(
-            self.get_lock_name(device_id),
+            device_id,
-            "lock",
             self._api.lock_return_activities,
             self._august_gateway.access_token,
             device_id,
@@ -366,14 +334,26 @@ class AugustData:
 
     def unlock(self, device_id):
         """Unlock the device."""
-        return _call_api_operation_that_requires_bridge(
+        return self._call_api_op_requires_bridge(
-            self.get_lock_name(device_id),
+            device_id,
-            "unlock",
             self._api.unlock_return_activities,
             self._august_gateway.access_token,
             device_id,
         )
 
+    def _call_api_op_requires_bridge(self, device_id, func, *args, **kwargs):
+        """Call an API that requires the bridge to be online and will change the device state."""
+        ret = None
+        try:
+            ret = func(*args, **kwargs)
+        except AugustApiHTTPError as err:
+            device_name = self.get_device_name(device_id)
+            if device_name is None:
+                device_name = f"DeviceID: {device_id}"
+            raise HomeAssistantError(f"{device_name}: {err}")
+
+        return ret
+
     def _filter_inoperative_locks(self):
         # Remove non-operative locks as there must
         # be a bridge (August Connect) for them to
@@ -400,16 +380,3 @@ class AugustData:
                 operative_locks.append(lock)
 
         self._locks = operative_locks
-
-
-def _call_api_operation_that_requires_bridge(
-    device_name, operation_name, func, *args, **kwargs
-):
-    """Call an API that requires the bridge to be online."""
-    ret = None
-    try:
-        ret = func(*args, **kwargs)
-    except AugustApiHTTPError as err:
-        raise HomeAssistantError(device_name + ": " + str(err))
-
-    return ret

homeassistant/components/august/activity.py

@@ -0,0 +1,141 @@
+"""Consume the august activity stream."""
+from functools import partial
+import logging
+
+from requests import RequestException
+
+from homeassistant.core import callback
+from homeassistant.helpers.dispatcher import async_dispatcher_send
+from homeassistant.helpers.event import async_track_time_interval
+from homeassistant.util.dt import utcnow
+
+from .const import ACTIVITY_UPDATE_INTERVAL, AUGUST_DEVICE_UPDATE
+
+_LOGGER = logging.getLogger(__name__)
+
+ACTIVITY_STREAM_FETCH_LIMIT = 10
+ACTIVITY_CATCH_UP_FETCH_LIMIT = 200
+
+
+class ActivityStream:
+    """August activity stream handler."""
+
+    def __init__(self, hass, api, august_gateway, house_ids):
+        """Init August activity stream object."""
+        self._hass = hass
+        self._august_gateway = august_gateway
+        self._api = api
+        self._house_ids = house_ids
+        self._latest_activities_by_id_type = {}
+        self._last_update_time = None
+        self._abort_async_track_time_interval = None
+
+    async def async_start(self):
+        """Start fetching updates from the activity stream."""
+        await self._async_update(utcnow)
+        self._abort_async_track_time_interval = async_track_time_interval(
+            self._hass, self._async_update, ACTIVITY_UPDATE_INTERVAL
+        )
+
+    @callback
+    def async_stop(self):
+        """Stop fetching updates from the activity stream."""
+        if self._abort_async_track_time_interval is None:
+            return
+        self._abort_async_track_time_interval()
+
+    @callback
+    def async_get_latest_device_activity(self, device_id, activity_types):
+        """Return latest activity that is one of the acitivty_types."""
+        if device_id not in self._latest_activities_by_id_type:
+            return None
+
+        latest_device_activities = self._latest_activities_by_id_type[device_id]
+        latest_activity = None
+
+        for activity_type in activity_types:
+            if activity_type in latest_device_activities:
+                if (
+                    latest_activity is not None
+                    and latest_device_activities[activity_type].activity_start_time
+                    <= latest_activity.activity_start_time
+                ):
+                    continue
+                latest_activity = latest_device_activities[activity_type]
+
+        return latest_activity
+
+    async def _async_update(self, time):
+        """Update the activity stream from August."""
+
+        # This is the only place we refresh the api token
+        await self._august_gateway.async_refresh_access_token_if_needed()
+        await self._update_device_activities(time)
+
+    async def _update_device_activities(self, time):
+        _LOGGER.debug("Start retrieving device activities")
+
+        limit = (
+            ACTIVITY_STREAM_FETCH_LIMIT
+            if self._last_update_time
+            else ACTIVITY_CATCH_UP_FETCH_LIMIT
+        )
+
+        for house_id in self._house_ids:
+            _LOGGER.debug("Updating device activity for house id %s", house_id)
+            try:
+                activities = await self._hass.async_add_executor_job(
+                    partial(
+                        self._api.get_house_activities,
+                        self._august_gateway.access_token,
+                        house_id,
+                        limit=limit,
+                    )
+                )
+            except RequestException as ex:
+                _LOGGER.error(
+                    "Request error trying to retrieve activity for house id %s: %s",
+                    house_id,
+                    ex,
+                )
+            _LOGGER.debug(
+                "Completed retrieving device activities for house id %s", house_id
+            )
+
+            updated_device_ids = self._process_newer_device_activities(activities)
+
+            if updated_device_ids:
+                for device_id in updated_device_ids:
+                    _LOGGER.debug(
+                        "async_dispatcher_send (from activity stream): AUGUST_DEVICE_UPDATE-%s",
+                        device_id,
+                    )
+                    async_dispatcher_send(
+                        self._hass, f"{AUGUST_DEVICE_UPDATE}-{device_id}"
+                    )
+
+        self._last_update_time = time
+
+    def _process_newer_device_activities(self, activities):
+        updated_device_ids = set()
+        for activity in activities:
+            self._latest_activities_by_id_type.setdefault(activity.device_id, {})
+
+            lastest_activity = self._latest_activities_by_id_type[
+                activity.device_id
+            ].get(activity.activity_type)
+
+            # Ignore activities that are older than the latest one
+            if (
+                lastest_activity
+                and lastest_activity.activity_start_time >= activity.activity_start_time
+            ):
+                continue
+
+            self._latest_activities_by_id_type[activity.device_id][
+                activity.activity_type
+            ] = activity
+
+            updated_device_ids.add(activity.device_id)
+
+        return updated_device_ids

homeassistant/components/august/binary_sensor.py

@@ -12,12 +12,24 @@ from homeassistant.components.binary_sensor import (
     DEVICE_CLASS_OCCUPANCY,
     BinarySensorDevice,
 )
+from homeassistant.core import callback
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
+from homeassistant.helpers.event import async_track_point_in_utc_time
+from homeassistant.util.dt import utcnow
 
-from .const import DATA_AUGUST, DEFAULT_NAME, DOMAIN
+from .const import (
+    AUGUST_DEVICE_UPDATE,
+    DATA_AUGUST,
+    DEFAULT_NAME,
+    DOMAIN,
+    MIN_TIME_BETWEEN_DETAIL_UPDATES,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
-SCAN_INTERVAL = timedelta(seconds=5)
+TIME_TO_DECLARE_DETECTION = timedelta(seconds=60)
+
+SCAN_INTERVAL = MIN_TIME_BETWEEN_DETAIL_UPDATES
 
 
 async def _async_retrieve_online_state(data, detail):
@@ -43,11 +55,13 @@ async def _async_retrieve_ding_state(data, detail):
 
 async def _async_activity_time_based_state(data, device_id, activity_types):
     """Get the latest state of the sensor."""
-    latest = await data.async_get_latest_device_activity(device_id, *activity_types)
+    latest = data.activity_stream.async_get_latest_device_activity(
+        device_id, activity_types
+    )
 
     if latest is not None:
         start = latest.activity_start_time
-        end = latest.activity_end_time + timedelta(seconds=45)
+        end = latest.activity_end_time + TIME_TO_DECLARE_DETECTION
         return start <= datetime.now() <= end
     return None
 
@@ -98,6 +112,7 @@ class AugustDoorBinarySensor(BinarySensorDevice):
 
     def __init__(self, data, sensor_type, door):
         """Initialize the sensor."""
+        self._undo_dispatch_subscription = None
         self._data = data
         self._sensor_type = sensor_type
         self._door = door
@@ -128,8 +143,8 @@ class AugustDoorBinarySensor(BinarySensorDevice):
 
     async def async_update(self):
         """Get the latest state of the sensor and update activity."""
-        door_activity = await self._data.async_get_latest_device_activity(
+        door_activity = self._data.activity_stream.async_get_latest_device_activity(
-            self._door.device_id, ActivityType.DOOR_OPERATION
+            self._door.device_id, [ActivityType.DOOR_OPERATION]
         )
         detail = await self._data.async_get_lock_detail(self._door.device_id)
 
@@ -162,12 +177,31 @@ class AugustDoorBinarySensor(BinarySensorDevice):
             "model": self._model,
         }
 
+    async def async_added_to_hass(self):
+        """Register callbacks."""
+
+        @callback
+        def update():
+            """Update the state."""
+            self.async_schedule_update_ha_state(True)
+
+        self._undo_dispatch_subscription = async_dispatcher_connect(
+            self.hass, f"{AUGUST_DEVICE_UPDATE}-{self._door.device_id}", update
+        )
+
+    async def async_will_remove_from_hass(self):
+        """Undo subscription."""
+        if self._undo_dispatch_subscription:
+            self._undo_dispatch_subscription()
+
 
 class AugustDoorbellBinarySensor(BinarySensorDevice):
     """Representation of an August binary sensor."""
 
     def __init__(self, data, sensor_type, doorbell):
         """Initialize the sensor."""
+        self._undo_dispatch_subscription = None
+        self._check_for_off_update_listener = None
         self._data = data
         self._sensor_type = sensor_type
         self._doorbell = doorbell
@@ -198,6 +232,7 @@ class AugustDoorbellBinarySensor(BinarySensorDevice):
 
     async def async_update(self):
         """Get the latest state of the sensor."""
+        self._cancel_any_pending_updates()
         async_state_provider = SENSOR_TYPES_DOORBELL[self._sensor_type][
             SENSOR_STATE_PROVIDER
         ]
@@ -217,6 +252,28 @@ class AugustDoorbellBinarySensor(BinarySensorDevice):
             self._firmware_version = detail.firmware_version
             self._model = detail.model
             self._state = await async_state_provider(self._data, detail)
+            if self._state and self.device_class != DEVICE_CLASS_CONNECTIVITY:
+                self._schedule_update_to_recheck_turn_off_sensor()
+
+    def _schedule_update_to_recheck_turn_off_sensor(self):
+        """Schedule an update to recheck the sensor to see if it is ready to turn off."""
+
+        @callback
+        def _scheduled_update(now):
+            """Timer callback for sensor update."""
+            _LOGGER.debug("%s: executing scheduled update", self.entity_id)
+            self.async_schedule_update_ha_state(True)
+            self._check_for_off_update_listener = None
+
+        self._check_for_off_update_listener = async_track_point_in_utc_time(
+            self.hass, _scheduled_update, utcnow() + TIME_TO_DECLARE_DETECTION
+        )
+
+    def _cancel_any_pending_updates(self):
+        """Cancel any updates to recheck a sensor to see if it is ready to turn off."""
+        if self._check_for_off_update_listener:
+            self._check_for_off_update_listener()
+            self._check_for_off_update_listener = None
 
     @property
     def unique_id(self) -> str:
@@ -236,3 +293,20 @@ class AugustDoorbellBinarySensor(BinarySensorDevice):
             "sw_version": self._firmware_version,
             "model": self._model,
         }
+
+    async def async_added_to_hass(self):
+        """Register callbacks."""
+
+        @callback
+        def update():
+            """Update the state."""
+            self.async_schedule_update_ha_state(True)
+
+        self._undo_dispatch_subscription = async_dispatcher_connect(
+            self.hass, f"{AUGUST_DEVICE_UPDATE}-{self._doorbell.device_id}", update
+        )
+
+    async def async_will_remove_from_hass(self):
+        """Undo subscription."""
+        if self._undo_dispatch_subscription:
+            self._undo_dispatch_subscription()

homeassistant/components/august/camera.py

@@ -1,14 +1,22 @@
-"""Support for August camera."""
+"""Support for August doorbell camera."""
-from datetime import timedelta
 
 from august.activity import ActivityType
 from august.util import update_doorbell_image_from_activity
 
 from homeassistant.components.camera import Camera
+from homeassistant.core import callback
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
 
-from .const import DATA_AUGUST, DEFAULT_NAME, DEFAULT_TIMEOUT, DOMAIN
+from .const import (
+    AUGUST_DEVICE_UPDATE,
+    DATA_AUGUST,
+    DEFAULT_NAME,
+    DEFAULT_TIMEOUT,
+    DOMAIN,
+    MIN_TIME_BETWEEN_DETAIL_UPDATES,
+)
 
-SCAN_INTERVAL = timedelta(seconds=5)
+SCAN_INTERVAL = MIN_TIME_BETWEEN_DETAIL_UPDATES
 
 
 async def async_setup_entry(hass, config_entry, async_add_entities):
@@ -28,6 +36,7 @@ class AugustCamera(Camera):
     def __init__(self, data, doorbell, timeout):
         """Initialize a August security camera."""
         super().__init__()
+        self._undo_dispatch_subscription = None
         self._data = data
         self._doorbell = doorbell
         self._doorbell_detail = None
@@ -67,8 +76,8 @@ class AugustCamera(Camera):
         self._doorbell_detail = await self._data.async_get_doorbell_detail(
             self._doorbell.device_id
         )
-        doorbell_activity = await self._data.async_get_latest_device_activity(
+        doorbell_activity = self._data.activity_stream.async_get_latest_device_activity(
-            self._doorbell.device_id, ActivityType.DOORBELL_MOTION
+            self._doorbell.device_id, [ActivityType.DOORBELL_MOTION]
         )
 
         if doorbell_activity is not None:
@@ -117,3 +126,20 @@ class AugustCamera(Camera):
             "sw_version": self._firmware_version,
             "model": self._model,
         }
+
+    async def async_added_to_hass(self):
+        """Register callbacks."""
+
+        @callback
+        def update():
+            """Update the state."""
+            self.async_schedule_update_ha_state(True)
+
+        self._undo_dispatch_subscription = async_dispatcher_connect(
+            self.hass, f"{AUGUST_DEVICE_UPDATE}-{self._doorbell.device_id}", update
+        )
+
+    async def async_will_remove_from_hass(self):
+        """Undo subscription."""
+        if self._undo_dispatch_subscription:
+            self._undo_dispatch_subscription()

homeassistant/components/august/const.py

@@ -8,6 +8,8 @@ CONF_ACCESS_TOKEN_CACHE_FILE = "access_token_cache_file"
 CONF_LOGIN_METHOD = "login_method"
 CONF_INSTALL_ID = "install_id"
 
+AUGUST_DEVICE_UPDATE = "august_devices_update"
+
 VERIFICATION_CODE_KEY = "verification_code"
 
 NOTIFICATION_ID = "august_notification"
@@ -20,16 +22,14 @@ DATA_AUGUST = "data_august"
 DEFAULT_NAME = "August"
 DOMAIN = "august"
 
-# Limit battery, online, and hardware updates to 1800 seconds
+# Limit battery, online, and hardware updates to hourly
 # in order to reduce the number of api requests and
 # avoid hitting rate limits
-MIN_TIME_BETWEEN_DETAIL_UPDATES = timedelta(seconds=1800)
+MIN_TIME_BETWEEN_DETAIL_UPDATES = timedelta(hours=1)
 
 # Activity needs to be checked more frequently as the
 # doorbell motion and rings are included here
-MIN_TIME_BETWEEN_ACTIVITY_UPDATES = timedelta(seconds=10)
+ACTIVITY_UPDATE_INTERVAL = timedelta(seconds=10)
-
-DEFAULT_SCAN_INTERVAL = timedelta(seconds=10)
 
 LOGIN_METHODS = ["phone", "email"]
 

homeassistant/components/august/lock.py

@@ -1,5 +1,4 @@
 """Support for August lock."""
-from datetime import timedelta
 import logging
 
 from august.activity import ActivityType
@@ -8,12 +7,20 @@ from august.util import update_lock_detail_from_activity
 
 from homeassistant.components.lock import LockDevice
 from homeassistant.const import ATTR_BATTERY_LEVEL
+from homeassistant.core import callback
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
 
-from .const import DATA_AUGUST, DEFAULT_NAME, DOMAIN
+from .const import (
+    AUGUST_DEVICE_UPDATE,
+    DATA_AUGUST,
+    DEFAULT_NAME,
+    DOMAIN,
+    MIN_TIME_BETWEEN_DETAIL_UPDATES,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
-SCAN_INTERVAL = timedelta(seconds=5)
+SCAN_INTERVAL = MIN_TIME_BETWEEN_DETAIL_UPDATES
 
 
 async def async_setup_entry(hass, config_entry, async_add_entities):
@@ -33,6 +40,7 @@ class AugustLock(LockDevice):
 
     def __init__(self, data, lock):
         """Initialize the lock."""
+        self._undo_dispatch_subscription = None
         self._data = data
         self._lock = lock
         self._lock_status = None
@@ -58,7 +66,9 @@ class AugustLock(LockDevice):
             update_lock_detail_from_activity(self._lock_detail, lock_activity)
 
         if self._update_lock_status_from_detail():
-            self.schedule_update_ha_state()
+            await self._data.async_signal_operation_changed_device_state(
+                self._lock.device_id
+            )
 
     def _update_lock_status_from_detail(self):
         detail = self._lock_detail
@@ -77,8 +87,8 @@ class AugustLock(LockDevice):
     async def async_update(self):
         """Get the latest state of the sensor and update activity."""
         self._lock_detail = await self._data.async_get_lock_detail(self._lock.device_id)
-        lock_activity = await self._data.async_get_latest_device_activity(
+        lock_activity = self._data.activity_stream.async_get_latest_device_activity(
-            self._lock.device_id, ActivityType.LOCK_OPERATION
+            self._lock.device_id, [ActivityType.LOCK_OPERATION]
         )
 
         if lock_activity is not None:
@@ -142,3 +152,20 @@ class AugustLock(LockDevice):
     def unique_id(self) -> str:
         """Get the unique id of the lock."""
         return f"{self._lock.device_id:s}_lock"
+
+    async def async_added_to_hass(self):
+        """Register callbacks."""
+
+        @callback
+        def update():
+            """Update the state."""
+            self.async_schedule_update_ha_state(True)
+
+        self._undo_dispatch_subscription = async_dispatcher_connect(
+            self.hass, f"{AUGUST_DEVICE_UPDATE}-{self._lock.device_id}", update
+        )
+
+    async def async_will_remove_from_hass(self):
+        """Undo subscription."""
+        if self._undo_dispatch_subscription:
+            self._undo_dispatch_subscription()

homeassistant/components/august/sensor.py

@@ -1,11 +1,10 @@
 """Support for August sensors."""
-from datetime import timedelta
 import logging
 
 from homeassistant.components.sensor import DEVICE_CLASS_BATTERY
 from homeassistant.helpers.entity import Entity
 
-from .const import DATA_AUGUST, DEFAULT_NAME, DOMAIN
+from .const import DATA_AUGUST, DEFAULT_NAME, DOMAIN, MIN_TIME_BETWEEN_DETAIL_UPDATES
 
 BATTERY_LEVEL_FULL = "Full"
 BATTERY_LEVEL_MEDIUM = "Medium"
@@ -13,7 +12,7 @@ BATTERY_LEVEL_LOW = "Low"
 
 _LOGGER = logging.getLogger(__name__)
 
-SCAN_INTERVAL = timedelta(seconds=5)
+SCAN_INTERVAL = MIN_TIME_BETWEEN_DETAIL_UPDATES
 
 
 def _retrieve_device_battery_state(detail):

tests/components/august/test_binary_sensor.py

@@ -1,7 +1,5 @@
 """The binary_sensor tests for the august platform."""
 
-import pytest
-
 from homeassistant.components.lock import DOMAIN as LOCK_DOMAIN
 from homeassistant.const import (
     ATTR_ENTITY_ID,
@@ -20,11 +18,6 @@ from tests.components.august.mocks import (
 )
 
 
-@pytest.mark.skip(
-    reason="The lock and doorsense can get out of sync due to update intervals, "
-    + "this is an existing bug which will be fixed with dispatcher events to tell "
-    + "all linked devices to update."
-)
 async def test_doorsense(hass):
     """Test creation of a lock with doorsense and bridge."""
     lock_one = await _mock_lock_from_fixture(
@@ -33,24 +26,32 @@ async def test_doorsense(hass):
     lock_details = [lock_one]
     await _create_august_with_devices(hass, lock_details)
 
-    binary_sensor_abc_name = hass.states.get("binary_sensor.abc_name_open")
+    binary_sensor_online_with_doorsense_name = hass.states.get(
-    assert binary_sensor_abc_name.state == STATE_ON
+        "binary_sensor.online_with_doorsense_name_open"
+    )
+    assert binary_sensor_online_with_doorsense_name.state == STATE_ON
 
     data = {}
-    data[ATTR_ENTITY_ID] = "lock.abc_name"
+    data[ATTR_ENTITY_ID] = "lock.online_with_doorsense_name"
     assert await hass.services.async_call(
         LOCK_DOMAIN, SERVICE_UNLOCK, data, blocking=True
     )
+    await hass.async_block_till_done()
 
-    binary_sensor_abc_name = hass.states.get("binary_sensor.abc_name_open")
+    binary_sensor_online_with_doorsense_name = hass.states.get(
-    assert binary_sensor_abc_name.state == STATE_ON
+        "binary_sensor.online_with_doorsense_name_open"
+    )
+    assert binary_sensor_online_with_doorsense_name.state == STATE_ON
 
     assert await hass.services.async_call(
         LOCK_DOMAIN, SERVICE_LOCK, data, blocking=True
     )
+    await hass.async_block_till_done()
 
-    binary_sensor_abc_name = hass.states.get("binary_sensor.abc_name_open")
+    binary_sensor_online_with_doorsense_name = hass.states.get(
-    assert binary_sensor_abc_name.state == STATE_OFF
+        "binary_sensor.online_with_doorsense_name_open"
+    )
+    assert binary_sensor_online_with_doorsense_name.state == STATE_OFF
 
 
 async def test_create_doorbell(hass):

tests/components/august/test_lock.py

@@ -28,8 +28,8 @@ async def test_lock_device_registry(hass):
     reg_device = device_registry.async_get_device(
         identifiers={("august", "online_with_doorsense")}, connections=set()
     )
-    assert "AUG-MD01" == reg_device.model
+    assert reg_device.model == "AUG-MD01"
-    assert "undefined-4.3.0-1.8.14" == reg_device.sw_version
+    assert reg_device.sw_version == "undefined-4.3.0-1.8.14"
 
 
 async def test_one_lock_operation(hass):