diff --git a/custom_components/panasonic_cc/panasonic/__init__.py b/custom_components/panasonic_cc/panasonic/__init__.py index 5c492a5..61b670c 100644 --- a/custom_components/panasonic_cc/panasonic/__init__.py +++ b/custom_components/panasonic_cc/panasonic/__init__.py @@ -24,6 +24,7 @@ ) from .const import DATA_COORDINATORS, ENERGY_COORDINATORS from .coordinator import ( + MAX_CONCURRENT_API_CALLS, PanasonicDeviceCoordinator, PanasonicDeviceEnergyCoordinator, ) @@ -63,6 +64,9 @@ async def async_setup_panasonic( devices = api.get_devices() _LOGGER.info("Got %s Panasonic devices", len(devices)) + api_semaphore = asyncio.Semaphore(MAX_CONCURRENT_API_CALLS) + hass.data[DOMAIN]["_api_semaphore"] = api_semaphore + data_coordinators: list[PanasonicDeviceCoordinator] = [] energy_coordinators: list[PanasonicDeviceEnergyCoordinator] = [] @@ -70,11 +74,11 @@ async def async_setup_panasonic( device_coordinators_uninitialized: list[tuple[PanasonicDeviceCoordinator, PanasonicDeviceInfo]] = [] for device in devices: try: - device_coordinator = PanasonicDeviceCoordinator(hass, config, api, device) + device_coordinator = PanasonicDeviceCoordinator(hass, config, api, device, api_semaphore) device_coordinators_uninitialized.append((device_coordinator, device)) if enable_daily_energy_sensor: energy_coordinators.append( - PanasonicDeviceEnergyCoordinator(hass, config, api, device) + PanasonicDeviceEnergyCoordinator(hass, config, api, device, api_semaphore) ) except Exception as exc: _LOGGER.warning("Failed to create coordinator for device %s: %s", device.name, exc, exc_info=True) diff --git a/custom_components/panasonic_cc/panasonic/coordinator.py b/custom_components/panasonic_cc/panasonic/coordinator.py index cde2d6c..1f059ea 100644 --- a/custom_components/panasonic_cc/panasonic/coordinator.py +++ b/custom_components/panasonic_cc/panasonic/coordinator.py @@ -1,6 +1,7 @@ """Coordinators for Panasonic Comfort Cloud devices.""" import asyncio import logging +import random from datetime import timedelta from aiohttp import ClientResponseError @@ -32,6 +33,9 @@ MAX_CONSECUTIVE_FAILURES = 5 BACKOFF_MULTIPLIER = 2 MAX_UPDATE_INTERVAL = 600 # seconds +MAX_CONCURRENT_API_CALLS = 3 +MIN_RATE_LIMIT_INTERVAL = 30 +BACKOFF_MULTIPLIER_RATE_LIMIT = 4 _LOGGER = logging.getLogger(__name__) @@ -63,6 +67,7 @@ def __init__( config: dict, api_client: ApiClient, device_info: PanasonicDeviceInfo, + api_semaphore: asyncio.Semaphore, ) -> None: """Initialize the coordinator.""" self._base_interval = config.get( @@ -84,6 +89,7 @@ def __init__( self._auth_failed = False self._last_error: FriendlyError | None = None self._last_command_error: FriendlyError | None = None + self._api_semaphore = api_semaphore @property def last_error(self) -> FriendlyError | None: @@ -141,7 +147,8 @@ def get_change_request_builder(self) -> ChangeRequestBuilder: async def async_apply_changes(self, request_builder: ChangeRequestBuilder) -> None: """Apply changes to the device.""" try: - await self._api_client.set_device_raw(self.device, request_builder.build()) + async with self._api_semaphore: + await self._api_client.set_device_raw(self.device, request_builder.build()) # Clear command error on success self._last_command_error = None except Exception as err: @@ -193,7 +200,9 @@ async def _async_update_data(self) -> int: try: if self._device is None: - self._device = await self._api_client.get_device(self._device_info) + async with self._api_semaphore: + await asyncio.sleep(random.uniform(0.0, 0.5)) + self._device = await self._api_client.get_device(self._device_info) _LOGGER.debug( "%s Device features - Nanoe: %s, Eco Navi: %s, AI Eco: %s", self._device_info.name, @@ -204,7 +213,10 @@ async def _async_update_data(self) -> int: self._update_id = 1 self._reset_backoff() return self._update_id - if await self._api_client.try_update_device(self._device): + async with self._api_semaphore: + await asyncio.sleep(random.uniform(0.0, 0.5)) + updated = await self._api_client.try_update_device(self._device) + if updated: self._update_id += 1 self._reset_backoff() return self._update_id @@ -238,10 +250,15 @@ def _reset_backoff(self) -> None: def _handle_failure(self, err: Exception | None = None) -> None: """Handle API failure with exponential backoff.""" self._consecutive_failures += 1 + multiplier = BACKOFF_MULTIPLIER + if err is not None and classify_error(err).category == ErrorCategory.RATE_LIMIT: + multiplier = BACKOFF_MULTIPLIER_RATE_LIMIT new_interval = min( - self._base_interval * (BACKOFF_MULTIPLIER ** self._consecutive_failures), + self._base_interval * (multiplier ** self._consecutive_failures), MAX_UPDATE_INTERVAL, ) + if err is not None and classify_error(err).category == ErrorCategory.RATE_LIMIT: + new_interval = max(new_interval, MIN_RATE_LIMIT_INTERVAL) self.update_interval = timedelta(seconds=new_interval) if err is not None: self._last_error = classify_error(err) @@ -273,6 +290,7 @@ def __init__( config: dict, api_client: ApiClient, device_info: PanasonicDeviceInfo, + api_semaphore: asyncio.Semaphore, ) -> None: """Initialize the coordinator.""" self._base_interval = config.get( @@ -291,6 +309,7 @@ def __init__( self._consecutive_failures = 0 self._auth_failed = False self._last_error: FriendlyError | None = None + self._api_semaphore = api_semaphore @property def last_error(self) -> FriendlyError | None: @@ -341,11 +360,16 @@ async def _async_update_data(self) -> int: try: if self._energy is None: - self._energy = await self._api_client.async_get_energy(self._device_info) + async with self._api_semaphore: + await asyncio.sleep(random.uniform(0.0, 0.5)) + self._energy = await self._api_client.async_get_energy(self._device_info) self._update_id = 1 self._reset_backoff() return self._update_id - if await self._api_client.async_try_update_energy(self._energy): + async with self._api_semaphore: + await asyncio.sleep(random.uniform(0.0, 0.5)) + updated = await self._api_client.async_try_update_energy(self._energy) + if updated: self._update_id += 1 self._reset_backoff() return self._update_id @@ -379,10 +403,15 @@ def _reset_backoff(self) -> None: def _handle_failure(self, err: Exception | None = None) -> None: """Handle API failure with exponential backoff.""" self._consecutive_failures += 1 + multiplier = BACKOFF_MULTIPLIER + if err is not None and classify_error(err).category == ErrorCategory.RATE_LIMIT: + multiplier = BACKOFF_MULTIPLIER_RATE_LIMIT new_interval = min( - self._base_interval * (BACKOFF_MULTIPLIER ** self._consecutive_failures), + self._base_interval * (multiplier ** self._consecutive_failures), MAX_UPDATE_INTERVAL, ) + if err is not None and classify_error(err).category == ErrorCategory.RATE_LIMIT: + new_interval = max(new_interval, MIN_RATE_LIMIT_INTERVAL) self.update_interval = timedelta(seconds=new_interval) if err is not None: self._last_error = classify_error(err)