Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Bump `imgw_pib` to version 1.0.9 and remove hydrological detail entities (#134668)

pull/133771/merge
Maciej Bieniek 2025-01-16 22:42:03 +00:00 committed by GitHub
parent e6c696933f
commit b0d3aa1c34
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 86 additions and 418 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
homeassistant/components/imgw_pib/__init__.py
View File

@ -9,16 +9,18 @@ from aiohttp import ClientError
from imgw_pib import ImgwPib
from imgw_pib.exceptions import ApiError
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_PLATFORM
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import CONF_STATION_ID
from .const import CONF_STATION_ID, DOMAIN
from .coordinator import ImgwPibDataUpdateCoordinator
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR]
PLATFORMS: list[Platform] = [Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
@ -42,7 +44,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ImgwPibConfigEntry) -> b
try:
imgwpib = await ImgwPib.create(
client_session, hydrological_station_id=station_id
client_session,
hydrological_station_id=station_id,
hydrological_details=False,
)
except (ClientError, TimeoutError, ApiError) as err:
raise ConfigEntryNotReady from err
@ -50,6 +54,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ImgwPibConfigEntry) -> b
coordinator = ImgwPibDataUpdateCoordinator(hass, imgwpib, station_id)
await coordinator.async_config_entry_first_refresh()
# Remove binary_sensor entities for which the endpoint has been blocked by IMGW-PIB API
entity_reg = er.async_get(hass)
for key in ("flood_warning", "flood_alarm"):
if entity_id := entity_reg.async_get_entity_id(
BINARY_SENSOR_PLATFORM, DOMAIN, f"{coordinator.station_id}_{key}"
):
entity_reg.async_remove(entity_id)
entry.runtime_data = ImgwPibData(coordinator)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)

82
homeassistant/components/imgw_pib/binary_sensor.py
View File

@ -1,82 +0,0 @@
"""IMGW-PIB binary sensor platform."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from imgw_pib.model import HydrologicalData
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ImgwPibConfigEntry
from .coordinator import ImgwPibDataUpdateCoordinator
from .entity import ImgwPibEntity
PARALLEL_UPDATES = 1
@dataclass(frozen=True, kw_only=True)
class ImgwPibBinarySensorEntityDescription(BinarySensorEntityDescription):
"""IMGW-PIB sensor entity description."""
value: Callable[[HydrologicalData], bool | None]
BINARY_SENSOR_TYPES: tuple[ImgwPibBinarySensorEntityDescription, ...] = (
ImgwPibBinarySensorEntityDescription(
key="flood_warning",
translation_key="flood_warning",
device_class=BinarySensorDeviceClass.SAFETY,
value=lambda data: data.flood_warning,
),
ImgwPibBinarySensorEntityDescription(
key="flood_alarm",
translation_key="flood_alarm",
device_class=BinarySensorDeviceClass.SAFETY,
value=lambda data: data.flood_alarm,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: ImgwPibConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Add a IMGW-PIB binary sensor entity from a config_entry."""
coordinator = entry.runtime_data.coordinator
async_add_entities(
ImgwPibBinarySensorEntity(coordinator, description)
for description in BINARY_SENSOR_TYPES
if getattr(coordinator.data, description.key) is not None
)
class ImgwPibBinarySensorEntity(ImgwPibEntity, BinarySensorEntity):
"""Define IMGW-PIB binary sensor entity."""
entity_description: ImgwPibBinarySensorEntityDescription
def __init__(
self,
coordinator: ImgwPibDataUpdateCoordinator,
description: ImgwPibBinarySensorEntityDescription,
) -> None:
"""Initialize."""
super().__init__(coordinator)
self._attr_unique_id = f"{coordinator.station_id}_{description.key}"
self.entity_description = description
@property
def is_on(self) -> bool | None:
"""Return true if the binary sensor is on."""
return self.entity_description.value(self.coordinator.data)

20
homeassistant/components/imgw_pib/icons.json
View File

