mirror of
https://github.com/Electric-Special/ha-core.git
synced 2026-03-21 05:06:13 +01:00
Hide asserts behind TYPE_CHECKING in Synology DSM (#152880)
This commit is contained in:
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
from itertools import chain
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from synology_dsm.api.surveillance_station import SynoSurveillanceStation
|
||||
from synology_dsm.api.surveillance_station.camera import SynoCamera
|
||||
@@ -177,10 +178,12 @@ async def async_remove_config_entry_device(
|
||||
"""Remove synology_dsm config entry from a device."""
|
||||
data = entry.runtime_data
|
||||
api = data.api
|
||||
assert api.information is not None
|
||||
if TYPE_CHECKING:
|
||||
assert api.information is not None
|
||||
serial = api.information.serial
|
||||
storage = api.storage
|
||||
assert storage is not None
|
||||
if TYPE_CHECKING:
|
||||
assert storage is not None
|
||||
all_cameras: list[SynoCamera] = []
|
||||
if api.surveillance_station is not None:
|
||||
# get_all_cameras does not do I/O
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from synology_dsm.api.core.security import SynoCoreSecurity
|
||||
from synology_dsm.api.storage.storage import SynoStorage
|
||||
@@ -68,7 +69,8 @@ async def async_setup_entry(
|
||||
data = entry.runtime_data
|
||||
api = data.api
|
||||
coordinator = data.coordinator_central
|
||||
assert api.storage is not None
|
||||
if TYPE_CHECKING:
|
||||
assert api.storage is not None
|
||||
|
||||
entities: list[SynoDSMSecurityBinarySensor | SynoDSMStorageBinarySensor] = [
|
||||
SynoDSMSecurityBinarySensor(api, coordinator, description)
|
||||
@@ -121,7 +123,8 @@ class SynoDSMSecurityBinarySensor(SynoDSMBinarySensor):
|
||||
@property
|
||||
def extra_state_attributes(self) -> dict[str, str]:
|
||||
"""Return security checks details."""
|
||||
assert self._api.security is not None
|
||||
if TYPE_CHECKING:
|
||||
assert self._api.security is not None
|
||||
return self._api.security.status_by_check
|
||||
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@ from __future__ import annotations
|
||||
from collections.abc import Callable, Coroutine
|
||||
from dataclasses import dataclass
|
||||
import logging
|
||||
from typing import Any, Final
|
||||
from typing import TYPE_CHECKING, Any, Final
|
||||
|
||||
from homeassistant.components.button import (
|
||||
ButtonDeviceClass,
|
||||
@@ -72,8 +72,9 @@ class SynologyDSMButton(ButtonEntity):
|
||||
"""Initialize the Synology DSM binary_sensor entity."""
|
||||
self.entity_description = description
|
||||
self.syno_api = api
|
||||
assert api.network is not None
|
||||
assert api.information is not None
|
||||
if TYPE_CHECKING:
|
||||
assert api.network is not None
|
||||
assert api.information is not None
|
||||
self._attr_name = f"{api.network.hostname} {description.name}"
|
||||
self._attr_unique_id = f"{api.information.serial}_{description.key}"
|
||||
self._attr_device_info = DeviceInfo(
|
||||
@@ -82,7 +83,8 @@ class SynologyDSMButton(ButtonEntity):
|
||||
|
||||
async def async_press(self) -> None:
|
||||
"""Triggers the Synology DSM button press service."""
|
||||
assert self.syno_api.network is not None
|
||||
if TYPE_CHECKING:
|
||||
assert self.syno_api.network is not None
|
||||
LOGGER.debug(
|
||||
"Trigger %s for %s",
|
||||
self.entity_description.key,
|
||||
|
||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from synology_dsm.api.surveillance_station import SynoCamera, SynoSurveillanceStation
|
||||
from synology_dsm.exceptions import (
|
||||
@@ -94,7 +95,8 @@ class SynoDSMCamera(SynologyDSMBaseEntity[SynologyDSMCameraUpdateCoordinator], C
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Return the device information."""
|
||||
information = self._api.information
|
||||
assert information is not None
|
||||
if TYPE_CHECKING:
|
||||
assert information is not None
|
||||
return DeviceInfo(
|
||||
identifiers={(DOMAIN, f"{information.serial}_{self.camera_data.id}")},
|
||||
name=self.camera_data.name,
|
||||
@@ -129,7 +131,8 @@ class SynoDSMCamera(SynologyDSMBaseEntity[SynologyDSMCameraUpdateCoordinator], C
|
||||
_LOGGER.debug("Update stream URL for camera %s", self.camera_data.name)
|
||||
self.stream.update_source(url)
|
||||
|
||||
assert self.platform.config_entry
|
||||
if TYPE_CHECKING:
|
||||
assert self.platform.config_entry
|
||||
self.async_on_remove(
|
||||
async_dispatcher_connect(
|
||||
self.hass,
|
||||
@@ -153,7 +156,8 @@ class SynoDSMCamera(SynologyDSMBaseEntity[SynologyDSMCameraUpdateCoordinator], C
|
||||
)
|
||||
if not self.available:
|
||||
return None
|
||||
assert self._api.surveillance_station is not None
|
||||
if TYPE_CHECKING:
|
||||
assert self._api.surveillance_station is not None
|
||||
try:
|
||||
return await self._api.surveillance_station.get_camera_image(
|
||||
self.entity_description.camera_id, self.snapshot_quality
|
||||
@@ -187,7 +191,8 @@ class SynoDSMCamera(SynologyDSMBaseEntity[SynologyDSMCameraUpdateCoordinator], C
|
||||
"SynoDSMCamera.enable_motion_detection(%s)",
|
||||
self.camera_data.name,
|
||||
)
|
||||
assert self._api.surveillance_station is not None
|
||||
if TYPE_CHECKING:
|
||||
assert self._api.surveillance_station is not None
|
||||
await self._api.surveillance_station.enable_motion_detection(
|
||||
self.entity_description.camera_id
|
||||
)
|
||||
@@ -198,7 +203,8 @@ class SynoDSMCamera(SynologyDSMBaseEntity[SynologyDSMCameraUpdateCoordinator], C
|
||||
"SynoDSMCamera.disable_motion_detection(%s)",
|
||||
self.camera_data.name,
|
||||
)
|
||||
assert self._api.surveillance_station is not None
|
||||
if TYPE_CHECKING:
|
||||
assert self._api.surveillance_station is not None
|
||||
await self._api.surveillance_station.disable_motion_detection(
|
||||
self.entity_description.camera_id
|
||||
)
|
||||
|
||||
@@ -6,7 +6,7 @@ from collections.abc import Awaitable, Callable, Coroutine
|
||||
from dataclasses import dataclass
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Any, Concatenate
|
||||
from typing import TYPE_CHECKING, Any, Concatenate
|
||||
|
||||
from synology_dsm.api.surveillance_station.camera import SynoCamera
|
||||
from synology_dsm.exceptions import (
|
||||
@@ -110,14 +110,16 @@ class SynologyDSMSwitchUpdateCoordinator(
|
||||
async def async_setup(self) -> None:
|
||||
"""Set up the coordinator initial data."""
|
||||
info = await self.api.dsm.surveillance_station.get_info()
|
||||
assert info is not None
|
||||
if TYPE_CHECKING:
|
||||
assert info is not None
|
||||
self.version = info["data"]["CMSMinVersion"]
|
||||
|
||||
@async_re_login_on_expired
|
||||
async def _async_update_data(self) -> dict[str, dict[str, Any]]:
|
||||
"""Fetch all data from api."""
|
||||
surveillance_station = self.api.surveillance_station
|
||||
assert surveillance_station is not None
|
||||
if TYPE_CHECKING:
|
||||
assert surveillance_station is not None
|
||||
return {
|
||||
"switches": {
|
||||
"home_mode": bool(await surveillance_station.get_home_mode_status())
|
||||
@@ -161,7 +163,8 @@ class SynologyDSMCameraUpdateCoordinator(
|
||||
async def _async_update_data(self) -> dict[str, dict[int, SynoCamera]]:
|
||||
"""Fetch all camera data from api."""
|
||||
surveillance_station = self.api.surveillance_station
|
||||
assert surveillance_station is not None
|
||||
if TYPE_CHECKING:
|
||||
assert surveillance_station is not None
|
||||
current_data: dict[int, SynoCamera] = {
|
||||
camera.id: camera for camera in surveillance_station.get_all_cameras()
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from homeassistant.helpers.device_registry import DeviceInfo
|
||||
from homeassistant.helpers.entity import EntityDescription
|
||||
@@ -47,8 +47,9 @@ class SynologyDSMBaseEntity[_CoordinatorT: SynologyDSMUpdateCoordinator[Any]](
|
||||
self._api = api
|
||||
information = api.information
|
||||
network = api.network
|
||||
assert information is not None
|
||||
assert network is not None
|
||||
if TYPE_CHECKING:
|
||||
assert information is not None
|
||||
assert network is not None
|
||||
|
||||
self._attr_unique_id: str = (
|
||||
f"{information.serial}_{description.api_key}:{description.key}"
|
||||
@@ -94,14 +95,17 @@ class SynologyDSMDeviceEntity(
|
||||
information = api.information
|
||||
network = api.network
|
||||
external_usb = api.external_usb
|
||||
assert information is not None
|
||||
assert storage is not None
|
||||
assert network is not None
|
||||
if TYPE_CHECKING:
|
||||
assert information is not None
|
||||
assert storage is not None
|
||||
assert network is not None
|
||||
|
||||
if "volume" in description.key:
|
||||
assert self._device_id is not None
|
||||
if TYPE_CHECKING:
|
||||
assert self._device_id is not None
|
||||
volume = storage.get_volume(self._device_id)
|
||||
assert volume is not None
|
||||
if TYPE_CHECKING:
|
||||
assert volume is not None
|
||||
# Volume does not have a name
|
||||
self._device_name = volume["id"].replace("_", " ").capitalize()
|
||||
self._device_manufacturer = "Synology"
|
||||
@@ -114,17 +118,20 @@ class SynologyDSMDeviceEntity(
|
||||
.replace("shr", "SHR")
|
||||
)
|
||||
elif "disk" in description.key:
|
||||
assert self._device_id is not None
|
||||
if TYPE_CHECKING:
|
||||
assert self._device_id is not None
|
||||
disk = storage.get_disk(self._device_id)
|
||||
assert disk is not None
|
||||
if TYPE_CHECKING:
|
||||
assert disk is not None
|
||||
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"]
|
||||
elif "device" in description.key:
|
||||
assert self._device_id is not None
|
||||
assert external_usb is not None
|
||||
if TYPE_CHECKING:
|
||||
assert self._device_id is not None
|
||||
assert external_usb is not None
|
||||
for device in external_usb.get_devices.values():
|
||||
if device.device_name == self._device_id:
|
||||
self._device_name = device.device_name
|
||||
@@ -133,8 +140,9 @@ class SynologyDSMDeviceEntity(
|
||||
self._device_type = device.device_type
|
||||
break
|
||||
elif "partition" in description.key:
|
||||
assert self._device_id is not None
|
||||
assert external_usb is not None
|
||||
if TYPE_CHECKING:
|
||||
assert self._device_id is not None
|
||||
assert external_usb is not None
|
||||
for device in external_usb.get_devices.values():
|
||||
for partition in device.device_partitions.values():
|
||||
if partition.partition_title == self._device_id:
|
||||
|
||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
from logging import getLogger
|
||||
import mimetypes
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from aiohttp import web
|
||||
from synology_dsm.api.photos import SynoPhotosAlbum, SynoPhotosItem
|
||||
@@ -121,9 +122,11 @@ class SynologyPhotosMediaSource(MediaSource):
|
||||
DOMAIN, identifier.unique_id
|
||||
)
|
||||
)
|
||||
assert entry
|
||||
if TYPE_CHECKING:
|
||||
assert entry
|
||||
diskstation = entry.runtime_data
|
||||
assert diskstation.api.photos is not None
|
||||
if TYPE_CHECKING:
|
||||
assert diskstation.api.photos is not None
|
||||
|
||||
if identifier.album_id is None:
|
||||
# Get Albums
|
||||
@@ -131,7 +134,8 @@ class SynologyPhotosMediaSource(MediaSource):
|
||||
albums = await diskstation.api.photos.get_albums()
|
||||
except SynologyDSMException:
|
||||
return []
|
||||
assert albums is not None
|
||||
if TYPE_CHECKING:
|
||||
assert albums is not None
|
||||
|
||||
ret = [
|
||||
BrowseMediaSource(
|
||||
@@ -190,7 +194,8 @@ class SynologyPhotosMediaSource(MediaSource):
|
||||
)
|
||||
except SynologyDSMException:
|
||||
return []
|
||||
assert album_items is not None
|
||||
if TYPE_CHECKING:
|
||||
assert album_items is not None
|
||||
|
||||
ret = []
|
||||
for album_item in album_items:
|
||||
@@ -249,7 +254,8 @@ class SynologyPhotosMediaSource(MediaSource):
|
||||
self, item: SynoPhotosItem, diskstation: SynologyDSMData
|
||||
) -> str | None:
|
||||
"""Get thumbnail."""
|
||||
assert diskstation.api.photos is not None
|
||||
if TYPE_CHECKING:
|
||||
assert diskstation.api.photos is not None
|
||||
|
||||
try:
|
||||
thumbnail = await diskstation.api.photos.get_item_thumbnail_url(item)
|
||||
@@ -290,9 +296,11 @@ class SynologyDsmMediaView(http.HomeAssistantView):
|
||||
DOMAIN, source_dir_id
|
||||
)
|
||||
)
|
||||
assert entry
|
||||
if TYPE_CHECKING:
|
||||
assert entry
|
||||
diskstation = entry.runtime_data
|
||||
assert diskstation.api.photos is not None
|
||||
if TYPE_CHECKING:
|
||||
assert diskstation.api.photos is not None
|
||||
item = SynoPhotosItem(image_id, "", "", "", cache_key, "xl", shared, passphrase)
|
||||
try:
|
||||
if passphrase:
|
||||
|
||||
@@ -4,7 +4,7 @@ from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta
|
||||
from typing import cast
|
||||
from typing import TYPE_CHECKING, cast
|
||||
|
||||
from synology_dsm.api.core.external_usb import (
|
||||
SynoCoreExternalUSB,
|
||||
@@ -345,7 +345,8 @@ async def async_setup_entry(
|
||||
api = data.api
|
||||
coordinator = data.coordinator_central
|
||||
storage = api.storage
|
||||
assert storage is not None
|
||||
if TYPE_CHECKING:
|
||||
assert storage is not None
|
||||
known_usb_devices: set[str] = set()
|
||||
|
||||
def _check_usb_devices() -> None:
|
||||
@@ -504,7 +505,8 @@ class SynoDSMExternalUSBSensor(SynologyDSMDeviceEntity, SynoDSMSensor):
|
||||
def native_value(self) -> StateType:
|
||||
"""Return the state."""
|
||||
external_usb = self._api.external_usb
|
||||
assert external_usb is not None
|
||||
if TYPE_CHECKING:
|
||||
assert external_usb is not None
|
||||
if "device" in self.entity_description.key:
|
||||
for device in external_usb.get_devices.values():
|
||||
if device.device_name == self._device_id:
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import cast
|
||||
from typing import TYPE_CHECKING, cast
|
||||
|
||||
from synology_dsm.exceptions import SynologyDSMException
|
||||
|
||||
@@ -27,7 +27,8 @@ async def _service_handler(call: ServiceCall) -> None:
|
||||
entry: SynologyDSMConfigEntry | None = (
|
||||
call.hass.config_entries.async_entry_for_domain_unique_id(DOMAIN, serial)
|
||||
)
|
||||
assert entry
|
||||
if TYPE_CHECKING:
|
||||
assert entry
|
||||
dsm_device = entry.runtime_data
|
||||
elif len(dsm_devices) == 1:
|
||||
dsm_device = next(iter(dsm_devices.values()))
|
||||
|
||||
@@ -4,7 +4,7 @@ from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import logging
|
||||
from typing import Any
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from synology_dsm.api.surveillance_station import SynoSurveillanceStation
|
||||
|
||||
@@ -45,7 +45,8 @@ async def async_setup_entry(
|
||||
"""Set up the Synology NAS switch."""
|
||||
data = entry.runtime_data
|
||||
if coordinator := data.coordinator_switches:
|
||||
assert coordinator.version is not None
|
||||
if TYPE_CHECKING:
|
||||
assert coordinator.version is not None
|
||||
async_add_entities(
|
||||
SynoDSMSurveillanceHomeModeToggle(
|
||||
data.api, coordinator.version, coordinator, description
|
||||
@@ -79,8 +80,9 @@ class SynoDSMSurveillanceHomeModeToggle(
|
||||
|
||||
async def async_turn_on(self, **kwargs: Any) -> None:
|
||||
"""Turn on Home mode."""
|
||||
assert self._api.surveillance_station is not None
|
||||
assert self._api.information
|
||||
if TYPE_CHECKING:
|
||||
assert self._api.surveillance_station is not None
|
||||
assert self._api.information
|
||||
_LOGGER.debug(
|
||||
"SynoDSMSurveillanceHomeModeToggle.turn_on(%s)",
|
||||
self._api.information.serial,
|
||||
@@ -90,8 +92,9 @@ class SynoDSMSurveillanceHomeModeToggle(
|
||||
|
||||
async def async_turn_off(self, **kwargs: Any) -> None:
|
||||
"""Turn off Home mode."""
|
||||
assert self._api.surveillance_station is not None
|
||||
assert self._api.information
|
||||
if TYPE_CHECKING:
|
||||
assert self._api.surveillance_station is not None
|
||||
assert self._api.information
|
||||
_LOGGER.debug(
|
||||
"SynoDSMSurveillanceHomeModeToggle.turn_off(%s)",
|
||||
self._api.information.serial,
|
||||
@@ -107,9 +110,10 @@ class SynoDSMSurveillanceHomeModeToggle(
|
||||
@property
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Return the device information."""
|
||||
assert self._api.surveillance_station is not None
|
||||
assert self._api.information is not None
|
||||
assert self._api.network is not None
|
||||
if TYPE_CHECKING:
|
||||
assert self._api.surveillance_station is not None
|
||||
assert self._api.information is not None
|
||||
assert self._api.network is not None
|
||||
return DeviceInfo(
|
||||
identifiers={
|
||||
(
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Final
|
||||
from typing import TYPE_CHECKING, Final
|
||||
|
||||
from synology_dsm.api.core.upgrade import SynoCoreUpgrade
|
||||
from yarl import URL
|
||||
@@ -63,13 +63,15 @@ class SynoDSMUpdateEntity(
|
||||
@property
|
||||
def installed_version(self) -> str | None:
|
||||
"""Version installed and in use."""
|
||||
assert self._api.information is not None
|
||||
if TYPE_CHECKING:
|
||||
assert self._api.information is not None
|
||||
return self._api.information.version_string
|
||||
|
||||
@property
|
||||
def latest_version(self) -> str | None:
|
||||
"""Latest version available for install."""
|
||||
assert self._api.upgrade is not None
|
||||
if TYPE_CHECKING:
|
||||
assert self._api.upgrade is not None
|
||||
if not self._api.upgrade.update_available:
|
||||
return self.installed_version
|
||||
return self._api.upgrade.available_version
|
||||
@@ -77,8 +79,9 @@ class SynoDSMUpdateEntity(
|
||||
@property
|
||||
def release_url(self) -> str | None:
|
||||
"""URL to the full release notes of the latest version available."""
|
||||
assert self._api.information is not None
|
||||
assert self._api.upgrade is not None
|
||||
if TYPE_CHECKING:
|
||||
assert self._api.information is not None
|
||||
assert self._api.upgrade is not None
|
||||
|
||||
if (details := self._api.upgrade.available_version_details) is None:
|
||||
return None
|
||||
|
||||
Reference in New Issue
Block a user