Merge pull request #28 from jdejaegh/21-warnings-should-be-visible-before-they-actually-start

Warnings should be visible before they actually start

custom_components/irm_kmi/coordinator.py

@@ -273,7 +273,7 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
             return None
 
         forecasts = list()
-        day = datetime.now()
+        day = datetime.now(tz=pytz.timezone('Europe/Brussels')).replace(hour=0, minute=0, second=0, microsecond=0)
 
         for f in data:
             if 'dateShow' in f:
@@ -282,6 +282,7 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
             hour = f.get('hour', None)
             if hour is None:
                 continue
+            day = day.replace(hour=int(hour))
 
             precipitation_probability = None
             if f.get('precipChance', None) is not None:
@@ -299,7 +300,7 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
                     pass
 
             forecast = Forecast(
-                datetime=day.strftime(f'%Y-%m-%dT{hour}:00:00'),
+                datetime=day.isoformat(),
                 condition=CDT_MAP.get((ww, f.get('dayNight', None)), None),
                 native_precipitation=f.get('precipQuantity', None),
                 native_temperature=f.get('temp', None),
@@ -347,10 +348,10 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
                     pass
 
             is_daytime = f.get('dayNight', None) == 'd'
-
+            now = datetime.now(pytz.timezone('Europe/Brussels'))
             forecast = IrmKmiForecast(
-                datetime=(datetime.now() + timedelta(days=n_days)).strftime('%Y-%m-%d')
+                datetime=(now + timedelta(days=n_days)).strftime('%Y-%m-%d')
-                if is_daytime else datetime.now().strftime('%Y-%m-%d'),
+                if is_daytime else now.strftime('%Y-%m-%d'),
                 condition=CDT_MAP.get((f.get('ww1', None), f.get('dayNight', None)), None),
                 native_precipitation=precipitation,
                 native_temperature=f.get('tempMax', None),

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_coordinator.py

@@ -113,7 +113,7 @@ async def test_daily_forecast(
     assert result[1] == expected
 
 
-@freeze_time(datetime.fromisoformat('2023-12-26T18:30:00.028724'))
+@freeze_time(datetime.fromisoformat('2023-12-26T18:30:00+01:00'))
 def test_hourly_forecast() -> None:
     api_data = get_api_data("forecast.json").get('for', {}).get('hourly')
     result = IrmKmiCoordinator.hourly_list_to_forecast(api_data)
@@ -122,7 +122,7 @@ def test_hourly_forecast() -> None:
     assert len(result) == 49
 
     expected = Forecast(
-        datetime='2023-12-27T02:00:00',
+        datetime='2023-12-27T02:00:00+01:00',
         condition=ATTR_CONDITION_RAINY,
         native_precipitation=.98,
         native_temperature=8,

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