Extract entity base classes into a new module for Synology DSM (#69796)

Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Michael 2022-04-24 10:26:34 +02:00 committed by GitHub
parent 009111151d
commit 781f18e0c3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 145 additions and 141 deletions

View File

@ -1177,6 +1177,7 @@ omit =
homeassistant/components/synology_dsm/camera.py
homeassistant/components/synology_dsm/diagnostics.py
homeassistant/components/synology_dsm/common.py
homeassistant/components/synology_dsm/entity.py
homeassistant/components/synology_dsm/sensor.py
homeassistant/components/synology_dsm/service.py
homeassistant/components/synology_dsm/switch.py

View File

@ -28,12 +28,7 @@ from homeassistant.helpers.device_registry import (
DeviceEntry,
async_get_registry as get_dev_reg,
)
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
UpdateFailed,
)
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .common import SynoApi
from .const import (
@ -49,16 +44,12 @@ from .const import (
SYNO_API,
SYSTEM_LOADED,
UNDO_UPDATE_LISTENER,
SynologyDSMEntityDescription,
)
from .service import async_setup_services
CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False)
ATTRIBUTION = "Data provided by Synology"
_LOGGER = logging.getLogger(__name__)
@ -223,104 +214,3 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
await hass.config_entries.async_reload(entry.entry_id)
class SynologyDSMBaseEntity(
CoordinatorEntity[DataUpdateCoordinator[dict[str, dict[str, Any]]]]
):
"""Representation of a Synology NAS entry."""
entity_description: SynologyDSMEntityDescription
unique_id: str
_attr_attribution = ATTRIBUTION
def __init__(
self,
api: SynoApi,
coordinator: DataUpdateCoordinator[dict[str, dict[str, Any]]],
description: SynologyDSMEntityDescription,
) -> None:
"""Initialize the Synology DSM entity."""
super().__init__(coordinator)
self.entity_description = description
self._api = api
self._attr_name = f"{api.network.hostname} {description.name}"
self._attr_unique_id: str = (
f"{api.information.serial}_{description.api_key}:{description.key}"
)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self._api.information.serial)},
name=self._api.network.hostname,
manufacturer="Synology",
model=self._api.information.model,
sw_version=self._api.information.version_string,
configuration_url=self._api.config_url,
)
async def async_added_to_hass(self) -> None:
"""Register entity for updates from API."""
self.async_on_remove(
self._api.subscribe(self.entity_description.api_key, self.unique_id)
)
await super().async_added_to_hass()
class SynologyDSMDeviceEntity(SynologyDSMBaseEntity):
"""Representation of a Synology NAS disk or volume entry."""
def __init__(
self,
api: SynoApi,
coordinator: DataUpdateCoordinator[dict[str, dict[str, Any]]],
description: SynologyDSMEntityDescription,
device_id: str | None = None,
) -> None:
"""Initialize the Synology DSM disk or volume entity."""
super().__init__(api, coordinator, description)
self._device_id = device_id
self._device_name: str | None = None
self._device_manufacturer: str | None = None
self._device_model: str | None = None
self._device_firmware: str | None = None
self._device_type = None
if "volume" in description.key:
volume = self._api.storage.get_volume(self._device_id)
# Volume does not have a name
self._device_name = volume["id"].replace("_", " ").capitalize()
self._device_manufacturer = "Synology"
self._device_model = self._api.information.model
self._device_firmware = self._api.information.version_string
self._device_type = (
volume["device_type"]
.replace("_", " ")
.replace("raid", "RAID")
.replace("shr", "SHR")
)
elif "disk" in description.key:
disk = self._api.storage.get_disk(self._device_id)
self._device_name = disk["name"]
self._device_manufacturer = disk["vendor"]
self._device_model = disk["model"].strip()
self._device_firmware = disk["firm"]
self._device_type = disk["diskType"]
self._attr_name = (
f"{self._api.network.hostname} ({self._device_name}) {description.name}"
)
self._attr_unique_id += f"_{self._device_id}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, f"{self._api.information.serial}_{self._device_id}")},
name=f"{self._api.network.hostname} ({self._device_name})",
manufacturer=self._device_manufacturer,
model=self._device_model,
sw_version=self._device_firmware,
via_device=(DOMAIN, self._api.information.serial),
configuration_url=self._api.config_url,
)
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self._api.storage # type: ignore [no-any-return]

View File

@ -21,8 +21,13 @@ from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import SynoApi, SynologyDSMBaseEntity, SynologyDSMDeviceEntity
from .const import COORDINATOR_CENTRAL, DOMAIN, SYNO_API, SynologyDSMEntityDescription
from . import SynoApi
from .const import COORDINATOR_CENTRAL, DOMAIN, SYNO_API
from .entity import (
SynologyDSMBaseEntity,
SynologyDSMDeviceEntity,
SynologyDSMEntityDescription,
)
@dataclass

View File

@ -21,15 +21,15 @@ from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import SynoApi, SynologyDSMBaseEntity
from . import SynoApi
from .const import (
CONF_SNAPSHOT_QUALITY,
COORDINATOR_CAMERAS,
DEFAULT_SNAPSHOT_QUALITY,
DOMAIN,
SYNO_API,
SynologyDSMEntityDescription,
)
from .entity import SynologyDSMBaseEntity, SynologyDSMEntityDescription
_LOGGER = logging.getLogger(__name__)

View File

