Implement binary sensor for weather warning

custom_components/irm_kmi/binary_sensor.py

@@ -1,16 +1,18 @@
 """Sensor to signal weather warning from the IRM KMI"""
-
+import datetime
 import logging
 
+import pytz
 from homeassistant.components import binary_sensor
-from homeassistant.components.binary_sensor import BinarySensorEntity, BinarySensorDeviceClass
+from homeassistant.components.binary_sensor import (BinarySensorDeviceClass,
+                                                    BinarySensorEntity)
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers.device_registry import DeviceInfo, DeviceEntryType
+from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
-from custom_components.irm_kmi import IrmKmiCoordinator, DOMAIN
+from custom_components.irm_kmi import DOMAIN, IrmKmiCoordinator
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -28,7 +30,6 @@ class IrmKmiWarning(CoordinatorEntity, BinarySensorEntity):
                  coordinator: IrmKmiCoordinator,
                  entry: ConfigEntry
                  ) -> None:
-        _LOGGER.info(f"{entry.entry_id}, {entry.title}")
         super().__init__(coordinator)
         BinarySensorEntity.__init__(self)
         self._attr_device_class = BinarySensorDeviceClass.SAFETY
@@ -44,5 +45,18 @@ class IrmKmiWarning(CoordinatorEntity, BinarySensorEntity):
 
     @property
     def is_on(self) -> bool | None:
-        # TODO return a real value but first, change implementation of coordinator to expose the data
-        return True
+        if self.coordinator.data.get('warnings') is None:
+            return False
+
+        now = datetime.datetime.now(tz=pytz.timezone(self.hass.config.time_zone))
+        for item in self.coordinator.data.get('warnings'):
+            if item.get('starts_at') < now < item.get('ends_at'):
+                return True
+
+        return False
+
+    @property
+    def extra_state_attributes(self) -> dict:
+        """Return the camera state attributes."""
+        attrs = {"warnings": self.coordinator.data.get('warnings', [])}
+        return attrs

custom_components/irm_kmi/const.py

@@ -121,3 +121,16 @@ IRM_KMI_TO_HA_CONDITION_MAP: Final = {
     (27, 'd'): ATTR_CONDITION_EXCEPTIONAL,
     (27, 'n'): ATTR_CONDITION_EXCEPTIONAL
 }
+
+MAP_WARNING_ID_TO_SLUG: Final = {
+    0: 'wind',
+    1: 'rain',
+    2: 'ice_or_snow',
+    3: 'thunder',
+    7: 'fog',
+    9: 'cold',
+    12: 'thunder_wind_rain',
+    13: 'thunderstorm_strong_gusts',
+    14: 'thunderstorm_large_rainfall',
+    15: 'storm_surge',
+    17: 'coldspell'}

custom_components/irm_kmi/coordinator.py

@@ -18,10 +18,11 @@ from homeassistant.helpers.update_coordinator import (DataUpdateCoordinator,
 from .api import IrmKmiApiClient, IrmKmiApiError
 from .const import CONF_DARK_MODE, CONF_STYLE, DOMAIN
 from .const import IRM_KMI_TO_HA_CONDITION_MAP as CDT_MAP
-from .const import (LANGS, OPTION_STYLE_SATELLITE, OUT_OF_BENELUX,
-                    STYLE_TO_PARAM_MAP)
+from .const import LANGS
+from .const import MAP_WARNING_ID_TO_SLUG as SLUG_MAP
+from .const import OPTION_STYLE_SATELLITE, OUT_OF_BENELUX, STYLE_TO_PARAM_MAP
 from .data import (AnimationFrameData, CurrentWeatherData, IrmKmiForecast,
-                   ProcessedCoordinatorData, RadarAnimationData)
+                   ProcessedCoordinatorData, RadarAnimationData, WarningData)
 from .rain_graph import RainGraph
 from .utils import disable_from_config, get_config_value
 
@@ -131,7 +132,8 @@ class IrmKmiCoordinator(DataUpdateCoordinator):
             current_weather=IrmKmiCoordinator.current_weather_from_data(api_data),
             daily_forecast=IrmKmiCoordinator.daily_list_to_forecast(api_data.get('for', {}).get('daily')),
             hourly_forecast=IrmKmiCoordinator.hourly_list_to_forecast(api_data.get('for', {}).get('hourly')),
