Cleanup unifiprotect entity model (#119746)

* Small cleanups to unifiprotect

* Small cleanups to unifiprotect

* Small cleanups to unifiprotect

* Small cleanups to unifiprotect

* tweak

* comments

* comments

* stale docstrings

* missed one

* remove dead code

* remove dead code

* remove dead code

* remove dead code

* cleanup
This commit is contained in:
J. Nick Koston 2024-06-15 21:02:03 -05:00 committed by GitHub
parent c0a680a80a
commit c519e12042
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 101 additions and 144 deletions

View file

@ -4,7 +4,6 @@ from __future__ import annotations
from collections.abc import Sequence from collections.abc import Sequence
import dataclasses import dataclasses
import logging
from uiprotect.data import ( from uiprotect.data import (
NVR, NVR,
@ -35,15 +34,14 @@ from .entity import (
ProtectNVREntity, ProtectNVREntity,
async_all_device_entities, async_all_device_entities,
) )
from .models import PermRequired, ProtectEventMixin, ProtectRequiredKeysMixin from .models import PermRequired, ProtectEntityDescription, ProtectEventMixin
_LOGGER = logging.getLogger(__name__)
_KEY_DOOR = "door" _KEY_DOOR = "door"
@dataclasses.dataclass(frozen=True, kw_only=True) @dataclasses.dataclass(frozen=True, kw_only=True)
class ProtectBinaryEntityDescription( class ProtectBinaryEntityDescription(
ProtectRequiredKeysMixin, BinarySensorEntityDescription ProtectEntityDescription, BinarySensorEntityDescription
): ):
"""Describes UniFi Protect Binary Sensor entity.""" """Describes UniFi Protect Binary Sensor entity."""
@ -613,7 +611,7 @@ DISK_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = (
), ),
) )
_MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectRequiredKeysMixin]] = { _MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectEntityDescription]] = {
ModelType.CAMERA: CAMERA_SENSORS, ModelType.CAMERA: CAMERA_SENSORS,
ModelType.LIGHT: LIGHT_SENSORS, ModelType.LIGHT: LIGHT_SENSORS,
ModelType.SENSOR: SENSE_SENSORS, ModelType.SENSOR: SENSE_SENSORS,
@ -621,7 +619,7 @@ _MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectRequiredKeysMixin]] = {
ModelType.VIEWPORT: VIEWER_SENSORS, ModelType.VIEWPORT: VIEWER_SENSORS,
} }
_MOUNTABLE_MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectRequiredKeysMixin]] = { _MOUNTABLE_MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectEntityDescription]] = {
ModelType.SENSOR: MOUNTABLE_SENSE_SENSORS, ModelType.SENSOR: MOUNTABLE_SENSE_SENSORS,
} }
@ -652,10 +650,9 @@ class MountableProtectDeviceBinarySensor(ProtectDeviceBinarySensor):
@callback @callback
def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None: def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None:
super()._async_update_device_from_protect(device) super()._async_update_device_from_protect(device)
updated_device = self.device
# UP Sense can be any of the 3 contact sensor device classes # UP Sense can be any of the 3 contact sensor device classes
self._attr_device_class = MOUNT_DEVICE_CLASS_MAP.get( self._attr_device_class = MOUNT_DEVICE_CLASS_MAP.get(
updated_device.mount_type, BinarySensorDeviceClass.DOOR self.device.mount_type, BinarySensorDeviceClass.DOOR
) )
@ -688,7 +685,6 @@ class ProtectDiskBinarySensor(ProtectNVREntity, BinarySensorEntity):
@callback @callback
def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None: def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None:
super()._async_update_device_from_protect(device) super()._async_update_device_from_protect(device)
slot = self._disk.slot slot = self._disk.slot
self._attr_available = False self._attr_available = False
@ -714,7 +710,7 @@ class ProtectEventBinarySensor(EventEntityMixin, BinarySensorEntity):
def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None: def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None:
super()._async_update_device_from_protect(device) super()._async_update_device_from_protect(device)
is_on = self.entity_description.get_is_on(self.device, self._event) is_on = self.entity_description.get_is_on(self.device, self._event)
self._attr_is_on: bool | None = is_on self._attr_is_on = is_on
if not is_on: if not is_on:
self._event = None self._event = None
self._attr_extra_state_attributes = {} self._attr_extra_state_attributes = {}

View file

