Fix Shelly Plus H&T sleep period on external power state change (#81121)

pull/81341/head
Shay Levy 2022-10-28 19:48:27 +03:00 committed by Paulus Schoutsen
parent aeecc93ad6
commit 1ef9e9e19a
2 changed files with 34 additions and 1 deletions

View File

@ -41,7 +41,12 @@ from .const import (
SLEEP_PERIOD_MULTIPLIER,
UPDATE_PERIOD_MULTIPLIER,
)
from .utils import device_update_info, get_block_device_name, get_rpc_device_name
from .utils import (
device_update_info,
get_block_device_name,
get_rpc_device_name,
get_rpc_device_wakeup_period,
)
@dataclass
@ -355,6 +360,24 @@ class ShellyRpcCoordinator(DataUpdateCoordinator):
LOGGER.debug("Reloading entry %s", self.name)
await self.hass.config_entries.async_reload(self.entry.entry_id)
def update_sleep_period(self) -> bool:
"""Check device sleep period & update if changed."""
if (
not self.device.initialized
or not (wakeup_period := get_rpc_device_wakeup_period(self.device.status))
or wakeup_period == self.entry.data.get(CONF_SLEEP_PERIOD)
):
return False
data = {**self.entry.data}
data[CONF_SLEEP_PERIOD] = wakeup_period
self.hass.config_entries.async_update_entry(self.entry, data=data)
update_interval = SLEEP_PERIOD_MULTIPLIER * wakeup_period
self.update_interval = timedelta(seconds=update_interval)
return True
@callback
def _async_device_updates_handler(self) -> None:
"""Handle device updates."""
@ -365,6 +388,8 @@ class ShellyRpcCoordinator(DataUpdateCoordinator):
):
return
self.update_sleep_period()
self._last_event = self.device.event
for event in self.device.event["events"]:
@ -393,6 +418,9 @@ class ShellyRpcCoordinator(DataUpdateCoordinator):
async def _async_update_data(self) -> None:
"""Fetch data."""
if self.update_sleep_period():
return
if sleep_period := self.entry.data.get(CONF_SLEEP_PERIOD):
# Sleeping device, no point polling it, just mark it unavailable
raise UpdateFailed(

View File

@ -272,6 +272,11 @@ def get_rpc_device_sleep_period(config: dict[str, Any]) -> int:
return cast(int, config["sys"].get("sleep", {}).get("wakeup_period", 0))
def get_rpc_device_wakeup_period(status: dict[str, Any]) -> int:
"""Return the device wakeup period in seconds or 0 for non sleeping devices."""
return cast(int, status["sys"].get("wakeup_period", 0))
def get_info_auth(info: dict[str, Any]) -> bool:
"""Return true if device has authorization enabled."""
return cast(bool, info.get("auth") or info.get("auth_en"))