-            animation=await self._async_animation_data(api_data=api_data)
+            animation=await self._async_animation_data(api_data=api_data),
+            warnings=self.warnings_from_data(api_data.get('for', {}).get('warning'))
         )
 
     async def download_images_from_api(self,
@@ -344,3 +346,37 @@ class IrmKmiCoordinator(DataUpdateCoordinator):
         return RainGraph(radar_animation, image_path, bg_size,
                          dark_mode=self._dark_mode,
                          tz=self.hass.config.time_zone)
+
+    def warnings_from_data(self, warning_data: list | None) -> List[WarningData] | None:
+        """Create a list of warning data instances based on the api data"""
+        if warning_data is None or not isinstance(warning_data, list) or len(warning_data) == 0:
+            return None
+
+        result = list()
+        for data in warning_data:
+            try:
+                warning_id = int(data.get('warningType', {}).get('id'))
+                start = datetime.fromisoformat(data.get('fromTimestamp', None))
+                end = datetime.fromisoformat(data.get('toTimestamp', None))
+            except TypeError | ValueError:
+                # Without this data, the warning is useless
+                continue
+
+            try:
+                level = int(data.get('warningLevel'))
+            except TypeError:
+                level = None
+
+            result.append(
+                WarningData(
+                    slug=SLUG_MAP.get(warning_id, 'unknown'),
+                    id=warning_id,
+                    level=level,
+                    friendly_name=data.get('warningType', {}).get('name', {}).get(self.hass.config.language),
+                    text=data.get('text', {}).get(self.hass.config.language),
+                    starts_at=start,
+                    ends_at=end
+                )
+            )
+
+        return result if len(result) > 0 else None

custom_components/irm_kmi/data.py

@@ -45,9 +45,21 @@ class RadarAnimationData(TypedDict, total=False):
     svg_animated: bytes | None
 
 
+class WarningData(TypedDict, total=False):
+    """Holds data about a specific warning"""
+    slug: str
+    id: int
+    level: int
+    friendly_name: str
+    text: str
+    starts_at: datetime
+    ends_at: datetime
+
+
 class ProcessedCoordinatorData(TypedDict, total=False):
     """Data class that will be exposed to the entities consuming data from an IrmKmiCoordinator"""
     current_weather: CurrentWeatherData
     hourly_forecast: List[Forecast] | None
     daily_forecast: List[IrmKmiForecast] | None
     animation: RadarAnimationData
+    warnings: List[WarningData] | None

tests/conftest.py

@@ -17,6 +17,10 @@ from custom_components.irm_kmi.const import (
     OPTION_DEPRECATED_FORECAST_NOT_USED, OPTION_STYLE_STD)
 
 
+def get_api_data(fixture: str) -> dict:
+    return json.loads(load_fixture(fixture))
+
+
 async def patched(url: str, params: dict | None = None) -> bytes:
     if "cdn.knmi.nl" in url:
         file_name = "tests/fixtures/clouds_nl.png"

tests/test_binary_sensor.py

@@ -0,0 +1,27 @@
+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

tests/test_config_flow.py

@@ -35,7 +35,6 @@ async def test_full_user_flow(
                     CONF_STYLE: OPTION_STYLE_STD,
                     CONF_DARK_MODE: False},
     )
-    print(result2)
     assert result2.get("type") == FlowResultType.CREATE_ENTRY
     assert result2.get("title") == "test home"
     assert result2.get("data") == {CONF_ZONE: ENTITY_ID_HOME,

tests/test_coordinator.py

@@ -1,4 +1,3 @@
-import json
 from datetime import datetime, timedelta
 
 from freezegun import freeze_time
@@ -6,15 +5,11 @@ from homeassistant.components.weather import (ATTR_CONDITION_CLOUDY,
                                               ATTR_CONDITION_PARTLYCLOUDY,
                                               ATTR_CONDITION_RAINY, Forecast)
 from homeassistant.core import HomeAssistant
-from pytest_homeassistant_custom_component.common import (MockConfigEntry,
-                                                          load_fixture)
+from pytest_homeassistant_custom_component.common import MockConfigEntry
 
 from custom_components.irm_kmi.coordinator import IrmKmiCoordinator
 from custom_components.irm_kmi.data import CurrentWeatherData, IrmKmiForecast
-
-
-def get_api_data(fixture: str) -> dict:
-    return json.loads(load_fixture(fixture))
+from tests.conftest import get_api_data
 
 
 async def test_jules_forgot_to_revert_update_interval_before_pushing(
@@ -26,6 +21,30 @@ async def test_jules_forgot_to_revert_update_interval_before_pushing(
     assert timedelta(minutes=5) <= coordinator.update_interval
 
 
+@freeze_time(datetime.fromisoformat('2024-01-12T07:10: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'))
+
+    assert isinstance(result, list)
+    assert len(result) == 2
+
+    first = result[0]
+
+    assert first.get('starts_at').replace(tzinfo=None) < datetime.now()
+    assert first.get('ends_at').replace(tzinfo=None) > datetime.now()
+
+    assert first.get('slug') == 'fog'
+    assert first.get('friendly_name') == 'Fog'
+    assert first.get('id') == 7
+    assert first.get('level') == 1
+
+
 @freeze_time(datetime.fromisoformat('2023-12-26T18:30:00'))
 def test_current_weather_be() -> None:
     api_data = get_api_data("forecast.json")