@ -23,7 +23,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DEVICES_THAT_ADOPT, DISPATCH_ADD, DOMAIN from .const import DEVICES_THAT_ADOPT, DISPATCH_ADD, DOMAIN
from .data import ProtectData, UFPConfigEntry from .data import ProtectData, UFPConfigEntry
from .entity import ProtectDeviceEntity, async_all_device_entities from .entity import ProtectDeviceEntity, async_all_device_entities
from .models import PermRequired, ProtectRequiredKeysMixin, ProtectSetableKeysMixin, T from .models import PermRequired, ProtectEntityDescription, ProtectSetableKeysMixin, T
from .utils import async_dispatch_id as _ufpd from .utils import async_dispatch_id as _ufpd
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -95,7 +95,7 @@ CHIME_BUTTONS: tuple[ProtectButtonEntityDescription, ...] = (
) )
_MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectRequiredKeysMixin]] = { _MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectEntityDescription]] = {
ModelType.CHIME: CHIME_BUTTONS, ModelType.CHIME: CHIME_BUTTONS,
ModelType.SENSOR: SENSOR_BUTTONS, ModelType.SENSOR: SENSOR_BUTTONS,
} }

View file

@ -158,6 +158,9 @@ async def async_setup_entry(
async_add_entities(_async_camera_entities(hass, entry, data)) async_add_entities(_async_camera_entities(hass, entry, data))
_EMPTY_CAMERA_FEATURES = CameraEntityFeature(0)
class ProtectCamera(ProtectDeviceEntity, Camera): class ProtectCamera(ProtectDeviceEntity, Camera):
"""A Ubiquiti UniFi Protect Camera.""" """A Ubiquiti UniFi Protect Camera."""
@ -206,13 +209,12 @@ class ProtectCamera(ProtectDeviceEntity, Camera):
rtsp_url = channel.rtsps_url if self._secure else channel.rtsp_url rtsp_url = channel.rtsps_url if self._secure else channel.rtsp_url
# _async_set_stream_source called by __init__ # _async_set_stream_source called by __init__
self._stream_source = ( # pylint: disable=attribute-defined-outside-init # pylint: disable-next=attribute-defined-outside-init
None if disable_stream else rtsp_url self._stream_source = None if disable_stream else rtsp_url
)
if self._stream_source: if self._stream_source:
self._attr_supported_features = CameraEntityFeature.STREAM self._attr_supported_features = CameraEntityFeature.STREAM
else: else:
self._attr_supported_features = CameraEntityFeature(0) self._attr_supported_features = _EMPTY_CAMERA_FEATURES
@callback @callback
def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None: def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None:

View file

@ -6,7 +6,7 @@ from collections.abc import Callable, Sequence
from functools import partial from functools import partial
import logging import logging
from operator import attrgetter from operator import attrgetter
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING
from uiprotect.data import ( from uiprotect.data import (
NVR, NVR,
@ -31,7 +31,7 @@ from .const import (
DOMAIN, DOMAIN,
) )
from .data import ProtectData from .data import ProtectData
from .models import PermRequired, ProtectEventMixin, ProtectRequiredKeysMixin from .models import PermRequired, ProtectEntityDescription, ProtectEventMixin
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -41,8 +41,8 @@ def _async_device_entities(
data: ProtectData, data: ProtectData,
klass: type[BaseProtectEntity], klass: type[BaseProtectEntity],
model_type: ModelType, model_type: ModelType,
descs: Sequence[ProtectRequiredKeysMixin], descs: Sequence[ProtectEntityDescription],
unadopted_descs: Sequence[ProtectRequiredKeysMixin] | None = None, unadopted_descs: Sequence[ProtectEntityDescription] | None = None,
ufp_device: ProtectAdoptableDeviceModel | None = None, ufp_device: ProtectAdoptableDeviceModel | None = None,
) -> list[BaseProtectEntity]: ) -> list[BaseProtectEntity]:
if not descs and not unadopted_descs: if not descs and not unadopted_descs:
@ -119,11 +119,11 @@ _ALL_MODEL_TYPES = (
@callback @callback
def _combine_model_descs( def _combine_model_descs(
model_type: ModelType, model_type: ModelType,
model_descriptions: dict[ModelType, Sequence[ProtectRequiredKeysMixin]] | None, model_descriptions: dict[ModelType, Sequence[ProtectEntityDescription]] | None,
all_descs: Sequence[ProtectRequiredKeysMixin] | None, all_descs: Sequence[ProtectEntityDescription] | None,
) -> list[ProtectRequiredKeysMixin]: ) -> list[ProtectEntityDescription]:
"""Combine all the descriptions with descriptions a model type.""" """Combine all the descriptions with descriptions a model type."""
descs: list[ProtectRequiredKeysMixin] = list(all_descs) if all_descs else [] descs: list[ProtectEntityDescription] = list(all_descs) if all_descs else []
if model_descriptions and (model_descs := model_descriptions.get(model_type)): if model_descriptions and (model_descs := model_descriptions.get(model_type)):
descs.extend(model_descs) descs.extend(model_descs)
return descs return descs
@ -133,10 +133,10 @@ def _combine_model_descs(
def async_all_device_entities( def async_all_device_entities(
data: ProtectData, data: ProtectData,
klass: type[BaseProtectEntity], klass: type[BaseProtectEntity],
model_descriptions: dict[ModelType, Sequence[ProtectRequiredKeysMixin]] model_descriptions: dict[ModelType, Sequence[ProtectEntityDescription]]
| None = None, | None = None,
all_descs: Sequence[ProtectRequiredKeysMixin] | None = None, all_descs: Sequence[ProtectEntityDescription] | None = None,
unadopted_descs: list[ProtectRequiredKeysMixin] | None = None, unadopted_descs: list[ProtectEntityDescription] | None = None,
ufp_device: ProtectAdoptableDeviceModel | None = None, ufp_device: ProtectAdoptableDeviceModel | None = None,
) -> list[BaseProtectEntity]: ) -> list[BaseProtectEntity]:
"""Generate a list of all the device entities.""" """Generate a list of all the device entities."""
@ -163,6 +163,7 @@ class BaseProtectEntity(Entity):
device: ProtectAdoptableDeviceModel | NVR device: ProtectAdoptableDeviceModel | NVR
_attr_should_poll = False _attr_should_poll = False
_attr_attribution = DEFAULT_ATTRIBUTION
_state_attrs: tuple[str, ...] = ("_attr_available",) _state_attrs: tuple[str, ...] = ("_attr_available",)
def __init__( def __init__(
@ -191,10 +192,9 @@ class BaseProtectEntity(Entity):
else "" else ""
) )
self._attr_name = f"{self.device.display_name} {name.title()}" self._attr_name = f"{self.device.display_name} {name.title()}"
if isinstance(description, ProtectRequiredKeysMixin): if isinstance(description, ProtectEntityDescription):
self._async_get_ufp_enabled = description.get_ufp_enabled self._async_get_ufp_enabled = description.get_ufp_enabled
self._attr_attribution = DEFAULT_ATTRIBUTION
self._async_set_device_info() self._async_set_device_info()
self._async_update_device_from_protect(device) self._async_update_device_from_protect(device)
self._state_getters = tuple( self._state_getters = tuple(
@ -301,8 +301,7 @@ class ProtectNVREntity(BaseProtectEntity):
@callback @callback
def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None: def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None:
data = self.data data = self.data
last_update_success = data.last_update_success if last_update_success := data.last_update_success:
if last_update_success:
self.device = data.api.bootstrap.nvr self.device = data.api.bootstrap.nvr
self._attr_available = last_update_success self._attr_available = last_update_success
@ -311,28 +310,18 @@ class ProtectNVREntity(BaseProtectEntity):
class EventEntityMixin(ProtectDeviceEntity): class EventEntityMixin(ProtectDeviceEntity):
"""Adds motion event attributes to sensor.""" """Adds motion event attributes to sensor."""
_unrecorded_attributes = frozenset({ATTR_EVENT_ID, ATTR_EVENT_SCORE})
entity_description: ProtectEventMixin entity_description: ProtectEventMixin
_unrecorded_attributes = frozenset({ATTR_EVENT_ID, ATTR_EVENT_SCORE})
def __init__( _event: Event | None = None
self,
*args: Any,
**kwarg: Any,
) -> None:
"""Init an sensor that has event thumbnails."""
super().__init__(*args, **kwarg)
self._event: Event | None = None
@callback @callback
def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None: def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None:
event = self.entity_description.get_event_obj(device) if (event := self.entity_description.get_event_obj(device)) is None:
if event is not None: self._attr_extra_state_attributes = {}
else:
self._attr_extra_state_attributes = { self._attr_extra_state_attributes = {
ATTR_EVENT_ID: event.id, ATTR_EVENT_ID: event.id,
ATTR_EVENT_SCORE: event.score, ATTR_EVENT_SCORE: event.score,
} }
else:
self._attr_extra_state_attributes = {}
self._event = event self._event = event
super()._async_update_device_from_protect(device) super()._async_update_device_from_protect(device)

View file

@ -5,8 +5,10 @@ from __future__ import annotations
from collections.abc import Callable, Coroutine from collections.abc import Callable, Coroutine
from dataclasses import dataclass from dataclasses import dataclass
from enum import Enum from enum import Enum
from functools import partial
import logging import logging
from typing import TYPE_CHECKING, Any, Generic, TypeVar from operator import attrgetter
from typing import Any, Generic, TypeVar
from uiprotect.data import NVR, Event, ProtectAdoptableDeviceModel from uiprotect.data import NVR, Event, ProtectAdoptableDeviceModel
@ -19,15 +21,6 @@ _LOGGER = logging.getLogger(__name__)
T = TypeVar("T", bound=ProtectAdoptableDeviceModel | NVR) T = TypeVar("T", bound=ProtectAdoptableDeviceModel | NVR)
def split_tuple(value: tuple[str, ...] | str | None) -> tuple[str, ...] | None:
"""Split string to tuple."""
if value is None:
return None
if TYPE_CHECKING:
assert isinstance(value, str)
return tuple(value.split("."))
class PermRequired(int, Enum): class PermRequired(int, Enum):
"""Type of permission level required for entity.""" """Type of permission level required for entity."""
@ -37,92 +30,83 @@ class PermRequired(int, Enum):
@dataclass(frozen=True, kw_only=True) @dataclass(frozen=True, kw_only=True)
class ProtectRequiredKeysMixin(EntityDescription, Generic[T]): class ProtectEntityDescription(EntityDescription, Generic[T]):
"""Mixin for required keys.""" """Base class for protect entity descriptions."""
# `ufp_required_field`, `ufp_value`, and `ufp_enabled` are defined as ufp_required_field: str | None = None
# a `str` in the dataclass, but `__post_init__` converts it to a ufp_value: str | None = None
# `tuple[str, ...]` to avoid doing it at run time in `get_nested_attr`
# which is usually called millions of times per day.
ufp_required_field: tuple[str, ...] | str | None = None
ufp_value: tuple[str, ...] | str | None = None
ufp_value_fn: Callable[[T], Any] | None = None ufp_value_fn: Callable[[T], Any] | None = None
ufp_enabled: tuple[str, ...] | str | None = None ufp_enabled: str | None = None
ufp_perm: PermRequired | None = None ufp_perm: PermRequired | None = None
def __post_init__(self) -> None:
"""Pre-convert strings to tuples for faster get_nested_attr."""
object.__setattr__(
self, "ufp_required_field", split_tuple(self.ufp_required_field)
)
object.__setattr__(self, "ufp_value", split_tuple(self.ufp_value))
object.__setattr__(self, "ufp_enabled", split_tuple(self.ufp_enabled))
def get_ufp_value(self, obj: T) -> Any: def get_ufp_value(self, obj: T) -> Any:
"""Return value from UniFi Protect device.""" """Return value from UniFi Protect device.
if (ufp_value := self.ufp_value) is not None:
if TYPE_CHECKING:
# `ufp_value` is defined as a `str` in the dataclass, but
# `__post_init__` converts it to a `tuple[str, ...]` to avoid
# doing it at run time in `get_nested_attr` which is usually called
# millions of times per day. This tells mypy that it's a tuple.
assert isinstance(ufp_value, tuple)
return get_nested_attr(obj, ufp_value)
if (ufp_value_fn := self.ufp_value_fn) is not None:
return ufp_value_fn(obj)
# reminder for future that one is required May be overridden by ufp_value or ufp_value_fn.
"""
# ufp_value or ufp_value_fn is required, the
# RuntimeError is to catch any issues in the code
# with new descriptions.
raise RuntimeError( # pragma: no cover raise RuntimeError( # pragma: no cover
"`ufp_value` or `ufp_value_fn` is required" "`ufp_value` or `ufp_value_fn` is required"
) )
def get_ufp_enabled(self, obj: T) -> bool: def has_required(self, obj: T) -> bool:
"""Return value from UniFi Protect device.""" """Return if required field is set.
if (ufp_enabled := self.ufp_enabled) is not None:
if TYPE_CHECKING: May be overridden by ufp_required_field.
# `ufp_enabled` is defined as a `str` in the dataclass, but """
# `__post_init__` converts it to a `tuple[str, ...]` to avoid
# doing it at run time in `get_nested_attr` which is usually called
# millions of times per day. This tells mypy that it's a tuple.
assert isinstance(ufp_enabled, tuple)
return bool(get_nested_attr(obj, ufp_enabled))
return True return True
def has_required(self, obj: T) -> bool: def get_ufp_enabled(self, obj: T) -> bool:
"""Return if has required field.""" """Return if entity is enabled.
if (ufp_required_field := self.ufp_required_field) is None:
return True May be overridden by ufp_enabled.
if TYPE_CHECKING: """
# `ufp_required_field` is defined as a `str` in the dataclass, but return True
# `__post_init__` converts it to a `tuple[str, ...]` to avoid
# doing it at run time in `get_nested_attr` which is usually called def __post_init__(self) -> None:
# millions of times per day. This tells mypy that it's a tuple. """Override get_ufp_value, has_required, and get_ufp_enabled if required."""
assert isinstance(ufp_required_field, tuple) _setter = partial(object.__setattr__, self)
return bool(get_nested_attr(obj, ufp_required_field)) if (_ufp_value := self.ufp_value) is not None:
ufp_value = tuple(_ufp_value.split("."))
_setter("get_ufp_value", partial(get_nested_attr, attrs=ufp_value))
elif (ufp_value_fn := self.ufp_value_fn) is not None:
_setter("get_ufp_value", ufp_value_fn)
if (_ufp_enabled := self.ufp_enabled) is not None:
ufp_enabled = tuple(_ufp_enabled.split("."))
_setter("get_ufp_enabled", partial(get_nested_attr, attrs=ufp_enabled))
if (_ufp_required_field := self.ufp_required_field) is not None:
ufp_required_field = tuple(_ufp_required_field.split("."))
_setter(
"has_required",
lambda obj: bool(get_nested_attr(obj, ufp_required_field)),
)
@dataclass(frozen=True, kw_only=True) @dataclass(frozen=True, kw_only=True)
class ProtectEventMixin(ProtectRequiredKeysMixin[T]): class ProtectEventMixin(ProtectEntityDescription[T]):
"""Mixin for events.""" """Mixin for events."""
ufp_event_obj: str | None = None ufp_event_obj: str | None = None
def get_event_obj(self, obj: T) -> Event | None: def get_event_obj(self, obj: T) -> Event | None:
"""Return value from UniFi Protect device.""" """Return value from UniFi Protect device."""
if self.ufp_event_obj is not None:
event: Event | None = getattr(obj, self.ufp_event_obj, None)
return event
return None return None
def __post_init__(self) -> None:
"""Override get_event_obj if ufp_event_obj is set."""
if (_ufp_event_obj := self.ufp_event_obj) is not None:
object.__setattr__(self, "get_event_obj", attrgetter(_ufp_event_obj))
super().__post_init__()
def get_is_on(self, obj: T, event: Event | None) -> bool: def get_is_on(self, obj: T, event: Event | None) -> bool:
"""Return value if event is active.""" """Return value if event is active."""
return event is not None and self.get_ufp_value(obj) return event is not None and self.get_ufp_value(obj)
@dataclass(frozen=True, kw_only=True) @dataclass(frozen=True, kw_only=True)
class ProtectSetableKeysMixin(ProtectRequiredKeysMixin[T]): class ProtectSetableKeysMixin(ProtectEntityDescription[T]):
"""Mixin for settable values.""" """Mixin for settable values."""
ufp_set_method: str | None = None ufp_set_method: str | None = None

View file

@ -5,7 +5,6 @@ from __future__ import annotations
from collections.abc import Sequence from collections.abc import Sequence
from dataclasses import dataclass from dataclasses import dataclass
from datetime import timedelta from datetime import timedelta
import logging
from uiprotect.data import ( from uiprotect.data import (
Camera, Camera,
@ -23,9 +22,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .data import ProtectData, UFPConfigEntry from .data import ProtectData, UFPConfigEntry
from .entity import ProtectDeviceEntity, async_all_device_entities from .entity import ProtectDeviceEntity, async_all_device_entities
from .models import PermRequired, ProtectRequiredKeysMixin, ProtectSetableKeysMixin, T from .models import PermRequired, ProtectEntityDescription, ProtectSetableKeysMixin, T
_LOGGER = logging.getLogger(__name__)
@dataclass(frozen=True, kw_only=True) @dataclass(frozen=True, kw_only=True)
@ -213,7 +210,7 @@ CHIME_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ufp_perm=PermRequired.WRITE, ufp_perm=PermRequired.WRITE,
), ),
) )
_MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectRequiredKeysMixin]] = { _MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectEntityDescription]] = {
ModelType.CAMERA: CAMERA_NUMBERS, ModelType.CAMERA: CAMERA_NUMBERS,
ModelType.LIGHT: LIGHT_NUMBERS, ModelType.LIGHT: LIGHT_NUMBERS,
ModelType.SENSOR: SENSE_NUMBERS, ModelType.SENSOR: SENSE_NUMBERS,

View file

@ -35,7 +35,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import TYPE_EMPTY_VALUE from .const import TYPE_EMPTY_VALUE
from .data import ProtectData, UFPConfigEntry from .data import ProtectData, UFPConfigEntry
from .entity import ProtectDeviceEntity, async_all_device_entities from .entity import ProtectDeviceEntity, async_all_device_entities
from .models import PermRequired, ProtectRequiredKeysMixin, ProtectSetableKeysMixin, T from .models import PermRequired, ProtectEntityDescription, ProtectSetableKeysMixin, T
from .utils import async_get_light_motion_current from .utils import async_get_light_motion_current
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -319,7 +319,7 @@ VIEWER_SELECTS: tuple[ProtectSelectEntityDescription, ...] = (
), ),
) )
_MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectRequiredKeysMixin]] = { _MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectEntityDescription]] = {
ModelType.CAMERA: CAMERA_SELECTS, ModelType.CAMERA: CAMERA_SELECTS,
ModelType.LIGHT: LIGHT_SELECTS, ModelType.LIGHT: LIGHT_SELECTS,
ModelType.SENSOR: SENSE_SELECTS, ModelType.SENSOR: SENSE_SELECTS,

View file

@ -47,7 +47,7 @@ from .entity import (
ProtectNVREntity, ProtectNVREntity,
async_all_device_entities, async_all_device_entities,
) )
from .models import PermRequired, ProtectEventMixin, ProtectRequiredKeysMixin, T from .models import PermRequired, ProtectEntityDescription, ProtectEventMixin, T
from .utils import async_get_light_motion_current from .utils import async_get_light_motion_current
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -56,7 +56,7 @@ OBJECT_TYPE_NONE = "none"
@dataclass(frozen=True, kw_only=True) @dataclass(frozen=True, kw_only=True)
class ProtectSensorEntityDescription( class ProtectSensorEntityDescription(
ProtectRequiredKeysMixin[T], SensorEntityDescription ProtectEntityDescription[T], SensorEntityDescription
): ):
"""Describes UniFi Protect Sensor entity.""" """Describes UniFi Protect Sensor entity."""
@ -608,7 +608,7 @@ VIEWER_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
), ),
) )
_MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectRequiredKeysMixin]] = { _MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectEntityDescription]] = {
ModelType.CAMERA: CAMERA_SENSORS + CAMERA_DISABLED_SENSORS, ModelType.CAMERA: CAMERA_SENSORS + CAMERA_DISABLED_SENSORS,
ModelType.SENSOR: SENSE_SENSORS, ModelType.SENSOR: SENSE_SENSORS,
ModelType.LIGHT: LIGHT_SENSORS, ModelType.LIGHT: LIGHT_SENSORS,

View file

@ -30,7 +30,7 @@ from .entity import (
ProtectNVREntity, ProtectNVREntity,
async_all_device_entities, async_all_device_entities,
) )
from .models import PermRequired, ProtectRequiredKeysMixin, ProtectSetableKeysMixin, T from .models import PermRequired, ProtectEntityDescription, ProtectSetableKeysMixin, T
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
ATTR_PREV_MIC = "prev_mic_level" ATTR_PREV_MIC = "prev_mic_level"
@ -459,7 +459,7 @@ NVR_SWITCHES: tuple[ProtectSwitchEntityDescription, ...] = (
), ),
) )
_MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectRequiredKeysMixin]] = { _MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectEntityDescription]] = {
ModelType.CAMERA: CAMERA_SWITCHES, ModelType.CAMERA: CAMERA_SWITCHES,
ModelType.LIGHT: LIGHT_SWITCHES, ModelType.LIGHT: LIGHT_SWITCHES,
ModelType.SENSOR: SENSE_SWITCHES, ModelType.SENSOR: SENSE_SWITCHES,
@ -467,7 +467,7 @@ _MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectRequiredKeysMixin]] = {
ModelType.VIEWPORT: VIEWER_SWITCHES, ModelType.VIEWPORT: VIEWER_SWITCHES,
} }
_PRIVACY_MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectRequiredKeysMixin]] = { _PRIVACY_MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectEntityDescription]] = {
ModelType.CAMERA: [PRIVACY_MODE_SWITCH] ModelType.CAMERA: [PRIVACY_MODE_SWITCH]
} }
@ -487,7 +487,6 @@ class ProtectSwitch(ProtectDeviceEntity, SwitchEntity):
"""Initialize an UniFi Protect Switch.""" """Initialize an UniFi Protect Switch."""
super().__init__(data, device, description) super().__init__(data, device, description)
self._attr_name = f"{self.device.display_name} {self.entity_description.name}" self._attr_name = f"{self.device.display_name} {self.entity_description.name}"
self._switch_type = self.entity_description.key
def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None: def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None:
super()._async_update_device_from_protect(device) super()._async_update_device_from_protect(device)
@ -539,21 +538,20 @@ class ProtectPrivacyModeSwitch(RestoreEntity, ProtectSwitch):
def __init__( def __init__(
self, self,
data: ProtectData, data: ProtectData,
device: ProtectAdoptableDeviceModel, device: Camera,
description: ProtectSwitchEntityDescription, description: ProtectSwitchEntityDescription,
) -> None: ) -> None:
"""Initialize an UniFi Protect Switch.""" """Initialize an UniFi Protect Switch."""
super().__init__(data, device, description) super().__init__(data, device, description)
if device.is_privacy_on:
if self.device.is_privacy_on:
extra_state = self.extra_state_attributes or {} extra_state = self.extra_state_attributes or {}
self._previous_mic_level = extra_state.get(ATTR_PREV_MIC, 100) self._previous_mic_level = extra_state.get(ATTR_PREV_MIC, 100)
self._previous_record_mode = extra_state.get( self._previous_record_mode = extra_state.get(
ATTR_PREV_RECORD, RecordingMode.ALWAYS ATTR_PREV_RECORD, RecordingMode.ALWAYS
) )
else: else:
self._previous_mic_level = self.device.mic_volume self._previous_mic_level = device.mic_volume
self._previous_record_mode = self.device.recording_settings.mode self._previous_record_mode = device.recording_settings.mode
@callback @callback
def _update_previous_attr(self) -> None: def _update_previous_attr(self) -> None:

