From e6932247928474aaf78748bdbb132b9bdb236fe9 Mon Sep 17 00:00:00 2001 From: Jules Dejaeghere Date: Sun, 12 May 2024 17:28:24 +0200 Subject: [PATCH] Add timestamp sensor for next weather warning --- custom_components/irm_kmi/sensor.py | 49 +++++++ .../irm_kmi/translations/en.json | 3 + .../irm_kmi/translations/fr.json | 3 + .../irm_kmi/translations/nl.json | 3 + tests/test_binary_sensor.py | 58 -------- tests/test_warning_sensors.py | 125 ++++++++++++++++++ 6 files changed, 183 insertions(+), 58 deletions(-) delete mode 100644 tests/test_binary_sensor.py create mode 100644 tests/test_warning_sensors.py diff --git a/custom_components/irm_kmi/sensor.py b/custom_components/irm_kmi/sensor.py index fd1e3fd..9a9d131 100644 --- a/custom_components/irm_kmi/sensor.py +++ b/custom_components/irm_kmi/sensor.py @@ -1,6 +1,8 @@ """Sensor for pollen from the IRM KMI""" +import datetime import logging +import pytz from homeassistant.components import sensor from homeassistant.components.sensor import SensorDeviceClass, SensorEntity from homeassistant.config_entries import ConfigEntry @@ -19,6 +21,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_e """Set up the sensor platform""" coordinator = hass.data[DOMAIN][entry.entry_id] async_add_entities([IrmKmiPollen(coordinator, entry, pollen.lower()) for pollen in POLLEN_NAMES]) + async_add_entities([IrmKmiNextWarning(coordinator, entry),]) class IrmKmiPollen(CoordinatorEntity, SensorEntity): @@ -45,3 +48,49 @@ class IrmKmiPollen(CoordinatorEntity, SensorEntity): def native_value(self) -> str | None: """Return the state of the sensor.""" return self.coordinator.data.get('pollen', {}).get(self._pollen, None) + + +class IrmKmiNextWarning(CoordinatorEntity, SensorEntity): + """Representation of the next weather warning""" + + _attr_has_entity_name = True + _attr_device_class = SensorDeviceClass.TIMESTAMP + + def __init__(self, + coordinator: IrmKmiCoordinator, + entry: ConfigEntry, + ) -> None: + super().__init__(coordinator) + SensorEntity.__init__(self) + self._attr_unique_id = f"{entry.entry_id}-next-warning" + self.entity_id = sensor.ENTITY_ID_FORMAT.format(f"{str(entry.title).lower()}_next_warning") + self._attr_device_info = coordinator.shared_device_info + self._attr_translation_key = f"next_warning" + + @property + def native_value(self) -> datetime.datetime | None: + """Return the timestamp for the start of the next warning. Is None when no future warning are available""" + if self.coordinator.data.get('warnings') is None: + return None + + now = datetime.datetime.now(tz=pytz.timezone(self.hass.config.time_zone)) + earliest_next = None + for item in self.coordinator.data.get('warnings'): + if now < item.get('starts_at'): + if earliest_next is None: + earliest_next = item.get('starts_at') + else: + earliest_next = min(earliest_next, item.get('starts_at')) + + return earliest_next + + @property + def extra_state_attributes(self) -> dict: + """Return the attributes related to all the future warnings.""" + now = datetime.datetime.now(tz=pytz.timezone(self.hass.config.time_zone)) + attrs = {"next_warnings": [w for w in self.coordinator.data.get('warnings', []) if now < w.get('starts_at')]} + + attrs["next_warnings_friendly_names"] = ", ".join( + [warning['friendly_name'] for warning in attrs['next_warnings'] if warning['friendly_name'] != '']) + + return attrs diff --git a/custom_components/irm_kmi/translations/en.json b/custom_components/irm_kmi/translations/en.json index b74cf82..847dcca 100644 --- a/custom_components/irm_kmi/translations/en.json +++ b/custom_components/irm_kmi/translations/en.json @@ -78,6 +78,9 @@ }, "entity": { "sensor": { + "next_warning": { + "name": "Next warning" + }, "pollen_alder": { "name": "Alder pollen", "state": { diff --git a/custom_components/irm_kmi/translations/fr.json b/custom_components/irm_kmi/translations/fr.json index 617b50f..6328e60 100644 --- a/custom_components/irm_kmi/translations/fr.json +++ b/custom_components/irm_kmi/translations/fr.json @@ -78,6 +78,9 @@ }, "entity": { "sensor": { + "next_warning": { + "name": "Prochain avertissement" + }, "pollen_alder": { "name": "Pollen d'aulne", "state": { diff --git a/custom_components/irm_kmi/translations/nl.json b/custom_components/irm_kmi/translations/nl.json index d5ada4c..0b206d0 100644 --- a/custom_components/irm_kmi/translations/nl.json +++ b/custom_components/irm_kmi/translations/nl.json @@ -78,6 +78,9 @@ }, "entity": { "sensor": { + "next_warning": { + "name": "Volgende waarschuwing" + }, "pollen_alder": { "name": "Elzenpollen", "state": { diff --git a/tests/test_binary_sensor.py b/tests/test_binary_sensor.py deleted file mode 100644 index a003c61..0000000 --- a/tests/test_binary_sensor.py +++ /dev/null @@ -1,58 +0,0 @@ -from datetime import datetime - -from freezegun import freeze_time -from homeassistant.core import HomeAssistant -from pytest_homeassistant_custom_component.common import MockConfigEntry - -from custom_components.irm_kmi import IrmKmiCoordinator -from custom_components.irm_kmi.binary_sensor import IrmKmiWarning -from tests.conftest import get_api_data - - -@freeze_time(datetime.fromisoformat('2024-01-12T07:55:00+01:00')) -async def test_warning_data( - hass: HomeAssistant, - mock_config_entry: MockConfigEntry -) -> None: - api_data = get_api_data("be_forecast_warning.json") - coordinator = IrmKmiCoordinator(hass, mock_config_entry) - - result = coordinator.warnings_from_data(api_data.get('for', {}).get('warning')) - - coordinator.data = {'warnings': result} - warning = IrmKmiWarning(coordinator, mock_config_entry) - warning.hass = hass - - assert warning.is_on - assert len(warning.extra_state_attributes['warnings']) == 2 - - for w in warning.extra_state_attributes['warnings']: - assert w['is_active'] - - assert warning.extra_state_attributes['active_warnings_friendly_names'] == "Fog, Ice or snow" - - -@freeze_time(datetime.fromisoformat('2024-01-12T07:55:00+01:00')) -async def test_warning_data( - hass: HomeAssistant, - mock_config_entry: MockConfigEntry -) -> None: - # When language is unknown, default to english setting - hass.config.language = "foo" - - api_data = get_api_data("be_forecast_warning.json") - coordinator = IrmKmiCoordinator(hass, mock_config_entry) - - result = coordinator.warnings_from_data(api_data.get('for', {}).get('warning')) - - coordinator.data = {'warnings': result} - warning = IrmKmiWarning(coordinator, mock_config_entry) - warning.hass = hass - - assert warning.is_on - assert len(warning.extra_state_attributes['warnings']) == 2 - - for w in warning.extra_state_attributes['warnings']: - assert w['is_active'] - - assert warning.extra_state_attributes['active_warnings_friendly_names'] == "Fog, Ice or snow" diff --git a/tests/test_warning_sensors.py b/tests/test_warning_sensors.py new file mode 100644 index 0000000..71f38e6 --- /dev/null +++ b/tests/test_warning_sensors.py @@ -0,0 +1,125 @@ +from datetime import datetime + +from freezegun import freeze_time +from homeassistant.core import HomeAssistant +from pytest_homeassistant_custom_component.common import MockConfigEntry + +from custom_components.irm_kmi import IrmKmiCoordinator +from custom_components.irm_kmi.binary_sensor import IrmKmiWarning +from custom_components.irm_kmi.sensor import IrmKmiNextWarning +from tests.conftest import get_api_data + + +@freeze_time(datetime.fromisoformat('2024-01-12T07:55:00+01:00')) +async def test_warning_data( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry +) -> None: + api_data = get_api_data("be_forecast_warning.json") + coordinator = IrmKmiCoordinator(hass, mock_config_entry) + + result = coordinator.warnings_from_data(api_data.get('for', {}).get('warning')) + + coordinator.data = {'warnings': result} + warning = IrmKmiWarning(coordinator, mock_config_entry) + warning.hass = hass + + assert warning.is_on + assert len(warning.extra_state_attributes['warnings']) == 2 + + for w in warning.extra_state_attributes['warnings']: + assert w['is_active'] + + assert warning.extra_state_attributes['active_warnings_friendly_names'] == "Fog, Ice or snow" + + +@freeze_time(datetime.fromisoformat('2024-01-12T07:55:00+01:00')) +async def test_warning_data( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry +) -> None: + # When language is unknown, default to english setting + hass.config.language = "foo" + + api_data = get_api_data("be_forecast_warning.json") + coordinator = IrmKmiCoordinator(hass, mock_config_entry) + + result = coordinator.warnings_from_data(api_data.get('for', {}).get('warning')) + + coordinator.data = {'warnings': result} + warning = IrmKmiWarning(coordinator, mock_config_entry) + warning.hass = hass + + assert warning.is_on + assert len(warning.extra_state_attributes['warnings']) == 2 + + for w in warning.extra_state_attributes['warnings']: + assert w['is_active'] + + assert warning.extra_state_attributes['active_warnings_friendly_names'] == "Fog, Ice or snow" + + +@freeze_time(datetime.fromisoformat('2024-01-11T20:00:00+01:00')) +async def test_next_warning_when_data_available( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry +) -> None: + api_data = get_api_data("be_forecast_warning.json") + coordinator = IrmKmiCoordinator(hass, mock_config_entry) + + result = coordinator.warnings_from_data(api_data.get('for', {}).get('warning')) + + coordinator.data = {'warnings': result} + warning = IrmKmiNextWarning(coordinator, mock_config_entry) + warning.hass = hass + + assert warning.state == "2024-01-12T06:00:00+00:00" + assert len(warning.extra_state_attributes['next_warnings']) == 2 + + assert warning.extra_state_attributes['next_warnings_friendly_names'] == "Fog, Ice or snow" + + +@freeze_time(datetime.fromisoformat('2024-01-12T07:30:00+01:00')) +async def test_next_warning_none_when_only_active_warnings( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry +) -> None: + api_data = get_api_data("be_forecast_warning.json") + coordinator = IrmKmiCoordinator(hass, mock_config_entry) + + result = coordinator.warnings_from_data(api_data.get('for', {}).get('warning')) + + coordinator.data = {'warnings': result} + warning = IrmKmiNextWarning(coordinator, mock_config_entry) + warning.hass = hass + + assert warning.state is None + assert len(warning.extra_state_attributes['next_warnings']) == 0 + + assert warning.extra_state_attributes['next_warnings_friendly_names'] == "" + + +@freeze_time(datetime.fromisoformat('2024-01-12T07:30:00+01:00')) +async def test_next_warning_none_when_no_warnings( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry +) -> None: + coordinator = IrmKmiCoordinator(hass, mock_config_entry) + + coordinator.data = {'warnings': []} + warning = IrmKmiNextWarning(coordinator, mock_config_entry) + warning.hass = hass + + assert warning.state is None + assert len(warning.extra_state_attributes['next_warnings']) == 0 + + assert warning.extra_state_attributes['next_warnings_friendly_names'] == "" + + coordinator.data = dict() + warning = IrmKmiNextWarning(coordinator, mock_config_entry) + warning.hass = hass + + assert warning.state is None + assert len(warning.extra_state_attributes['next_warnings']) == 0 + + assert warning.extra_state_attributes['next_warnings_friendly_names'] == ""