jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 08:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add water price sensor to suez water (#130141)

Browse Source 
* Suez water: add water price sensor

* sensor description

* clean up
This commit is contained in:
jb101010-2 2024-11-09 10:57:22 +01:00 committed by GitHub
parent d11012b2b7
commit 701f35488c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 175 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
homeassistant/components/suez_water/coordinator.py
View File

@ -1,6 +1,11 @@
"""Suez water update coordinator."""
from pysuez import AggregatedData, PySuezError, SuezClient
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import date
from typing import Any
from pysuez import PySuezError, SuezClient
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
@ -11,7 +16,28 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
from .const import CONF_COUNTER_ID, DATA_REFRESH_INTERVAL, DOMAIN
class SuezWaterCoordinator(DataUpdateCoordinator[AggregatedData]):
@dataclass
class SuezWaterAggregatedAttributes:
"""Class containing aggregated sensor extra attributes."""
this_month_consumption: dict[date, float]
previous_month_consumption: dict[date, float]
last_year_overall: dict[str, float]
this_year_overall: dict[str, float]
history: dict[date, float]
highest_monthly_consumption: float
@dataclass
class SuezWaterData:
"""Class used to hold all fetch data from suez api."""
aggregated_value: float
aggregated_attr: Mapping[str, Any]
price: float
class SuezWaterCoordinator(DataUpdateCoordinator[SuezWaterData]):
"""Suez water coordinator."""
_suez_client: SuezClient
@ -37,10 +63,22 @@ class SuezWaterCoordinator(DataUpdateCoordinator[AggregatedData]):
if not await self._suez_client.check_credentials():
raise ConfigEntryError("Invalid credentials for suez water")
async def _async_update_data(self) -> AggregatedData:
async def _async_update_data(self) -> SuezWaterData:
"""Fetch data from API endpoint."""
try:
data = await self._suez_client.fetch_aggregated_data()
aggregated = await self._suez_client.fetch_aggregated_data()
data = SuezWaterData(
aggregated_value=aggregated.value,
aggregated_attr={
"this_month_consumption": aggregated.current_month,
"previous_month_consumption": aggregated.previous_month,
"highest_monthly_consumption": aggregated.highest_monthly_consumption,
"last_year_overall": aggregated.previous_year,
"this_year_overall": aggregated.current_year,
"history": aggregated.history,
},
price=(await self._suez_client.get_price()).price,
)
except PySuezError as err:
_LOGGER.exception(err)
raise UpdateFailed(

94
homeassistant/components/suez_water/sensor.py
View File

@ -2,19 +2,53 @@
from __future__ import annotations
from collections.abc import Mapping
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from typing import Any
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
from pysuez.const import ATTRIBUTION
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfVolume
from homeassistant.const import CURRENCY_EURO, UnitOfVolume
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import CONF_COUNTER_ID, DOMAIN
from .coordinator import SuezWaterCoordinator
from .coordinator import SuezWaterCoordinator, SuezWaterData
@dataclass(frozen=True, kw_only=True)
class SuezWaterSensorEntityDescription(SensorEntityDescription):
"""Describes Suez water sensor entity."""
value_fn: Callable[[SuezWaterData], float | str | None]
attr_fn: Callable[[SuezWaterData], Mapping[str, Any] | None] = lambda _: None
SENSORS: tuple[SuezWaterSensorEntityDescription, ...] = (
SuezWaterSensorEntityDescription(
key="water_usage_yesterday",
translation_key="water_usage_yesterday",
native_unit_of_measurement=UnitOfVolume.LITERS,
device_class=SensorDeviceClass.WATER,
value_fn=lambda suez_data: suez_data.aggregated_value,
attr_fn=lambda suez_data: suez_data.aggregated_attr,
),
SuezWaterSensorEntityDescription(
key="water_price",
translation_key="water_price",
native_unit_of_measurement=CURRENCY_EURO,
device_class=SensorDeviceClass.MONETARY,
value_fn=lambda suez_data: suez_data.price,
),
)
async def async_setup_entry(
@ -24,46 +58,42 @@ async def async_setup_entry(
) -> None:
"""Set up Suez Water sensor from a config entry."""
coordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities([SuezAggregatedSensor(coordinator, entry.data[CONF_COUNTER_ID])])
counter_id = entry.data[CONF_COUNTER_ID]
async_add_entities(
SuezWaterSensor(coordinator, counter_id, description) for description in SENSORS
)
class SuezAggregatedSensor(CoordinatorEntity[SuezWaterCoordinator], SensorEntity):
"""Representation of a Sensor."""
class SuezWaterSensor(CoordinatorEntity[SuezWaterCoordinator], SensorEntity):
"""Representation of a Suez water sensor."""
_attr_has_entity_name = True
_attr_translation_key = "water_usage_yesterday"
_attr_native_unit_of_measurement = UnitOfVolume.LITERS
_attr_device_class = SensorDeviceClass.WATER
_attr_attribution = ATTRIBUTION
entity_description: SuezWaterSensorEntityDescription
def __init__(self, coordinator: SuezWaterCoordinator, counter_id: int) -> None:
"""Initialize the data object."""
def __init__(
self,
coordinator: SuezWaterCoordinator,
counter_id: int,
entity_description: SuezWaterSensorEntityDescription,
) -> None:
"""Initialize the suez water sensor entity."""
super().__init__(coordinator)
self._attr_extra_state_attributes = {}
self._attr_unique_id = f"{counter_id}_water_usage_yesterday"
self._attr_unique_id = f"{counter_id}_{entity_description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, str(counter_id))},
entry_type=DeviceEntryType.SERVICE,
manufacturer="Suez",
)
self.entity_description = entity_description
@property
def native_value(self) -> float:
"""Return the current daily usage."""
return self.coordinator.data.value
def native_value(self) -> float | str | None:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.coordinator.data)
@property
def attribution(self) -> str:
"""Return data attribution message."""
return self.coordinator.data.attribution
@property
def extra_state_attributes(self) -> Mapping[str, Any]:
"""Return aggregated data."""
return {
"this_month_consumption": self.coordinator.data.current_month,
"previous_month_consumption": self.coordinator.data.previous_month,
"highest_monthly_consumption": self.coordinator.data.highest_monthly_consumption,
"last_year_overall": self.coordinator.data.previous_year,
"this_year_overall": self.coordinator.data.current_year,
"history": self.coordinator.data.history,
}
def extra_state_attributes(self) -> Mapping[str, Any] | None:
"""Return extra state of the sensor."""
return self.entity_description.attr_fn(self.coordinator.data)

3
homeassistant/components/suez_water/strings.json
View File

@ -23,6 +23,9 @@
"sensor": {
"water_usage_yesterday": {
"name": "Water usage yesterday"
},
"water_price": {
"name": "Water price"
}
}
}