@ -1,14 +1,12 @@
"""Constants for Synology DSM."""
from __future__ import annotations
from dataclasses import dataclass
from synology_dsm.api.surveillance_station.const import SNAPSHOT_PROFILE_BALANCED
from homeassistant.const import Platform
from homeassistant.helpers.entity import EntityDescription
DOMAIN = "synology_dsm"
ATTRIBUTION = "Data provided by Synology"
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.BUTTON,
@ -52,15 +50,3 @@ SERVICES = [
SERVICE_REBOOT,
SERVICE_SHUTDOWN,
]
@dataclass
class SynologyDSMRequiredKeysMixin:
"""Mixin for required keys."""
api_key: str
@dataclass
class SynologyDSMEntityDescription(EntityDescription, SynologyDSMRequiredKeysMixin):
"""Generic Synology DSM entity description."""

View File

@ -0,0 +1,122 @@
"""Entities for Synology DSM."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from homeassistant.helpers.entity import DeviceInfo, EntityDescription
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .common import SynoApi
from .const import ATTRIBUTION, DOMAIN
@dataclass
class SynologyDSMRequiredKeysMixin:
"""Mixin for required keys."""
api_key: str
@dataclass
class SynologyDSMEntityDescription(EntityDescription, SynologyDSMRequiredKeysMixin):
"""Generic Synology DSM entity description."""
class SynologyDSMBaseEntity(
CoordinatorEntity[DataUpdateCoordinator[dict[str, dict[str, Any]]]]
):
"""Representation of a Synology NAS entry."""
entity_description: SynologyDSMEntityDescription
unique_id: str
_attr_attribution = ATTRIBUTION
def __init__(
self,
api: SynoApi,
coordinator: DataUpdateCoordinator[dict[str, dict[str, Any]]],
description: SynologyDSMEntityDescription,
) -> None:
"""Initialize the Synology DSM entity."""
super().__init__(coordinator)
self.entity_description = description
self._api = api
self._attr_name = f"{api.network.hostname} {description.name}"
self._attr_unique_id: str = (
f"{api.information.serial}_{description.api_key}:{description.key}"
)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self._api.information.serial)},
name=self._api.network.hostname,
manufacturer="Synology",
model=self._api.information.model,
sw_version=self._api.information.version_string,
configuration_url=self._api.config_url,
)
async def async_added_to_hass(self) -> None:
"""Register entity for updates from API."""
self.async_on_remove(
self._api.subscribe(self.entity_description.api_key, self.unique_id)
)
await super().async_added_to_hass()
class SynologyDSMDeviceEntity(SynologyDSMBaseEntity):
"""Representation of a Synology NAS disk or volume entry."""
def __init__(
self,
api: SynoApi,
coordinator: DataUpdateCoordinator[dict[str, dict[str, Any]]],
description: SynologyDSMEntityDescription,
device_id: str | None = None,
) -> None:
"""Initialize the Synology DSM disk or volume entity."""
super().__init__(api, coordinator, description)
self._device_id = device_id
self._device_name: str | None = None
self._device_manufacturer: str | None = None
self._device_model: str | None = None
self._device_firmware: str | None = None
self._device_type = None
if "volume" in description.key:
volume = self._api.storage.get_volume(self._device_id)
# Volume does not have a name
self._device_name = volume["id"].replace("_", " ").capitalize()
self._device_manufacturer = "Synology"
self._device_model = self._api.information.model
self._device_firmware = self._api.information.version_string
self._device_type = (
volume["device_type"]
.replace("_", " ")
.replace("raid", "RAID")
.replace("shr", "SHR")
)
elif "disk" in description.key:
disk = self._api.storage.get_disk(self._device_id)
self._device_name = disk["name"]
self._device_manufacturer = disk["vendor"]
self._device_model = disk["model"].strip()
self._device_firmware = disk["firm"]
self._device_type = disk["diskType"]
self._attr_name = (
f"{self._api.network.hostname} ({self._device_name}) {description.name}"
)
self._attr_unique_id += f"_{self._device_id}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, f"{self._api.information.serial}_{self._device_id}")},
name=f"{self._api.network.hostname} ({self._device_name})",
manufacturer=self._device_manufacturer,
model=self._device_model,
sw_version=self._device_firmware,
via_device=(DOMAIN, self._api.information.serial),
configuration_url=self._api.config_url,
)

View File

@ -30,13 +30,11 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from homeassistant.util.dt import utcnow
from . import SynoApi, SynologyDSMBaseEntity, SynologyDSMDeviceEntity
from .const import (
CONF_VOLUMES,
COORDINATOR_CENTRAL,
DOMAIN,
ENTITY_UNIT_LOAD,
SYNO_API,
from . import SynoApi
from .const import CONF_VOLUMES, COORDINATOR_CENTRAL, DOMAIN, ENTITY_UNIT_LOAD, SYNO_API
from .entity import (
SynologyDSMBaseEntity,
SynologyDSMDeviceEntity,
SynologyDSMEntityDescription,
)

View File

@ -14,8 +14,9 @@ from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import SynoApi, SynologyDSMBaseEntity
from .const import COORDINATOR_SWITCHES, DOMAIN, SYNO_API, SynologyDSMEntityDescription
from . import SynoApi
from .const import COORDINATOR_SWITCHES, DOMAIN, SYNO_API
from .entity import SynologyDSMBaseEntity, SynologyDSMEntityDescription
_LOGGER = logging.getLogger(__name__)

View File

@ -13,8 +13,9 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import SynoApi, SynologyDSMBaseEntity
from .const import COORDINATOR_CENTRAL, DOMAIN, SYNO_API, SynologyDSMEntityDescription
from . import SynoApi
from .const import COORDINATOR_CENTRAL, DOMAIN, SYNO_API
from .entity import SynologyDSMBaseEntity, SynologyDSMEntityDescription
@dataclass