View file

@ -18,9 +18,9 @@ from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .data import ProtectData, UFPConfigEntry from .data import UFPConfigEntry
from .entity import ProtectDeviceEntity, async_all_device_entities from .entity import ProtectDeviceEntity, async_all_device_entities
from .models import PermRequired, ProtectRequiredKeysMixin, ProtectSetableKeysMixin, T from .models import PermRequired, ProtectEntityDescription, ProtectSetableKeysMixin, T
@dataclass(frozen=True, kw_only=True) @dataclass(frozen=True, kw_only=True)
@ -50,7 +50,7 @@ CAMERA: tuple[ProtectTextEntityDescription, ...] = (
), ),
) )
_MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectRequiredKeysMixin]] = { _MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectEntityDescription]] = {
ModelType.CAMERA: CAMERA, ModelType.CAMERA: CAMERA,
} }
@ -88,15 +88,6 @@ class ProtectDeviceText(ProtectDeviceEntity, TextEntity):
entity_description: ProtectTextEntityDescription entity_description: ProtectTextEntityDescription
_state_attrs = ("_attr_available", "_attr_native_value") _state_attrs = ("_attr_available", "_attr_native_value")
def __init__(
self,
data: ProtectData,
device: ProtectAdoptableDeviceModel,
description: ProtectTextEntityDescription,
) -> None:
"""Initialize an UniFi Protect sensor."""
super().__init__(data, device, description)
@callback @callback
def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None: def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None:
super()._async_update_device_from_protect(device) super()._async_update_device_from_protect(device)