Add number entities for velux opening device limitations (#174388)

This commit is contained in:
wollew
2026-07-09 10:35:41 +02:00
committed by GitHub
parent 3d8285dccd
commit 1e5cc1e307
10 changed files with 791 additions and 20 deletions
+2 -2
View File
@@ -2,7 +2,7 @@
import dataclasses
from pyvlx import PyVLX, PyVLXException, Window
from pyvlx import OpeningDevice, PyVLX, PyVLXException
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
@@ -75,7 +75,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: VeluxConfigEntry) -> boo
limitation_coordinators: dict[int, VeluxLimitationCoordinator] = {}
for node in pyvlx.nodes:
if isinstance(node, Window) and node.rain_sensor:
if isinstance(node, OpeningDevice):
coordinator = VeluxLimitationCoordinator(hass, entry, node)
# do not await coordinator.async_config_entry_first_refresh() here to avoid doing
# it for disabled entities, the entities will call it when they are added to hass
@@ -60,7 +60,8 @@ class VeluxRainSensor(
"""Called when the entity is added to Home Assistant."""
await super().async_added_to_hass()
# Get initial state as we didn't do it on coordinator initialization to avoid doing it for disabled entities
await self.coordinator.async_request_refresh()
if self.coordinator.data is None:
await self.coordinator.async_request_refresh()
@property
@override
@@ -69,8 +70,8 @@ class VeluxRainSensor(
# Velux windows with rain sensors report an opening
# limitation when rain is detected. So far we've
# seen 89, 91, 93 (most cases) or 100 (Velux GPU).
# It probably makes sense to
# assume that any large enough limitation (we use >=89) means rain is detected.
# It probably makes sense to assume that any large
# enough limitation (we use >=89) means rain is detected.
# Documentation on this is non-existent AFAIK.
if self.coordinator.data is None:
return None
+10 -2
View File
@@ -21,6 +21,7 @@ class VeluxLimitationData:
"""Data for one opening device's limitations."""
limitation_min: Position
limitation_max: Position
class VeluxLimitationCoordinator(DataUpdateCoordinator[VeluxLimitationData | None]):
@@ -44,9 +45,16 @@ class VeluxLimitationCoordinator(DataUpdateCoordinator[VeluxLimitationData | Non
@override
async def _async_update_data(self) -> VeluxLimitationData:
"""Fetch limitation min data from the device."""
"""Fetch limitation min and max from the device."""
try:
min_pos = await self.node.get_limitation_min()
max_pos = await self.node.get_limitation_max()
LOGGER.debug(
"Fetched limitations for %s: pyvlx_min=%s%% pyvlx_max=%s%%",
self.node.name,
min_pos.position_percent,
max_pos.position_percent,
)
except (OSError, PyVLXException) as err:
raise UpdateFailed(f"Error fetching limitations: {err}") from err
return VeluxLimitationData(limitation_min=min_pos)
return VeluxLimitationData(limitation_min=min_pos, limitation_max=max_pos)
-1
View File
@@ -68,7 +68,6 @@ class VeluxEntity(Entity):
def __init__(self, node: Node, config_entry_id: str) -> None:
"""Initialize the Velux device."""
self.node = node
self._attr_unique_id = velux_unique_id(node, config_entry_id)
self._attr_device_info = velux_device_info(node, config_entry_id)
+193 -7
View File
@@ -1,16 +1,24 @@
"""Support for Velux exterior heating number entities."""
"""Support for Velux exterior heating and cover open/closed number entities."""
from dataclasses import replace
from typing import override
from pyvlx import ExteriorHeating, Intensity
from pyvlx import ExteriorHeating, Intensity, OpeningDevice, Position
from homeassistant.components.number import NumberEntity
from homeassistant.const import PERCENTAGE
from homeassistant.components.number import NumberEntity, NumberMode
from homeassistant.const import PERCENTAGE, EntityCategory, UnitOfRatio
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import VeluxConfigEntry
from .entity import VeluxEntity, wrap_pyvlx_call_exceptions
from .coordinator import VeluxLimitationCoordinator
from .entity import (
VeluxEntity,
velux_device_info,
velux_unique_id,
wrap_pyvlx_call_exceptions,
)
PARALLEL_UPDATES = 1
@@ -22,11 +30,22 @@ async def async_setup_entry(
) -> None:
"""Set up number entities for the Velux platform."""
pyvlx = config_entry.runtime_data.pyvlx
async_add_entities(
limitation_coordinators = config_entry.runtime_data.limitation_coordinators
entities: list[NumberEntity] = [
VeluxExteriorHeatingNumber(node, config_entry.entry_id)
for node in pyvlx.nodes
if isinstance(node, ExteriorHeating)
)
]
for node in pyvlx.nodes:
if isinstance(node, OpeningDevice):
coordinator = limitation_coordinators[node.node_id]
entities.extend(
[
VeluxOpenPositionLimitNumber(coordinator, config_entry.entry_id),
VeluxClosedPositionLimitNumber(coordinator, config_entry.entry_id),
]
)
async_add_entities(entities)
class VeluxExteriorHeatingNumber(VeluxEntity, NumberEntity):
@@ -56,3 +75,170 @@ class VeluxExteriorHeatingNumber(VeluxEntity, NumberEntity):
Intensity(intensity_percent=round(value)),
wait_for_completion=True,
)
class VeluxPositionLimitNumber(
CoordinatorEntity[VeluxLimitationCoordinator], NumberEntity
):
"""Shared behavior for Velux limitation number entities.
Home Assistant expresses cover position as opening percentage, while pyvlx
uses the opposite direction. These entities expose HA-side open/closed
position limits and convert to pyvlx positions only at the API boundary.
"""
_attr_entity_category = EntityCategory.CONFIG
_attr_entity_registry_enabled_default = False
_attr_mode = NumberMode.BOX
_attr_native_step = 1
_attr_native_unit_of_measurement = UnitOfRatio.PERCENTAGE
_attr_has_entity_name = True
_limitation_kind: str
def __init__(
self, coordinator: VeluxLimitationCoordinator, config_entry_id: str
) -> None:
"""Initialize Velux limitation number."""
super().__init__(coordinator)
node = coordinator.node
unique_id = velux_unique_id(node, config_entry_id)
self._attr_unique_id = f"{unique_id}_{self._limitation_kind}_limitation"
self._attr_translation_key = f"{self._limitation_kind}_position_limitation"
self._attr_device_info = velux_device_info(node, config_entry_id)
@override
async def async_added_to_hass(self) -> None:
"""Request an immediate refresh when the entity is first added."""
await super().async_added_to_hass()
# Get initial state as we didn't do it on coordinator initialization to avoid doing it for disabled entities
if self.coordinator.data is None:
await self.coordinator.async_request_refresh()
@property
@override
def available(self) -> bool:
"""Return False until coordinator has successfully populated data.
The entity is only available once the coordinator has successfully
fetched data at least once.
"""
if self.coordinator.data is None:
return False
return super().available
@property
@override
def native_value(self) -> float | None:
"""Return the current limitation in Home Assistant semantics."""
if position := self._get_pyvlx_limit():
return 100 - position.position_percent
return None
@wrap_pyvlx_call_exceptions
@override
async def async_set_native_value(self, value: float) -> None:
"""Set the limitation in Home Assistant semantics."""
# this will only be called if the entity is available, so coordinator.data is not None
await self._async_set_pyvlx_limitation(
Position(position_percent=100 - round(value))
)
def _get_pyvlx_limit(self) -> Position | None:
"""Get the pyvlx limitation backing this HA-side entity."""
raise NotImplementedError
def _updated_pyvlx_limits(
self, updated_position: Position, current_min: Position, current_max: Position
) -> tuple[Position, Position]:
"""Return pyvlx min/max values with this entity's side updated."""
raise NotImplementedError
async def _async_set_pyvlx_limitation(self, position: Position) -> None:
"""Set pyvlx limitations while preserving the unchanged side."""
assert self.coordinator.data is not None # checked in async_set_native_value
current_min = self.coordinator.data.limitation_min
current_max = self.coordinator.data.limitation_max
position_min, position_max = self._updated_pyvlx_limits(
position, current_min, current_max
)
await self.coordinator.node.set_position_limitations(
position_min=position_min,
position_max=position_max,
)
self.coordinator.async_set_updated_data(
replace(
self.coordinator.data,
limitation_min=position_min,
limitation_max=position_max,
)
)
class VeluxClosedPositionLimitNumber(VeluxPositionLimitNumber):
"""Representation of the closed position limit."""
_attr_native_min_value = 0
_limitation_kind = "closed"
def _sibling_value(self) -> float | None:
"""Return the sibling open limit value, or None if unknown."""
return (
100 - self.coordinator.data.limitation_min.position_percent
if self.coordinator.data
else None
)
@property
@override
def native_max_value(self) -> float:
"""Return the upper bound: the current open limit (or 100 if unknown)."""
sibling_value = self._sibling_value()
return sibling_value if sibling_value is not None else 100
@override
def _get_pyvlx_limit(self) -> Position | None:
"""Get the pyvlx max limit backing the HA closed position limit."""
return self.coordinator.data.limitation_max if self.coordinator.data else None
@override
def _updated_pyvlx_limits(
self, updated_position: Position, current_min: Position, current_max: Position
) -> tuple[Position, Position]:
"""Update pyvlx max and preserve pyvlx min for HA closed limit changes."""
return current_min, updated_position
class VeluxOpenPositionLimitNumber(VeluxPositionLimitNumber):
"""Representation of the open position limit."""
_attr_native_max_value = 100
_limitation_kind = "open"
def _sibling_value(self) -> float | None:
"""Return the sibling close limit value, or None if unknown."""
return (
100 - self.coordinator.data.limitation_max.position_percent
if self.coordinator.data
else None
)
@property
@override
def native_min_value(self) -> float:
"""Return the lower bound: the current closed limit (or 0 if unknown)."""
sibling_value = self._sibling_value()
return sibling_value if sibling_value is not None else 0
@override
def _get_pyvlx_limit(self) -> Position | None:
"""Get the pyvlx min limit backing the HA open position limit."""
return self.coordinator.data.limitation_min if self.coordinator.data else None
@override
def _updated_pyvlx_limits(
self, updated_position: Position, current_min: Position, current_max: Position
) -> tuple[Position, Position]:
"""Update pyvlx min and preserve pyvlx max for HA open limit changes."""
return updated_position, current_max
@@ -53,6 +53,14 @@
"dual_roller_shutter_upper": {
"name": "Upper shutter"
}
},
"number": {
"closed_position_limitation": {
"name": "Closed position limit"
},
"open_position_limitation": {
"name": "Open position limit"
}
}
},
"exceptions": {
+11
View File
@@ -72,6 +72,8 @@ def mock_window() -> AsyncMock:
window.rain_sensor = True
window.serial_number = "123456789"
window.get_limitation_min.return_value = MagicMock(position_percent=0)
window.get_limitation_max.return_value = MagicMock(position_percent=100)
window.set_position_limitations = AsyncMock()
window.device_updated_cbs = []
window.is_opening = False
window.is_closing = False
@@ -98,6 +100,9 @@ def mock_dual_roller_shutter() -> AsyncMock:
position_percent=30, closed=False, known=True
)
cover.position = MagicMock(position_percent=30, closed=False, known=True)
cover.get_limitation_min.return_value = MagicMock(position_percent=0)
cover.get_limitation_max.return_value = MagicMock(position_percent=100)
cover.set_position_limitations = AsyncMock()
cover.pyvlx = MagicMock()
return cover
@@ -120,6 +125,9 @@ def mock_blind() -> AsyncMock:
blind.close_orientation = AsyncMock()
blind.stop_orientation = AsyncMock()
blind.set_orientation = AsyncMock()
blind.get_limitation_min.return_value = MagicMock(position_percent=0)
blind.get_limitation_max.return_value = MagicMock(position_percent=100)
blind.set_position_limitations = AsyncMock()
blind.pyvlx = MagicMock()
return blind
@@ -191,6 +199,9 @@ def mock_cover_type(request: pytest.FixtureRequest) -> AsyncMock:
cover.position_lower_curtain = MagicMock(
position_percent=30, closed=False, known=True
)
cover.get_limitation_min.return_value = MagicMock(position_percent=0)
cover.get_limitation_max.return_value = MagicMock(position_percent=100)
cover.set_position_limitations = AsyncMock()
cover.pyvlx = MagicMock()
return cover
@@ -1,4 +1,244 @@
# serializer version: 1
# name: test_number_setup[number.test_blind_closed_position_limit-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<NumberEntityCapabilityAttribute.MAX: 'max'>: 100,
<NumberEntityCapabilityAttribute.MIN: 'min'>: 0,
<NumberEntityCapabilityAttribute.MODE: 'mode'>: <NumberMode.BOX: 'box'>,
<NumberEntityCapabilityAttribute.STEP: 'step'>: 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.test_blind_closed_position_limit',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Closed position limit',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Closed position limit',
'platform': 'velux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'closed_position_limitation',
'unique_id': '4711_closed_limitation',
'unit_of_measurement': <UnitOfRatio.PERCENTAGE: '%'>,
})
# ---
# name: test_number_setup[number.test_blind_closed_position_limit-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Test Blind Closed position limit',
<NumberEntityCapabilityAttribute.MAX: 'max'>: 100,
<NumberEntityCapabilityAttribute.MIN: 'min'>: 0,
<NumberEntityCapabilityAttribute.MODE: 'mode'>: <NumberMode.BOX: 'box'>,
<NumberEntityCapabilityAttribute.STEP: 'step'>: 1,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfRatio.PERCENTAGE: '%'>,
}),
'context': <ANY>,
'entity_id': 'number.test_blind_closed_position_limit',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_number_setup[number.test_blind_open_position_limit-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<NumberEntityCapabilityAttribute.MAX: 'max'>: 100,
<NumberEntityCapabilityAttribute.MIN: 'min'>: 0,
<NumberEntityCapabilityAttribute.MODE: 'mode'>: <NumberMode.BOX: 'box'>,
<NumberEntityCapabilityAttribute.STEP: 'step'>: 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.test_blind_open_position_limit',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Open position limit',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Open position limit',
'platform': 'velux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'open_position_limitation',
'unique_id': '4711_open_limitation',
'unit_of_measurement': <UnitOfRatio.PERCENTAGE: '%'>,
})
# ---
# name: test_number_setup[number.test_blind_open_position_limit-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Test Blind Open position limit',
<NumberEntityCapabilityAttribute.MAX: 'max'>: 100,
<NumberEntityCapabilityAttribute.MIN: 'min'>: 0,
<NumberEntityCapabilityAttribute.MODE: 'mode'>: <NumberMode.BOX: 'box'>,
<NumberEntityCapabilityAttribute.STEP: 'step'>: 1,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfRatio.PERCENTAGE: '%'>,
}),
'context': <ANY>,
'entity_id': 'number.test_blind_open_position_limit',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '100',
})
# ---
# name: test_number_setup[number.test_dual_roller_shutter_closed_position_limit-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<NumberEntityCapabilityAttribute.MAX: 'max'>: 100,
<NumberEntityCapabilityAttribute.MIN: 'min'>: 0,
<NumberEntityCapabilityAttribute.MODE: 'mode'>: <NumberMode.BOX: 'box'>,
<NumberEntityCapabilityAttribute.STEP: 'step'>: 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.test_dual_roller_shutter_closed_position_limit',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Closed position limit',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Closed position limit',
'platform': 'velux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'closed_position_limitation',
'unique_id': '987654321_closed_limitation',
'unit_of_measurement': <UnitOfRatio.PERCENTAGE: '%'>,
})
# ---
# name: test_number_setup[number.test_dual_roller_shutter_closed_position_limit-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Test Dual Roller Shutter Closed position limit',
<NumberEntityCapabilityAttribute.MAX: 'max'>: 100,
<NumberEntityCapabilityAttribute.MIN: 'min'>: 0,
<NumberEntityCapabilityAttribute.MODE: 'mode'>: <NumberMode.BOX: 'box'>,
<NumberEntityCapabilityAttribute.STEP: 'step'>: 1,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfRatio.PERCENTAGE: '%'>,
}),
'context': <ANY>,
'entity_id': 'number.test_dual_roller_shutter_closed_position_limit',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_number_setup[number.test_dual_roller_shutter_open_position_limit-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<NumberEntityCapabilityAttribute.MAX: 'max'>: 100,
<NumberEntityCapabilityAttribute.MIN: 'min'>: 0,
<NumberEntityCapabilityAttribute.MODE: 'mode'>: <NumberMode.BOX: 'box'>,
<NumberEntityCapabilityAttribute.STEP: 'step'>: 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.test_dual_roller_shutter_open_position_limit',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Open position limit',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Open position limit',
'platform': 'velux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'open_position_limitation',
'unique_id': '987654321_open_limitation',
'unit_of_measurement': <UnitOfRatio.PERCENTAGE: '%'>,
})
# ---
# name: test_number_setup[number.test_dual_roller_shutter_open_position_limit-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Test Dual Roller Shutter Open position limit',
<NumberEntityCapabilityAttribute.MAX: 'max'>: 100,
<NumberEntityCapabilityAttribute.MIN: 'min'>: 0,
<NumberEntityCapabilityAttribute.MODE: 'mode'>: <NumberMode.BOX: 'box'>,
<NumberEntityCapabilityAttribute.STEP: 'step'>: 1,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfRatio.PERCENTAGE: '%'>,
}),
'context': <ANY>,
'entity_id': 'number.test_dual_roller_shutter_open_position_limit',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '100',
})
# ---
# name: test_number_setup[number.test_exterior_heating-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -59,3 +299,123 @@
'state': '33',
})
# ---
# name: test_number_setup[number.test_window_closed_position_limit-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<NumberEntityCapabilityAttribute.MAX: 'max'>: 100,
<NumberEntityCapabilityAttribute.MIN: 'min'>: 0,
<NumberEntityCapabilityAttribute.MODE: 'mode'>: <NumberMode.BOX: 'box'>,
<NumberEntityCapabilityAttribute.STEP: 'step'>: 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.test_window_closed_position_limit',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Closed position limit',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Closed position limit',
'platform': 'velux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'closed_position_limitation',
'unique_id': '123456789_closed_limitation',
'unit_of_measurement': <UnitOfRatio.PERCENTAGE: '%'>,
})
# ---
# name: test_number_setup[number.test_window_closed_position_limit-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Test Window Closed position limit',
<NumberEntityCapabilityAttribute.MAX: 'max'>: 100,
<NumberEntityCapabilityAttribute.MIN: 'min'>: 0,
<NumberEntityCapabilityAttribute.MODE: 'mode'>: <NumberMode.BOX: 'box'>,
<NumberEntityCapabilityAttribute.STEP: 'step'>: 1,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfRatio.PERCENTAGE: '%'>,
}),
'context': <ANY>,
'entity_id': 'number.test_window_closed_position_limit',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_number_setup[number.test_window_open_position_limit-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<NumberEntityCapabilityAttribute.MAX: 'max'>: 100,
<NumberEntityCapabilityAttribute.MIN: 'min'>: 0,
<NumberEntityCapabilityAttribute.MODE: 'mode'>: <NumberMode.BOX: 'box'>,
<NumberEntityCapabilityAttribute.STEP: 'step'>: 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.test_window_open_position_limit',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Open position limit',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Open position limit',
'platform': 'velux',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'open_position_limitation',
'unique_id': '123456789_open_limitation',
'unit_of_measurement': <UnitOfRatio.PERCENTAGE: '%'>,
})
# ---
# name: test_number_setup[number.test_window_open_position_limit-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Test Window Open position limit',
<NumberEntityCapabilityAttribute.MAX: 'max'>: 100,
<NumberEntityCapabilityAttribute.MIN: 'min'>: 0,
<NumberEntityCapabilityAttribute.MODE: 'mode'>: <NumberMode.BOX: 'box'>,
<NumberEntityCapabilityAttribute.STEP: 'step'>: 1,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfRatio.PERCENTAGE: '%'>,
}),
'context': <ANY>,
'entity_id': 'number.test_window_open_position_limit',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '100',
})
# ---
@@ -133,6 +133,7 @@ async def test_rain_sensor_unavailability(
# Simulate communication error
mock_window.get_limitation_min.side_effect = PyVLXException("Connection failed")
mock_window.get_limitation_max.side_effect = PyVLXException("Connection failed")
await update_polled_entities(hass, freezer)
# Entity should now be unavailable
@@ -143,6 +144,7 @@ async def test_rain_sensor_unavailability(
# Simulate recovery
mock_window.get_limitation_min.side_effect = None
mock_window.get_limitation_min.return_value.position_percent = 0
mock_window.get_limitation_max.side_effect = None
await update_polled_entities(hass, freezer)
# Entity should be available again
state = hass.states.get(test_entity_id)
+201 -5
View File
@@ -1,9 +1,11 @@
"""Test Velux number entities."""
from unittest.mock import AsyncMock
from unittest.mock import AsyncMock, MagicMock
from freezegun.api import FrozenDateTimeFactory
import pytest
from pyvlx import Intensity
from pyvlx import Intensity, PyVLXException
from pyvlx.opening_device import Position
from homeassistant.components.number import (
ATTR_VALUE,
@@ -11,12 +13,12 @@ from homeassistant.components.number import (
SERVICE_SET_VALUE,
)
from homeassistant.components.velux.const import DOMAIN
from homeassistant.const import STATE_UNKNOWN, Platform
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from . import update_callback_entity
from . import update_callback_entity, update_polled_entities
from tests.common import MockConfigEntry, SnapshotAssertion, snapshot_platform
@@ -34,6 +36,7 @@ def get_number_entity_id(mock: AsyncMock) -> str:
return f"number.{mock.name.lower().replace(' ', '_')}"
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_number_setup(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
@@ -49,7 +52,7 @@ async def test_number_setup(
)
async def test_number_device_association(
async def test_heating_entity_number_device_association(
hass: HomeAssistant,
mock_exterior_heating: AsyncMock,
entity_registry: er.EntityRegistry,
@@ -127,3 +130,196 @@ async def test_set_invalid_value_fails(
)
mock_exterior_heating.set_intensity.assert_not_awaited()
def closed_limit_entity_id(mock: AsyncMock) -> str:
"""Return entity ID of the closed position limit entity."""
return f"number.{mock.name.lower().replace(' ', '_')}_closed_position_limit"
def open_limit_entity_id(mock: AsyncMock) -> str:
"""Return entity ID of the open position limit entity."""
return f"number.{mock.name.lower().replace(' ', '_')}_open_position_limit"
async def test_limitation_entity_number_device_association(
hass: HomeAssistant,
mock_window: AsyncMock,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Ensure limitation number entity is associated with a device."""
entity_id = closed_limit_entity_id(mock_window)
entry = entity_registry.async_get(entity_id)
assert entry is not None
assert entry.device_id is not None
device_entry = device_registry.async_get(entry.device_id)
assert device_entry is not None
assert (DOMAIN, mock_window.serial_number) in device_entry.identifiers
@pytest.mark.parametrize("mock_pyvlx", ["mock_window"], indirect=True)
async def test_limitation_entities_created(
hass: HomeAssistant,
mock_window: AsyncMock,
entity_registry: er.EntityRegistry,
) -> None:
"""Open and closed position limit entities are created disabled by default."""
for get_entity_id in (closed_limit_entity_id, open_limit_entity_id):
entry = entity_registry.async_get(get_entity_id(mock_window))
assert entry is not None
assert entry.disabled_by == er.RegistryEntryDisabler.INTEGRATION
@pytest.mark.parametrize("mock_pyvlx", ["mock_window"], indirect=True)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_limitation_entities_enabled_state(
hass: HomeAssistant,
mock_window: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""After enabling, open/closed limit entities reflect HA-side opening semantics."""
# HA minimum opening comes from pyvlx max, HA maximum opening comes from pyvlx min.
mock_window.get_limitation_min.return_value = MagicMock(position_percent=0)
mock_window.get_limitation_max.return_value = MagicMock(position_percent=100)
await update_polled_entities(hass, freezer)
assert hass.states.get(closed_limit_entity_id(mock_window)).state == "0"
assert hass.states.get(open_limit_entity_id(mock_window)).state == "100"
mock_window.get_limitation_min.return_value = MagicMock(position_percent=50)
mock_window.get_limitation_max.return_value = MagicMock(position_percent=30)
await update_polled_entities(hass, freezer)
assert hass.states.get(closed_limit_entity_id(mock_window)).state == "70"
assert hass.states.get(open_limit_entity_id(mock_window)).state == "50"
@pytest.mark.parametrize("mock_pyvlx", ["mock_window"], indirect=True)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_limitation_entity_bounds_follow_sibling_value(
hass: HomeAssistant,
mock_window: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Reported min/max bounds track the sibling limitation value."""
mock_window.get_limitation_min.return_value = MagicMock(position_percent=20)
mock_window.get_limitation_max.return_value = MagicMock(position_percent=70)
await update_polled_entities(hass, freezer)
closed_state = hass.states.get(closed_limit_entity_id(mock_window))
open_state = hass.states.get(open_limit_entity_id(mock_window))
assert closed_state is not None
assert open_state is not None
assert closed_state.attributes["max"] == 80
assert open_state.attributes["min"] == 30
mock_window.get_limitation_min.return_value = MagicMock(position_percent=40)
mock_window.get_limitation_max.return_value = MagicMock(position_percent=90)
await update_polled_entities(hass, freezer)
closed_state = hass.states.get(closed_limit_entity_id(mock_window))
open_state = hass.states.get(open_limit_entity_id(mock_window))
assert closed_state is not None
assert open_state is not None
assert closed_state.attributes["max"] == 60
assert open_state.attributes["min"] == 10
@pytest.mark.parametrize("mock_pyvlx", ["mock_window"], indirect=True)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_set_min_limitation(
hass: HomeAssistant,
mock_window: AsyncMock,
) -> None:
"""Setting HA minimum opening updates pyvlx max and preserves pyvlx min."""
entity_id = closed_limit_entity_id(mock_window)
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,
{ATTR_VALUE: 40, "entity_id": entity_id},
blocking=True,
)
mock_window.set_position_limitations.assert_awaited_once_with(
position_min=mock_window.get_limitation_min.return_value,
position_max=Position(position_percent=60),
)
@pytest.mark.parametrize("mock_pyvlx", ["mock_window"], indirect=True)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_set_limitation_updates_state_optimistically(
hass: HomeAssistant,
mock_window: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Setting a limitation updates the entity state before the next refresh."""
mock_window.get_limitation_min.return_value = MagicMock(position_percent=0)
mock_window.get_limitation_max.return_value = MagicMock(position_percent=100)
await update_polled_entities(hass, freezer)
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,
{ATTR_VALUE: 40, "entity_id": closed_limit_entity_id(mock_window)},
blocking=True,
)
assert hass.states.get(closed_limit_entity_id(mock_window)).state == "40"
@pytest.mark.parametrize("mock_pyvlx", ["mock_window"], indirect=True)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_set_max_limitation(
hass: HomeAssistant,
mock_window: AsyncMock,
) -> None:
"""Setting HA maximum opening updates pyvlx min and preserves pyvlx max."""
entity_id = open_limit_entity_id(mock_window)
await hass.services.async_call(
NUMBER_DOMAIN,
SERVICE_SET_VALUE,
{ATTR_VALUE: 70, "entity_id": entity_id},
blocking=True,
)
mock_window.set_position_limitations.assert_awaited_once_with(
position_min=Position(position_percent=30),
position_max=mock_window.get_limitation_max.return_value,
)
@pytest.mark.parametrize("mock_pyvlx", ["mock_window"], indirect=True)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_limitation_unavailable_on_error(
hass: HomeAssistant,
mock_window: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Entities become unavailable when pyvlx raises an exception."""
mock_window.get_limitation_min.side_effect = PyVLXException("Connection lost")
mock_window.get_limitation_max.side_effect = PyVLXException("Connection lost")
await update_polled_entities(hass, freezer)
assert (
hass.states.get(closed_limit_entity_id(mock_window)).state == STATE_UNAVAILABLE
)
assert hass.states.get(open_limit_entity_id(mock_window)).state == STATE_UNAVAILABLE
# Recovery
mock_window.get_limitation_min.side_effect = None
mock_window.get_limitation_min.return_value = MagicMock(position_percent=0)
mock_window.get_limitation_max.side_effect = None
mock_window.get_limitation_max.return_value = MagicMock(position_percent=0)
await update_polled_entities(hass, freezer)
assert (
hass.states.get(closed_limit_entity_id(mock_window)).state != STATE_UNAVAILABLE
)
assert hass.states.get(open_limit_entity_id(mock_window)).state != STATE_UNAVAILABLE