Compare commits

...

4 Commits

Author SHA1 Message Date
Paul Bottein
85336eadd6 Address copilot feedbacks 2026-02-05 14:08:09 +01:00
Paul Bottein
d037b2073d Add tests 2026-02-05 13:26:10 +01:00
Paul Bottein
ff3abb5b0b Refactor code 2026-02-05 12:28:40 +01:00
Paul Bottein
c014d32cac Group Overkiz sub-devices by placeOID 2026-02-05 12:23:09 +01:00
2 changed files with 262 additions and 8 deletions

View File

@@ -58,18 +58,95 @@ class OverkizEntity(CoordinatorEntity[OverkizDataUpdateCoordinator]):
"""Return Overkiz device linked to this entity."""
return self.coordinator.data[self.device_url]
def _get_sibling_devices(self) -> list[Device]:
"""Return sibling devices sharing the same base device URL."""
prefix = f"{self.base_device_url}#"
return [
device
for device in self.coordinator.data.values()
if device.device_url != self.device_url
and device.device_url.startswith(prefix)
]
def _has_siblings_with_different_place_oid(self) -> bool:
"""Check if sibling devices have different placeOIDs.
Returns True if siblings have different place_oid values, indicating
devices should be grouped by placeOID rather than by base URL.
"""
my_place_oid = self.device.place_oid
if not my_place_oid:
return False
return any(
sibling.place_oid and sibling.place_oid != my_place_oid
for sibling in self._get_sibling_devices()
)
def _get_device_index(self, device_url: str) -> int | None:
"""Extract numeric index from device URL (e.g., 'io://gw/123#4' -> 4)."""
suffix = device_url.split("#")[-1]
return int(suffix) if suffix.isdigit() else None
def _is_main_device_for_place_oid(self) -> bool:
"""Check if this device is the main device for its placeOID group.
The device with the lowest URL index among siblings sharing the same
placeOID is considered the main device and provides full device info.
"""
my_place_oid = self.device.place_oid
if not my_place_oid:
return True
my_index = self._get_device_index(self.device_url)
if my_index is None:
return True
return not any(
(sibling_index := self._get_device_index(sibling.device_url)) is not None
and sibling_index < my_index
for sibling in self._get_sibling_devices()
if sibling.place_oid == my_place_oid
)
def _get_via_device_id(self, use_place_oid_grouping: bool) -> str:
"""Return the via_device identifier for device registry hierarchy.
Sub-devices link to the main actuator (#1 device) when using placeOID
grouping, otherwise they link directly to the gateway.
"""
gateway_id = self.executor.get_gateway_id()
if not use_place_oid_grouping or self.device_url.endswith("#1"):
return gateway_id
main_device = self.coordinator.data.get(f"{self.base_device_url}#1")
if main_device and main_device.place_oid:
return f"{self.base_device_url}#{main_device.place_oid}"
return gateway_id
def generate_device_info(self) -> DeviceInfo:
"""Return device registry information for this entity."""
# Some devices, such as the Smart Thermostat have several devices
# in one physical device, with same device url, terminated by '#' and a number.
# In this case, we use the base device url as the device identifier.
if self.is_sub_device:
# Only return the url of the base device, to inherit device name
# and model from parent device.
# Some devices, such as the Smart Thermostat, have several sub-devices
# sharing the same base URL (terminated by '#' and a number).
use_place_oid_grouping = self._has_siblings_with_different_place_oid()
# Sub-devices without placeOID grouping inherit info from parent device
if self.is_sub_device and not use_place_oid_grouping:
return DeviceInfo(
identifiers={(DOMAIN, self.executor.base_device_url)},
)
# Determine identifier based on grouping strategy
if use_place_oid_grouping:
identifier = f"{self.base_device_url}#{self.device.place_oid}"
# Non-main devices only reference the identifier
if not self._is_main_device_for_place_oid():
return DeviceInfo(identifiers={(DOMAIN, identifier)})
else:
identifier = self.executor.base_device_url
manufacturer = (
self.executor.select_attribute(OverkizAttribute.CORE_MANUFACTURER)
or self.executor.select_state(OverkizState.CORE_MANUFACTURER_NAME)
@@ -92,7 +169,7 @@ class OverkizEntity(CoordinatorEntity[OverkizDataUpdateCoordinator]):
)
return DeviceInfo(
identifiers={(DOMAIN, self.executor.base_device_url)},
identifiers={(DOMAIN, identifier)},
name=self.device.label,
manufacturer=str(manufacturer),
model=str(model),
@@ -102,7 +179,7 @@ class OverkizEntity(CoordinatorEntity[OverkizDataUpdateCoordinator]):
),
hw_version=self.device.controllable_name,
suggested_area=suggested_area,
via_device=(DOMAIN, self.executor.get_gateway_id()),
via_device=(DOMAIN, self._get_via_device_id(use_place_oid_grouping)),
configuration_url=self.coordinator.client.server.configuration_url,
)

View File

