Add UniFi Protect number platform (#63220)

pull/63231/head
Christopher Bailey 2022-01-02 14:47:25 -05:00 committed by GitHub
parent 66c2259f5c
commit d364186571
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 444 additions and 3 deletions

View File

@ -46,5 +46,6 @@ PLATFORMS = [
Platform.CAMERA,
Platform.LIGHT,
Platform.MEDIA_PLAYER,
Platform.NUMBER,
Platform.SWITCH,
]

View File

@ -0,0 +1,191 @@
"""This component provides number entities for UniFi Protect."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
import logging
from pyunifiprotect.data.devices import Camera, Light
from homeassistant.components.number import NumberEntity, NumberEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity import EntityCategory
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"
@dataclass
class NumberKeysMixin:
"""Mixin for required keys."""
ufp_max: int
ufp_min: int
ufp_step: int
ufp_set_function: str
@dataclass
class ProtectNumberEntityDescription(
ProtectRequiredKeysMixin, NumberEntityDescription, NumberKeysMixin
):
"""Describes UniFi Protect Number entity."""
CAMERA_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ProtectNumberEntityDescription(
key=_KEY_WDR,
name="Wide Dynamic Range",
icon="mdi:state-machine",
entity_category=EntityCategory.CONFIG,
ufp_min=0,
ufp_max=3,
ufp_step=1,
ufp_required_field="feature_flags.has_wdr",
ufp_value="isp_settings.wdr",
ufp_set_function="set_wdr_level",
),
ProtectNumberEntityDescription(
key=_KEY_MIC_LEVEL,
name="Microphone Level",
icon="mdi:microphone",
entity_category=EntityCategory.CONFIG,
ufp_min=0,
ufp_max=100,
ufp_step=1,
ufp_required_field="feature_flags.has_mic",
ufp_value="mic_volume",
ufp_set_function="set_mic_volume",
),
ProtectNumberEntityDescription(
key=_KEY_ZOOM_POS,
name="Zoom Position",
icon="mdi:magnify-plus-outline",
entity_category=EntityCategory.CONFIG,
ufp_min=0,
ufp_max=100,
ufp_step=1,
ufp_required_field="feature_flags.can_optical_zoom",
ufp_value="isp_settings.zoom_position",
ufp_set_function="set_camera_zoom",
),
ProtectNumberEntityDescription(
key=_KEY_CHIME,
name="Chime Duration",
icon="mdi:camera-timer",
entity_category=EntityCategory.CONFIG,
ufp_min=0,
ufp_max=10000,
ufp_step=100,
ufp_required_field="feature_flags.has_chime",
ufp_value="chime_duration",
ufp_set_function="set_chime_duration",
),
)
LIGHT_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ProtectNumberEntityDescription(
key=_KEY_SENSITIVITY,
name="Motion Sensitivity",
icon="mdi:walk",
entity_category=EntityCategory.CONFIG,
ufp_min=0,
ufp_max=100,
ufp_step=1,
ufp_required_field=None,
ufp_value="light_device_settings.pir_sensitivity",
ufp_set_function="set_sensitivity",
),
ProtectNumberEntityDescription(
key=_KEY_DURATION,
name="Auto-shutoff Duration",
icon="mdi:camera-timer",
entity_category=EntityCategory.CONFIG,
ufp_min=15,
ufp_max=900,
ufp_step=15,
ufp_required_field=None,
ufp_value="light_device_settings.pir_duration",
ufp_set_function="set_duration",
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up number entities for UniFi Protect integration."""
data: ProtectData = hass.data[DOMAIN][entry.entry_id]
entities: list[ProtectDeviceEntity] = async_all_device_entities(
data,
ProtectNumbers,
camera_descs=CAMERA_NUMBERS,
light_descs=LIGHT_NUMBERS,
)
async_add_entities(entities)
class ProtectNumbers(ProtectDeviceEntity, NumberEntity):
"""A UniFi Protect Number Entity."""
def __init__(
self,
data: ProtectData,
device: Camera | Light,
description: ProtectNumberEntityDescription,
) -> None:
"""Initialize the Number Entities."""
self.device: Camera | Light = device
self.entity_description: ProtectNumberEntityDescription = description
super().__init__(data)
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
@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
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)

View File

@ -187,9 +187,11 @@ def ids_from_device_description(
) -> tuple[str, str]:
"""Return expected unique_id and entity_id for a give platform/device/description combination."""
entity_name = device.name.lower().replace(":", "").replace(" ", "_")
entity_name = (
device.name.lower().replace(":", "").replace(" ", "_").replace("-", "_")
)
description_entity_name = (
description.name.lower().replace(":", "").replace(" ", "_")
description.name.lower().replace(":", "").replace(" ", "_").replace("-", "_")
)
unique_id = f"{device.id}_{description.key}"

View File

