Add sensor platform to LetPot integration (#138491)

* Add sensor platform to LetPot integration

* Handle support in description supported_fn, use common string

* Update homeassistant/components/letpot/switch.py

* Update homeassistant/components/letpot/sensor.py

* Update homeassistant/components/letpot/sensor.py

* Update homeassistant/components/letpot/strings.json

* Fix translation key in snapshot

* snapshot no quotes

---------

Co-authored-by: Josef Zweck <josef@zweck.dev>
This commit is contained in:
Joris Pelgröm 2025-02-14 13:57:27 +01:00 committed by GitHub
parent 48f58c7d49
commit 371490a470
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 288 additions and 31 deletions

View File

@ -22,7 +22,7 @@ from .const import (
) )
from .coordinator import LetPotConfigEntry, LetPotDeviceCoordinator from .coordinator import LetPotConfigEntry, LetPotDeviceCoordinator
PLATFORMS: list[Platform] = [Platform.SWITCH, Platform.TIME] PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.SWITCH, Platform.TIME]
async def async_setup_entry(hass: HomeAssistant, entry: LetPotConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: LetPotConfigEntry) -> bool:

View File

@ -1,18 +1,27 @@
"""Base class for LetPot entities.""" """Base class for LetPot entities."""
from collections.abc import Callable, Coroutine from collections.abc import Callable, Coroutine
from dataclasses import dataclass
from typing import Any, Concatenate from typing import Any, Concatenate
from letpot.exceptions import LetPotConnectionException, LetPotException from letpot.exceptions import LetPotConnectionException, LetPotException
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN from .const import DOMAIN
from .coordinator import LetPotDeviceCoordinator from .coordinator import LetPotDeviceCoordinator
@dataclass(frozen=True, kw_only=True)
class LetPotEntityDescription(EntityDescription):
"""Description for all LetPot entities."""
supported_fn: Callable[[LetPotDeviceCoordinator], bool] = lambda _: True
class LetPotEntity(CoordinatorEntity[LetPotDeviceCoordinator]): class LetPotEntity(CoordinatorEntity[LetPotDeviceCoordinator]):
"""Defines a base LetPot entity.""" """Defines a base LetPot entity."""

View File

