Reduce code needed to check unifiprotect attrs (#119706)

* Reduce code needed to check unifiprotect attrs

* Apply suggestions from code review

* Update homeassistant/components/unifiprotect/manifest.json

* Apply suggestions from code review

* revert

* adjust

* tweak

* make mypy happy
This commit is contained in:
J. Nick Koston 2024-06-14 14:29:18 -05:00 committed by GitHub
parent f8bf357811
commit c0ff2d866f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 49 additions and 196 deletions

View file

@ -5,7 +5,6 @@ from __future__ import annotations
from collections.abc import Sequence from collections.abc import Sequence
import dataclasses import dataclasses
import logging import logging
from typing import Any
from uiprotect.data import ( from uiprotect.data import (
NVR, NVR,
@ -632,26 +631,23 @@ class ProtectDeviceBinarySensor(ProtectDeviceEntity, BinarySensorEntity):
device: Camera | Light | Sensor device: Camera | Light | Sensor
entity_description: ProtectBinaryEntityDescription entity_description: ProtectBinaryEntityDescription
_state_attrs: tuple[str, ...] = ("_attr_available", "_attr_is_on")
@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)
self._attr_is_on = self.entity_description.get_ufp_value(self.device) self._attr_is_on = self.entity_description.get_ufp_value(self.device)
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (self._attr_available, self._attr_is_on)
class MountableProtectDeviceBinarySensor(ProtectDeviceBinarySensor): class MountableProtectDeviceBinarySensor(ProtectDeviceBinarySensor):
"""A UniFi Protect Device Binary Sensor that can change device class at runtime.""" """A UniFi Protect Device Binary Sensor that can change device class at runtime."""
device: Sensor device: Sensor
_state_attrs: tuple[str, ...] = (
"_attr_available",
"_attr_is_on",
"_attr_device_class",
)
@callback @callback
def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None: def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None:
@ -662,21 +658,13 @@ class MountableProtectDeviceBinarySensor(ProtectDeviceBinarySensor):
updated_device.mount_type, BinarySensorDeviceClass.DOOR updated_device.mount_type, BinarySensorDeviceClass.DOOR
) )
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (self._attr_available, self._attr_is_on, self._attr_device_class)
class ProtectDiskBinarySensor(ProtectNVREntity, BinarySensorEntity): class ProtectDiskBinarySensor(ProtectNVREntity, BinarySensorEntity):
"""A UniFi Protect NVR Disk Binary Sensor.""" """A UniFi Protect NVR Disk Binary Sensor."""
_disk: UOSDisk _disk: UOSDisk
entity_description: ProtectBinaryEntityDescription entity_description: ProtectBinaryEntityDescription
_state_attrs = ("_attr_available", "_attr_is_on")
def __init__( def __init__(
self, self,
@ -715,21 +703,12 @@ class ProtectDiskBinarySensor(ProtectNVREntity, BinarySensorEntity):
self._attr_is_on = not self._disk.is_healthy self._attr_is_on = not self._disk.is_healthy
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (self._attr_available, self._attr_is_on)
class ProtectEventBinarySensor(EventEntityMixin, BinarySensorEntity): class ProtectEventBinarySensor(EventEntityMixin, BinarySensorEntity):
"""A UniFi Protect Device Binary Sensor for events.""" """A UniFi Protect Device Binary Sensor for events."""
entity_description: ProtectBinaryEventEntityDescription entity_description: ProtectBinaryEventEntityDescription
_state_attrs = ("_attr_available", "_attr_is_on", "_attr_extra_state_attributes")
@callback @callback
def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None: def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None:
@ -740,20 +719,6 @@ class ProtectEventBinarySensor(EventEntityMixin, BinarySensorEntity):
self._event = None self._event = None
self._attr_extra_state_attributes = {} self._attr_extra_state_attributes = {}
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (
self._attr_available,
self._attr_is_on,
self._attr_extra_state_attributes,
)
MODEL_DESCRIPTIONS_WITH_CLASS = ( MODEL_DESCRIPTIONS_WITH_CLASS = (
(_MODEL_DESCRIPTIONS, ProtectDeviceBinarySensor), (_MODEL_DESCRIPTIONS, ProtectDeviceBinarySensor),

View file

@ -186,7 +186,6 @@ class ProtectButton(ProtectDeviceEntity, ButtonEntity):
@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)
if self.entity_description.key == KEY_ADOPT: if self.entity_description.key == KEY_ADOPT:
device = self.device device = self.device
self._attr_available = device.can_adopt and device.can_create( self._attr_available = device.can_adopt and device.can_create(
@ -195,6 +194,5 @@ class ProtectButton(ProtectDeviceEntity, ButtonEntity):
async def async_press(self) -> None: async def async_press(self) -> None:
"""Press the button.""" """Press the button."""
if self.entity_description.ufp_press is not None: if self.entity_description.ufp_press is not None:
await getattr(self.device, self.entity_description.ufp_press)() await getattr(self.device, self.entity_description.ufp_press)()

View file

@ -3,7 +3,6 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Any
from typing_extensions import Generator from typing_extensions import Generator
from uiprotect.data import ( from uiprotect.data import (
@ -163,6 +162,11 @@ class ProtectCamera(ProtectDeviceEntity, Camera):
"""A Ubiquiti UniFi Protect Camera.""" """A Ubiquiti UniFi Protect Camera."""
device: UFPCamera device: UFPCamera
_state_attrs = (
"_attr_available",
"_attr_is_recording",
"_attr_motion_detection_enabled",
)
def __init__( def __init__(
self, self,
@ -210,20 +214,6 @@ class ProtectCamera(ProtectDeviceEntity, Camera):
else: else:
self._attr_supported_features = CameraEntityFeature(0) self._attr_supported_features = CameraEntityFeature(0)
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (
self._attr_available,
self._attr_is_recording,
self._attr_motion_detection_enabled,
)
@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)

View file

@ -3,7 +3,9 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Callable, Sequence from collections.abc import Callable, Sequence
from functools import partial
import logging import logging
from operator import attrgetter
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
from uiprotect.data import ( from uiprotect.data import (
@ -161,6 +163,7 @@ class BaseProtectEntity(Entity):
device: ProtectAdoptableDeviceModel | NVR device: ProtectAdoptableDeviceModel | NVR
_attr_should_poll = False _attr_should_poll = False
_state_attrs: tuple[str, ...] = ("_attr_available",)
def __init__( def __init__(
self, self,
@ -194,6 +197,9 @@ class BaseProtectEntity(Entity):
self._attr_attribution = DEFAULT_ATTRIBUTION 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(
partial(attrgetter(attr), self) for attr in self._state_attrs
)
async def async_update(self) -> None: async def async_update(self) -> None:
"""Update the entity. """Update the entity.
@ -233,24 +239,18 @@ class BaseProtectEntity(Entity):
and (not async_get_ufp_enabled or async_get_ufp_enabled(device)) and (not async_get_ufp_enabled or async_get_ufp_enabled(device))
) )
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (self._attr_available,)
@callback @callback
def _async_updated_event(self, device: ProtectAdoptableDeviceModel | NVR) -> None: def _async_updated_event(self, device: ProtectAdoptableDeviceModel | NVR) -> None:
"""When device is updated from Protect.""" """When device is updated from Protect."""
previous_attrs = [getter() for getter in self._state_getters]
previous_attrs = self._async_get_state_attrs()
self._async_update_device_from_protect(device) self._async_update_device_from_protect(device)
current_attrs = self._async_get_state_attrs() changed = False
if previous_attrs != current_attrs: for idx, getter in enumerate(self._state_getters):
if previous_attrs[idx] != getter():
changed = True
break
if changed:
if _LOGGER.isEnabledFor(logging.DEBUG): if _LOGGER.isEnabledFor(logging.DEBUG):
device_name = device.name or "" device_name = device.name or ""
if hasattr(self, "entity_description") and self.entity_description.name: if hasattr(self, "entity_description") and self.entity_description.name:
@ -261,7 +261,7 @@ class BaseProtectEntity(Entity):
device_name, device_name,
device.mac, device.mac,
previous_attrs, previous_attrs,
current_attrs, tuple((getattr(self, attr)) for attr in self._state_attrs),
) )
self.async_write_ha_state() self.async_write_ha_state()

View file

@ -63,16 +63,7 @@ class ProtectLight(ProtectDeviceEntity, LightEntity):
_attr_icon = "mdi:spotlight-beam" _attr_icon = "mdi:spotlight-beam"
_attr_color_mode = ColorMode.BRIGHTNESS _attr_color_mode = ColorMode.BRIGHTNESS
_attr_supported_color_modes = {ColorMode.BRIGHTNESS} _attr_supported_color_modes = {ColorMode.BRIGHTNESS}
_state_attrs = ("_attr_available", "_attr_is_on", "_attr_brightness")
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (self._attr_available, self._attr_is_on, self._attr_brightness)
@callback @callback
def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None: def _async_update_device_from_protect(self, device: ProtectModelWithId) -> None:

View file

@ -49,6 +49,13 @@ class ProtectLock(ProtectDeviceEntity, LockEntity):
device: Doorlock device: Doorlock
entity_description: LockEntityDescription entity_description: LockEntityDescription
_state_attrs = (
"_attr_available",
"_attr_is_locked",
"_attr_is_locking",
"_attr_is_unlocking",
"_attr_is_jammed",
)
def __init__( def __init__(
self, self,
@ -64,22 +71,6 @@ class ProtectLock(ProtectDeviceEntity, LockEntity):
self._attr_name = f"{self.device.display_name} Lock" self._attr_name = f"{self.device.display_name} Lock"
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (
self._attr_available,
self._attr_is_locked,
self._attr_is_locking,
self._attr_is_unlocking,
self._attr_is_jammed,
)
@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)

View file

@ -69,6 +69,7 @@ class ProtectMediaPlayer(ProtectDeviceEntity, MediaPlayerEntity):
| MediaPlayerEntityFeature.STOP | MediaPlayerEntityFeature.STOP
| MediaPlayerEntityFeature.BROWSE_MEDIA | MediaPlayerEntityFeature.BROWSE_MEDIA
) )
_state_attrs = ("_attr_available", "_attr_state", "_attr_volume_level")
def __init__( def __init__(
self, self,
@ -107,16 +108,6 @@ class ProtectMediaPlayer(ProtectDeviceEntity, MediaPlayerEntity):
) )
self._attr_available = is_connected and updated_device.feature_flags.has_speaker self._attr_available = is_connected and updated_device.feature_flags.has_speaker
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (self._attr_available, self._attr_state, self._attr_volume_level)
async def async_set_volume_level(self, volume: float) -> None: async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1.""" """Set volume level, range 0..1."""

View file

@ -6,7 +6,6 @@ from collections.abc import Sequence
from dataclasses import dataclass from dataclasses import dataclass
from datetime import timedelta from datetime import timedelta
import logging import logging
from typing import Any
from uiprotect.data import ( from uiprotect.data import (
Camera, Camera,
@ -257,6 +256,7 @@ class ProtectNumbers(ProtectDeviceEntity, NumberEntity):
device: Camera | Light device: Camera | Light
entity_description: ProtectNumberEntityDescription entity_description: ProtectNumberEntityDescription
_state_attrs = ("_attr_available", "_attr_native_value")
def __init__( def __init__(
self, self,
@ -278,13 +278,3 @@ class ProtectNumbers(ProtectDeviceEntity, NumberEntity):
async def async_set_native_value(self, value: float) -> None: async def async_set_native_value(self, value: float) -> None:
"""Set new value.""" """Set new value."""
await self.entity_description.ufp_set(self.device, value) await self.entity_description.ufp_set(self.device, value)
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (self._attr_available, self._attr_native_value)

View file

@ -358,6 +358,7 @@ class ProtectSelects(ProtectDeviceEntity, SelectEntity):
device: Camera | Light | Viewer device: Camera | Light | Viewer
entity_description: ProtectSelectEntityDescription entity_description: ProtectSelectEntityDescription
_state_attrs = ("_attr_available", "_attr_options", "_attr_current_option")
def __init__( def __init__(
self, self,
@ -418,13 +419,3 @@ class ProtectSelects(ProtectDeviceEntity, SelectEntity):
if self.entity_description.ufp_enum_type is not None: if self.entity_description.ufp_enum_type is not None:
unifi_value = self.entity_description.ufp_enum_type(unifi_value) unifi_value = self.entity_description.ufp_enum_type(unifi_value)
await self.entity_description.ufp_set(self.device, unifi_value) await self.entity_description.ufp_set(self.device, unifi_value)
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (self._attr_available, self._attr_options, self._attr_current_option)

View file

@ -702,60 +702,33 @@ class ProtectDeviceSensor(ProtectDeviceEntity, SensorEntity):
"""A Ubiquiti UniFi Protect Sensor.""" """A Ubiquiti UniFi Protect Sensor."""
entity_description: ProtectSensorEntityDescription entity_description: ProtectSensorEntityDescription
_state_attrs = ("_attr_available", "_attr_native_value")
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)
self._attr_native_value = self.entity_description.get_ufp_value(self.device) self._attr_native_value = self.entity_description.get_ufp_value(self.device)
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (self._attr_available, self._attr_native_value)
class ProtectNVRSensor(ProtectNVREntity, SensorEntity): class ProtectNVRSensor(ProtectNVREntity, SensorEntity):
"""A Ubiquiti UniFi Protect Sensor.""" """A Ubiquiti UniFi Protect Sensor."""
entity_description: ProtectSensorEntityDescription entity_description: ProtectSensorEntityDescription
_state_attrs = ("_attr_available", "_attr_native_value")
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)
self._attr_native_value = self.entity_description.get_ufp_value(self.device) self._attr_native_value = self.entity_description.get_ufp_value(self.device)
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (self._attr_available, self._attr_native_value)
class ProtectEventSensor(EventEntityMixin, SensorEntity): class ProtectEventSensor(EventEntityMixin, SensorEntity):
"""A UniFi Protect Device Sensor with access tokens.""" """A UniFi Protect Device Sensor with access tokens."""
entity_description: ProtectSensorEventEntityDescription entity_description: ProtectSensorEventEntityDescription
_state_attrs = (
@callback "_attr_available",
def _async_get_state_attrs(self) -> tuple[Any, ...]: "_attr_native_value",
"""Retrieve data that goes into the current state of the entity. "_attr_extra_state_attributes",
)
Called before and after updating entity and state is only written if there
is a change.
"""
return (
self._attr_available,
self._attr_native_value,
self._attr_extra_state_attributes,
)
class ProtectLicensePlateEventSensor(ProtectEventSensor): class ProtectLicensePlateEventSensor(ProtectEventSensor):

View file

@ -476,6 +476,7 @@ class ProtectSwitch(ProtectDeviceEntity, SwitchEntity):
"""A UniFi Protect Switch.""" """A UniFi Protect Switch."""
entity_description: ProtectSwitchEntityDescription entity_description: ProtectSwitchEntityDescription
_state_attrs = ("_attr_available", "_attr_is_on")
def __init__( def __init__(
self, self,
@ -500,20 +501,12 @@ class ProtectSwitch(ProtectDeviceEntity, SwitchEntity):
"""Turn the device off.""" """Turn the device off."""
await self.entity_description.ufp_set(self.device, False) await self.entity_description.ufp_set(self.device, False)
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (self._attr_available, self._attr_is_on)
class ProtectNVRSwitch(ProtectNVREntity, SwitchEntity): class ProtectNVRSwitch(ProtectNVREntity, SwitchEntity):
"""A UniFi Protect NVR Switch.""" """A UniFi Protect NVR Switch."""
entity_description: ProtectSwitchEntityDescription entity_description: ProtectSwitchEntityDescription
_state_attrs = ("_attr_available", "_attr_is_on")
def __init__( def __init__(
self, self,
@ -537,15 +530,6 @@ class ProtectNVRSwitch(ProtectNVREntity, SwitchEntity):
"""Turn the device off.""" """Turn the device off."""
await self.entity_description.ufp_set(self.device, False) await self.entity_description.ufp_set(self.device, False)
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (self._attr_available, self._attr_is_on)
class ProtectPrivacyModeSwitch(RestoreEntity, ProtectSwitch): class ProtectPrivacyModeSwitch(RestoreEntity, ProtectSwitch):
"""A UniFi Protect Switch.""" """A UniFi Protect Switch."""

View file

@ -4,7 +4,6 @@ from __future__ import annotations
from collections.abc import Sequence from collections.abc import Sequence
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any
from uiprotect.data import ( from uiprotect.data import (
Camera, Camera,
@ -87,6 +86,7 @@ class ProtectDeviceText(ProtectDeviceEntity, TextEntity):
"""A Ubiquiti UniFi Protect Sensor.""" """A Ubiquiti UniFi Protect Sensor."""
entity_description: ProtectTextEntityDescription entity_description: ProtectTextEntityDescription
_state_attrs = ("_attr_available", "_attr_native_value")
def __init__( def __init__(
self, self,
@ -102,17 +102,6 @@ class ProtectDeviceText(ProtectDeviceEntity, TextEntity):
super()._async_update_device_from_protect(device) super()._async_update_device_from_protect(device)
self._attr_native_value = self.entity_description.get_ufp_value(self.device) self._attr_native_value = self.entity_description.get_ufp_value(self.device)
@callback
def _async_get_state_attrs(self) -> tuple[Any, ...]:
"""Retrieve data that goes into the current state of the entity.
Called before and after updating entity and state is only written if there
is a change.
"""
return (self._attr_available, self._attr_native_value)
async def async_set_value(self, value: str) -> None: async def async_set_value(self, value: str) -> None:
"""Change the value.""" """Change the value."""
await self.entity_description.ufp_set(self.device, value) await self.entity_description.ufp_set(self.device, value)