starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-26 01:11:51 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add UniFi Protect numbers to the API key only mode (#182726)

Browse Source
This commit is contained in:
Raphael Hehl
2026-09-20 16:49:55 +02:00
committed by GitHub
parent 94a4a22bc2
commit 123a469bd5
3 changed files with 328 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/unifiprotect/const.py
+1
View File
@@ -89,6 +89,7 @@ PUBLIC_ONLY_PLATFORMS = [
Platform.CAMERA,
Platform.EVENT,
Platform.LIGHT,
Platform.NUMBER,
Platform.SENSOR,
Platform.SIREN,
Platform.SWITCH,
homeassistant/components/unifiprotect/number.py
+27 -6
View File
@@ -7,11 +7,16 @@ import logging
from typing import override
from uiprotect.data import Camera, Chime, Light, ModelType, ProtectAdoptableDeviceModel
from uiprotect.data.public_devices import PublicLight, SensorFeatureCapability
from uiprotect.data.public_devices import (
PublicDeviceModel,
PublicLight,
SensorFeatureCapability,
)
from homeassistant.components.number import NumberEntity, NumberEntityDescription
from homeassistant.const import PERCENTAGE, EntityCategory, Platform, UnitOfTime
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .data import ProtectData, ProtectDeviceType, UFPConfigEntry
@@ -79,7 +84,9 @@ CAMERA_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ufp_min=1,
ufp_max=100,
ufp_step=1,
ufp_required_field="has_mic",
# The public setter refuses a camera whose only microphone is a
# hot-plugged module, so the gate is the built-in flag in both modes.
ufp_required_field="feature_flags.has_mic",
ufp_public_value="mic_volume",
ufp_set_method="set_mic_volume",
ufp_perm=PermRequired.WRITE,
@@ -285,28 +292,42 @@ async def async_setup_entry(
entities += _async_all_chime_ring_volume_entities(data, device)
async_add_entities(entities)
@callback
def _add_new_public_device(device: PublicDeviceModel) -> None:
async_add_entities(
async_all_device_entities(
data,
ProtectNumbers,
model_descriptions=_MODEL_DESCRIPTIONS,
public_device=device,
)
)
data.async_subscribe_adopt(_add_new_device)
entry.async_on_unload(
async_dispatcher_connect(hass, data.public_add_signal, _add_new_public_device)
)
entities = async_all_device_entities(
data,
ProtectNumbers,
model_descriptions=_MODEL_DESCRIPTIONS,
)
# Add ring volume entities for all chimes
entities += _async_all_chime_ring_volume_entities(data)
if not data.api.is_public_only:
# The ring volume per paired camera is a private-only chime setting.
entities += _async_all_chime_ring_volume_entities(data)
async_add_entities(entities)
class ProtectNumbers(ProtectDeviceEntity, NumberEntity):
"""A UniFi Protect Number Entity."""
device: Camera | Light
entity_description: ProtectNumberEntityDescription
_state_attrs = ("_attr_available", "_attr_native_value")
def __init__(
self,
data: ProtectData,
device: Camera | Light,
device: ProtectDeviceType,
description: ProtectNumberEntityDescription,
) -> None:
"""Initialize the Number Entities."""
tests/components/unifiprotect/test_number.py
+300 -1
View File
@@ -1,6 +1,9 @@
"""Test the UniFi Protect number platform."""
from collections.abc import Callable, Coroutine
from datetime import timedelta
from functools import partial
from typing import Any
from unittest.mock import AsyncMock, Mock, patch
import pytest
@@ -11,9 +14,13 @@ from uiprotect.data import (
IRLEDMode,
Light,
Permission,
ProtectAdoptableDeviceModel,
RingSetting,
Sensor,
WSAction,
)
from uiprotect.data.devices import Hotplug
from uiprotect.data.public_devices import PublicChime, SensorFeatureCapability
from homeassistant.components.unifiprotect.const import DEFAULT_ATTRIBUTION, DOMAIN
from homeassistant.components.unifiprotect.number import (
@@ -22,6 +29,7 @@ from homeassistant.components.unifiprotect.number import (
SENSE_NUMBERS,
ProtectNumberEntityDescription,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
ATTR_ATTRIBUTION,
ATTR_ENTITY_ID,
@@ -30,9 +38,10 @@ from homeassistant.const import (
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers import device_registry as dr, entity_registry as er
from . import patch_ufp_method
from .conftest import UNIFI_MAC
from .utils import (
MockUFPFixture,
adopt_devices,
@@ -154,6 +163,26 @@ async def test_number_setup_camera_none(
assert_entity_counts(hass, Platform.NUMBER, 0, 0)
async def test_number_no_mic_level_for_hot_plugged_mic(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
ufp: MockUFPFixture,
camera: Camera,
) -> None:
"""A camera whose only microphone is hot-plugged gets no microphone level.
``Camera.has_mic`` counts the hot-plugged module, but the public setter the
number writes through refuses such a camera, so the built-in flag gates it.
"""
camera.feature_flags.has_mic = False
camera.feature_flags.hotplug = Hotplug(audio=True)
assert camera.has_mic
await init_entry(hass, ufp, [camera])
assert "mic_level" not in _number_keys(entity_registry, camera.mac)
async def test_number_setup_camera_missing_attr(
hass: HomeAssistant, ufp: MockUFPFixture, camera: Camera
) -> None:
@@ -734,3 +763,273 @@ async def test_chime_ring_volume_unavailable_when_unpaired(
state = hass.states.get(entity_id)
assert state
assert state.state == "unavailable"
def _number_keys(entity_registry: er.EntityRegistry, mac: str) -> set[str]:
"""Return the description keys of the numbers registered for a device."""
prefix = f"{mac}_"
return {
entry.unique_id.removeprefix(prefix)
for entry in entity_registry.entities.values()
if entry.domain == Platform.NUMBER and entry.unique_id.startswith(prefix)
}
def _make_streamless_public_camera(camera: Camera, **kwargs: Any) -> Mock:
"""Build a public camera without RTSPS streams (snapshot-only)."""
public = make_public_camera(camera, **kwargs)
public.rtsps_streams = None
return public
@pytest.mark.parametrize(
("fixture_name", "make", "key", "value", "setter", "present_keys", "absent_keys"),
[
pytest.param(
"camera",
partial(_make_streamless_public_camera, mic_volume=42),
"mic_level",
"42",
"set_mic_volume",
set(),
{"wdr_value", "zoom_position", "chime_duration", "icr_lux"},
id="camera",
),
pytest.param(
"doorbell",
partial(_make_streamless_public_camera, mic_volume=42),
"mic_level",
"42",
"set_mic_volume",
set(),
{"system_sounds_volume", "doorbell_ring_volume", "chime_duration"},
id="doorbell",
),
pytest.param(
"light",
partial(make_public_light, pir_sensitivity=30),
"sensitivity",
"30",
"set_sensitivity",
{"duration"},
set(),
id="light",
),
pytest.param(
"sensor_all",
partial(
make_public_sensor,
motion_sensitivity=42,
capabilities={SensorFeatureCapability.MOTION},
),
"sensitivity",
"42",
"set_motion_sensitivity",
set(),
set(),
id="sensor",
),
],
)
async def test_public_only_number_end_to_end(
hass: HomeAssistant,
request: pytest.FixtureRequest,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
ufp_public_only: MockUFPFixture,
setup_public_only: Callable[[], Coroutine[Any, Any, None]],
fixture_name: str,
make: Callable[[ProtectAdoptableDeviceModel], Mock],
key: str,
value: str,
setter: str,
present_keys: set[str],
absent_keys: set[str],
) -> None:
"""A public-only entry builds the migrated numbers from the public object.
Private-only numbers are absent, the device is registered from public
identity and a new value goes to the public setter.
"""
device = request.getfixturevalue(fixture_name)
public = make(device)
store = getattr(ufp_public_only.api.public_bootstrap, f"{device.model.value}s")
store[device.id] = public
await setup_public_only()
assert ufp_public_only.entry.state is ConfigEntryState.LOADED
keys = _number_keys(entity_registry, device.mac)
assert key in keys
assert present_keys <= keys
assert not keys & absent_keys
entity_id = entity_registry.async_get_entity_id(
Platform.NUMBER, DOMAIN, f"{device.mac}_{key}"
)
assert entity_id
assert hass.states.get(entity_id).state == value
entry = entity_registry.async_get(entity_id)
assert entry
device_entry = device_registry.async_get(entry.device_id)
assert device_entry
assert device_entry.model == public.type
nvr_device = device_registry.async_get_device_by_identifier(
(DOMAIN, UNIFI_MAC), ufp_public_only.entry.entry_id
)
assert nvr_device
assert device_entry.via_device_id == nvr_device.id
await hass.services.async_call(
"number", "set_value", {ATTR_ENTITY_ID: entity_id, "value": 55}, blocking=True
)
getattr(public, setter).assert_awaited_once_with(55.0)
async def test_public_only_number_light_duration_setter(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
light: Light,
ufp_public_only: MockUFPFixture,
setup_public_only: Callable[[], Coroutine[Any, Any, None]],
) -> None:
"""The auto-shutoff duration is written as a timedelta to the public light."""
public = make_public_light(light, pir_duration_ms=30000)
ufp_public_only.api.public_bootstrap.lights[light.id] = public
await setup_public_only()
entity_id = entity_registry.async_get_entity_id(
Platform.NUMBER, DOMAIN, f"{light.mac}_duration"
)
assert entity_id
assert hass.states.get(entity_id).state == "30"
await hass.services.async_call(
"number", "set_value", {ATTR_ENTITY_ID: entity_id, "value": 45}, blocking=True
)
public.set_duration.assert_awaited_once_with(timedelta(seconds=45))
async def test_public_only_number_chime_has_no_numbers(
hass: HomeAssistant,
chime: Chime,
ufp_public_only: MockUFPFixture,
setup_public_only: Callable[[], Coroutine[Any, Any, None]],
) -> None:
"""Chime volumes are private-only settings, so a public chime yields nothing."""
public = Mock(spec=PublicChime)
public.id = chime.id
public.mac = chime.mac
public.name = chime.name
public.model = chime.model
public.state = DeviceState.CONNECTED
ufp_public_only.api.public_bootstrap.chimes[chime.id] = public
await setup_public_only()
assert ufp_public_only.entry.state is ConfigEntryState.LOADED
assert_entity_counts(hass, Platform.NUMBER, 0, 0)
def _make_public_camera_without_mic(camera: Camera) -> Mock:
"""Build a public camera whose feature flags carry no built-in microphone."""
public = _make_streamless_public_camera(camera)
public.feature_flags.has_mic = False
return public
@pytest.mark.parametrize(
("fixture_name", "make"),
[
pytest.param(
"camera", _make_public_camera_without_mic, id="camera_without_mic"
),
pytest.param(
"sensor_all",
partial(
make_public_sensor, capabilities={SensorFeatureCapability.TEMPERATURE}
),
id="sensor_without_motion",
),
],
)
async def test_public_only_number_gated_out(
request: pytest.FixtureRequest,
entity_registry: er.EntityRegistry,
ufp_public_only: MockUFPFixture,
setup_public_only: Callable[[], Coroutine[Any, Any, None]],
fixture_name: str,
make: Callable[[ProtectAdoptableDeviceModel], Mock],
) -> None:
"""A device failing the public gate gets no number.
The camera gate is the built-in microphone flag, the sense gate the
motion capability.
"""
device = request.getfixturevalue(fixture_name)
store = getattr(ufp_public_only.api.public_bootstrap, f"{device.model.value}s")
store[device.id] = make(device)
await setup_public_only()
assert _number_keys(entity_registry, device.mac) == set()
async def test_public_only_number_added_after_setup(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
light: Light,
ufp_public_only: MockUFPFixture,
setup_public_only: Callable[[], Coroutine[Any, Any, None]],
caplog: pytest.LogCaptureFixture,
) -> None:
"""In public-only mode a light added later gets its numbers from its add frame.
The public devices websocket ``add`` frame is the only discovery signal
without a local user; a re-delivered frame must not add a second time.
"""
await setup_public_only()
assert_entity_counts(hass, Platform.NUMBER, 0, 0)
public = make_public_light(light)
ufp_public_only.api.public_bootstrap.lights[light.id] = public
msg = public_device_ws_message(public)
msg.action = WSAction.ADD
ufp_public_only.devices_ws_subscription(msg)
await hass.async_block_till_done()
assert _number_keys(entity_registry, light.mac) == {
"sensitivity",
"duration",
}
count = len(hass.states.async_entity_ids(Platform.NUMBER.value))
ufp_public_only.devices_ws_subscription(msg)
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(Platform.NUMBER.value)) == count
assert "already exists" not in caplog.text
async def test_public_only_number_sense_registry_cleanup(
entity_registry: er.EntityRegistry,
sensor_all: Sensor,
ufp_public_only: MockUFPFixture,
setup_public_only: Callable[[], Coroutine[Any, Any, None]],
) -> None:
"""The capability cleanup runs without a private bootstrap."""
stale = entity_registry.async_get_or_create(
Platform.NUMBER,
DOMAIN,
f"{sensor_all.mac}_sensitivity",
config_entry=ufp_public_only.entry,
)
ufp_public_only.api.public_bootstrap.sensors[sensor_all.id] = make_public_sensor(
sensor_all, capabilities={SensorFeatureCapability.TEMPERATURE}
)
await setup_public_only()
assert entity_registry.async_get(stale.entity_id) is None