jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 11:17:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor EntityDescriptions for UniFi Protect (#63716)

Browse Source
This commit is contained in:
Christopher Bailey 2022-01-09 23:37:24 -05:00 committed by GitHub
parent b658c053ec
commit d8ba90fb8a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 465 additions and 507 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

108
homeassistant/components/unifiprotect/binary_sensor.py
View File

@ -4,7 +4,6 @@ from __future__ import annotations
from copy import copy
from dataclasses import dataclass
import logging
from typing import Any
from pyunifiprotect.data import NVR, Camera, Event, Light, Sensor
@ -39,26 +38,21 @@ class ProtectBinaryEntityDescription(
):
"""Describes UniFi Protect Binary Sensor entity."""
_KEY_DOORBELL = "doorbell"
_KEY_MOTION = "motion"
_KEY_DOOR = "door"
_KEY_DARK = "dark"
_KEY_BATTERY_LOW = "battery_low"
_KEY_DISK_HEALTH = "disk_health"
ufp_last_trip_value: str | None = None
CAMERA_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = (
ProtectBinaryEntityDescription(
key=_KEY_DOORBELL,
key="doorbell",
name="Doorbell",
device_class=BinarySensorDeviceClass.OCCUPANCY,
icon="mdi:doorbell-video",
ufp_required_field="feature_flags.has_chime",
ufp_value="is_ringing",
ufp_last_trip_value="last_ring",
),
ProtectBinaryEntityDescription(
key=_KEY_DARK,
key="dark",
name="Is Dark",
icon="mdi:brightness-6",
ufp_value="is_dark",
@ -67,54 +61,58 @@ CAMERA_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = (
LIGHT_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = (
ProtectBinaryEntityDescription(
key=_KEY_DARK,
key="dark",
name="Is Dark",
icon="mdi:brightness-6",
ufp_value="is_dark",
),
ProtectBinaryEntityDescription(
key=_KEY_MOTION,
key="motion",
name="Motion Detected",
device_class=BinarySensorDeviceClass.MOTION,
ufp_value="is_pir_motion_detected",
ufp_last_trip_value="last_motion",
),
)
SENSE_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = (
ProtectBinaryEntityDescription(
key=_KEY_DOOR,
key="door",
name="Door",
device_class=BinarySensorDeviceClass.DOOR,
ufp_value="is_opened",
ufp_last_trip_value="open_status_changed_at",
),
ProtectBinaryEntityDescription(
key=_KEY_BATTERY_LOW,
key="battery_low",
name="Battery low",
device_class=BinarySensorDeviceClass.BATTERY,
entity_category=EntityCategory.DIAGNOSTIC,
ufp_value="battery_status.is_low",
),
ProtectBinaryEntityDescription(
key=_KEY_MOTION,
key="motion",
name="Motion Detected",
device_class=BinarySensorDeviceClass.MOTION,
ufp_value="is_motion_detected",
ufp_last_trip_value="motion_detected_at",
),
)
MOTION_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = (
ProtectBinaryEntityDescription(
key=_KEY_MOTION,
key="motion",
name="Motion",
device_class=BinarySensorDeviceClass.MOTION,
ufp_value="is_motion_detected",
ufp_last_trip_value="last_motion",
),
)
DISK_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = (
ProtectBinaryEntityDescription(
key=_KEY_DISK_HEALTH,
key="disk_health",
name="Disk {index} Health",
device_class=BinarySensorDeviceClass.PROBLEM,
entity_category=EntityCategory.DIAGNOSTIC,
@ -181,65 +179,30 @@ def _async_nvr_entities(
class ProtectDeviceBinarySensor(ProtectDeviceEntity, BinarySensorEntity):
"""A UniFi Protect Device Binary Sensor."""
def __init__(
self,
data: ProtectData,
description: ProtectBinaryEntityDescription,
device: Camera | Light | Sensor | None = None,
) -> None:
"""Initialize the Binary Sensor."""
if device and not hasattr(self, "device"):
self.device: Camera | Light | Sensor = device
self.entity_description: ProtectBinaryEntityDescription = description
super().__init__(data)
@callback
def _async_update_extra_attrs_from_protect(self) -> dict[str, Any]:
attrs: dict[str, Any] = {}
key = self.entity_description.key
if key == _KEY_DARK:
return attrs
if isinstance(self.device, Camera):
if key == _KEY_DOORBELL:
attrs[ATTR_LAST_TRIP_TIME] = self.device.last_ring
elif key == _KEY_MOTION:
attrs[ATTR_LAST_TRIP_TIME] = self.device.last_motion
elif isinstance(self.device, Sensor):
if key in (_KEY_MOTION, _KEY_DOOR):
if key == _KEY_MOTION:
last_trip = self.device.motion_detected_at
else:
last_trip = self.device.open_status_changed_at
attrs[ATTR_LAST_TRIP_TIME] = last_trip
elif isinstance(self.device, Light):
if key == _KEY_MOTION:
attrs[ATTR_LAST_TRIP_TIME] = self.device.last_motion
return attrs
device: Camera | Light | Sensor
entity_description: ProtectBinaryEntityDescription
@callback
def _async_update_device_from_protect(self) -> None:
super()._async_update_device_from_protect()
assert self.entity_description.ufp_value is not None
self._attr_is_on = get_nested_attr(
self.device, self.entity_description.ufp_value
)
attrs = self.extra_state_attributes or {}
self._attr_extra_state_attributes = {
**attrs,
**self._async_update_extra_attrs_from_protect(),
}
self._attr_is_on = self.entity_description.get_ufp_value(self.device)
if self.entity_description.ufp_last_trip_value is not None:
last_trip = get_nested_attr(
self.device, self.entity_description.ufp_last_trip_value
)
attrs = self.extra_state_attributes or {}
self._attr_extra_state_attributes = {
**attrs,
ATTR_LAST_TRIP_TIME: last_trip,
}
class ProtectDiskBinarySensor(ProtectNVREntity, BinarySensorEntity):
"""A UniFi Protect NVR Disk Binary Sensor."""
entity_description: ProtectBinaryEntityDescription
def __init__(
self,
data: ProtectData,
@ -252,8 +215,7 @@ class ProtectDiskBinarySensor(ProtectNVREntity, BinarySensorEntity):
description.key = f"{description.key}_{index}"
description.name = (description.name or "{index}").format(index=index)
self._index = index
self.entity_description: ProtectBinaryEntityDescription = description
super().__init__(data, device)
super().__init__(data, device, description)
@callback
def _async_update_device_from_protect(self) -> None:
@ -271,15 +233,7 @@ class ProtectDiskBinarySensor(ProtectNVREntity, BinarySensorEntity):
class ProtectEventBinarySensor(EventThumbnailMixin, ProtectDeviceBinarySensor):
"""A UniFi Protect Device Binary Sensor with access tokens."""
def __init__(
self,
data: ProtectData,
device: Camera,
description: ProtectBinaryEntityDescription,
) -> None:
"""Init a binary sensor that uses access tokens."""
self.device: Camera = device
super().__init__(data, description=description)
device: Camera
@callback
def _async_get_event(self) -> Event | None:

5
homeassistant/components/unifiprotect/camera.py
View File

@ -88,6 +88,8 @@ async def async_setup_entry(
class ProtectCamera(ProtectDeviceEntity, Camera):
"""A Ubiquiti UniFi Protect Camera."""
device: UFPCamera
def __init__(
self,
data: ProtectData,
@ -98,12 +100,11 @@ class ProtectCamera(ProtectDeviceEntity, Camera):
disable_stream: bool,
) -> None:
"""Initialize an UniFi camera."""
self.device: UFPCamera = camera
self.channel = channel
self._secure = secure
self._disable_stream = disable_stream
self._last_image: bytes | None = None
super().__init__(data)
super().__init__(data, camera)
if self._secure:
self._attr_unique_id = f"{self.device.id}_{self.channel.id}"

28
homeassistant/components/unifiprotect/entity.py
View File

@ -109,30 +109,26 @@ def async_all_device_entities(
class ProtectDeviceEntity(Entity):
"""Base class for UniFi protect entities."""
device: ProtectAdoptableDeviceModel
_attr_should_poll = False
def __init__(
self,
data: ProtectData,
device: ProtectAdoptableDeviceModel | None = None,
device: ProtectAdoptableDeviceModel,
description: EntityDescription | None = None,
) -> None:
"""Initialize the entity."""
super().__init__()
self.data: ProtectData = data
if device and not hasattr(self, "device"):
self.device: ProtectAdoptableDeviceModel = device
if description and not hasattr(self, "entity_description"):
self.entity_description = description
elif hasattr(self, "entity_description"):
description = self.entity_description
self.device = device
if description is None:
self._attr_unique_id = f"{self.device.id}"
self._attr_name = f"{self.device.name}"
else:
self.entity_description = description
self._attr_unique_id = f"{self.device.id}_{description.key}"
name = description.name or ""
self._attr_name = f"{self.device.name} {name.title()}"
@ -191,6 +187,9 @@ class ProtectDeviceEntity(Entity):
class ProtectNVREntity(ProtectDeviceEntity):
"""Base class for unifi protect entities."""
# separate subclass on purpose
device: NVR # type: ignore[assignment]
def __init__(
self,
entry: ProtectData,
@ -198,9 +197,7 @@ class ProtectNVREntity(ProtectDeviceEntity):
description: EntityDescription | None = None,
) -> None:
"""Initialize the entity."""
# ProtectNVREntity is intentionally a separate base class
self.device: NVR = device # type: ignore
super().__init__(entry, description=description)
super().__init__(entry, device, description) # type: ignore[arg-type]
@callback
def _async_set_device_info(self) -> None:
@ -222,13 +219,12 @@ class ProtectNVREntity(ProtectDeviceEntity):
self._attr_available = self.data.last_update_success
class AccessTokenMixin(Entity):
class AccessTokenMixin(ProtectDeviceEntity):
"""Adds access_token attribute and provides access tokens for use for anonymous views."""
@property
def access_tokens(self) -> deque[str]:
"""Get valid access_tokens for current entity."""
assert isinstance(self, ProtectDeviceEntity)
return self.data.async_get_or_create_access_tokens(self.entity_id)
@callback
@ -247,7 +243,6 @@ class AccessTokenMixin(Entity):
@callback
def async_cleanup_tokens(self) -> None:
"""Clean up any remaining tokens on removal."""
assert isinstance(self, ProtectDeviceEntity)
if self.entity_id in self.data.access_tokens:
del self.data.access_tokens[self.entity_id]
@ -307,8 +302,7 @@ class EventThumbnailMixin(AccessTokenMixin):
@callback
def _async_update_device_from_protect(self) -> None:
assert isinstance(self, ProtectDeviceEntity)
super()._async_update_device_from_protect() # type: ignore
super()._async_update_device_from_protect()
self._event = self._async_get_event()
attrs = self.extra_state_attributes or {}

11
homeassistant/components/unifiprotect/light.py
View File

@ -56,18 +56,11 @@ def hass_to_unifi_brightness(value: int) -> int:
class ProtectLight(ProtectDeviceEntity, LightEntity):
"""A Ubiquiti UniFi Protect Light Entity."""
device: Light
_attr_icon = "mdi:spotlight-beam"
_attr_supported_features = SUPPORT_BRIGHTNESS
def __init__(
self,
data: ProtectData,
device: Light,
) -> None:
"""Initialize an UniFi light."""
self.device: Light = device
super().__init__(data)
@callback
def _async_update_device_from_protect(self) -> None:
super()._async_update_device_from_protect()

14
homeassistant/components/unifiprotect/media_player.py
View File

@ -55,18 +55,22 @@ async def async_setup_entry(
class ProtectMediaPlayer(ProtectDeviceEntity, MediaPlayerEntity):
"""A Ubiquiti UniFi Protect Speaker."""
device: Camera
entity_description: MediaPlayerEntityDescription
def __init__(
self,
data: ProtectData,
camera: Camera,
) -> None:
"""Initialize an UniFi speaker."""
self.device: Camera = camera
self.entity_description = MediaPlayerEntityDescription(
key="speaker", device_class=MediaPlayerDeviceClass.SPEAKER
super().__init__(
data,
camera,
MediaPlayerEntityDescription(
key="speaker", device_class=MediaPlayerDeviceClass.SPEAKER
),
)
super().__init__(data)
self._attr_name = f"{self.device.name} Speaker"
self._attr_supported_features = (

43
homeassistant/components/unifiprotect/models.py
View File

@ -1,7 +1,18 @@
"""The unifiprotect integration models."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
from dataclasses import dataclass
import logging
from typing import Any
from pyunifiprotect.data import NVR, ProtectAdoptableDeviceModel
from homeassistant.helpers.entity import EntityDescription
from .utils import get_nested_attr
_LOGGER = logging.getLogger(__name__)
@dataclass
@ -10,3 +21,35 @@ class ProtectRequiredKeysMixin:
ufp_required_field: str | None = None
ufp_value: str | None = None
ufp_value_fn: Callable[[ProtectAdoptableDeviceModel | NVR], Any] | None = None
def get_ufp_value(self, obj: ProtectAdoptableDeviceModel | NVR) -> Any:
"""Return value from UniFi Protect device."""
if self.ufp_value is not None:
return get_nested_attr(obj, self.ufp_value)
if self.ufp_value_fn is not None:
return self.ufp_value_fn(obj)
# reminder for future that one is required
raise RuntimeError( # pragma: no cover
"`ufp_value` or `ufp_value_fn` is required"
)
@dataclass
class ProtectSetableKeysMixin(ProtectRequiredKeysMixin):
"""Mixin to for settable values."""
ufp_set_method: str | None = None
ufp_set_method_fn: Callable[
[ProtectAdoptableDeviceModel, Any], Coroutine[Any, Any, None]
] | None = None
async def ufp_set(self, obj: ProtectAdoptableDeviceModel, value: Any) -> None:
"""Set value for UniFi Protect device."""
assert isinstance(self, EntityDescription)
_LOGGER.debug("Setting %s to %s for %s", self.name, value, obj.name)
if self.ufp_set_method is not None:
await getattr(obj, self.ufp_set_method)(value)
elif self.ufp_set_method_fn is not None:
await self.ufp_set_method_fn(obj, value)

86
homeassistant/components/unifiprotect/number.py
View File

@ -3,7 +3,7 @@ from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import Any
from pyunifiprotect.data.devices import Camera, Light
@ -16,17 +16,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .data import ProtectData
from .entity import ProtectDeviceEntity, async_all_device_entities
from .models import ProtectRequiredKeysMixin
from .utils import get_nested_attr
_LOGGER = logging.getLogger(__name__)
_KEY_WDR = "wdr_value"
_KEY_MIC_LEVEL = "mic_level"
_KEY_ZOOM_POS = "zoom_position"
_KEY_SENSITIVITY = "sensitivity"
_KEY_DURATION = "duration"
_KEY_CHIME = "chime_duration"
from .models import ProtectSetableKeysMixin
@dataclass
@ -36,19 +26,28 @@ class NumberKeysMixin:
ufp_max: int
ufp_min: int
ufp_step: int
ufp_set_function: str
@dataclass
class ProtectNumberEntityDescription(
ProtectRequiredKeysMixin, NumberEntityDescription, NumberKeysMixin
ProtectSetableKeysMixin, NumberEntityDescription, NumberKeysMixin
):
"""Describes UniFi Protect Number entity."""
def _get_pir_duration(obj: Any) -> int:
assert isinstance(obj, Light)
return int(obj.light_device_settings.pir_duration.total_seconds())
async def _set_pir_duration(obj: Any, value: float) -> None:
assert isinstance(obj, Light)
await obj.set_duration(timedelta(seconds=value))
CAMERA_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ProtectNumberEntityDescription(
key=_KEY_WDR,
key="wdr_value",
name="Wide Dynamic Range",
icon="mdi:state-machine",
entity_category=EntityCategory.CONFIG,
@ -57,10 +56,10 @@ CAMERA_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ufp_step=1,
ufp_required_field="feature_flags.has_wdr",
ufp_value="isp_settings.wdr",
ufp_set_function="set_wdr_level",
ufp_set_method="set_wdr_level",
),
ProtectNumberEntityDescription(
key=_KEY_MIC_LEVEL,
key="mic_level",
name="Microphone Level",
icon="mdi:microphone",
entity_category=EntityCategory.CONFIG,
@ -69,10 +68,10 @@ CAMERA_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ufp_step=1,
ufp_required_field="feature_flags.has_mic",
ufp_value="mic_volume",
ufp_set_function="set_mic_volume",
ufp_set_method="set_mic_volume",
),
ProtectNumberEntityDescription(
key=_KEY_ZOOM_POS,
key="zoom_position",
name="Zoom Level",
icon="mdi:magnify-plus-outline",
entity_category=EntityCategory.CONFIG,
@ -81,10 +80,10 @@ CAMERA_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ufp_step=1,
ufp_required_field="feature_flags.can_optical_zoom",
ufp_value="isp_settings.zoom_position",
ufp_set_function="set_camera_zoom",
ufp_set_method="set_camera_zoom",
),
ProtectNumberEntityDescription(
key=_KEY_CHIME,
key="duration",
name="Chime Duration",
icon="mdi:camera-timer",
entity_category=EntityCategory.CONFIG,
@ -93,13 +92,13 @@ CAMERA_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ufp_step=100,
ufp_required_field="feature_flags.has_chime",
ufp_value="chime_duration",
ufp_set_function="set_chime_duration",
ufp_set_method="set_chime_duration",
),
)
LIGHT_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ProtectNumberEntityDescription(
key=_KEY_SENSITIVITY,
key="sensitivity",
name="Motion Sensitivity",
icon="mdi:walk",
entity_category=EntityCategory.CONFIG,
@ -108,10 +107,10 @@ LIGHT_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ufp_step=1,
ufp_required_field=None,
ufp_value="light_device_settings.pir_sensitivity",
ufp_set_function="set_sensitivity",
ufp_set_method="set_sensitivity",
),
ProtectNumberEntityDescription(
key=_KEY_DURATION,
key="duration",
name="Auto-shutoff Duration",
icon="mdi:camera-timer",
entity_category=EntityCategory.CONFIG,
@ -119,8 +118,8 @@ LIGHT_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ufp_max=900,
ufp_step=15,
ufp_required_field=None,
ufp_value="light_device_settings.pir_duration",
ufp_set_function="set_duration",
ufp_value_fn=_get_pir_duration,
ufp_set_method_fn=_set_pir_duration,
),
)
@ -145,6 +144,9 @@ async def async_setup_entry(
class ProtectNumbers(ProtectDeviceEntity, NumberEntity):
"""A UniFi Protect Number Entity."""
device: Camera | Light
entity_description: ProtectNumberEntityDescription
def __init__(
self,
data: ProtectData,
@ -152,9 +154,7 @@ class ProtectNumbers(ProtectDeviceEntity, NumberEntity):
description: ProtectNumberEntityDescription,
) -> None:
"""Initialize the Number Entities."""
self.device: Camera | Light = device
self.entity_description: ProtectNumberEntityDescription = description
super().__init__(data)
super().__init__(data, device, description)
self._attr_max_value = self.entity_description.ufp_max
self._attr_min_value = self.entity_description.ufp_min
self._attr_step = self.entity_description.ufp_step
@ -162,30 +162,8 @@ class ProtectNumbers(ProtectDeviceEntity, NumberEntity):
@callback
def _async_update_device_from_protect(self) -> None:
super()._async_update_device_from_protect()
assert self.entity_description.ufp_value is not None
value: float | timedelta = get_nested_attr(
self.device, self.entity_description.ufp_value
)
if isinstance(value, timedelta):
self._attr_value = int(value.total_seconds())
else:
self._attr_value = value
self._attr_value = self.entity_description.get_ufp_value(self.device)
async def async_set_value(self, value: float) -> None:
"""Set new value."""
function = self.entity_description.ufp_set_function
_LOGGER.debug(
"Calling %s to set %s for %s",
function,
value,
self.device.name,
)
set_value: float | timedelta = value
if self.entity_description.key == _KEY_DURATION:
set_value = timedelta(seconds=value)
await getattr(self.device, function)(set_value)
await self.entity_description.ufp_set(self.device, value)

258
homeassistant/components/unifiprotect/select.py
View File

@ -1,12 +1,14 @@
"""This component provides select entities for UniFi Protect."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import timedelta
from enum import Enum
import logging
from typing import Any, Final
from pyunifiprotect.api import ProtectApiClient
from pyunifiprotect.data import (
Camera,
DoorbellMessageType,
@ -14,11 +16,9 @@ from pyunifiprotect.data import (
Light,
LightModeEnableType,
LightModeType,
Liveview,
RecordingMode,
Viewer,
)
from pyunifiprotect.data.devices import LCDMessage
import voluptuous as vol
from homeassistant.components.select import SelectEntity, SelectEntityDescription
@ -33,17 +33,10 @@ from homeassistant.util.dt import utcnow
from .const import ATTR_DURATION, ATTR_MESSAGE, DOMAIN, TYPE_EMPTY_VALUE
from .data import ProtectData
from .entity import ProtectDeviceEntity, async_all_device_entities
from .models import ProtectRequiredKeysMixin
from .utils import get_nested_attr
from .models import ProtectSetableKeysMixin
_LOGGER = logging.getLogger(__name__)
_KEY_IR = "infrared"
_KEY_REC_MODE = "recording_mode"
_KEY_VIEWER = "viewer"
_KEY_LIGHT_MOTION = "light_motion"
_KEY_DOORBELL_TEXT = "doorbell_text"
_KEY_PAIRED_CAMERA = "paired_camera"
INFRARED_MODES = [
{"id": IRLEDMode.AUTO.value, "name": "Auto"},
@ -93,27 +86,112 @@ SET_DOORBELL_LCD_MESSAGE_SCHEMA = vol.Schema(
@dataclass
class ProtectSelectEntityDescription(ProtectRequiredKeysMixin, SelectEntityDescription):
class ProtectSelectEntityDescription(ProtectSetableKeysMixin, SelectEntityDescription):
"""Describes UniFi Protect Select entity."""
ufp_options: list[dict[str, Any]] | None = None
ufp_options_callable: Callable[
[ProtectApiClient], list[dict[str, Any]]
] | None = None
ufp_enum_type: type[Enum] | None = None
ufp_set_function: str | None = None
ufp_set_method: str | None = None
def _get_viewer_options(api: ProtectApiClient) -> list[dict[str, Any]]:
return [
{"id": item.id, "name": item.name} for item in api.bootstrap.liveviews.values()
]
def _get_doorbell_options(api: ProtectApiClient) -> list[dict[str, Any]]:
default_message = api.bootstrap.nvr.doorbell_settings.default_message_text
messages = api.bootstrap.nvr.doorbell_settings.all_messages
built_messages = ({"id": item.type.value, "name": item.text} for item in messages)
return [
{"id": "", "name": f"Default Message ({default_message})"},
*built_messages,
]
def _get_paired_camera_options(api: ProtectApiClient) -> list[dict[str, Any]]:
options = [{"id": TYPE_EMPTY_VALUE, "name": "Not Paired"}]
for camera in api.bootstrap.cameras.values():
options.append({"id": camera.id, "name": camera.name})
return options
def _get_viewer_current(obj: Any) -> str:
assert isinstance(obj, Viewer)
return obj.liveview_id
def _get_light_motion_current(obj: Any) -> str:
assert isinstance(obj, Light)
# a bit of extra to allow On Motion Always/Dark
if (
obj.light_mode_settings.mode == LightModeType.MOTION
and obj.light_mode_settings.enable_at == LightModeEnableType.DARK
):
return f"{LightModeType.MOTION.value}Dark"
return obj.light_mode_settings.mode.value
def _get_doorbell_current(obj: Any) -> str | None:
assert isinstance(obj, Camera)
if obj.lcd_message is None:
return None
return obj.lcd_message.text
async def _set_light_mode(obj: Any, mode: str) -> None:
assert isinstance(obj, Light)
lightmode, timing = LIGHT_MODE_TO_SETTINGS[mode]
await obj.set_light_settings(
LightModeType(lightmode),
enable_at=None if timing is None else LightModeEnableType(timing),
)
async def _set_paired_camera(obj: Any, camera_id: str) -> None:
assert isinstance(obj, Light)
if camera_id == TYPE_EMPTY_VALUE:
camera: Camera | None = None
else:
camera = obj.api.bootstrap.cameras.get(camera_id)
await obj.set_paired_camera(camera)
async def _set_doorbell_message(obj: Any, message: str) -> None:
assert isinstance(obj, Camera)
if message.startswith(DoorbellMessageType.CUSTOM_MESSAGE.value):
await obj.set_lcd_text(DoorbellMessageType.CUSTOM_MESSAGE, text=message)
elif message == TYPE_EMPTY_VALUE:
await obj.set_lcd_text(None)
else:
await obj.set_lcd_text(DoorbellMessageType(message))
async def _set_liveview(obj: Any, liveview_id: str) -> None:
assert isinstance(obj, Viewer)
liveview = obj.api.bootstrap.liveviews[liveview_id]
await obj.set_liveview(liveview)
CAMERA_SELECTS: tuple[ProtectSelectEntityDescription, ...] = (
ProtectSelectEntityDescription(
key=_KEY_REC_MODE,
key="recording_mode",
name="Recording Mode",
icon="mdi:video-outline",
entity_category=EntityCategory.CONFIG,
ufp_options=DEVICE_RECORDING_MODES,
ufp_enum_type=RecordingMode,
ufp_value="recording_settings.mode",
ufp_set_function="set_recording_mode",
ufp_set_method="set_recording_mode",
),
ProtectSelectEntityDescription(
key=_KEY_IR,
key="infrared",
name="Infrared Mode",
icon="mdi:circle-opacity",
entity_category=EntityCategory.CONFIG,
@ -121,16 +199,18 @@ CAMERA_SELECTS: tuple[ProtectSelectEntityDescription, ...] = (
ufp_options=INFRARED_MODES,
ufp_enum_type=IRLEDMode,
ufp_value="isp_settings.ir_led_mode",
ufp_set_function="set_ir_led_model",
ufp_set_method="set_ir_led_model",
),
ProtectSelectEntityDescription(
key=_KEY_DOORBELL_TEXT,
key="doorbell_text",
name="Doorbell Text",
icon="mdi:card-text",
entity_category=EntityCategory.CONFIG,
device_class=DEVICE_CLASS_LCD_MESSAGE,
ufp_required_field="feature_flags.has_lcd_screen",
ufp_value="lcd_message",
ufp_value_fn=_get_doorbell_current,
ufp_options_callable=_get_doorbell_options,
ufp_set_method_fn=_set_doorbell_message,
),
)
@ -141,25 +221,29 @@ LIGHT_SELECTS: tuple[ProtectSelectEntityDescription, ...] = (
icon="mdi:spotlight",
entity_category=EntityCategory.CONFIG,
ufp_options=MOTION_MODE_TO_LIGHT_MODE,
ufp_value="light_mode_settings.mode",
ufp_value_fn=_get_light_motion_current,
ufp_set_method_fn=_set_light_mode,
),
ProtectSelectEntityDescription(
key=_KEY_PAIRED_CAMERA,
key="paired_camera",
name="Paired Camera",
icon="mdi:cctv",
entity_category=EntityCategory.CONFIG,
ufp_value="camera_id",
ufp_options_callable=_get_paired_camera_options,
ufp_set_method_fn=_set_paired_camera,
),
)
VIEWER_SELECTS: tuple[ProtectSelectEntityDescription, ...] = (
ProtectSelectEntityDescription(
key=_KEY_VIEWER,
key="viewer",
name="Liveview",
icon="mdi:view-dashboard",
entity_category=None,
ufp_value="liveview",
ufp_set_function="set_liveview",
ufp_options_callable=_get_viewer_options,
ufp_value_fn=_get_viewer_current,
ufp_set_method_fn=_set_liveview,
),
)
@ -191,6 +275,9 @@ async def async_setup_entry(
class ProtectSelects(ProtectDeviceEntity, SelectEntity):
"""A UniFi Protect Select Entity."""
device: Camera | Light | Viewer
entity_description: ProtectSelectEntityDescription
def __init__(
self,
data: ProtectData,
@ -198,66 +285,33 @@ class ProtectSelects(ProtectDeviceEntity, SelectEntity):
description: ProtectSelectEntityDescription,
) -> None:
"""Initialize the unifi protect select entity."""
assert description.ufp_value is not None
self.device: Camera | Light | Viewer = device
self.entity_description: ProtectSelectEntityDescription = description
super().__init__(data)
super().__init__(data, device, description)
self._attr_name = f"{self.device.name} {self.entity_description.name}"
options = description.ufp_options
if options is not None:
self._attr_options = [item["name"] for item in options]
self._hass_to_unifi_options: dict[str, Any] = {
item["name"]: item["id"] for item in options
}
self._unifi_to_hass_options: dict[Any, str] = {
item["id"]: item["name"] for item in options
}
self._async_set_dynamic_options()
self._async_set_options()
@callback
def _async_update_device_from_protect(self) -> None:
super()._async_update_device_from_protect()
# entities with categories are not exposed for voice and safe to update dynamically
if self.entity_description.entity_category is not None:
if (
self.entity_description.entity_category is not None
and self.entity_description.ufp_options_callable is not None
):
_LOGGER.debug(
"Updating dynamic select options for %s", self.entity_description.name
)
self._async_set_dynamic_options()
self._async_set_options()
@callback
def _async_set_dynamic_options(self) -> None:
"""Options that do not actually update dynamically.
def _async_set_options(self) -> None:
"""Set options attributes from UniFi Protect device."""
This is due to possible downstream platforms dependencies on these options.
"""
if self.entity_description.ufp_options is not None:
return
if self.entity_description.key == _KEY_VIEWER:
options = [
{"id": item.id, "name": item.name}
for item in self.data.api.bootstrap.liveviews.values()
]
elif self.entity_description.key == _KEY_DOORBELL_TEXT:
default_message = (
self.data.api.bootstrap.nvr.doorbell_settings.default_message_text
)
messages = self.data.api.bootstrap.nvr.doorbell_settings.all_messages
built_messages = (
{"id": item.type.value, "name": item.text} for item in messages
)
options = [
{"id": "", "name": f"Default Message ({default_message})"},
*built_messages,
]
elif self.entity_description.key == _KEY_PAIRED_CAMERA:
options = [{"id": TYPE_EMPTY_VALUE, "name": "Not Paired"}]
for camera in self.data.api.bootstrap.cameras.values():
options.append({"id": camera.id, "name": camera.name})
options = self.entity_description.ufp_options
else:
assert self.entity_description.ufp_options_callable is not None
options = self.entity_description.ufp_options_callable(self.data.api)
self._attr_options = [item["name"] for item in options]
self._hass_to_unifi_options = {item["name"]: item["id"] for item in options}
@ -267,79 +321,29 @@ class ProtectSelects(ProtectDeviceEntity, SelectEntity):
def current_option(self) -> str:
"""Return the current selected option."""
assert self.entity_description.ufp_value is not None
unifi_value = get_nested_attr(self.device, self.entity_description.ufp_value)
unifi_value = self.entity_description.get_ufp_value(self.device)
if unifi_value is None:
unifi_value = TYPE_EMPTY_VALUE
elif isinstance(unifi_value, Liveview):
unifi_value = unifi_value.id
elif self.entity_description.key == _KEY_LIGHT_MOTION:
assert isinstance(self.device, Light)
# a bit of extra to allow On Motion Always/Dark
if (
self.device.light_mode_settings.mode == LightModeType.MOTION
and self.device.light_mode_settings.enable_at
== LightModeEnableType.DARK
):
unifi_value = f"{LightModeType.MOTION.value}Dark"
elif self.entity_description.key == _KEY_DOORBELL_TEXT:
assert isinstance(unifi_value, LCDMessage)
return unifi_value.text
return self._unifi_to_hass_options.get(unifi_value, unifi_value)
async def async_select_option(self, option: str) -> None:
"""Change the Select Entity Option."""
if isinstance(self.device, Light):
if self.entity_description.key == _KEY_LIGHT_MOTION:
lightmode, timing = LIGHT_MODE_TO_SETTINGS[option]
_LOGGER.debug("Changing Light Mode to %s", option)
await self.device.set_light_settings(
LightModeType(lightmode),
enable_at=None if timing is None else LightModeEnableType(timing),
)
return
unifi_value = self._hass_to_unifi_options[option]
if self.entity_description.key == _KEY_PAIRED_CAMERA:
if unifi_value == TYPE_EMPTY_VALUE:
unifi_value = None
camera = self.data.api.bootstrap.cameras.get(unifi_value)
await self.device.set_paired_camera(camera)
_LOGGER.debug("Changed Paired Camera to to: %s", option)
return
# Light Motion is a bit different
if self.entity_description.key == _KEY_LIGHT_MOTION:
assert self.entity_description.ufp_set_method_fn is not None
await self.entity_description.ufp_set_method_fn(self.device, option)
return
unifi_value = self._hass_to_unifi_options[option]
if isinstance(self.device, Camera):
if self.entity_description.key == _KEY_DOORBELL_TEXT:
if unifi_value.startswith(DoorbellMessageType.CUSTOM_MESSAGE.value):
await self.device.set_lcd_text(
DoorbellMessageType.CUSTOM_MESSAGE, text=option
)
elif unifi_value == TYPE_EMPTY_VALUE:
await self.device.set_lcd_text(None)
else:
await self.device.set_lcd_text(DoorbellMessageType(unifi_value))
_LOGGER.debug("Changed Doorbell LCD Text to: %s", option)
return
if self.entity_description.ufp_enum_type is not None:
unifi_value = self.entity_description.ufp_enum_type(unifi_value)
elif self.entity_description.key == _KEY_VIEWER:
unifi_value = self.data.api.bootstrap.liveviews[unifi_value]
_LOGGER.debug("%s set to: %s", self.entity_description.key, option)
assert self.entity_description.ufp_set_function
coro = getattr(self.device, self.entity_description.ufp_set_function)
await coro(unifi_value)
await self.entity_description.ufp_set(self.device, unifi_value)
async def async_set_doorbell_message(self, message: str, duration: str) -> None:
"""Set LCD Message on Doorbell display."""
if self.entity_description.key != _KEY_DOORBELL_TEXT:
if self.entity_description.device_class != DEVICE_CLASS_LCD_MESSAGE:
raise HomeAssistantError("Not a doorbell text select entity")
assert isinstance(self.device, Camera)

189
homeassistant/components/unifiprotect/sensor.py
View File

@ -2,7 +2,7 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta
from datetime import datetime
import logging
from typing import Any
@ -28,7 +28,7 @@ from homeassistant.const import (
TIME_SECONDS,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity import Entity, EntityCategory
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
@ -40,7 +40,6 @@ from .entity import (
async_all_device_entities,
)
from .models import ProtectRequiredKeysMixin
from .utils import get_nested_attr
_LOGGER = logging.getLogger(__name__)
DETECTED_OBJECT_NONE = "none"
@ -53,50 +52,54 @@ class ProtectSensorEntityDescription(ProtectRequiredKeysMixin, SensorEntityDescr
precision: int | None = None
def get_ufp_value(self, obj: ProtectAdoptableDeviceModel | NVR) -> Any:
"""Return value from UniFi Protect device."""
value = super().get_ufp_value(obj)
_KEY_UPTIME = "uptime"
_KEY_BLE = "ble_signal"
_KEY_WIRED = "phy_rate"
_KEY_WIFI = "wifi_signal"
if isinstance(value, float) and self.precision:
value = round(value, self.precision)
return value
_KEY_RX = "stats_rx"
_KEY_TX = "stats_tx"
_KEY_OLDEST = "oldest_recording"
_KEY_USED = "storage_used"
_KEY_WRITE_RATE = "write_rate"
_KEY_VOLTAGE = "voltage"
_KEY_BATTERY = "battery_level"
_KEY_LIGHT = "light_level"
_KEY_HUMIDITY = "humidity_level"
_KEY_TEMP = "temperature_level"
def _get_uptime(obj: ProtectAdoptableDeviceModel | NVR) -> datetime | None:
if obj.up_since is None:
return None
_KEY_CPU = "cpu_utilization"
_KEY_CPU_TEMP = "cpu_temperature"
_KEY_MEMORY = "memory_utilization"
_KEY_DISK = "storage_utilization"
_KEY_RECORD_ROTATE = "record_rotating"
_KEY_RECORD_TIMELAPSE = "record_timelapse"
_KEY_RECORD_DETECTIONS = "record_detections"
_KEY_RES_HD = "resolution_HD"
_KEY_RES_4K = "resolution_4K"
_KEY_RES_FREE = "resolution_free"
_KEY_CAPACITY = "record_capacity"
# up_since can vary slightly over time
# truncate to ensure no extra state_change events fire
return obj.up_since.replace(second=0, microsecond=0)
def _get_nvr_recording_capacity(obj: Any) -> int:
assert isinstance(obj, NVR)
if obj.storage_stats.capacity is None:
return 0
return int(obj.storage_stats.capacity.total_seconds())
def _get_nvr_memory(obj: Any) -> float | None:
assert isinstance(obj, NVR)
memory = obj.system_info.memory
if memory.available is None or memory.total is None:
return None
return (1 - memory.available / memory.total) * 100
_KEY_OBJECT = "detected_object"
ALL_DEVICES_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ProtectSensorEntityDescription(
key=_KEY_UPTIME,
key="uptime",
name="Uptime",
icon="mdi:clock",
device_class=SensorDeviceClass.TIMESTAMP,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
ufp_value="up_since",
ufp_value_fn=_get_uptime,
),
ProtectSensorEntityDescription(
key=_KEY_BLE,
key="ble_signal",
name="Bluetooth Signal Strength",
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
@ -107,7 +110,7 @@ ALL_DEVICES_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ufp_required_field="bluetooth_connection_state.signal_strength",
),
ProtectSensorEntityDescription(
key=_KEY_WIRED,
key="phy_rate",
name="Link Speed",
native_unit_of_measurement=DATA_RATE_MEGABITS_PER_SECOND,
entity_category=EntityCategory.DIAGNOSTIC,
@ -117,7 +120,7 @@ ALL_DEVICES_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ufp_required_field="wired_connection_state.phy_rate",
),
ProtectSensorEntityDescription(
key=_KEY_WIFI,
key="wifi_signal",
name="WiFi Signal Strength",
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
@ -131,14 +134,14 @@ ALL_DEVICES_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
CAMERA_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ProtectSensorEntityDescription(
key=_KEY_OLDEST,
key="oldest_recording",
name="Oldest Recording",
device_class=SensorDeviceClass.TIMESTAMP,
entity_category=EntityCategory.DIAGNOSTIC,
ufp_value="stats.video.recording_start",
),
ProtectSensorEntityDescription(
key=_KEY_USED,
key="storage_used",
name="Storage Used",
native_unit_of_measurement=DATA_BYTES,
entity_category=EntityCategory.DIAGNOSTIC,
@ -146,7 +149,7 @@ CAMERA_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ufp_value="stats.storage.used",
),
ProtectSensorEntityDescription(
key=_KEY_WRITE_RATE,
key="write_rate",
name="Disk Write Rate",
native_unit_of_measurement=DATA_RATE_BYTES_PER_SECOND,
entity_category=EntityCategory.DIAGNOSTIC,
@ -155,7 +158,7 @@ CAMERA_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
precision=2,
),
ProtectSensorEntityDescription(
key=_KEY_VOLTAGE,
key="voltage",
name="Voltage",
device_class=SensorDeviceClass.VOLTAGE,
native_unit_of_measurement=ELECTRIC_POTENTIAL_VOLT,
@ -171,7 +174,7 @@ CAMERA_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
CAMERA_DISABLED_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ProtectSensorEntityDescription(
key=_KEY_RX,
key="stats_rx",
name="Received Data",
native_unit_of_measurement=DATA_BYTES,
entity_registry_enabled_default=False,
@ -180,7 +183,7 @@ CAMERA_DISABLED_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ufp_value="stats.rx_bytes",
),
ProtectSensorEntityDescription(
key=_KEY_TX,
key="stats_tx",
name="Transferred Data",
native_unit_of_measurement=DATA_BYTES,
entity_registry_enabled_default=False,
@ -192,7 +195,7 @@ CAMERA_DISABLED_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
SENSE_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ProtectSensorEntityDescription(
key=_KEY_BATTERY,
key="battery_level",
name="Battery Level",
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
@ -201,7 +204,7 @@ SENSE_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ufp_value="battery_status.percentage",
),
ProtectSensorEntityDescription(
key=_KEY_LIGHT,
key="light_level",
name="Light Level",
native_unit_of_measurement=LIGHT_LUX,
device_class=SensorDeviceClass.ILLUMINANCE,
@ -209,7 +212,7 @@ SENSE_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ufp_value="stats.light.value",
),
ProtectSensorEntityDescription(
key=_KEY_HUMIDITY,
key="humidity_level",
name="Humidity Level",
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.HUMIDITY,
@ -217,7 +220,7 @@ SENSE_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ufp_value="stats.humidity.value",
),
ProtectSensorEntityDescription(
key=_KEY_TEMP,
key="temperature_level",
name="Temperature",
native_unit_of_measurement=TEMP_CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
@ -228,15 +231,15 @@ SENSE_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
NVR_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ProtectSensorEntityDescription(
key=_KEY_UPTIME,
key="uptime",
name="Uptime",
icon="mdi:clock",
device_class=SensorDeviceClass.TIMESTAMP,
entity_category=EntityCategory.DIAGNOSTIC,
ufp_value="up_since",
ufp_value_fn=_get_uptime,
),
ProtectSensorEntityDescription(
key=_KEY_DISK,
key="storage_utilization",
name="Storage Utilization",
native_unit_of_measurement=PERCENTAGE,
icon="mdi:harddisk",
@ -246,7 +249,7 @@ NVR_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
precision=2,
),
ProtectSensorEntityDescription(
key=_KEY_RECORD_TIMELAPSE,
key="record_rotating",
name="Type: Timelapse Video",
native_unit_of_measurement=PERCENTAGE,
icon="mdi:server",
@ -256,7 +259,7 @@ NVR_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
precision=2,
),
ProtectSensorEntityDescription(
key=_KEY_RECORD_ROTATE,
key="record_timelapse",
name="Type: Continuous Video",
native_unit_of_measurement=PERCENTAGE,
icon="mdi:server",
@ -266,7 +269,7 @@ NVR_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
precision=2,
),
ProtectSensorEntityDescription(
key=_KEY_RECORD_DETECTIONS,
key="record_detections",
name="Type: Detections Video",
native_unit_of_measurement=PERCENTAGE,
icon="mdi:server",
@ -276,7 +279,7 @@ NVR_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
precision=2,
),
ProtectSensorEntityDescription(
key=_KEY_RES_HD,
key="resolution_HD",
name="Resolution: HD Video",
native_unit_of_measurement=PERCENTAGE,
icon="mdi:cctv",
@ -286,7 +289,7 @@ NVR_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
precision=2,
),
ProtectSensorEntityDescription(
key=_KEY_RES_4K,
key="resolution_4K",
name="Resolution: 4K Video",
native_unit_of_measurement=PERCENTAGE,
icon="mdi:cctv",
@ -296,7 +299,7 @@ NVR_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
precision=2,
),
ProtectSensorEntityDescription(
key=_KEY_RES_FREE,
key="resolution_free",
name="Resolution: Free Space",
native_unit_of_measurement=PERCENTAGE,
icon="mdi:cctv",
@ -306,19 +309,19 @@ NVR_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
precision=2,
),
ProtectSensorEntityDescription(
key=_KEY_CAPACITY,
key="record_capacity",
name="Recording Capacity",
native_unit_of_measurement=TIME_SECONDS,
icon="mdi:record-rec",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
ufp_value="storage_stats.capacity",
ufp_value_fn=_get_nvr_recording_capacity,
),
)
NVR_DISABLED_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ProtectSensorEntityDescription(
key=_KEY_CPU,
key="cpu_utilization",
name="CPU Utilization",
native_unit_of_measurement=PERCENTAGE,
icon="mdi:speedometer",
@ -328,7 +331,7 @@ NVR_DISABLED_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ufp_value="system_info.cpu.average_load",
),
ProtectSensorEntityDescription(
key=_KEY_CPU_TEMP,
key="cpu_temperature",
name="CPU Temperature",
native_unit_of_measurement=TEMP_CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
@ -338,20 +341,21 @@ NVR_DISABLED_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ufp_value="system_info.cpu.temperature",
),
ProtectSensorEntityDescription(
key=_KEY_MEMORY,
key="memory_utilization",
name="Memory Utilization",
native_unit_of_measurement=PERCENTAGE,
icon="mdi:memory",
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
ufp_value_fn=_get_nvr_memory,
precision=2,
),
)
MOTION_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ProtectSensorEntityDescription(
key=_KEY_OBJECT,
key="detected_object",
name="Detected Object",
device_class=DEVICE_CLASS_DETECTION,
),
@ -411,28 +415,11 @@ def _async_nvr_entities(
return entities
class SensorValueMixin(Entity):
"""A mixin to provide sensor values."""
@callback
def _clean_sensor_value(self, value: Any) -> Any:
if isinstance(value, timedelta):
value = int(value.total_seconds())
elif isinstance(value, datetime):
# UniFi Protect value can vary slightly over time
# truncate to ensure no extra state_change events fire
value = value.replace(second=0, microsecond=0)
assert isinstance(self.entity_description, ProtectSensorEntityDescription)
if isinstance(value, float) and self.entity_description.precision:
value = round(value, self.entity_description.precision)
return value
class ProtectDeviceSensor(SensorValueMixin, ProtectDeviceEntity, SensorEntity):
class ProtectDeviceSensor(ProtectDeviceEntity, SensorEntity):
"""A Ubiquiti UniFi Protect Sensor."""
entity_description: ProtectSensorEntityDescription
def __init__(
self,
data: ProtectData,
@ -440,21 +427,15 @@ class ProtectDeviceSensor(SensorValueMixin, ProtectDeviceEntity, SensorEntity):
description: ProtectSensorEntityDescription,
) -> None:
"""Initialize an UniFi Protect sensor."""
self.entity_description: ProtectSensorEntityDescription = description
super().__init__(data, device)
super().__init__(data, device, description)
@callback
def _async_update_device_from_protect(self) -> None:
super()._async_update_device_from_protect()
if self.entity_description.ufp_value is None:
return
value = get_nested_attr(self.device, self.entity_description.ufp_value)
self._attr_native_value = self._clean_sensor_value(value)
self._attr_native_value = self.entity_description.get_ufp_value(self.device)
class ProtectNVRSensor(SensorValueMixin, ProtectNVREntity, SensorEntity):
class ProtectNVRSensor(ProtectNVREntity, SensorEntity):
"""A Ubiquiti UniFi Protect Sensor."""
entity_description: ProtectSensorEntityDescription
@ -466,38 +447,18 @@ class ProtectNVRSensor(SensorValueMixin, ProtectNVREntity, SensorEntity):
description: ProtectSensorEntityDescription,
) -> None:
"""Initialize an UniFi Protect sensor."""
self.entity_description = description
super().__init__(data, device)
super().__init__(data, device, description)
@callback
def _async_update_device_from_protect(self) -> None:
super()._async_update_device_from_protect()
# _KEY_MEMORY
if self.entity_description.ufp_value is None:
memory = self.device.system_info.memory
if memory.available is None or memory.total is None:
self._attr_available = False
return
value = (1 - memory.available / memory.total) * 100
else:
value = get_nested_attr(self.device, self.entity_description.ufp_value)
self._attr_native_value = self._clean_sensor_value(value)
self._attr_native_value = self.entity_description.get_ufp_value(self.device)
class ProtectEventSensor(EventThumbnailMixin, ProtectDeviceSensor):
class ProtectEventSensor(ProtectDeviceSensor, EventThumbnailMixin):
"""A UniFi Protect Device Sensor with access tokens."""
def __init__(
self,
data: ProtectData,
device: Camera,
description: ProtectSensorEntityDescription,
) -> None:
"""Init an sensor that uses access tokens."""
self.device: Camera = device
super().__init__(data, device, description)
device: Camera
@callback
def _async_get_event(self) -> Event | None:
@ -515,8 +476,8 @@ class ProtectEventSensor(EventThumbnailMixin, ProtectDeviceSensor):
@callback
def _async_update_device_from_protect(self) -> None:
super()._async_update_device_from_protect()
# do not call ProtectDeviceSensor method since we want event to get value here
EventThumbnailMixin._async_update_device_from_protect(self)
if self._event is None:
self._attr_native_value = DETECTED_OBJECT_NONE
else:

122
homeassistant/components/unifiprotect/switch.py
View File

@ -17,70 +17,71 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .data import ProtectData
from .entity import ProtectDeviceEntity, async_all_device_entities
from .models import ProtectRequiredKeysMixin
from .utils import get_nested_attr
from .models import ProtectSetableKeysMixin
_LOGGER = logging.getLogger(__name__)
@dataclass
class ProtectSwitchEntityDescription(ProtectRequiredKeysMixin, SwitchEntityDescription):
class ProtectSwitchEntityDescription(ProtectSetableKeysMixin, SwitchEntityDescription):
"""Describes UniFi Protect Switch entity."""
ufp_set_function: str | None = None
_KEY_STATUS_LIGHT = "status_light"
_KEY_HDR_MODE = "hdr_mode"
_KEY_HIGH_FPS = "high_fps"
_KEY_PRIVACY_MODE = "privacy_mode"
_KEY_SYSTEM_SOUNDS = "system_sounds"
_KEY_OSD_NAME = "osd_name"
_KEY_OSD_DATE = "osd_date"
_KEY_OSD_LOGO = "osd_logo"
_KEY_OSD_BITRATE = "osd_bitrate"
_KEY_SMART_PERSON = "smart_person"
_KEY_SMART_VEHICLE = "smart_vehicle"
_KEY_SSH = "ssh"
def _get_is_highfps(obj: Any) -> bool:
assert isinstance(obj, Camera)
return bool(obj.video_mode == VideoMode.HIGH_FPS)
async def _set_highfps(obj: Any, value: bool) -> None:
assert isinstance(obj, Camera)
if value:
await obj.set_video_mode(VideoMode.HIGH_FPS)
else:
await obj.set_video_mode(VideoMode.DEFAULT)
ALL_DEVICES_SWITCHES: tuple[ProtectSwitchEntityDescription, ...] = (
ProtectSwitchEntityDescription(
key=_KEY_SSH,
key="ssh",
name="SSH Enabled",
icon="mdi:lock",
entity_registry_enabled_default=False,
entity_category=EntityCategory.CONFIG,
ufp_value="is_ssh_enabled",
ufp_set_function="set_ssh",
ufp_set_method="set_ssh",
),
)
CAMERA_SWITCHES: tuple[ProtectSwitchEntityDescription, ...] = (
ProtectSwitchEntityDescription(
key=_KEY_STATUS_LIGHT,
key="status_light",
name="Status Light On",
icon="mdi:led-on",
entity_category=EntityCategory.CONFIG,
ufp_required_field="feature_flags.has_led_status",
ufp_value="led_settings.is_enabled",
ufp_set_function="set_status_light",
ufp_set_method="set_status_light",
),
ProtectSwitchEntityDescription(
key=_KEY_HDR_MODE,
key="hdr_mode",
name="HDR Mode",
icon="mdi:brightness-7",
entity_category=EntityCategory.CONFIG,
ufp_required_field="feature_flags.has_hdr",
ufp_value="hdr_mode",
ufp_set_function="set_hdr",
ufp_set_method="set_hdr",
),
ProtectSwitchEntityDescription(
key=_KEY_HIGH_FPS,
key="high_fps",
name="High FPS",
icon="mdi:video-high-definition",
entity_category=EntityCategory.CONFIG,
ufp_required_field="feature_flags.has_highfps",
ufp_value="video_mode",
ufp_value_fn=_get_is_highfps,
ufp_set_method_fn=_set_highfps,
),
ProtectSwitchEntityDescription(
key=_KEY_PRIVACY_MODE,
@ -91,75 +92,75 @@ CAMERA_SWITCHES: tuple[ProtectSwitchEntityDescription, ...] = (
ufp_value="is_privacy_on",
),
ProtectSwitchEntityDescription(
key=_KEY_SYSTEM_SOUNDS,
key="system_sounds",
name="System Sounds",
icon="mdi:speaker",
entity_category=EntityCategory.CONFIG,
ufp_required_field="feature_flags.has_speaker",
ufp_value="speaker_settings.are_system_sounds_enabled",
ufp_set_function="set_system_sounds",
ufp_set_method="set_system_sounds",
),
ProtectSwitchEntityDescription(
key=_KEY_OSD_NAME,
key="osd_name",
name="Overlay: Show Name",
icon="mdi:fullscreen",
entity_category=EntityCategory.CONFIG,
ufp_value="osd_settings.is_name_enabled",
ufp_set_function="set_osd_name",
ufp_set_method="set_osd_name",
),
ProtectSwitchEntityDescription(
key=_KEY_OSD_DATE,
key="osd_date",
name="Overlay: Show Date",
icon="mdi:fullscreen",
entity_category=EntityCategory.CONFIG,
ufp_value="osd_settings.is_date_enabled",
ufp_set_function="set_osd_date",
ufp_set_method="set_osd_date",
),
ProtectSwitchEntityDescription(
key=_KEY_OSD_LOGO,
key="osd_logo",
name="Overlay: Show Logo",
icon="mdi:fullscreen",
entity_category=EntityCategory.CONFIG,
ufp_value="osd_settings.is_logo_enabled",
ufp_set_function="set_osd_logo",
ufp_set_method="set_osd_logo",
),
ProtectSwitchEntityDescription(
key=_KEY_OSD_BITRATE,
key="osd_bitrate",
name="Overlay: Show Bitrate",
icon="mdi:fullscreen",
entity_category=EntityCategory.CONFIG,
ufp_value="osd_settings.is_debug_enabled",
ufp_set_function="set_osd_bitrate",
ufp_set_method="set_osd_bitrate",
),
ProtectSwitchEntityDescription(
key=_KEY_SMART_PERSON,
key="smart_person",
name="Detections: Person",
icon="mdi:walk",
entity_category=EntityCategory.CONFIG,
ufp_required_field="feature_flags.has_smart_detect",
ufp_value="is_person_detection_on",
ufp_set_function="set_person_detection",
ufp_set_method="set_person_detection",
),
ProtectSwitchEntityDescription(
key=_KEY_SMART_VEHICLE,
key="smart_vehicle",
name="Detections: Vehicle",
icon="mdi:car",
entity_category=EntityCategory.CONFIG,
ufp_required_field="feature_flags.has_smart_detect",
ufp_value="is_vehicle_detection_on",
ufp_set_function="set_vehicle_detection",
ufp_set_method="set_vehicle_detection",
),
)
LIGHT_SWITCHES: tuple[ProtectSwitchEntityDescription, ...] = (
ProtectSwitchEntityDescription(
key=_KEY_STATUS_LIGHT,
key="status_light",
name="Status Light On",
icon="mdi:led-on",
entity_category=EntityCategory.CONFIG,
ufp_value="light_device_settings.is_indicator_enabled",
ufp_set_function="set_status_light",
ufp_set_method="set_status_light",
),
)
@ -184,6 +185,8 @@ async def async_setup_entry(
class ProtectSwitch(ProtectDeviceEntity, SwitchEntity):
"""A UniFi Protect Switch."""
entity_description: ProtectSwitchEntityDescription
def __init__(
self,
data: ProtectData,
@ -191,8 +194,7 @@ class ProtectSwitch(ProtectDeviceEntity, SwitchEntity):
description: ProtectSwitchEntityDescription,
) -> None:
"""Initialize an UniFi Protect Switch."""
self.entity_description: ProtectSwitchEntityDescription = description
super().__init__(data, device)
super().__init__(data, device, description)
self._attr_name = f"{self.device.name} {self.entity_description.name}"
self._switch_type = self.entity_description.key
@ -210,44 +212,26 @@ class ProtectSwitch(ProtectDeviceEntity, SwitchEntity):
@property
def is_on(self) -> bool:
"""Return true if device is on."""
assert self.entity_description.ufp_value is not None
ufp_value = get_nested_attr(self.device, self.entity_description.ufp_value)
if self._switch_type == _KEY_HIGH_FPS:
return bool(ufp_value == VideoMode.HIGH_FPS)
return ufp_value is True
return self.entity_description.get_ufp_value(self.device) is True
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the device on."""
if self.entity_description.ufp_set_function is not None:
await getattr(self.device, self.entity_description.ufp_set_function)(True)
return
assert isinstance(self.device, Camera)
if self._switch_type == _KEY_HIGH_FPS:
_LOGGER.debug("Turning on High FPS mode")
await self.device.set_video_mode(VideoMode.HIGH_FPS)
return
if self._switch_type == _KEY_PRIVACY_MODE:
_LOGGER.debug("Turning Privacy Mode on for %s", self.device.name)
assert isinstance(self.device, Camera)
self._previous_mic_level = self.device.mic_volume
self._previous_record_mode = self.device.recording_settings.mode
await self.device.set_privacy(True, 0, RecordingMode.NEVER)
else:
await self.entity_description.ufp_set(self.device, True)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the device off."""
if self.entity_description.ufp_set_function is not None:
await getattr(self.device, self.entity_description.ufp_set_function)(False)
return
assert isinstance(self.device, Camera)
if self._switch_type == _KEY_HIGH_FPS:
_LOGGER.debug("Turning off High FPS mode")
await self.device.set_video_mode(VideoMode.DEFAULT)
elif self._switch_type == _KEY_PRIVACY_MODE:
_LOGGER.debug("Turning Privacy Mode off for %s", self.device.name)
if self._switch_type == _KEY_PRIVACY_MODE:
assert isinstance(self.device, Camera)
_LOGGER.debug("Setting Privacy Mode to false for %s", self.device.name)
await self.device.set_privacy(
False, self._previous_mic_level, self._previous_record_mode
)
else:
await self.entity_description.ufp_set(self.device, False)

53
tests/components/unifiprotect/test_number.py
View File

@ -10,7 +10,6 @@ from pyunifiprotect.data import Camera, Light
from homeassistant.components.unifiprotect.const import DEFAULT_ATTRIBUTION
from homeassistant.components.unifiprotect.number import (
_KEY_DURATION,
CAMERA_NUMBERS,
LIGHT_NUMBERS,
ProtectNumberEntityDescription,
@ -79,7 +78,7 @@ async def camera_fixture(
camera_obj.isp_settings.wdr = 0
camera_obj.mic_volume = 0
camera_obj.isp_settings.zoom_position = 0
camera_obj.chime_duration = timedelta(seconds=0)
camera_obj.chime_duration = 0
mock_entry.api.bootstrap.reset_objects()
mock_entry.api.bootstrap.cameras = {
@ -199,17 +198,14 @@ async def test_number_setup_camera_missing_attr(
assert_entity_counts(hass, Platform.NUMBER, 0, 0)
@pytest.mark.parametrize("description", LIGHT_NUMBERS)
async def test_number_light_simple(
hass: HomeAssistant, light: Light, description: ProtectNumberEntityDescription
):
"""Tests all simple numbers for lights."""
async def test_number_light_sensitivity(hass: HomeAssistant, light: Light):
"""Test sensitivity number entity for lights."""
assert description.ufp_set_function is not None
description = LIGHT_NUMBERS[0]
assert description.ufp_set_method is not None
light.__fields__[description.ufp_set_function] = Mock()
setattr(light, description.ufp_set_function, AsyncMock())
set_method = getattr(light, description.ufp_set_function)
light.__fields__["set_sensitivity"] = Mock()
light.set_sensitivity = AsyncMock()
_, entity_id = ids_from_device_description(Platform.NUMBER, light, description)
@ -217,10 +213,24 @@ async def test_number_light_simple(
"number", "set_value", {ATTR_ENTITY_ID: entity_id, "value": 15.0}, blocking=True
)
if description.key == _KEY_DURATION:
set_method.assert_called_once_with(timedelta(seconds=15.0))
else:
set_method.assert_called_once_with(15.0)
light.set_sensitivity.assert_called_once_with(15.0)
async def test_number_light_duration(hass: HomeAssistant, light: Light):
"""Test chime duration number entity for lights."""
description = LIGHT_NUMBERS[1]
light.__fields__["set_duration"] = Mock()
light.set_duration = AsyncMock()
_, entity_id = ids_from_device_description(Platform.NUMBER, light, description)
await hass.services.async_call(
"number", "set_value", {ATTR_ENTITY_ID: entity_id, "value": 15.0}, blocking=True
)
light.set_duration.assert_called_once_with(timedelta(seconds=15.0))
@pytest.mark.parametrize("description", CAMERA_NUMBERS)
@ -229,11 +239,11 @@ async def test_number_camera_simple(
):
"""Tests all simple numbers for cameras."""
assert description.ufp_set_function is not None
assert description.ufp_set_method is not None
camera.__fields__[description.ufp_set_function] = Mock()
setattr(camera, description.ufp_set_function, AsyncMock())
set_method = getattr(camera, description.ufp_set_function)
camera.__fields__[description.ufp_set_method] = Mock()
setattr(camera, description.ufp_set_method, AsyncMock())
set_method = getattr(camera, description.ufp_set_method)
_, entity_id = ids_from_device_description(Platform.NUMBER, camera, description)
@ -241,7 +251,4 @@ async def test_number_camera_simple(
"number", "set_value", {ATTR_ENTITY_ID: entity_id, "value": 1.0}, blocking=True
)
if description.key == _KEY_DURATION:
set_method.assert_called_once_with(timedelta(seconds=1.0))
else:
set_method.assert_called_once_with(1.0)
set_method.assert_called_once_with(1.0)

47
tests/components/unifiprotect/test_sensor.py
View File

@ -26,7 +26,7 @@ from homeassistant.components.unifiprotect.sensor import (
NVR_SENSORS,
SENSE_SENSORS,
)
from homeassistant.const import ATTR_ATTRIBUTION, STATE_UNAVAILABLE, Platform
from homeassistant.const import ATTR_ATTRIBUTION, STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
@ -242,15 +242,17 @@ async def test_sensor_setup_nvr(
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
async def test_sensor_nvr_memory_unavaiable(
async def test_sensor_nvr_missing_values(
hass: HomeAssistant, mock_entry: MockEntityFixture, now: datetime
):
"""Test memory sensor for NVR if no data available."""
"""Test NVR sensor sensors if no data available."""
mock_entry.api.bootstrap.reset_objects()
nvr: NVR = mock_entry.api.bootstrap.nvr
nvr.system_info.memory.available = None
nvr.system_info.memory.total = None
nvr.up_since = None
nvr.storage_stats.capacity = None
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
await hass.async_block_till_done()
@ -260,6 +262,39 @@ async def test_sensor_nvr_memory_unavaiable(
entity_registry = er.async_get(hass)
# Uptime
description = NVR_SENSORS[0]
unique_id, entity_id = ids_from_device_description(
Platform.SENSOR, nvr, description
)
entity = entity_registry.async_get(entity_id)
assert entity
assert entity.unique_id == unique_id
await enable_entity(hass, mock_entry.entry.entry_id, entity_id)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_UNKNOWN
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
# Memory
description = NVR_SENSORS[8]
unique_id, entity_id = ids_from_device_description(
Platform.SENSOR, nvr, description
)
entity = entity_registry.async_get(entity_id)
assert entity
assert entity.unique_id == unique_id
state = hass.states.get(entity_id)
assert state
assert state.state == "0"
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
# Memory
description = NVR_DISABLED_SENSORS[2]
unique_id, entity_id = ids_from_device_description(
Platform.SENSOR, nvr, description
@ -267,14 +302,14 @@ async def test_sensor_nvr_memory_unavaiable(
entity = entity_registry.async_get(entity_id)
assert entity
assert entity.disabled is not description.entity_registry_enabled_default
assert entity.disabled is True
assert entity.unique_id == unique_id
await enable_entity(hass, mock_entry.entry.entry_id, entity_id)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_UNAVAILABLE
assert state.state == STATE_UNKNOWN
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
@ -286,7 +321,7 @@ async def test_sensor_setup_camera(
entity_registry = er.async_get(hass)
expected_values = (
now.replace(second=0, microsecond=0).isoformat(),
now.replace(microsecond=0).isoformat(),
"100",
"100.0",
"20.0",

8
tests/components/unifiprotect/test_switch.py
View File

@ -384,11 +384,11 @@ async def test_switch_camera_simple(
if description.name in ("High FPS", "Privacy Mode"):
return
assert description.ufp_set_function is not None
assert description.ufp_set_method is not None
camera.__fields__[description.ufp_set_function] = Mock()
setattr(camera, description.ufp_set_function, AsyncMock())
set_method = getattr(camera, description.ufp_set_function)
camera.__fields__[description.ufp_set_method] = Mock()
setattr(camera, description.ufp_set_method, AsyncMock())
set_method = getattr(camera, description.ufp_set_method)
_, entity_id = ids_from_device_description(Platform.SWITCH, camera, description)