Improve reolink generic typing (#88786)

Co-authored-by: starkillerOG <starkiller.og@gmail.com>
pull/89263/head
Marc Mueller 2023-03-06 21:54:34 +01:00 committed by GitHub
parent 83fa4c6c60
commit 84034959ba
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 49 additions and 40 deletions

View File

@ -6,6 +6,7 @@ import asyncio
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import Literal
from aiohttp import ClientConnectorError
import async_timeout
@ -43,8 +44,8 @@ class ReolinkData:
"""Data for the Reolink integration."""
host: ReolinkHost
device_coordinator: DataUpdateCoordinator
firmware_coordinator: DataUpdateCoordinator
device_coordinator: DataUpdateCoordinator[None]
firmware_coordinator: DataUpdateCoordinator[str | Literal[False]]
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
@ -74,7 +75,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, host.stop)
)
async def async_device_config_update():
async def async_device_config_update() -> None:
"""Update the host state cache and renew the ONVIF-subscription."""
async with async_timeout.timeout(host.api.timeout):
try:
@ -87,7 +88,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
async with async_timeout.timeout(host.api.timeout):
await host.renew()
async def async_check_firmware_update():
async def async_check_firmware_update() -> str | Literal[False]:
"""Check for firmware updates."""
if not host.api.supported(None, "update"):
return False

View File

