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

Drop UniFi Protect helpers that uiprotect 16.12.0 now provides (#182688)

Browse Source
This commit is contained in:
Raphael Hehl
2026-09-19 17:10:33 +02:00
committed by GitHub
parent 3f5a758f5f
commit d91529ad52
6 changed files with 26 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/unifiprotect/number.py
+3 -13
View File
@@ -4,14 +4,10 @@ from collections.abc import Sequence
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import cast, override
from typing import override
from uiprotect.data import Camera, Chime, Light, ModelType, ProtectAdoptableDeviceModel
from uiprotect.data.public_devices import (
PublicDeviceModel,
PublicLight,
SensorFeatureCapability,
)
from uiprotect.data.public_devices import PublicLight, SensorFeatureCapability
from homeassistant.components.number import NumberEntity, NumberEntityDescription
from homeassistant.const import PERCENTAGE, EntityCategory, Platform, UnitOfTime
@@ -46,12 +42,6 @@ class ProtectNumberEntityDescription(
ufp_step: int | float
def _get_pir_duration_public(obj: PublicDeviceModel) -> int | None:
# Public API reports the PIR auto-shutoff duration in milliseconds.
duration = cast(PublicLight, obj).light_device_settings.pir_duration
return None if duration is None else round(duration / 1000)
async def _set_pir_duration(obj: PublicLight, value: float) -> None:
await obj.set_duration(timedelta(seconds=value))
@@ -185,7 +175,7 @@ LIGHT_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ufp_min=15,
ufp_max=900,
ufp_step=15,
ufp_public_value_fn=_get_pir_duration_public,
ufp_public_value="light_device_settings.pir_duration_seconds",
ufp_set_method_fn=_set_pir_duration,
ufp_perm=PermRequired.WRITE,
),
homeassistant/components/unifiprotect/select.py
+2 -13
View File
@@ -4,7 +4,7 @@ from collections.abc import Callable, Sequence
from dataclasses import dataclass
from enum import Enum
import logging
from typing import Any, cast, override
from typing import Any, override
from uiprotect.api import ProtectApiClient
from uiprotect.data import (
@@ -27,7 +27,6 @@ from uiprotect.data import (
)
from uiprotect.data.public_devices import (
PublicCamera,
PublicDeviceModel,
PublicLight,
SensorFeatureCapability,
)
@@ -223,16 +222,6 @@ _HDR_MODE_MAP = {
"always": PublicHdrMode.ON,
"off": PublicHdrMode.OFF,
}
_HDR_MODE_MAP_INVERSE = {v: k for k, v in _HDR_MODE_MAP.items()}
def _get_hdr_mode_public(obj: PublicDeviceModel) -> str | None:
"""Return the HDR option id from the public camera's ``hdr_type``.
``hdr_type`` is non-optional on the public model; ``.get`` still yields
``None`` for any value missing from the map.
"""
return _HDR_MODE_MAP_INVERSE.get(cast(PublicCamera, obj).hdr_type)
async def _set_hdr_mode(obj: PublicCamera, mode: str) -> None:
@@ -298,7 +287,7 @@ CAMERA_SELECTS: tuple[ProtectSelectEntityDescription, ...] = (
entity_category=EntityCategory.CONFIG,
ufp_required_field="feature_flags.has_hdr",
ufp_options=HDR_MODES,
ufp_public_value_fn=_get_hdr_mode_public,
ufp_public_value="hdr_mode_display",
ufp_set_method_fn=_set_hdr_mode,
ufp_perm=PermRequired.WRITE,
),
homeassistant/components/unifiprotect/sensor.py
+3 -13
View File
@@ -6,7 +6,7 @@ from datetime import datetime
from functools import partial
import logging
import operator
from typing import Any, cast, override
from typing import Any, override
from uiprotect.data import (
NVR,
@@ -19,12 +19,7 @@ from uiprotect.data import (
ProtectDeviceModel,
Sensor,
)
from uiprotect.data.public_devices import (
PublicDeviceModel,
PublicLight,
SensorFeatureCapability,
)
from uiprotect.utils import convert_to_datetime
from uiprotect.data.public_devices import PublicDeviceModel, SensorFeatureCapability
from homeassistant.components.sensor import (
SensorDeviceClass,
@@ -100,11 +95,6 @@ class ProtectSensorEventEntityDescription(
"""Describes UniFi Protect Sensor entity."""
def _get_last_motion_public(obj: PublicDeviceModel) -> datetime | None:
# Public API reports last motion as a JS epoch (ms); private side a datetime.
return convert_to_datetime(cast(PublicLight, obj).last_motion)
def _get_uptime(obj: ProtectDeviceModel) -> datetime | None:
if obj.up_since is None:
return None
@@ -523,7 +513,7 @@ LIGHT_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
key="motion_last_trip_time",
translation_key="last_motion_detected",
device_class=SensorDeviceClass.TIMESTAMP,
ufp_public_value_fn=_get_last_motion_public,
ufp_public_value="last_motion_dt",
entity_registry_enabled_default=False,
),
ProtectSensorEntityDescription(
homeassistant/components/unifiprotect/siren.py
+5 -51
View File
@@ -1,6 +1,5 @@
"""UniFi Protect siren platform (Public API)."""
from datetime import datetime
import logging
from typing import Any, override
@@ -12,14 +11,12 @@ from homeassistant.components.siren import (
SirenEntity,
SirenEntityFeature,
)
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.event import async_call_later
from homeassistant.util import dt as dt_util
from .const import DEFAULT_ATTRIBUTION, DEFAULT_BRAND, DOMAIN
from .data import ProtectData, UFPConfigEntry
@@ -89,7 +86,6 @@ class ProtectSiren(SirenEntity):
via_device_id=data.nvr_device_id,
)
self._siren_mac = siren.mac
self._cancel_scheduled_off: CALLBACK_TYPE | None = None
self._update_from_siren(siren)
@property
@@ -115,11 +111,9 @@ class ProtectSiren(SirenEntity):
The state is always re-read from the public bootstrap: the library
merges WS updates into it before dispatching, and ``None`` carries no
object to read.
object to read. A timed run ending is announced by the library as a
regular update.
"""
# Cancel any previous auto-off timer before scheduling a new one.
self._cancel_off_timer()
prev_state = (self._attr_available, self._attr_is_on)
if (siren := self._siren) is None:
@@ -129,38 +123,9 @@ class ProtectSiren(SirenEntity):
else:
self._update_from_siren(siren)
# The server never emits a WS message when a timed run expires, so
# we must schedule our own callback. Both activated_at and
# duration are in milliseconds in the WS payload.
status = siren.siren_status
if (
status.is_active
and status.activated_at is not None
and status.duration is not None
):
delay = (
status.activated_at + status.duration
) / 1000 - dt_util.utcnow().timestamp()
if delay <= 0:
# Already expired (e.g. stale bootstrap after a reconnect):
# override the is_active=True from the payload immediately
# so we never briefly write ON into the state machine.
self._attr_is_on = False
else:
self._cancel_scheduled_off = async_call_later(
self.hass, delay, self._async_scheduled_off
)
if (self._attr_available, self._attr_is_on) != prev_state:
self.async_write_ha_state()
@callback
def _async_scheduled_off(self, _now: datetime) -> None:
"""Timed siren run has expired — push state to OFF."""
self._cancel_scheduled_off = None
self._attr_is_on = False
self.async_write_ha_state()
@override
async def async_added_to_hass(self) -> None:
"""Subscribe to public WS updates dispatched by ProtectData."""
@@ -168,20 +133,10 @@ class ProtectSiren(SirenEntity):
self.async_on_remove(
self.data.async_subscribe_public(self._siren_mac, self._async_updated)
)
self.async_on_remove(self._cancel_off_timer)
# Refresh from the bootstrap: a WS update or delete that landed between
# entity construction and this subscription would otherwise be missed,
# and an already-active timed run needs its auto-off timer scheduled so
# a siren that was running when HA started does not remain stuck ON.
# entity construction and this subscription would otherwise be missed.
self._async_updated(None)
@callback
def _cancel_off_timer(self) -> None:
"""Cancel the pending auto-off timer if any."""
if self._cancel_scheduled_off is not None:
self._cancel_scheduled_off()
self._cancel_scheduled_off = None
@async_ufp_instance_command
@override
async def async_turn_on(self, **kwargs: Any) -> None:
@@ -235,7 +190,6 @@ class ProtectSiren(SirenEntity):
)
await siren.stop()
# The server does not emit a WS event after a manual stop, so we set
# the state optimistically and cancel any pending auto-off timer.
self._cancel_off_timer()
# the state optimistically.
self._attr_is_on = False
self.async_write_ha_state()
tests/components/unifiprotect/test_siren.py
+11 -199
View File
@@ -1,19 +1,11 @@
"""Tests for the UniFi Protect siren (Public API) entities."""
from collections.abc import Callable, Coroutine
from datetime import timedelta
from typing import Any
from unittest.mock import AsyncMock, Mock, PropertyMock, patch
import pytest
from uiprotect.data import (
DeviceState,
ModelType,
PublicSirenStatus,
Siren,
SirenDuration,
WSAction,
)
from uiprotect.data import DeviceState, ModelType, Siren, SirenDuration, WSAction
from uiprotect.exceptions import (
BadRequest,
ClientError,
@@ -44,7 +36,6 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.util import dt as dt_util
from .utils import (
MockUFPFixture,
@@ -53,8 +44,6 @@ from .utils import (
make_public_bootstrap,
)
from tests.common import async_fire_time_changed
SIREN_ID = "siren-id-1"
SIREN_MAC = "AA:BB:CC:DD:EE:02"
SIREN_NAME = "Garage Siren"
@@ -66,11 +55,6 @@ def _make_siren(
*, is_active: bool = False, state: DeviceState = DeviceState.CONNECTED
) -> Mock:
"""Build a mock :class:`Siren`."""
status = Mock(spec=PublicSirenStatus)
status.is_active = is_active
status.activated_at = None
status.duration = None
status.turn_off_at = None
siren = Mock(spec=Siren)
siren.id = SIREN_ID
siren.mac = SIREN_MAC
@@ -78,7 +62,6 @@ def _make_siren(
siren.model = ModelType.SIREN
siren.state = state
siren.volume = 50
siren.siren_status = status
siren.is_active = is_active
siren.play = AsyncMock()
siren.stop = AsyncMock()
@@ -408,7 +391,7 @@ async def test_siren_state_updates_from_public_ws(
ufp_with_siren: MockUFPFixture,
siren: Mock,
) -> None:
"""A public devices WS update for the siren refreshes the entity state."""
"""Public devices WS updates flip the entity on and back off."""
await init_entry(hass, ufp_with_siren, [])
state = hass.states.get(SIREN_ENTITY_ID)
@@ -426,6 +409,15 @@ async def test_siren_state_updates_from_public_ws(
assert state is not None
assert state.state == STATE_ON
# A timed run ending arrives the same way, as an update with the flag off.
siren.is_active = False
ufp_with_siren.devices_ws_subscription(_make_ws_msg(siren))
await hass.async_block_till_done()
state = hass.states.get(SIREN_ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
async def test_siren_ws_update_no_state_change(
hass: HomeAssistant,
@@ -477,149 +469,6 @@ async def test_siren_availability_follows_websocket_state(
assert state.state == STATE_OFF
async def test_siren_auto_off_after_timed_duration(
hass: HomeAssistant,
ufp_with_siren: MockUFPFixture,
siren: Mock,
) -> None:
"""State flips to OFF automatically when a timed duration expires.
The public devices WS never sends an 'off' event for timed runs, so the
entity must schedule its own callback via async_call_later.
"""
await init_entry(hass, ufp_with_siren, [])
state = hass.states.get(SIREN_ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
# Simulate a WS update: siren becomes active for 10 seconds.
now = dt_util.utcnow()
active_status = Mock(spec=PublicSirenStatus)
active_status.is_active = True
active_status.activated_at = int(now.timestamp() * 1000)
active_status.duration = 10000
active_status.turn_off_at = (
None # implementation uses activated_at+duration directly
)
siren.is_active = True
siren.siren_status = active_status
mock_msg = _make_ws_msg(siren)
assert ufp_with_siren.devices_ws_subscription is not None
ufp_with_siren.devices_ws_subscription(mock_msg)
await hass.async_block_till_done()
state = hass.states.get(SIREN_ENTITY_ID)
assert state is not None
assert state.state == STATE_ON
# Advance HA time past turn_off_at — the scheduled callback should fire.
async_fire_time_changed(hass, now + timedelta(seconds=11))
await hass.async_block_till_done()
state = hass.states.get(SIREN_ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
async def test_siren_turn_off_cancels_scheduled_timer(
hass: HomeAssistant,
ufp_with_siren: MockUFPFixture,
siren: Mock,
) -> None:
"""Manual turn_off cancels the pending auto-off timer.
When a timed run is active the entity holds a scheduled callback. A
manual turn_off must cancel that callback so the timer never fires and
the state stays OFF afterwards.
"""
await init_entry(hass, ufp_with_siren, [])
# Start a timed run — schedules an auto-off callback 30 s from now.
now = dt_util.utcnow()
active_status = Mock(spec=PublicSirenStatus)
active_status.is_active = True
active_status.activated_at = int(now.timestamp() * 1000)
active_status.duration = 30000 # 30 s — won't expire on its own
active_status.turn_off_at = None
siren.is_active = True
siren.siren_status = active_status
mock_msg = _make_ws_msg(siren)
assert ufp_with_siren.devices_ws_subscription is not None
ufp_with_siren.devices_ws_subscription(mock_msg)
await hass.async_block_till_done()
state = hass.states.get(SIREN_ENTITY_ID)
assert state is not None
assert state.state == STATE_ON
# Manually turn off — must cancel the scheduled timer.
await hass.services.async_call(
SIREN_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: SIREN_ENTITY_ID},
blocking=True,
)
state = hass.states.get(SIREN_ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
# Advance time past the original timer — state must stay OFF.
async_fire_time_changed(hass, now + timedelta(seconds=35))
await hass.async_block_till_done()
state = hass.states.get(SIREN_ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
async def test_siren_auto_off_when_already_expired_at_update(
hass: HomeAssistant,
ufp_with_siren: MockUFPFixture,
siren: Mock,
) -> None:
"""State flips to OFF when a WS update arrives with an already-expired duration.
On reconnect, the public bootstrap may still report is_active=True with an
activated_at+duration that is already in the past. The entity must treat
delay<=0 as immediately expired and set its state to OFF immediately.
"""
await init_entry(hass, ufp_with_siren, [])
state = hass.states.get(SIREN_ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
# Build a status whose turn-off time is 5 seconds in the PAST.
now = dt_util.utcnow()
expired_activated_at = int((now.timestamp() - 15) * 1000) # 15 s ago
expired_status = Mock(spec=PublicSirenStatus)
expired_status.is_active = True
expired_status.activated_at = expired_activated_at
expired_status.duration = 10000 # 10 s → expired 5 s ago
expired_status.turn_off_at = None
siren.is_active = True
siren.siren_status = expired_status
mock_msg = _make_ws_msg(siren)
assert ufp_with_siren.devices_ws_subscription is not None
ufp_with_siren.devices_ws_subscription(mock_msg)
await hass.async_block_till_done()
# Entity stays OFF: delay<=0 overrides is_active=True inline, so the state
# machine never sees ON.
state = hass.states.get(SIREN_ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
@pytest.mark.parametrize(
"state",
[DeviceState.DISCONNECTED, DeviceState.CONNECTING, DeviceState.UNKNOWN],
@@ -692,43 +541,6 @@ async def test_siren_unavailable_on_delete_event(
assert state.state == STATE_UNAVAILABLE
async def test_siren_auto_off_timer_scheduled_at_startup(
hass: HomeAssistant,
ufp_with_siren: MockUFPFixture,
siren: Mock,
) -> None:
"""Auto-off timer is scheduled for an already-active siren.
If a timed run is already in progress when HA starts, the entity must
schedule its own auto-off callback immediately (not wait for a WS update)
so the siren does not remain stuck ON after the run expires.
"""
# Configure the siren as already active with 10 s remaining.
now = dt_util.utcnow()
active_status = Mock(spec=PublicSirenStatus)
active_status.is_active = True
active_status.activated_at = int(now.timestamp() * 1000)
active_status.duration = 10000
active_status.turn_off_at = None
siren.is_active = True
siren.siren_status = active_status
await init_entry(hass, ufp_with_siren, [])
state = hass.states.get(SIREN_ENTITY_ID)
assert state is not None
assert state.state == STATE_ON
# Advance HA time past the expiry — the startup-scheduled timer must fire.
async_fire_time_changed(hass, now + timedelta(seconds=11))
await hass.async_block_till_done()
state = hass.states.get(SIREN_ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
@pytest.fixture(name="setup_hybrid")
def setup_hybrid_fixture(
hass: HomeAssistant, ufp: MockUFPFixture
tests/components/unifiprotect/utils.py
+2
View File
@@ -557,6 +557,7 @@ def make_public_light(
lds.pir_sensitivity if pir_sensitivity is None else pir_sensitivity
),
)
public.last_motion_dt = PublicLight.last_motion_dt.fget(public)
return public
@@ -677,6 +678,7 @@ def make_public_camera(
if hdr_type is None
else hdr_type
)
public.hdr_mode_display = PublicCamera.hdr_mode_display.fget(public)
flags = camera.feature_flags
public.has_package_camera = flags.has_package_camera
# Spec'd so a private-only flag reads as absent.