common.py

Go to the documentation of this file.

"""The Synology DSM component."""
 
from __future__ import annotations
 
import asyncio
from collections.abc import Callable
from contextlib import suppress
import logging
 
from synology_dsm import SynologyDSM
from synology_dsm.api.core.security import SynoCoreSecurity
from synology_dsm.api.core.system import SynoCoreSystem
from synology_dsm.api.core.upgrade import SynoCoreUpgrade
from synology_dsm.api.core.utilization import SynoCoreUtilization
from synology_dsm.api.dsm.information import SynoDSMInformation
from synology_dsm.api.dsm.network import SynoDSMNetwork
from synology_dsm.api.photos import SynoPhotos
from synology_dsm.api.storage.storage import SynoStorage
from synology_dsm.api.surveillance_station import SynoSurveillanceStation
from synology_dsm.exceptions import (
    SynologyDSMAPIErrorException,
    SynologyDSMException,
    SynologyDSMRequestException,
)
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    CONF_HOST,
    CONF_PASSWORD,
    CONF_PORT,
    CONF_SSL,
    CONF_USERNAME,
    CONF_VERIFY_SSL,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
from .const import (
    CONF_DEVICE_TOKEN,
    DEFAULT_TIMEOUT,
    EXCEPTION_DETAILS,
    EXCEPTION_UNKNOWN,
    SYNOLOGY_CONNECTION_EXCEPTIONS,
)
 