@ -1,5 +1,10 @@
{ {
"entity": { "entity": {
"sensor": {
"water_level": {
"default": "mdi:water-percent"
}
},
"switch": { "switch": {
"alarm_sound": { "alarm_sound": {
"default": "mdi:bell-ring", "default": "mdi:bell-ring",

View File

@ -59,8 +59,8 @@ rules:
docs-troubleshooting: todo docs-troubleshooting: todo
docs-use-cases: todo docs-use-cases: todo
dynamic-devices: todo dynamic-devices: todo
entity-category: todo entity-category: done
entity-device-class: todo entity-device-class: done
entity-disabled-by-default: todo entity-disabled-by-default: todo
entity-translations: done entity-translations: done
exception-translations: done exception-translations: done

View File

@ -0,0 +1,110 @@
"""Support for LetPot sensor entities."""
from collections.abc import Callable
from dataclasses import dataclass
from letpot.models import DeviceFeature, LetPotDeviceStatus, TemperatureUnit
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import PERCENTAGE, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from .coordinator import LetPotConfigEntry, LetPotDeviceCoordinator
from .entity import LetPotEntity, LetPotEntityDescription
# Coordinator is used to centralize the data updates
PARALLEL_UPDATES = 0
LETPOT_TEMPERATURE_UNIT_HA_UNIT = {
TemperatureUnit.CELSIUS: UnitOfTemperature.CELSIUS,
TemperatureUnit.FAHRENHEIT: UnitOfTemperature.FAHRENHEIT,
}
@dataclass(frozen=True, kw_only=True)
class LetPotSensorEntityDescription(LetPotEntityDescription, SensorEntityDescription):
"""Describes a LetPot sensor entity."""
native_unit_of_measurement_fn: Callable[[LetPotDeviceStatus], str | None]
value_fn: Callable[[LetPotDeviceStatus], StateType]
SENSORS: tuple[LetPotSensorEntityDescription, ...] = (
LetPotSensorEntityDescription(
key="temperature",
value_fn=lambda status: status.temperature_value,
native_unit_of_measurement_fn=(
lambda status: LETPOT_TEMPERATURE_UNIT_HA_UNIT[
status.temperature_unit or TemperatureUnit.CELSIUS
]
),
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
supported_fn=(
lambda coordinator: DeviceFeature.TEMPERATURE
in coordinator.device_client.device_features
),
),
LetPotSensorEntityDescription(
key="water_level",
translation_key="water_level",
value_fn=lambda status: status.water_level,
native_unit_of_measurement_fn=lambda _: PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
supported_fn=(
lambda coordinator: DeviceFeature.WATER_LEVEL
in coordinator.device_client.device_features
),
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: LetPotConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up LetPot sensor entities based on a device features."""
coordinators = entry.runtime_data
async_add_entities(
LetPotSensorEntity(coordinator, description)
for description in SENSORS
for coordinator in coordinators
if description.supported_fn(coordinator)
)
class LetPotSensorEntity(LetPotEntity, SensorEntity):
"""Defines a LetPot sensor entity."""
entity_description: LetPotSensorEntityDescription
def __init__(
self,
coordinator: LetPotDeviceCoordinator,
description: LetPotSensorEntityDescription,
) -> None:
"""Initialize LetPot sensor entity."""
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = f"{coordinator.config_entry.unique_id}_{coordinator.device.serial_number}_{description.key}"
@property
def native_unit_of_measurement(self) -> str | None:
"""Return the native unit of measurement."""
return self.entity_description.native_unit_of_measurement_fn(
self.coordinator.data
)
@property
def native_value(self) -> StateType:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.coordinator.data)

View File

@ -32,6 +32,11 @@
} }
}, },
"entity": { "entity": {
"sensor": {
"water_level": {
"name": "Water level"
}
},
"switch": { "switch": {
"alarm_sound": { "alarm_sound": {
"name": "Alarm sound" "name": "Alarm sound"

View File

@ -13,7 +13,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import LetPotConfigEntry, LetPotDeviceCoordinator from .coordinator import LetPotConfigEntry, LetPotDeviceCoordinator
from .entity import LetPotEntity, exception_handler from .entity import LetPotEntity, LetPotEntityDescription, exception_handler
# Each change pushes a 'full' device status with the change. The library will cache # Each change pushes a 'full' device status with the change. The library will cache
# pending changes to avoid overwriting, but try to avoid a lot of parallelism. # pending changes to avoid overwriting, but try to avoid a lot of parallelism.
@ -21,14 +21,33 @@ PARALLEL_UPDATES = 1
@dataclass(frozen=True, kw_only=True) @dataclass(frozen=True, kw_only=True)
class LetPotSwitchEntityDescription(SwitchEntityDescription): class LetPotSwitchEntityDescription(LetPotEntityDescription, SwitchEntityDescription):
"""Describes a LetPot switch entity.""" """Describes a LetPot switch entity."""
value_fn: Callable[[LetPotDeviceStatus], bool | None] value_fn: Callable[[LetPotDeviceStatus], bool | None]
set_value_fn: Callable[[LetPotDeviceClient, bool], Coroutine[Any, Any, None]] set_value_fn: Callable[[LetPotDeviceClient, bool], Coroutine[Any, Any, None]]
BASE_SWITCHES: tuple[LetPotSwitchEntityDescription, ...] = ( SWITCHES: tuple[LetPotSwitchEntityDescription, ...] = (
LetPotSwitchEntityDescription(
key="alarm_sound",
translation_key="alarm_sound",
value_fn=lambda status: status.system_sound,
set_value_fn=lambda device_client, value: device_client.set_sound(value),
entity_category=EntityCategory.CONFIG,
supported_fn=lambda coordinator: coordinator.data.system_sound is not None,
),
LetPotSwitchEntityDescription(
key="auto_mode",
translation_key="auto_mode",
value_fn=lambda status: status.water_mode == 1,
set_value_fn=lambda device_client, value: device_client.set_water_mode(value),
entity_category=EntityCategory.CONFIG,
supported_fn=(
lambda coordinator: DeviceFeature.PUMP_AUTO
in coordinator.device_client.device_features
),
),
LetPotSwitchEntityDescription( LetPotSwitchEntityDescription(
key="power", key="power",
translation_key="power", translation_key="power",
@ -44,20 +63,6 @@ BASE_SWITCHES: tuple[LetPotSwitchEntityDescription, ...] = (
entity_category=EntityCategory.CONFIG, entity_category=EntityCategory.CONFIG,
), ),
) )
ALARM_SWITCH: LetPotSwitchEntityDescription = LetPotSwitchEntityDescription(
key="alarm_sound",
translation_key="alarm_sound",
value_fn=lambda status: status.system_sound,
set_value_fn=lambda device_client, value: device_client.set_sound(value),
entity_category=EntityCategory.CONFIG,
)
AUTO_MODE_SWITCH: LetPotSwitchEntityDescription = LetPotSwitchEntityDescription(
key="auto_mode",
translation_key="auto_mode",
value_fn=lambda status: status.water_mode == 1,
set_value_fn=lambda device_client, value: device_client.set_water_mode(value),
entity_category=EntityCategory.CONFIG,
)
async def async_setup_entry( async def async_setup_entry(
@ -69,19 +74,10 @@ async def async_setup_entry(
coordinators = entry.runtime_data coordinators = entry.runtime_data
entities: list[SwitchEntity] = [ entities: list[SwitchEntity] = [
LetPotSwitchEntity(coordinator, description) LetPotSwitchEntity(coordinator, description)
for description in BASE_SWITCHES for description in SWITCHES
for coordinator in coordinators for coordinator in coordinators
if description.supported_fn(coordinator)
] ]
entities.extend(
LetPotSwitchEntity(coordinator, ALARM_SWITCH)
for coordinator in coordinators
if coordinator.data.system_sound is not None
)
entities.extend(
LetPotSwitchEntity(coordinator, AUTO_MODE_SWITCH)
for coordinator in coordinators
if DeviceFeature.PUMP_AUTO in coordinator.device_client.device_features
)
async_add_entities(entities) async_add_entities(entities)

View File

@ -0,0 +1,104 @@
# serializer version: 1
# name: test_all_entities[sensor.garden_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.garden_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Temperature',
'platform': 'letpot',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'a1b2c3d4e5f6a1b2c3d4e5f6_LPH63ABCD_temperature',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_all_entities[sensor.garden_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'Garden Temperature',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.garden_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '18',
})
# ---
# name: test_all_entities[sensor.garden_water_level-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.garden_water_level',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Water level',
'platform': 'letpot',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'water_level',
'unique_id': 'a1b2c3d4e5f6a1b2c3d4e5f6_LPH63ABCD_water_level',
'unit_of_measurement': '%',
})
# ---
# name: test_all_entities[sensor.garden_water_level-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Garden Water level',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.garden_water_level',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '100',
})
# ---

View File

@ -0,0 +1,28 @@
"""Test sensor entities for the LetPot integration."""
from unittest.mock import MagicMock, patch
from syrupy import SnapshotAssertion
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
async def test_all_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_client: MagicMock,
mock_device_client: MagicMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test sensor entities."""
with patch("homeassistant.components.letpot.PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)