forked from megakid/ha_octopus_intelligent
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
3 changed files
with
207 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
104 changes: 104 additions & 0 deletions
104
custom_components/octopus_intelligent/persistent_data.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,104 @@ | ||
| """Persistent data storage for the integration, based on the HASS helpers.storage.Store class.""" | ||
| import logging | ||
| from dataclasses import asdict, dataclass, InitVar | ||
| from typing import Any | ||
|
|
||
| from homeassistant.const import EVENT_HOMEASSISTANT_STOP | ||
| from homeassistant.core import CALLBACK_TYPE, Event, HomeAssistant | ||
| from homeassistant.exceptions import IntegrationError | ||
| from homeassistant.helpers.storage import Store | ||
|
|
||
| from .const import DOMAIN | ||
|
|
||
| _LOGGER = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| @dataclass | ||
| class PersistentData: | ||
| """JSON-serialisable data persistence backed by the HASS helpers.storage.Store class. | ||
| Frequently persisting data to "disk" can have undesired side effects when HASS is | ||
| running on some edge devices like the Raspberry Pi, whose storage consists of an | ||
| SD card that wears when data is written, eventually leading to hardware failure. | ||
| For this reason, by default, the data is only saved when the HASS STOP event is | ||
| fired, indicating that Home Assistant is quitting / restarting. This includes the | ||
| frontend web UI 'Restart' command, the "docker container stop" command, CTRL-C on | ||
| the command line, and generally when the HASS process receives the SIGTERM signal. | ||
| """ | ||
| # Note: InitVar fields are not persisted. They become arguments to __post_init__(). | ||
| hass: InitVar[HomeAssistant] | ||
| account_id: InitVar[str] | ||
|
|
||
| # --------------------------------------------- | ||
| # Start of JSON-serialisable persistent fields. | ||
| # | ||
| last_seen_planned_dispatch_source: str = "smart-charge" | ||
| # | ||
| # End of JSON-serialisable persistent fields. | ||
| # --------------------------------------------- | ||
|
|
||
| def __post_init__(self, hass: HomeAssistant, account_id: str): | ||
| self._hass = hass | ||
| self._store = Store[dict[str, Any]]( | ||
| hass=hass, | ||
| key=f"{DOMAIN}.{account_id}", | ||
| version=1, | ||
| minor_version=1, | ||
| ) | ||
| self._stop_event_listener: CALLBACK_TYPE | None = None | ||
| self.auto_save = True | ||
|
|
||
| @property | ||
| def auto_save(self) -> bool: | ||
| """Return whether auto saving is enabled.""" | ||
| return bool(self._stop_event_listener) | ||
|
|
||
| @auto_save.setter | ||
| def auto_save(self, enable: bool): | ||
| """Enable/disable automatically calling self.save() on the HASS STOP event.""" | ||
|
|
||
| async def _on_hass_stop(_: Event): | ||
| await self.save(raise_on_error=False) | ||
|
|
||
| if enable: | ||
| self._stop_event_listener = self._hass.bus.async_listen( | ||
| EVENT_HOMEASSISTANT_STOP, _on_hass_stop | ||
| ) | ||
| elif self._stop_event_listener: | ||
| self._stop_event_listener() | ||
| self._stop_event_listener = None | ||
|
|
||
| async def load(self): | ||
| """Load the data from persistent storage.""" | ||
| try: | ||
| data: dict[str, Any] = await self._store.async_load() or {} | ||
| except Exception as ex: # pylint: disable=broad-exception-caught | ||
| data = {} | ||
| _LOGGER.error( | ||
| "Using default values for persistent data because of an error: %s", ex | ||
| ) | ||
| # Explicitly save each field separately instead of using some '**data' | ||
| # unpacking syntax in order to be future-proof against schema changes | ||
| # that may add, remove or rename data fields. | ||
| self.last_seen_planned_dispatch_source = data.get( | ||
| "last_seen_planned_dispatch_source", self.last_seen_planned_dispatch_source | ||
| ) | ||
|
|
||
| async def save(self, raise_on_error=False): | ||
| """Save the data to persistent storage.""" | ||
| try: | ||
| await self._store.async_save(asdict(self)) | ||
| except Exception as ex: # pylint: disable=broad-exception-caught | ||
| msg = f"Error saving persistent data: {ex}" | ||
| if raise_on_error: | ||
| raise IntegrationError(msg) from ex | ||
| _LOGGER.error(msg) | ||
|
|
||
| async def remove(self, disable_auto_save=True): | ||
| """Remove the data from persistent storage (delete the JSON file on disk).""" | ||
| if disable_auto_save: | ||
| self.auto_save = False | ||
| try: | ||
| await self._store.async_remove() | ||
| except Exception as ex: # pylint: disable=broad-exception-caught | ||
| _LOGGER.error("Error removing persistent data: %s", ex) |