Add alarm and connectivity binary_sensors to myuplink (#111643)

* Add alarm and connectivity binary_sensors

* Get is_on for correct system

* Make coverage 100% in binary_sensor

* Address review comments

* Revert dict comprehension for now
This commit is contained in:
Åke Strandberg 2024-05-15 17:13:56 +02:00 committed by GitHub
parent 8eaf471dd2
commit c4c96be880
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 182 additions and 11 deletions

View file

@ -1,8 +1,9 @@
"""Binary sensors for myUplink.""" """Binary sensors for myUplink."""
from myuplink import DevicePoint from myuplink import DeviceConnectionState, DevicePoint
from homeassistant.components.binary_sensor import ( from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity, BinarySensorEntity,
BinarySensorEntityDescription, BinarySensorEntityDescription,
) )
@ -13,7 +14,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import MyUplinkDataCoordinator from . import MyUplinkDataCoordinator
from .const import DOMAIN from .const import DOMAIN
from .entity import MyUplinkEntity from .entity import MyUplinkEntity, MyUplinkSystemEntity
from .helpers import find_matching_platform from .helpers import find_matching_platform
CATEGORY_BASED_DESCRIPTIONS: dict[str, dict[str, BinarySensorEntityDescription]] = { CATEGORY_BASED_DESCRIPTIONS: dict[str, dict[str, BinarySensorEntityDescription]] = {
@ -25,6 +26,17 @@ CATEGORY_BASED_DESCRIPTIONS: dict[str, dict[str, BinarySensorEntityDescription]]
}, },
} }
CONNECTED_BINARY_SENSOR_DESCRIPTION = BinarySensorEntityDescription(
key="connected_state",
device_class=BinarySensorDeviceClass.CONNECTIVITY,
)
ALARM_BINARY_SENSOR_DESCRIPTION = BinarySensorEntityDescription(
key="has_alarm",
device_class=BinarySensorDeviceClass.PROBLEM,
translation_key="alarm",
)
def get_description(device_point: DevicePoint) -> BinarySensorEntityDescription | None: def get_description(device_point: DevicePoint) -> BinarySensorEntityDescription | None:
"""Get description for a device point. """Get description for a device point.
@ -46,7 +58,7 @@ async def async_setup_entry(
entities: list[BinarySensorEntity] = [] entities: list[BinarySensorEntity] = []
coordinator: MyUplinkDataCoordinator = hass.data[DOMAIN][config_entry.entry_id] coordinator: MyUplinkDataCoordinator = hass.data[DOMAIN][config_entry.entry_id]
# Setup device point sensors # Setup device point bound sensors
for device_id, point_data in coordinator.data.points.items(): for device_id, point_data in coordinator.data.points.items():
for point_id, device_point in point_data.items(): for point_id, device_point in point_data.items():
if find_matching_platform(device_point) == Platform.BINARY_SENSOR: if find_matching_platform(device_point) == Platform.BINARY_SENSOR:
@ -61,11 +73,37 @@ async def async_setup_entry(
unique_id_suffix=point_id, unique_id_suffix=point_id,
) )
) )
# Setup device bound sensors
entities.extend(
MyUplinkDeviceBinarySensor(
coordinator=coordinator,
device_id=device.id,
entity_description=CONNECTED_BINARY_SENSOR_DESCRIPTION,
unique_id_suffix="connection_state",
)
for system in coordinator.data.systems
for device in system.devices
)
# Setup system bound sensors
for system in coordinator.data.systems:
device_id = system.devices[0].id
entities.append(
MyUplinkSystemBinarySensor(
coordinator=coordinator,
device_id=device_id,
system_id=system.id,
entity_description=ALARM_BINARY_SENSOR_DESCRIPTION,
unique_id_suffix="has_alarm",
)
)
async_add_entities(entities) async_add_entities(entities)
class MyUplinkDevicePointBinarySensor(MyUplinkEntity, BinarySensorEntity): class MyUplinkDevicePointBinarySensor(MyUplinkEntity, BinarySensorEntity):
"""Representation of a myUplink device point binary sensor.""" """Representation of a myUplink device point bound binary sensor."""
def __init__( def __init__(
self, self,
@ -94,3 +132,73 @@ class MyUplinkDevicePointBinarySensor(MyUplinkEntity, BinarySensorEntity):
"""Binary sensor state value.""" """Binary sensor state value."""
device_point = self.coordinator.data.points[self.device_id][self.point_id] device_point = self.coordinator.data.points[self.device_id][self.point_id]
return int(device_point.value) != 0 return int(device_point.value) != 0
@property
def available(self) -> bool:
"""Return device data availability."""
return super().available and (
self.coordinator.data.devices[self.device_id].connectionState
== DeviceConnectionState.Connected
)
class MyUplinkDeviceBinarySensor(MyUplinkEntity, BinarySensorEntity):
"""Representation of a myUplink device bound binary sensor."""
def __init__(
self,
coordinator: MyUplinkDataCoordinator,
device_id: str,
entity_description: BinarySensorEntityDescription | None,
unique_id_suffix: str,
) -> None:
"""Initialize the binary_sensor."""
super().__init__(
coordinator=coordinator,
device_id=device_id,
unique_id_suffix=unique_id_suffix,
)
if entity_description is not None:
self.entity_description = entity_description
@property
def is_on(self) -> bool:
"""Binary sensor state value."""
return (
self.coordinator.data.devices[self.device_id].connectionState
== DeviceConnectionState.Connected
)
class MyUplinkSystemBinarySensor(MyUplinkSystemEntity, BinarySensorEntity):
"""Representation of a myUplink system bound binary sensor."""
def __init__(
self,
coordinator: MyUplinkDataCoordinator,
system_id: str,
device_id: str,
entity_description: BinarySensorEntityDescription | None,
unique_id_suffix: str,
) -> None:
"""Initialize the binary_sensor."""
super().__init__(
coordinator=coordinator,
system_id=system_id,
device_id=device_id,
unique_id_suffix=unique_id_suffix,
)
if entity_description is not None:
self.entity_description = entity_description
@property
def is_on(self) -> bool | None:
"""Binary sensor state value."""
retval = None
for system in self.coordinator.data.systems:
if system.id == self.system_id:
retval = system.has_alarm
break
return retval

View file

@ -8,7 +8,7 @@ from .coordinator import MyUplinkDataCoordinator
class MyUplinkEntity(CoordinatorEntity[MyUplinkDataCoordinator]): class MyUplinkEntity(CoordinatorEntity[MyUplinkDataCoordinator]):
"""Representation of a sensor.""" """Representation of myuplink entity."""
_attr_has_entity_name = True _attr_has_entity_name = True
@ -18,7 +18,7 @@ class MyUplinkEntity(CoordinatorEntity[MyUplinkDataCoordinator]):
device_id: str, device_id: str,
unique_id_suffix: str, unique_id_suffix: str,
) -> None: ) -> None:
"""Initialize the sensor.""" """Initialize the entity."""
super().__init__(coordinator=coordinator) super().__init__(coordinator=coordinator)
# Internal properties # Internal properties
@ -27,3 +27,27 @@ class MyUplinkEntity(CoordinatorEntity[MyUplinkDataCoordinator]):
# Basic values # Basic values
self._attr_unique_id = f"{device_id}-{unique_id_suffix}" self._attr_unique_id = f"{device_id}-{unique_id_suffix}"
self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, device_id)}) self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, device_id)})
class MyUplinkSystemEntity(MyUplinkEntity):
"""Representation of a system bound entity."""
def __init__(
self,
coordinator: MyUplinkDataCoordinator,
system_id: str,
device_id: str,
unique_id_suffix: str,
) -> None:
"""Initialize the entity."""
super().__init__(
coordinator=coordinator,
device_id=device_id,
unique_id_suffix=unique_id_suffix,
)
# Internal properties
self.system_id = system_id
# Basic values
self._attr_unique_id = f"{system_id}-{unique_id_suffix}"