@@ -0,0 +1,177 @@
"""Tests for Overkiz entity."""
from unittest.mock import Mock
import pytest
from homeassistant.components.overkiz.entity import OverkizEntity
def _create_mock_device(
device_url: str, place_oid: str | None, label: str = "Device"
) -> Mock:
"""Create a mock device with the given properties."""
device = Mock()
device.device_url = device_url
device.place_oid = place_oid
device.label = label
device.available = True
device.states = []
device.widget = Mock(value="TestWidget")
device.controllable_name = "test:Component"
return device
def _create_mock_entity(device: Mock, all_devices: list[Mock]) -> Mock:
"""Create a mock entity with the given device and coordinator data."""
entity = Mock(spec=OverkizEntity)
entity.device = device
entity.device_url = device.device_url
entity.base_device_url = device.device_url.split("#")[0]
entity.coordinator = Mock()
entity.coordinator.data = {d.device_url: d for d in all_devices}
prefix = f"{entity.base_device_url}#"
entity._get_sibling_devices = lambda: [
d
for d in all_devices
if d.device_url != device.device_url and d.device_url.startswith(prefix)
]
entity._get_device_index = lambda url: (
int(url.split("#")[-1]) if url.split("#")[-1].isdigit() else None
)
return entity
@pytest.mark.parametrize(
("place_oids", "expected"),
[
(["place-a", "place-b"], True),
(["place-a", "place-a"], False),
],
ids=["different_place_oids", "same_place_oids"],
)
def test_has_siblings_with_different_place_oid(
place_oids: list[str], expected: bool
) -> None:
"""Test detection of siblings with different placeOIDs."""
devices = [
_create_mock_device("io://gateway/123#1", place_oids[0], "Device 1"),
_create_mock_device("io://gateway/123#2", place_oids[1], "Device 2"),
]
entity = _create_mock_entity(devices[0], devices)
result = OverkizEntity._has_siblings_with_different_place_oid(entity)
assert result is expected
@pytest.mark.parametrize(
("device_index", "expected"),
[
(0, True),
(1, False),
],
ids=["lowest_index_is_main", "higher_index_not_main"],
)
def test_is_main_device_for_place_oid(device_index: int, expected: bool) -> None:
"""Test main device detection for placeOID group."""
devices = [
_create_mock_device("io://gateway/123#1", "place-a", "Device 1"),
_create_mock_device("io://gateway/123#4", "place-a", "Device 4"),
]
entity = _create_mock_entity(devices[device_index], devices)
result = OverkizEntity._is_main_device_for_place_oid(entity)
assert result is expected
def test_get_via_device_id_sub_device_links_to_main() -> None:
"""Test sub-device links to main actuator with placeOID grouping."""
devices = [
_create_mock_device("io://gateway/123#1", "place-a", "Actuator"),
_create_mock_device("io://gateway/123#2", "place-b", "Zone"),
]
entity = _create_mock_entity(devices[1], devices)
entity.executor = Mock()
entity.executor.get_gateway_id = Mock(return_value="gateway-id")
result = OverkizEntity._get_via_device_id(entity, use_place_oid_grouping=True)
assert result == "io://gateway/123#place-a"
def test_get_via_device_id_main_device_links_to_gateway() -> None:
"""Test main device (#1) links to gateway."""
devices = [
_create_mock_device("io://gateway/123#1", "place-a", "Actuator"),
]
entity = _create_mock_entity(devices[0], devices)
entity.executor = Mock()
entity.executor.get_gateway_id = Mock(return_value="gateway-id")
result = OverkizEntity._get_via_device_id(entity, use_place_oid_grouping=True)
assert result == "gateway-id"
def test_has_siblings_with_no_place_oid() -> None:
"""Test device with no placeOID returns False."""
devices = [
_create_mock_device("io://gateway/123#1", None, "Device 1"),
_create_mock_device("io://gateway/123#2", "place-b", "Device 2"),
]
entity = _create_mock_entity(devices[0], devices)
result = OverkizEntity._has_siblings_with_different_place_oid(entity)
assert result is False
def test_is_main_device_with_no_place_oid() -> None:
"""Test device with no placeOID is always considered main."""
devices = [
_create_mock_device("io://gateway/123#2", None, "Device 2"),
_create_mock_device("io://gateway/123#1", "place-a", "Device 1"),
]
entity = _create_mock_entity(devices[0], devices)
result = OverkizEntity._is_main_device_for_place_oid(entity)
assert result is True
def test_get_via_device_id_main_device_without_place_oid() -> None:
"""Test fallback to gateway when #1 device has no placeOID."""
devices = [
_create_mock_device("io://gateway/123#1", None, "Actuator"),
_create_mock_device("io://gateway/123#2", "place-b", "Zone"),
]
entity = _create_mock_entity(devices[1], devices)
entity.executor = Mock()
entity.executor.get_gateway_id = Mock(return_value="gateway-id")
result = OverkizEntity._get_via_device_id(entity, use_place_oid_grouping=True)
assert result == "gateway-id"
@pytest.mark.parametrize(
("device_url", "expected"),
[
("io://gateway/123#4", 4),
("io://gateway/123#10", 10),
("io://gateway/123#abc", None),
("io://gateway/123#", None),
],
ids=["single_digit", "multi_digit", "non_numeric", "empty_suffix"],
)
def test_get_device_index(device_url: str, expected: int | None) -> None:
"""Test extracting numeric index from device URL."""
device = _create_mock_device(device_url, "place-a")
entity = _create_mock_entity(device, [device])
result = OverkizEntity._get_device_index(entity, device_url)
assert result == expected