@ -1,26 +1,6 @@
{
"entity": {
"binary_sensor": {
"flood_warning": {
"default": "mdi:check-circle",
"state": {
"on": "mdi:home-flood"
}
},
"flood_alarm": {
"default": "mdi:check-circle",
"state": {
"on": "mdi:home-flood"
}
}
},
"sensor": {
"flood_warning_level": {
"default": "mdi:alert-outline"
},
"flood_alarm_level": {
"default": "mdi:alert"
},
"water_level": {
"default": "mdi:waves"
},

2
homeassistant/components/imgw_pib/manifest.json
View File

@ -5,5 +5,5 @@
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/imgw_pib",
"iot_class": "cloud_polling",
"requirements": ["imgw_pib==1.0.7"]
"requirements": ["imgw_pib==1.0.9"]
}

33
homeassistant/components/imgw_pib/sensor.py
View File

@ -8,17 +8,20 @@ from dataclasses import dataclass
from imgw_pib.model import HydrologicalData
from homeassistant.components.sensor import (
DOMAIN as SENSOR_PLATFORM,
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import EntityCategory, UnitOfLength, UnitOfTemperature
from homeassistant.const import UnitOfLength, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from . import ImgwPibConfigEntry
from .const import DOMAIN
from .coordinator import ImgwPibDataUpdateCoordinator
from .entity import ImgwPibEntity
@ -33,26 +36,6 @@ class ImgwPibSensorEntityDescription(SensorEntityDescription):
SENSOR_TYPES: tuple[ImgwPibSensorEntityDescription, ...] = (
ImgwPibSensorEntityDescription(
key="flood_alarm_level",
translation_key="flood_alarm_level",
native_unit_of_measurement=UnitOfLength.CENTIMETERS,
device_class=SensorDeviceClass.DISTANCE,
entity_category=EntityCategory.DIAGNOSTIC,
suggested_display_precision=0,
entity_registry_enabled_default=False,
value=lambda data: data.flood_alarm_level.value,
),
ImgwPibSensorEntityDescription(
key="flood_warning_level",
translation_key="flood_warning_level",
native_unit_of_measurement=UnitOfLength.CENTIMETERS,
device_class=SensorDeviceClass.DISTANCE,
entity_category=EntityCategory.DIAGNOSTIC,
suggested_display_precision=0,
entity_registry_enabled_default=False,
value=lambda data: data.flood_warning_level.value,
),
ImgwPibSensorEntityDescription(
key="water_level",
translation_key="water_level",
@ -82,6 +65,14 @@ async def async_setup_entry(
"""Add a IMGW-PIB sensor entity from a config_entry."""
coordinator = entry.runtime_data.coordinator
# Remove entities for which the endpoint has been blocked by IMGW-PIB API
entity_reg = er.async_get(hass)
for key in ("flood_warning_level", "flood_alarm_level"):
if entity_id := entity_reg.async_get_entity_id(
SENSOR_PLATFORM, DOMAIN, f"{coordinator.station_id}_{key}"
):
entity_reg.async_remove(entity_id)
async_add_entities(
ImgwPibSensorEntity(coordinator, description)
for description in SENSOR_TYPES

14
homeassistant/components/imgw_pib/strings.json
View File

@ -17,21 +17,7 @@
}
},
"entity": {
"binary_sensor": {
"flood_alarm": {
"name": "Flood alarm"
},
"flood_warning": {
"name": "Flood warning"
}
},
"sensor": {
"flood_alarm_level": {
"name": "Flood alarm level"
},
"flood_warning_level": {
"name": "Flood warning level"
},
"water_level": {
"name": "Water level"
},

2
requirements_all.txt generated
View File

@ -1205,7 +1205,7 @@ igloohome-api==0.0.6
ihcsdk==2.8.5
# homeassistant.components.imgw_pib
imgw_pib==1.0.7
imgw_pib==1.0.9
# homeassistant.components.incomfort
incomfort-client==0.6.4

2
requirements_test_all.txt generated
View File

@ -1019,7 +1019,7 @@ ifaddr==0.2.0
igloohome-api==0.0.6
# homeassistant.components.imgw_pib
imgw_pib==1.0.7
imgw_pib==1.0.9
# homeassistant.components.incomfort
incomfort-client==0.6.4

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

@ -16,11 +16,11 @@ HYDROLOGICAL_DATA = HydrologicalData(
river="River Name",
station_id="123",
water_level=SensorData(name="Water Level", value=526.0),
flood_alarm_level=SensorData(name="Flood Alarm Level", value=630.0),
flood_warning_level=SensorData(name="Flood Warning Level", value=590.0),
flood_alarm_level=SensorData(name="Flood Alarm Level", value=None),
flood_warning_level=SensorData(name="Flood Warning Level", value=None),
water_temperature=SensorData(name="Water Temperature", value=10.8),
flood_alarm=False,
flood_warning=False,
flood_alarm=None,
flood_warning=None,
water_level_measurement_date=datetime(2024, 4, 27, 10, 0, tzinfo=UTC),
water_temperature_measurement_date=datetime(2024, 4, 27, 10, 10, tzinfo=UTC),
)

97
tests/components/imgw_pib/snapshots/test_binary_sensor.ambr
View File

@ -1,97 +0,0 @@
# serializer version: 1
# name: test_binary_sensor[binary_sensor.river_name_station_name_flood_alarm-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.river_name_station_name_flood_alarm',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.SAFETY: 'safety'>,
'original_icon': None,
'original_name': 'Flood alarm',
'platform': 'imgw_pib',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'flood_alarm',
'unique_id': '123_flood_alarm',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[binary_sensor.river_name_station_name_flood_alarm-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by IMGW-PIB',
'device_class': 'safety',
'friendly_name': 'River Name (Station Name) Flood alarm',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.river_name_station_name_flood_alarm',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_binary_sensor[binary_sensor.river_name_station_name_flood_warning-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.river_name_station_name_flood_warning',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.SAFETY: 'safety'>,
'original_icon': None,
'original_name': 'Flood warning',
'platform': 'imgw_pib',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'flood_warning',
'unique_id': '123_flood_warning',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[binary_sensor.river_name_station_name_flood_warning-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by IMGW-PIB',
'device_class': 'safety',
'friendly_name': 'River Name (Station Name) Flood warning',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.river_name_station_name_flood_warning',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---

8
tests/components/imgw_pib/snapshots/test_diagnostics.ambr
View File

@ -20,17 +20,17 @@
'version': 1,
}),
'hydrological_data': dict({
'flood_alarm': False,
'flood_alarm': None,
'flood_alarm_level': dict({
'name': 'Flood Alarm Level',
'unit': None,
'value': 630.0,
'value': None,
}),
'flood_warning': False,
'flood_warning': None,
'flood_warning_level': dict({
'name': 'Flood Warning Level',
'unit': None,
'value': 590.0,
'value': None,
}),
'river': 'River Name',
'station': 'Station Name',

104
tests/components/imgw_pib/snapshots/test_sensor.ambr
View File

@ -1,108 +1,4 @@
# serializer version: 1
# name: test_sensor[sensor.river_name_station_name_flood_alarm_level-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': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.river_name_station_name_flood_alarm_level',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
}),
'original_device_class': <SensorDeviceClass.DISTANCE: 'distance'>,
'original_icon': None,
'original_name': 'Flood alarm level',
'platform': 'imgw_pib',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'flood_alarm_level',
'unique_id': '123_flood_alarm_level',
'unit_of_measurement': <UnitOfLength.CENTIMETERS: 'cm'>,
})
# ---
# name: test_sensor[sensor.river_name_station_name_flood_alarm_level-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by IMGW-PIB',
'device_class': 'distance',
'friendly_name': 'River Name (Station Name) Flood alarm level',
'unit_of_measurement': <UnitOfLength.CENTIMETERS: 'cm'>,
}),
'context': <ANY>,
'entity_id': 'sensor.river_name_station_name_flood_alarm_level',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '630.0',
})
# ---
# name: test_sensor[sensor.river_name_station_name_flood_warning_level-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': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.river_name_station_name_flood_warning_level',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
}),
'original_device_class': <SensorDeviceClass.DISTANCE: 'distance'>,
'original_icon': None,
'original_name': 'Flood warning level',
'platform': 'imgw_pib',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'flood_warning_level',
'unique_id': '123_flood_warning_level',
'unit_of_measurement': <UnitOfLength.CENTIMETERS: 'cm'>,
})
# ---
# name: test_sensor[sensor.river_name_station_name_flood_warning_level-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by IMGW-PIB',
'device_class': 'distance',
'friendly_name': 'River Name (Station Name) Flood warning level',
'unit_of_measurement': <UnitOfLength.CENTIMETERS: 'cm'>,
}),
'context': <ANY>,
'entity_id': 'sensor.river_name_station_name_flood_warning_level',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '590.0',
})
# ---
# name: test_sensor[sensor.river_name_station_name_water_level-entry]
EntityRegistryEntrySnapshot({
'aliases': set({

65
tests/components/imgw_pib/test_binary_sensor.py
View File

@ -1,65 +0,0 @@
"""Test the IMGW-PIB binary sensor platform."""
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
from imgw_pib import ApiError
from syrupy import SnapshotAssertion
from homeassistant.components.imgw_pib.const import UPDATE_INTERVAL
from homeassistant.const import STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import init_integration
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
ENTITY_ID = "binary_sensor.river_name_station_name_flood_alarm"
async def test_binary_sensor(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
mock_imgw_pib_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test states of the binary sensor."""
with patch("homeassistant.components.imgw_pib.PLATFORMS", [Platform.BINARY_SENSOR]):
await init_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_availability(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_imgw_pib_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Ensure that we mark the entities unavailable correctly when service is offline."""
await init_integration(hass, mock_config_entry)
state = hass.states.get(ENTITY_ID)
assert state
assert state.state != STATE_UNAVAILABLE
assert state.state == "off"
mock_imgw_pib_client.get_hydrological_data.side_effect = ApiError("API Error")
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state
assert state.state == STATE_UNAVAILABLE
mock_imgw_pib_client.get_hydrological_data.side_effect = None
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state
assert state.state != STATE_UNAVAILABLE
assert state.state == "off"

24
tests/components/imgw_pib/test_init.py
View File

@ -4,9 +4,11 @@ from unittest.mock import AsyncMock, patch
from imgw_pib import ApiError
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_PLATFORM
from homeassistant.components.imgw_pib.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import init_integration
@ -43,3 +45,25 @@ async def test_unload_entry(
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
assert not hass.data.get(DOMAIN)
async def test_remove_binary_sensor_entity(
hass: HomeAssistant,
mock_imgw_pib_client: AsyncMock,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test removing a binary_sensor entity."""
entity_id = "binary_sensor.river_name_station_name_flood_alarm"
entity_registry.async_get_or_create(
BINARY_SENSOR_PLATFORM,
DOMAIN,
"123_flood_alarm",
suggested_object_id=entity_id.rsplit(".", maxsplit=1)[-1],
config_entry=mock_config_entry,
)
await init_integration(hass, mock_config_entry)
assert hass.states.get(entity_id) is None

25
tests/components/imgw_pib/test_sensor.py
View File

@ -7,7 +7,8 @@ from imgw_pib import ApiError
import pytest
from syrupy import SnapshotAssertion
from homeassistant.components.imgw_pib.const import UPDATE_INTERVAL
from homeassistant.components.imgw_pib.const import DOMAIN, UPDATE_INTERVAL
from homeassistant.components.sensor import DOMAIN as SENSOR_PLATFORM
from homeassistant.const import STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
@ -65,3 +66,25 @@ async def test_availability(
assert state
assert state.state != STATE_UNAVAILABLE
assert state.state == "526.0"
async def test_remove_entity(
hass: HomeAssistant,
mock_imgw_pib_client: AsyncMock,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test removing entity."""
entity_id = "sensor.river_name_station_name_flood_alarm_level"
entity_registry.async_get_or_create(
SENSOR_PLATFORM,
DOMAIN,
"123_flood_alarm_level",
suggested_object_id=entity_id.rsplit(".", maxsplit=1)[-1],
config_entry=mock_config_entry,
)
await init_integration(hass, mock_config_entry)
assert hass.states.get(entity_id) is None