@ -0,0 +1,247 @@
"""Test the UniFi Protect number platform."""
# pylint: disable=protected-access
from __future__ import annotations
from datetime import timedelta
from unittest.mock import AsyncMock, Mock
import pytest
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,
)
from homeassistant.const import ATTR_ATTRIBUTION, ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .conftest import (
MockEntityFixture,
assert_entity_counts,
ids_from_device_description,
)
@pytest.fixture(name="light")
async def light_fixture(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_light: Light
):
"""Fixture for a single light for testing the number platform."""
# disable pydantic validation so mocking can happen
Light.__config__.validate_assignment = False
light_obj = mock_light.copy(deep=True)
light_obj._api = mock_entry.api
light_obj.name = "Test Light"
light_obj.light_device_settings.pir_sensitivity = 45
light_obj.light_device_settings.pir_duration = timedelta(seconds=45)
mock_entry.api.bootstrap.cameras = {}
mock_entry.api.bootstrap.lights = {
light_obj.id: light_obj,
}
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
await hass.async_block_till_done()
assert_entity_counts(hass, Platform.NUMBER, 2, 2)
yield light_obj
Light.__config__.validate_assignment = True
@pytest.fixture(name="camera")
async def camera_fixture(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_camera: Camera
):
"""Fixture for a single camera for testing the number platform."""
# disable pydantic validation so mocking can happen
Camera.__config__.validate_assignment = False
camera_obj = mock_camera.copy(deep=True)
camera_obj._api = mock_entry.api
camera_obj.channels[0]._api = mock_entry.api
camera_obj.channels[1]._api = mock_entry.api
camera_obj.channels[2]._api = mock_entry.api
camera_obj.name = "Test Camera"
camera_obj.feature_flags.has_chime = True
camera_obj.feature_flags.can_optical_zoom = True
camera_obj.feature_flags.has_mic = True
# has_wdr is an the inverse of has HDR
camera_obj.feature_flags.has_hdr = False
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)
mock_entry.api.bootstrap.lights = {}
mock_entry.api.bootstrap.cameras = {
camera_obj.id: camera_obj,
}
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
await hass.async_block_till_done()
assert_entity_counts(hass, Platform.NUMBER, 4, 4)
yield camera_obj
Camera.__config__.validate_assignment = True
async def test_number_setup_light(
hass: HomeAssistant,
light: Light,
):
"""Test number entity setup for light devices."""
entity_registry = er.async_get(hass)
for description in LIGHT_NUMBERS:
unique_id, entity_id = ids_from_device_description(
Platform.NUMBER, light, 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 == "45"
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
async def test_number_setup_camera_all(
hass: HomeAssistant,
camera: Camera,
):
"""Test number entity setup for camera devices (all features)."""
entity_registry = er.async_get(hass)
for description in CAMERA_NUMBERS:
unique_id, entity_id = ids_from_device_description(
Platform.NUMBER, camera, 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
async def test_number_setup_camera_none(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_camera: Camera
):
"""Test number entity setup for camera devices (no features)."""
camera_obj = mock_camera.copy(deep=True)
camera_obj._api = mock_entry.api
camera_obj.channels[0]._api = mock_entry.api
camera_obj.channels[1]._api = mock_entry.api
camera_obj.channels[2]._api = mock_entry.api
camera_obj.name = "Test Camera"
camera_obj.feature_flags.has_chime = False
camera_obj.feature_flags.can_optical_zoom = False
camera_obj.feature_flags.has_mic = False
# has_wdr is an the inverse of has HDR
camera_obj.feature_flags.has_hdr = True
mock_entry.api.bootstrap.lights = {}
mock_entry.api.bootstrap.cameras = {
camera_obj.id: camera_obj,
}
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
await hass.async_block_till_done()
assert_entity_counts(hass, Platform.NUMBER, 0, 0)
async def test_number_setup_camera_missing_attr(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_camera: Camera
):
"""Test number entity setup for camera devices (no features, bad attrs)."""
# disable pydantic validation so mocking can happen
Camera.__config__.validate_assignment = False
camera_obj = mock_camera.copy(deep=True)
camera_obj._api = mock_entry.api
camera_obj.channels[0]._api = mock_entry.api
camera_obj.channels[1]._api = mock_entry.api
camera_obj.channels[2]._api = mock_entry.api
camera_obj.name = "Test Camera"
camera_obj.feature_flags = None
Camera.__config__.validate_assignment = True
mock_entry.api.bootstrap.lights = {}
mock_entry.api.bootstrap.cameras = {
camera_obj.id: camera_obj,
}
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
await hass.async_block_till_done()
assert_entity_counts(hass, Platform.NUMBER, 0, 0)
@pytest.mark.parametrize("description", LIGHT_NUMBERS)
async def test_switch_light_simple(
hass: HomeAssistant, light: Light, description: ProtectNumberEntityDescription
):
"""Tests all simple switches for lights."""
assert description.ufp_set_function 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)
_, 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
)
if description.key == _KEY_DURATION:
set_method.assert_called_once_with(timedelta(seconds=15.0))
else:
set_method.assert_called_once_with(15.0)
@pytest.mark.parametrize("description", CAMERA_NUMBERS)
async def test_switch_camera_simple(
hass: HomeAssistant, camera: Camera, description: ProtectNumberEntityDescription
):
"""Tests all simple switches for cameras."""
assert description.ufp_set_function 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)
_, entity_id = ids_from_device_description(Platform.NUMBER, camera, description)
await hass.services.async_call(
"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)

View File

@ -1,4 +1,4 @@
"""Test the UniFi Protect light platform."""
"""Test the UniFi Protect switch platform."""
# pylint: disable=protected-access
from __future__ import annotations