LOGGER = logging.getLogger(__name__)
 
 
class SynoApi:
    """Class to interface with Synology DSM API."""
 
    dsm: SynologyDSM
 
    def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
        """Initialize the API wrapper class."""
        self._hass_hass = hass
        self._entry_entry = entry
        if entry.data.get(CONF_SSL):
            self.config_urlconfig_url = f"https://{entry.data[CONF_HOST]}:{entry.data[CONF_PORT]}"
        else:
            self.config_urlconfig_url = f"http://{entry.data[CONF_HOST]}:{entry.data[CONF_PORT]}"
 
        # DSM APIs
        self.informationinformation: SynoDSMInformation | None = None
        self.networknetwork: SynoDSMNetwork | None = None
        self.securitysecurity: SynoCoreSecurity | None = None
        self.storagestorage: SynoStorage | None = None
        self.photosphotos: SynoPhotos | None = None
        self.surveillance_stationsurveillance_station: SynoSurveillanceStation | None = None
        self.systemsystem: SynoCoreSystem | None = None
        self.upgradeupgrade: SynoCoreUpgrade | None = None
        self.utilisationutilisation: SynoCoreUtilization | None = None
 
        # Should we fetch them
        self._fetching_entities: dict[str, set[str]] = {}
        self._with_information_with_information = True
        self._with_security_with_security = True
        self._with_storage_with_storage = True
        self._with_photos_with_photos = True
        self._with_surveillance_station_with_surveillance_station = True
        self._with_system_with_system = True
        self._with_upgrade_with_upgrade = True
        self._with_utilisation_with_utilisation = True
 
        self._login_future_login_future: asyncio.Future[None] | None = None
 
    async def async_login(self) -> None:
        """Login to the Synology DSM API.
 
        This function will only login once if called multiple times
        by multiple different callers.
 
        If a login is already in progress, the function will await the
        login to complete before returning.
        """
        if self._login_future_login_future:
            return await self._login_future_login_future
 
        self._login_future_login_future = self._hass_hass.loop.create_future()
        try:
            await self.dsmdsm.login()
            self._login_future_login_future.set_result(None)
        except BaseException as err:
            if not self._login_future_login_future.done():
                self._login_future_login_future.set_exception(err)
            with suppress(BaseException):
                # Clear the flag as its normal that nothing
                # will wait for this future to be resolved
                # if there are no concurrent login attempts
                await self._login_future_login_future
            raise
        finally:
            self._login_future_login_future = None
 
    async def async_setup(self) -> None:
        """Start interacting with the NAS."""
        session = async_get_clientsession(self._hass_hass, self._entry_entry.data[CONF_VERIFY_SSL])
        self.dsmdsm = SynologyDSM(
            session,
            self._entry_entry.data[CONF_HOST],
            self._entry_entry.data[CONF_PORT],
            self._entry_entry.data[CONF_USERNAME],
            self._entry_entry.data[CONF_PASSWORD],
            self._entry_entry.data[CONF_SSL],
            timeout=DEFAULT_TIMEOUT,
            device_token=self._entry_entry.data.get(CONF_DEVICE_TOKEN),
        )
        await self.async_loginasync_login()
 
        # check if surveillance station is used
        self._with_surveillance_station_with_surveillance_station = bool(
            self.dsmdsm.apis.get(SynoSurveillanceStation.CAMERA_API_KEY)
        )
        if self._with_surveillance_station_with_surveillance_station:
            try:
                await self.dsmdsm.surveillance_station.update()
            except SYNOLOGY_CONNECTION_EXCEPTIONS:
                self._with_surveillance_station_with_surveillance_station = False
                self.dsmdsm.reset(SynoSurveillanceStation.API_KEY)
                LOGGER.warning(
                    "Surveillance Station found, but disabled due to missing user"
                    " permissions"
                )
 
        LOGGER.debug(
            "State of Surveillance_station during setup of '%s': %s",
            self._entry_entry.unique_id,
            self._with_surveillance_station_with_surveillance_station,
        )
 
        # check if upgrade is available
        try:
            await self.dsmdsm.upgrade.update()
        except SYNOLOGY_CONNECTION_EXCEPTIONS as ex:
            self._with_upgrade_with_upgrade = False
            self.dsmdsm.reset(SynoCoreUpgrade.API_KEY)
            LOGGER.debug("Disabled fetching upgrade data during setup: %s", ex)
 
        await self._fetch_device_configuration_fetch_device_configuration()
 
        try:
            await self._update_update()
        except SYNOLOGY_CONNECTION_EXCEPTIONS as err:
            LOGGER.debug(
                "Connection error during setup of '%s' with exception: %s",
                self._entry_entry.unique_id,
                err,
            )
            raise
 
    @callback
    def subscribe(self, api_key: str, unique_id: str) -> Callable[[], None]:
        """Subscribe an entity to API fetches."""
        LOGGER.debug("Subscribe new entity: %s", unique_id)
        if api_key not in self._fetching_entities:
            self._fetching_entities[api_key] = set()
        self._fetching_entities[api_key].add(unique_id)
 
        @callback
        def unsubscribe() -> None:
            """Unsubscribe an entity from API fetches (when disable)."""
            LOGGER.debug("Unsubscribe entity: %s", unique_id)
            self._fetching_entities[api_key].remove(unique_id)
            if len(self._fetching_entities[api_key]) == 0:
                self._fetching_entities.pop(api_key)
 
        return unsubscribe
 
    def _setup_api_requests(self) -> None:
        """Determine if we should fetch each API, if one entity needs it."""
        # Entities not added yet, fetch all
        if not self._fetching_entities:
            LOGGER.debug(
                "Entities not added yet, fetch all for '%s'", self._entry_entry.unique_id
            )
            return
 
        # surveillance_station is updated by own coordinator
        if self.surveillance_stationsurveillance_station:
            self.dsmdsm.reset(self.surveillance_stationsurveillance_station)
 
        # Determine if we should fetch an API
        self._with_system_with_system = bool(self.dsmdsm.apis.get(SynoCoreSystem.API_KEY))
        self._with_security_with_security = bool(
            self._fetching_entities.get(SynoCoreSecurity.API_KEY)
        )
        self._with_storage_with_storage = bool(self._fetching_entities.get(SynoStorage.API_KEY))
        self._with_photos_with_photos = bool(self._fetching_entities.get(SynoStorage.API_KEY))
        self._with_upgrade_with_upgrade = bool(self._fetching_entities.get(SynoCoreUpgrade.API_KEY))
        self._with_utilisation_with_utilisation = bool(
            self._fetching_entities.get(SynoCoreUtilization.API_KEY)
        )
        self._with_information_with_information = bool(
            self._fetching_entities.get(SynoDSMInformation.API_KEY)
        )
 
        # Reset not used API, information is not reset since it's used in device_info
        if not self._with_security_with_security:
            LOGGER.debug(
                "Disable security api from being updated for '%s'",
                self._entry_entry.unique_id,
            )
            if self.securitysecurity:
                self.dsmdsm.reset(self.securitysecurity)
            self.securitysecurity = None
 
        if not self._with_photos_with_photos:
            LOGGER.debug(
                "Disable photos api from being updated or '%s'", self._entry_entry.unique_id
            )
            if self.photosphotos:
                self.dsmdsm.reset(self.photosphotos)
            self.photosphotos = None
 
        if not self._with_storage_with_storage:
            LOGGER.debug(
                "Disable storage api from being updatedf or '%s'", self._entry_entry.unique_id
            )
            if self.storagestorage:
                self.dsmdsm.reset(self.storagestorage)
            self.storagestorage = None
 
        if not self._with_system_with_system:
            LOGGER.debug(
                "Disable system api from being updated for '%s'", self._entry_entry.unique_id
            )
            if self.systemsystem:
                self.dsmdsm.reset(self.systemsystem)
            self.systemsystem = None
 
        if not self._with_upgrade_with_upgrade:
            LOGGER.debug(
                "Disable upgrade api from being updated for '%s'", self._entry_entry.unique_id
            )
            if self.upgradeupgrade:
                self.dsmdsm.reset(self.upgradeupgrade)
            self.upgradeupgrade = None
 
        if not self._with_utilisation_with_utilisation:
            LOGGER.debug(
                "Disable utilisation api from being updated for '%s'",
                self._entry_entry.unique_id,
            )
            if self.utilisationutilisation:
                self.dsmdsm.reset(self.utilisationutilisation)
            self.utilisationutilisation = None
 
    async def _fetch_device_configuration(self) -> None:
        """Fetch initial device config."""
        self.informationinformation = self.dsmdsm.information
        self.networknetwork = self.dsmdsm.network
        await self.networknetwork.update()
 
        if self._with_security_with_security:
            LOGGER.debug("Enable security api updates for '%s'", self._entry_entry.unique_id)
            self.securitysecurity = self.dsmdsm.security
 
        if self._with_photos_with_photos:
            LOGGER.debug("Enable photos api updates for '%s'", self._entry_entry.unique_id)
            self.photosphotos = self.dsmdsm.photos
 
        if self._with_storage_with_storage:
            LOGGER.debug("Enable storage api updates for '%s'", self._entry_entry.unique_id)
            self.storagestorage = self.dsmdsm.storage
 
        if self._with_upgrade_with_upgrade:
            LOGGER.debug("Enable upgrade api updates for '%s'", self._entry_entry.unique_id)
            self.upgradeupgrade = self.dsmdsm.upgrade
 
        if self._with_system_with_system:
            LOGGER.debug("Enable system api updates for '%s'", self._entry_entry.unique_id)
            self.systemsystem = self.dsmdsm.system
 
        if self._with_utilisation_with_utilisation:
            LOGGER.debug(
                "Enable utilisation api updates for '%s'", self._entry_entry.unique_id
            )
            self.utilisationutilisation = self.dsmdsm.utilisation
 
        if self._with_surveillance_station_with_surveillance_station:
            LOGGER.debug(
                "Enable surveillance_station api updates for '%s'",
                self._entry_entry.unique_id,
            )
            self.surveillance_stationsurveillance_station = self.dsmdsm.surveillance_station
 
    async def _syno_api_executer(self, api_call: Callable) -> None:
        """Synology api call wrapper."""
        try:
            await api_call()
        except (SynologyDSMAPIErrorException, SynologyDSMRequestException) as err:
            LOGGER.debug(
                "Error from '%s': %s", self._entry_entry.unique_id, err, exc_info=True
            )
            raise
 
    async def async_reboot(self) -> None:
        """Reboot NAS."""
        if self.systemsystem:
            await self._syno_api_executer_syno_api_executer(self.systemsystem.reboot)
 
    async def async_shutdown(self) -> None:
        """Shutdown NAS."""
        if self.systemsystem:
            await self._syno_api_executer_syno_api_executer(self.systemsystem.shutdown)
 
    async def async_unload(self) -> None:
        """Stop interacting with the NAS and prepare for removal from hass."""
        # ignore API errors during logout
        with suppress(SynologyDSMException):
            await self._syno_api_executer_syno_api_executer(self.dsmdsm.logout)
 
    async def async_update(self) -> None:
        """Update function for updating API information."""
        await self._update_update()
 
    async def _update(self) -> None:
        """Update function for updating API information."""
        LOGGER.debug("Start data update for '%s'", self._entry_entry.unique_id)
        self._setup_api_requests_setup_api_requests()
        await self.dsmdsm.update(self._with_information_with_information)
 
 
def raise_config_entry_auth_error(err: Exception) -> None:
    """Raise ConfigEntryAuthFailed if error is related to authentication."""
    if err.args[0] and isinstance(err.args[0], dict):
        details = err.args[0].get(EXCEPTION_DETAILS, EXCEPTION_UNKNOWN)
    else:
        details = EXCEPTION_UNKNOWN
    raise ConfigEntryAuthFailed(f"reason: {details}") from err