Merge pull request #4 from jdejaegh/warnings

Add binary sensor for weather warnings

README.md

@@ -5,8 +5,6 @@ The data is collected via their non-public mobile application API.
 
 Although the provider is Belgian, the data is available for Belgium 🇧🇪, Luxembourg 🇱🇺, and The Netherlands 🇳🇱
 
-**Note: this is still under development, new versions may not be backward compatible.**
-
 ## Installing via HACS
 
 1. Go to HACS > Integrations
@@ -23,6 +21,7 @@ This integration provides the following things:
 - A weather entity with current weather conditions
 - Weather forecasts (hourly, daily and twice-daily) [using the service `weather.get_forecasts`](https://www.home-assistant.io/integrations/weather/#service-weatherget_forecasts)
 - A camera entity for rain radar and short-term rain previsions
+- A binary sensor for weather warnings
 
 The following options are available:
 
@@ -33,6 +32,8 @@ The following options are available:
 
 <details>
 <summary>Show screenshots</summary>
+<img src="https://github.com/jdejaegh/irm-kmi-ha/raw/main/img/sensors.png"/>  <br>
+<img src="https://github.com/jdejaegh/irm-kmi-ha/raw/main/img/forecast.png"/>  <br>
 <img src="https://github.com/jdejaegh/irm-kmi-ha/raw/main/img/camera_light.png"/>  <br>
 <img src="https://github.com/jdejaegh/irm-kmi-ha/raw/main/img/camera_dark.png"/>  <br>
 <img src="https://github.com/jdejaegh/irm-kmi-ha/raw/main/img/camera_sat.png"/>  
@@ -45,8 +46,7 @@ weather condition is taken into account in this integration.
 <br><img src="https://github.com/jdejaegh/irm-kmi-ha/raw/main/img/monday.png" height="150" alt="Example of two weather conditions">
 
 2. The trends for 14 days are not shown
-3. The warnings shown in the app are not shown by the integration
-4. The provider only has data for Belgium, Luxembourg and The Netherlands 
+3. The provider only has data for Belgium, Luxembourg and The Netherlands 
 
 ## Mapping between IRM KMI and Home Assistant weather conditions
 
@@ -71,6 +71,39 @@ Mapping was established based on my own interpretation of the icons and conditio
 | windy-variant   | Wind and clouds                   |                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |                                                                               |
 
 
+## Warning details
+
+The warning binary sensor is on if a warning is currently relevant (i.e. warning start time < current time < warning end time). 
+Warnings may be issued by the IRM KMI ahead of time but the binary sensor is only on when at least one of the issued warnings is relevant.
+
+The binary sensor has an additional attribute called `warnings`, with a list of warnings for the current location. 
+Warnings in the list may be warning issued ahead of time. 
+
+Each element in the list has the following attributes:
+ * `slug: str`: warning slug type, can be used for automation and does not change with language setting. Example:  `ice_or_snow`
+ * `id: int`: internal id for the warning type used by the IRM KMI api.  
+ * `level: int`: warning severity, from 1 (lower risk) to 3 (higher risk)
+ * `friendly_name: str`: language specific name for the warning type.  Examples: `Ice or snow`, `Chute de neige ou verglas`, `Sneeuw of ijzel`, `Glätte`
+ * `text: str`: language specific additional information about the warning
+ * `starts_at: datetime`: time at which the warning starts being relevant
+ * `ends_at: datetime`: time at which the warning stops being relevant
+
+The following table summarizes the different known warning types.  Other warning types may be returned and will have `unknown` as slug.  Feel free to open an issue with the id and the English friendly name to have it added to this integration.
+
+| Warning slug                | Warning id | Friendly name (en, fr, nl, de)                                                           |
+|-----------------------------|------------|------------------------------------------------------------------------------------------|
+| wind                        | 0          | Wind, Vent, Wind, Wind                                                                   |
+| rain                        | 1          | Rain, Pluie, Regen, Regen                                                                |
+| ice_or_snow                 | 2          | Ice or snow, Chute de neige ou verglas, Sneeuw of ijzel, Glätte                          |
+| thunder                     | 3          | Thunder, Orage, Onweer, Gewitter                                                         |
+| fog                         | 7          | Fog, Brouillard, Mist, Nebel                                                             |
+| cold                        | 9          | Cold, Froid, Koude, Kalt                                                                 |
+| thunder_wind_rain           | 12         | Thunder Wind Rain, Orage, rafales et averses, Onweer Wind Regen, Gewitter Windböen Regen |
+| thunderstorm_strong_gusts   | 13         | Thunderstorm & strong gusts, Orage et rafales, Onweer en wind, Gewitter und Windböen     |
+| thunderstorm_large_rainfall | 14         | Thunderstorm & large rainfall, Orage et averses, Onweer en regen, Gewitter und Regen     |
+| storm_surge                 | 15         | Storm surge, Marée forte, Stormtij, Sturmflut                                            |
+| coldspell                   | 17         | Coldspell, Vague de froid, Koude, Koude                                                  |
+
 ## Disclaimer
 
 This is a personal project and isn't in any way affiliated with, sponsored or endorsed by [The Royal Meteorological 

custom_components/irm_kmi/binary_sensor.py

@@ -0,0 +1,62 @@
+"""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 (BinarySensorDeviceClass,
+                                                    BinarySensorEntity)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+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 DOMAIN, IrmKmiCoordinator
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback):
+    """Set up the binary platform"""
+    coordinator = hass.data[DOMAIN][entry.entry_id]
+    async_add_entities([IrmKmiWarning(coordinator, entry)])
+
+
+class IrmKmiWarning(CoordinatorEntity, BinarySensorEntity):
+    """Representation of a weather warning binary sensor"""
+
+    def __init__(self,
+                 coordinator: IrmKmiCoordinator,
+                 entry: ConfigEntry
+                 ) -> None:
+        super().__init__(coordinator)
+        BinarySensorEntity.__init__(self)
+        self._attr_device_class = BinarySensorDeviceClass.SAFETY
+        self._attr_unique_id = entry.entry_id
+        self.entity_id = binary_sensor.ENTITY_ID_FORMAT.format(f"weather_warning_{str(entry.title).lower()}")
+        self._attr_name = f"Warning {entry.title}"
+        self._attr_device_info = DeviceInfo(
+            entry_type=DeviceEntryType.SERVICE,
+            identifiers={(DOMAIN, entry.entry_id)},
+            manufacturer="IRM KMI",
+            name=f"Warning {entry.title}"
+        )
+
+    @property
+    def is_on(self) -> bool | None:
+        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

@@ -15,7 +15,7 @@ from homeassistant.components.weather import (ATTR_CONDITION_CLEAR_NIGHT,
 from homeassistant.const import Platform
 
 DOMAIN: Final = 'irm_kmi'
-PLATFORMS: Final = [Platform.WEATHER, Platform.CAMERA]
+PLATFORMS: Final = [Platform.WEATHER, Platform.CAMERA, Platform.BINARY_SENSOR]
 CONFIG_FLOW_VERSION = 3
 
 OUT_OF_BENELUX: Final = ["außerhalb der Benelux (Brussels)",
@@ -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

@@ -9,6 +9,7 @@ class IrmKmiForecast(Forecast):
     """Forecast class with additional attributes for IRM KMI"""
 
     # TODO: add condition_2 as well and evolution to match data from the API?
+    # TODO: remove the _fr and _nl to have only one 'text' attribute
     text_fr: str | None
     text_nl: str | None
 
@@ -45,9 +46,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

requirements_tests.txt

@@ -1,6 +1,6 @@
-homeassistant==2024.1.2
+homeassistant==2024.1.3
 pytest
-pytest_homeassistant_custom_component @ git+https://github.com/MatthewFlamm/pytest-homeassistant-custom-component
+pytest_homeassistant_custom_component==0.13.89
 freezegun
 Pillow==10.1.0
 isort

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