Speed up coordinator refresh with parallel fetches and landscape lock (#401)

Periodic refreshes were taking ~2s because independent endpoints were
awaited sequentially and parallel per-zone landscape lookups each fired
a duplicate HTTP request to the same device-level endpoint.

- Gather `devices` and `timer_programs` concurrently in the coordinator
- Fan out per-device history/landscape bundles across all devices in
  parallel (previously the outer device loop was sequential)
- Add a per-device `asyncio.Lock` around `_refresh_landscapes` so
  concurrent callers share one HTTP request instead of duplicating N

Author: sebr

custom_components/bhyve/coordinator.py

@@ -30,7 +30,7 @@
     from homeassistant.core import HomeAssistant
 
     from .pybhyve import BHyveClient
-    from .pybhyve.typings import BHyveDevice, BHyveTimerProgram
+    from .pybhyve.typings import BHyveDevice
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -58,63 +58,34 @@ def __init__(
     async def _async_update_data(self) -> dict[str, Any]:
         """Fetch data from API (periodic polling)."""
         try:
-            # Fetch devices
-            devices: list[BHyveDevice] = await self.client.devices
+            # devices and timer_programs are independent endpoints; fetch in
+            # parallel so the refresh is bounded by the slower of the two.
+            devices, programs = await asyncio.gather(
+                self.client.devices,
+                self.client.timer_programs,
+            )
+
+            # Fan out per-device history/landscape fetches across all devices in
+            # parallel. Previously this loop was sequential across devices, so
+            # multi-device accounts paid N times the per-device roundtrip.
+            device_results = await asyncio.gather(
+                *[self._fetch_device_bundle(device) for device in devices],
+                return_exceptions=True,
+            )
 
-            # Build data structure
             data: dict[str, Any] = {
                 "devices": {},
                 "programs": {},
             }
 
-            # Process each device - fetch history and landscapes in parallel per device
-            for device in devices:
-                device_id = device.get("id")
-                device_type = device.get("type")
-                if not device_id:
+            for result in device_results:
+                if isinstance(result, BaseException):
+                    _LOGGER.debug("Error fetching device bundle: %s", result)
                     continue
+                if result is None:
+                    continue
+                data["devices"][result["device"]["id"]] = result
 
-                # Only fetch history and landscapes for sprinkler_timer devices
-                # Flood sensors and other device types don't have these features
-                if device_type == DEVICE_SPRINKLER:
-                    # Fetch history and landscapes in parallel for this device
-                    history_task = self._fetch_device_history(device_id)
-                    landscapes_task = self._fetch_landscapes(device_id, device)
-
-                    history, landscapes = await asyncio.gather(
-                        history_task,
-                        landscapes_task,
-                        return_exceptions=True,
-                    )
-
-                    # Handle exceptions from parallel tasks
-                    if isinstance(history, Exception):
-                        _LOGGER.debug(
-                            "Error fetching history for device %s: %s",
-                            device_id,
-                            history,
-                        )
-                        history = []
-                    if isinstance(landscapes, Exception):
-                        _LOGGER.debug(
-                            "Error fetching landscapes for device %s: %s",
-                            device_id,
-                            landscapes,
-                        )
-                        landscapes = {}
-                else:
-                    # Non-sprinkler devices don't have history or landscapes
-                    history = []
-                    landscapes = {}
-
-                data["devices"][device_id] = {
-                    "device": device,
-                    "history": history,
-                    "landscapes": landscapes,
-                }
-
-            # Fetch programs
-            programs: list[BHyveTimerProgram] = await self.client.timer_programs
             for program in programs:
                 program_id = program.get("id")
                 if program_id:
@@ -129,6 +100,41 @@ async def _async_update_data(self) -> dict[str, Any]:
             msg = f"Error communicating with API: {err}"
             raise UpdateFailed(msg) from err
 
+    async def _fetch_device_bundle(self, device: BHyveDevice) -> dict[str, Any] | None:
+        """Fetch history + landscapes for a single device."""
+        device_id = device.get("id")
+        if not device_id:
+            return None
+
+        if device.get("type") == DEVICE_SPRINKLER:
+            history, landscapes = await asyncio.gather(
+                self._fetch_device_history(device_id),
+                self._fetch_landscapes(device_id, device),
+                return_exceptions=True,
+            )
+
+            if isinstance(history, Exception):
+                _LOGGER.debug(
+                    "Error fetching history for device %s: %s", device_id, history
+                )
+                history = []
+            if isinstance(landscapes, Exception):
+                _LOGGER.debug(
+                    "Error fetching landscapes for device %s: %s",
+                    device_id,
+                    landscapes,
+                )
+                landscapes = {}
+        else:
+            history = []
+            landscapes = {}
+
+        return {
+            "device": device,
+            "history": history,
+            "landscapes": landscapes,
+        }
+
     async def _fetch_device_history(self, device_id: str) -> list[dict[str, Any]]:
         """Fetch watering history for a device."""
         try:

custom_components/bhyve/pybhyve/client.py

@@ -1,5 +1,6 @@
 """Define an object to interact with the REST API."""
 
+import asyncio
 import logging
 import re
 import time
@@ -50,6 +51,7 @@ def __init__(self, username: str, password: str, session: ClientSession) -> None
 
         self._landscapes: dict[str, list[BHyveZoneLandscape]] = {}
         self._last_poll_landscapes: dict[str, float] = {}
+        self._landscape_locks: dict[str, asyncio.Lock] = {}
 
     async def _request(
         self,
@@ -149,20 +151,29 @@ async def _refresh_device_history(
     async def _refresh_landscapes(
         self, device_id: str, *, force_update: bool = False
     ) -> None:
-        now = time.time()
-        if force_update:
-            _LOGGER.debug("Forcing landscape refresh %s", device_id)
-        elif now - self._last_poll_landscapes.get(device_id, 0) < API_POLL_PERIOD:
-            return
-
-        device_landscapes = await self._request(
-            "get",
-            f"{LANDSCAPE_DESCRIPTIONS_PATH}/{device_id}",
-            params={"t": str(time.time())},
-        )
-
-        self._landscapes[device_id] = device_landscapes or []
-        self._last_poll_landscapes[device_id] = now
+        # Single-flight: concurrent callers for the same device share one HTTP
+        # request. Without this, a coordinator refresh that fans out per-zone
+        # landscape lookups for N zones fires N duplicate requests.
+        lock = self._landscape_locks.get(device_id)
+        if lock is None:
+            lock = asyncio.Lock()
+            self._landscape_locks[device_id] = lock
+
+        async with lock:
+            now = time.time()
+            if force_update:
+                _LOGGER.debug("Forcing landscape refresh %s", device_id)
+            elif now - self._last_poll_landscapes.get(device_id, 0) < API_POLL_PERIOD:
+                return
+
+            device_landscapes = await self._request(
+                "get",
+                f"{LANDSCAPE_DESCRIPTIONS_PATH}/{device_id}",
+                params={"t": str(time.time())},
+            )
+
+            self._landscapes[device_id] = device_landscapes or []
+            self._last_poll_landscapes[device_id] = now
 
     async def _async_ws_handler(self, async_callback: Callable, data: Any) -> None:
         """Process incoming websocket message."""

tests/test_client.py

@@ -1,5 +1,6 @@
 """Test BHyve pybhyve client."""
 
+import asyncio
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
@@ -106,3 +107,33 @@ async def test_rate_limits_same_device(self, client: BHyveClient) -> None:
         await client.get_landscape("device-a", 1)
 
         assert client._request.await_count == 1
+
+    async def test_concurrent_calls_dedupe_to_single_request(
+        self, client: BHyveClient
+    ) -> None:
+        """
+        Concurrent get_landscape calls for the same device share one HTTP request.
+
+        The coordinator fans out per-zone landscape lookups in parallel. Without
+        single-flight locking, N zones trigger N duplicate requests to the same
+        endpoint.
+        """
+        call_started = asyncio.Event()
+        release = asyncio.Event()
+
+        async def slow_request(*_args: object, **_kwargs: object) -> list:
+            call_started.set()
+            await release.wait()
+            return [{"station": 1}, {"station": 2}, {"station": 3}]
+
+        client._request = AsyncMock(side_effect=slow_request)
+
+        tasks = [
+            asyncio.create_task(client.get_landscape("device-a", zone_id))
+            for zone_id in (1, 2, 3)
+        ]
+        await call_started.wait()
+        release.set()
+        await asyncio.gather(*tasks)
+
+        assert client._request.await_count == 1