@ -24,7 +24,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ReolinkData
from .const import DOMAIN
from .entity import ReolinkCoordinatorEntity
from .entity import ReolinkChannelCoordinatorEntity
@dataclass
@ -113,7 +113,7 @@ async def async_setup_entry(
async_add_entities(entities)
class ReolinkBinarySensorEntity(ReolinkCoordinatorEntity, BinarySensorEntity):
class ReolinkBinarySensorEntity(ReolinkChannelCoordinatorEntity, BinarySensorEntity):
"""Base binary-sensor class for Reolink IP camera motion sensors."""
entity_description: ReolinkBinarySensorEntityDescription

View File

@ -15,7 +15,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ReolinkData
from .const import DOMAIN
from .entity import ReolinkCoordinatorEntity
from .entity import ReolinkChannelCoordinatorEntity
@dataclass
@ -112,7 +112,7 @@ async def async_setup_entry(
)
class ReolinkButtonEntity(ReolinkCoordinatorEntity, ButtonEntity):
class ReolinkButtonEntity(ReolinkChannelCoordinatorEntity, ButtonEntity):
"""Base button entity class for Reolink IP cameras."""
entity_description: ReolinkButtonEntityDescription

View File

@ -10,7 +10,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ReolinkData
from .const import DOMAIN
from .entity import ReolinkCoordinatorEntity
from .entity import ReolinkChannelCoordinatorEntity
_LOGGER = logging.getLogger(__name__)
@ -39,7 +39,7 @@ async def async_setup_entry(
async_add_entities(cameras)
class ReolinkCamera(ReolinkCoordinatorEntity, Camera):
class ReolinkCamera(ReolinkChannelCoordinatorEntity, Camera):
"""An implementation of a Reolink IP camera."""
_attr_supported_features: CameraEntityFeature = CameraEntityFeature.STREAM
@ -51,7 +51,7 @@ class ReolinkCamera(ReolinkCoordinatorEntity, Camera):
stream: str,
) -> None:
"""Initialize Reolink camera stream."""
ReolinkCoordinatorEntity.__init__(self, reolink_data, channel)
ReolinkChannelCoordinatorEntity.__init__(self, reolink_data, channel)
Camera.__init__(self)
self._stream = stream

View File

@ -1,6 +1,8 @@
"""Reolink parent entity class."""
from __future__ import annotations
from typing import TypeVar
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import (
@ -11,24 +13,20 @@ from homeassistant.helpers.update_coordinator import (
from . import ReolinkData
from .const import DOMAIN
_T = TypeVar("_T")
class ReolinkBaseCoordinatorEntity(CoordinatorEntity):
"""Parent class for entities that control the Reolink NVR itself, without a channel.
A camera connected directly to HomeAssistant without using a NVR is in the reolink API
basically a NVR with a single channel that has the camera connected to that channel.
"""
class ReolinkBaseCoordinatorEntity(CoordinatorEntity[DataUpdateCoordinator[_T]]):
"""Parent class fo Reolink entities."""
_attr_has_entity_name = True
def __init__(
self,
reolink_data: ReolinkData,
coordinator: DataUpdateCoordinator | None = None,
coordinator: DataUpdateCoordinator[_T],
) -> None:
"""Initialize ReolinkBaseCoordinatorEntity for a NVR entity without a channel."""
if coordinator is None:
coordinator = reolink_data.device_coordinator
"""Initialize ReolinkBaseCoordinatorEntity."""
super().__init__(coordinator)
self._host = reolink_data.host
@ -52,17 +50,28 @@ class ReolinkBaseCoordinatorEntity(CoordinatorEntity):
return self._host.api.session_active and super().available
class ReolinkCoordinatorEntity(ReolinkBaseCoordinatorEntity):
class ReolinkHostCoordinatorEntity(ReolinkBaseCoordinatorEntity[None]):
"""Parent class for entities that control the Reolink NVR itself, without a channel.
A camera connected directly to HomeAssistant without using a NVR is in the reolink API
basically a NVR with a single channel that has the camera connected to that channel.
"""
def __init__(self, reolink_data: ReolinkData) -> None:
"""Initialize ReolinkHostCoordinatorEntity."""
super().__init__(reolink_data, reolink_data.device_coordinator)
class ReolinkChannelCoordinatorEntity(ReolinkHostCoordinatorEntity):
"""Parent class for Reolink hardware camera entities connected to a channel of the NVR."""
def __init__(
self,
reolink_data: ReolinkData,
channel: int,
coordinator: DataUpdateCoordinator | None = None,
) -> None:
"""Initialize ReolinkCoordinatorEntity for a hardware camera connected to a channel of the NVR."""
super().__init__(reolink_data, coordinator)
"""Initialize ReolinkChannelCoordinatorEntity for a hardware camera connected to a channel of the NVR."""
super().__init__(reolink_data)
self._channel = channel

View File

@ -20,7 +20,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ReolinkData
from .const import DOMAIN
from .entity import ReolinkCoordinatorEntity
from .entity import ReolinkChannelCoordinatorEntity
@dataclass
@ -89,7 +89,7 @@ async def async_setup_entry(
)
class ReolinkLightEntity(ReolinkCoordinatorEntity, LightEntity):
class ReolinkLightEntity(ReolinkChannelCoordinatorEntity, LightEntity):
"""Base light entity class for Reolink IP cameras."""
entity_description: ReolinkLightEntityDescription

View File

@ -19,7 +19,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ReolinkData
from .const import DOMAIN
from .entity import ReolinkCoordinatorEntity
from .entity import ReolinkChannelCoordinatorEntity
@dataclass
@ -194,7 +194,7 @@ async def async_setup_entry(
)
class ReolinkNumberEntity(ReolinkCoordinatorEntity, NumberEntity):
class ReolinkNumberEntity(ReolinkChannelCoordinatorEntity, NumberEntity):
"""Base number entity class for Reolink IP cameras."""
entity_description: ReolinkNumberEntityDescription

View File

@ -15,7 +15,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ReolinkData
from .const import DOMAIN
from .entity import ReolinkCoordinatorEntity
from .entity import ReolinkChannelCoordinatorEntity
@dataclass
@ -86,7 +86,7 @@ async def async_setup_entry(
)
class ReolinkSelectEntity(ReolinkCoordinatorEntity, SelectEntity):
class ReolinkSelectEntity(ReolinkChannelCoordinatorEntity, SelectEntity):
"""Base select entity class for Reolink IP cameras."""
entity_description: ReolinkSelectEntityDescription

View File

@ -20,7 +20,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ReolinkData
from .const import DOMAIN
from .entity import ReolinkCoordinatorEntity
from .entity import ReolinkChannelCoordinatorEntity
@dataclass
@ -56,7 +56,7 @@ async def async_setup_entry(
)
class ReolinkSirenEntity(ReolinkCoordinatorEntity, SirenEntity):
class ReolinkSirenEntity(ReolinkChannelCoordinatorEntity, SirenEntity):
"""Base siren entity class for Reolink IP cameras."""
_attr_supported_features = (

View File

@ -15,7 +15,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ReolinkData
from .const import DOMAIN
from .entity import ReolinkBaseCoordinatorEntity, ReolinkCoordinatorEntity
from .entity import ReolinkChannelCoordinatorEntity, ReolinkHostCoordinatorEntity
@dataclass
@ -172,7 +172,7 @@ async def async_setup_entry(
async_add_entities(entities)
class ReolinkSwitchEntity(ReolinkCoordinatorEntity, SwitchEntity):
class ReolinkSwitchEntity(ReolinkChannelCoordinatorEntity, SwitchEntity):
"""Base switch entity class for Reolink IP cameras."""
entity_description: ReolinkSwitchEntityDescription
@ -207,7 +207,7 @@ class ReolinkSwitchEntity(ReolinkCoordinatorEntity, SwitchEntity):
self.async_write_ha_state()
class ReolinkNVRSwitchEntity(ReolinkBaseCoordinatorEntity, SwitchEntity):
class ReolinkNVRSwitchEntity(ReolinkHostCoordinatorEntity, SwitchEntity):
"""Switch entity class for Reolink NVR features."""
entity_description: ReolinkNVRSwitchEntityDescription

View File

@ -2,7 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
from typing import Any, Literal
from reolink_aio.exceptions import ReolinkError
@ -34,7 +34,9 @@ async def async_setup_entry(
async_add_entities([ReolinkUpdateEntity(reolink_data)])
class ReolinkUpdateEntity(ReolinkBaseCoordinatorEntity, UpdateEntity):
class ReolinkUpdateEntity(
ReolinkBaseCoordinatorEntity[str | Literal[False]], UpdateEntity
):
"""Update entity for a Netgear device."""
_attr_device_class = UpdateDeviceClass.FIRMWARE
@ -59,9 +61,6 @@ class ReolinkUpdateEntity(ReolinkBaseCoordinatorEntity, UpdateEntity):
@property
def latest_version(self) -> str | None:
"""Latest version available for install."""
if self.coordinator.data is None:
return None
if not self.coordinator.data:
return self.installed_version