View file

@ -25,5 +25,12 @@
"create_entry": { "create_entry": {
"default": "[%key:common::config_flow::create_entry::authenticated%]" "default": "[%key:common::config_flow::create_entry::authenticated%]"
} }
},
"entity": {
"binary_sensor": {
"alarm": {
"name": "Alarm"
}
}
} }
} }

View file

@ -2,6 +2,9 @@
from unittest.mock import MagicMock from unittest.mock import MagicMock
import pytest
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from . import setup_integration from . import setup_integration
@ -9,17 +12,46 @@ from . import setup_integration
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
# Test one entity from each of binary_sensor classes.
@pytest.mark.parametrize(
("entity_id", "friendly_name", "test_attributes", "expected_state"),
[
(
"binary_sensor.gotham_city_pump_heating_medium_gp1",
"Gotham City Pump: Heating medium (GP1)",
True,
STATE_ON,
),
(
"binary_sensor.gotham_city_connectivity",
"Gotham City Connectivity",
False,
STATE_ON,
),
(
"binary_sensor.gotham_city_alarm",
"Gotham City Pump: Alarm",
False,
STATE_OFF,
),
],
)
async def test_sensor_states( async def test_sensor_states(
hass: HomeAssistant, hass: HomeAssistant,
mock_myuplink_client: MagicMock, mock_myuplink_client: MagicMock,
mock_config_entry: MockConfigEntry, mock_config_entry: MockConfigEntry,
entity_id: str,
friendly_name: str,
test_attributes: bool,
expected_state: str,
) -> None: ) -> None:
"""Test sensor state.""" """Test sensor state."""
await setup_integration(hass, mock_config_entry) await setup_integration(hass, mock_config_entry)
state = hass.states.get("binary_sensor.gotham_city_pump_heating_medium_gp1") state = hass.states.get(entity_id)
assert state is not None assert state is not None
assert state.state == "on" assert state.state == expected_state
assert state.attributes == { if test_attributes:
"friendly_name": "Gotham City Pump: Heating medium (GP1)", assert state.attributes == {
} "friendly_name": friendly_name,
}