Add timestamp sensor for next weather warning

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

custom_components/irm_kmi/translations/en.json

@@ -78,6 +78,9 @@
   },
   "entity": {
     "sensor": {
+      "next_warning": {
+        "name": "Next warning"
+      },
       "pollen_alder": {
         "name": "Alder pollen",
         "state": {

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": {

custom_components/irm_kmi/translations/nl.json

@@ -78,6 +78,9 @@
   },
   "entity": {
     "sensor": {
+      "next_warning": {
+        "name": "Volgende waarschuwing"
+      },
       "pollen_alder": {
         "name": "Elzenpollen",
         "state": {

tests/test_binary_sensor.py

@@ -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"

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'] == ""