8
tests/components/suez_water/conftest.py
View File

@ -3,10 +3,11 @@
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
from pysuez import AggregatedData, PriceResult
from pysuez.const import ATTRIBUTION
import pytest
from homeassistant.components.suez_water.const import DOMAIN
from homeassistant.components.suez_water.coordinator import AggregatedData
from tests.common import MockConfigEntry
@ -38,7 +39,7 @@ def mock_setup_entry() -> Generator[AsyncMock]:
@pytest.fixture(name="suez_client")
def mock_suez_data() -> Generator[AsyncMock]:
def mock_suez_client() -> Generator[AsyncMock]:
"""Create mock for suez_water external api."""
with (
patch(
@ -64,7 +65,7 @@ def mock_suez_data() -> Generator[AsyncMock]:
},
current_year=1500,
previous_year=1000,
attribution="suez water mock test",
attribution=ATTRIBUTION,
highest_monthly_consumption=2558,
history={
"2024-01-01": 130,
@ -75,4 +76,5 @@ def mock_suez_data() -> Generator[AsyncMock]:
)
suez_client.fetch_aggregated_data.return_value = result
suez_client.get_price.return_value = PriceResult("4.74")
yield suez_client

51
tests/components/suez_water/snapshots/test_sensor.ambr
View File

@ -1,4 +1,53 @@
# serializer version: 1
# name: test_sensors_valid_state[sensor.suez_mock_device_water_price-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.suez_mock_device_water_price',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.MONETARY: 'monetary'>,
'original_icon': None,
'original_name': 'Water price',
'platform': 'suez_water',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'water_price',
'unique_id': 'test-counter_water_price',
'unit_of_measurement': '€',
})
# ---
# name: test_sensors_valid_state[sensor.suez_mock_device_water_price-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by toutsurmoneau.fr',
'device_class': 'monetary',
'friendly_name': 'Suez mock device Water price',
'unit_of_measurement': '€',
}),
'context': <ANY>,
'entity_id': 'sensor.suez_mock_device_water_price',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '4.74',
})
# ---
# name: test_sensors_valid_state[sensor.suez_mock_device_water_usage_yesterday-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
@ -35,7 +84,7 @@
# name: test_sensors_valid_state[sensor.suez_mock_device_water_usage_yesterday-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'suez water mock test',
'attribution': 'Data provided by toutsurmoneau.fr',
'device_class': 'water',
'friendly_name': 'Suez mock device Water usage yesterday',
'highest_monthly_consumption': 2558,

21
tests/components/suez_water/test_sensor.py
View File

@ -3,6 +3,7 @@
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy import SnapshotAssertion
from homeassistant.components.suez_water.const import DATA_REFRESH_INTERVAL
@ -32,11 +33,13 @@ async def test_sensors_valid_state(
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize("method", [("fetch_aggregated_data"), ("get_price")])
async def test_sensors_failed_update(
hass: HomeAssistant,
suez_client: AsyncMock,
mock_config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
method: str,
) -> None:
"""Test that suez_water sensor reflect failure when api fails."""
@ -45,18 +48,20 @@ async def test_sensors_failed_update(
assert mock_config_entry.state is ConfigEntryState.LOADED
entity_ids = await hass.async_add_executor_job(hass.states.entity_ids)
assert len(entity_ids) == 1
assert len(entity_ids) == 2
state = hass.states.get(entity_ids[0])
assert entity_ids[0]
assert state.state != STATE_UNAVAILABLE
for entity in entity_ids:
state = hass.states.get(entity)
assert entity
assert state.state != STATE_UNAVAILABLE
suez_client.fetch_aggregated_data.side_effect = PySuezError("Should fail to update")
getattr(suez_client, method).side_effect = PySuezError("Should fail to update")
freezer.tick(DATA_REFRESH_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done(True)
state = hass.states.get(entity_ids[0])
assert state
assert state.state == STATE_UNAVAILABLE
for entity in entity_ids:
state = hass.states.get(entity)
assert entity
assert state.state == STATE_UNAVAILABLE