Minimal version, getting current temp and condition

Author: jdejaegh

custom_components/irm_kmi/__init__.py

@@ -1 +1 @@
-"""Integration for IRM KMI weather"""
+"""Integration for IRM KMI weather"""

custom_components/irm_kmi/api.py

@@ -0,0 +1,96 @@
+"""API Client for IRM KMI weather"""
+from __future__ import annotations
+
+import asyncio
+import socket
+
+import aiohttp
+import async_timeout
+import hashlib
+from datetime import datetime
+
+
+class IrmKmiApiError(Exception):
+    """Exception to indicate a general API error."""
+
+
+class IrmKmiApiCommunicationError(
+    IrmKmiApiError
+):
+    """Exception to indicate a communication error."""
+
+
+class IrmKmiApiParametersError(
+    IrmKmiApiError
+):
+    """Exception to indicate a parameter error."""
+
+
+class IrmKmiApiAuthenticationError(
+    IrmKmiApiError
+):
+    """Exception to indicate an authentication error."""
+
+
+def _api_key(method_name: str):
+    """Get API key."""
+    return hashlib.md5(f"r9EnW374jkJ9acc;{method_name};{datetime.now().strftime('%d/%m/%Y')}".encode()).hexdigest()
+
+
+class IrmKmiApiClient:
+    """Sample API Client."""
+
+    def __init__(
+            self,
+            session: aiohttp.ClientSession,
+    ) -> None:
+        """Sample API Client."""
+        self._session = session
+        self._base_url = "https://app.meteo.be/services/appv4/"
+
+    async def get_forecasts_city(self, city_id: int) -> any:
+        """Get forecasts for given city."""
+        return await self._api_wrapper(
+            params={"ins": city_id,
+                    "s": "getForecasts"}
+        )
+
+    async def _api_wrapper(
+            self,
+            params: dict,
+            path: str = "",
+            method: str = "get",
+            data: dict | None = None,
+            headers: dict | None = None
+    ) -> any:
+        """Get information from the API."""
+
+        if 's' not in params:
+            raise IrmKmiApiParametersError("No query provided as 's' argument for API")
+        else:
+            params['k'] = _api_key(params['s'])
+
+        try:
+            async with async_timeout.timeout(10):
+                response = await self._session.request(
+                    method=method,
+                    url=f"{self._base_url}{path}",
+                    headers=headers,
+                    json=data,
+                    params=params
+                )
+                response.raise_for_status()
+                return await response.json()
+
+        except asyncio.TimeoutError as exception:
+            raise IrmKmiApiCommunicationError(
+                "Timeout error fetching information",
+            ) from exception
+        except (aiohttp.ClientError, socket.gaierror) as exception:
+            raise IrmKmiApiCommunicationError(
+                "Error fetching information",
+            ) from exception
+        except Exception as exception:  # pylint: disable=broad-except
+            raise IrmKmiApiError(
+                "Something really wrong happened!"
+            ) from exception

custom_components/irm_kmi/const.py

@@ -0,0 +1,75 @@
+from homeassistant.components.weather import (
+    ATTR_CONDITION_CLOUDY,
+    ATTR_CONDITION_FOG,
+    ATTR_CONDITION_SNOWY_RAINY,
+    ATTR_CONDITION_LIGHTNING_RAINY,
+    ATTR_CONDITION_PARTLYCLOUDY,
+    ATTR_CONDITION_POURING,
+    ATTR_CONDITION_RAINY,
+    ATTR_CONDITION_SNOWY,
+    ATTR_CONDITION_SUNNY,
+    ATTR_CONDITION_CLEAR_NIGHT,
+    ATTR_CONDITION_EXCEPTIONAL
+)
+
+DOMAIN = 'irm_kmi'
+
+# map ('ww', 'dayNight') tuple from IRM KMI to HA conditions
+IRM_KMI_TO_HA_CONDITION_MAP = {
+    (0, 'd'): ATTR_CONDITION_SUNNY,
+    (0, 'n'): ATTR_CONDITION_CLEAR_NIGHT,
+    (1, 'd'): ATTR_CONDITION_PARTLYCLOUDY,
+    (1, 'n'): ATTR_CONDITION_PARTLYCLOUDY,
+    (2, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
+    (2, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
+    (3, 'd'): ATTR_CONDITION_CLOUDY,
+    (3, 'n'): ATTR_CONDITION_CLOUDY,
+    (4, 'd'): ATTR_CONDITION_POURING,
+    (4, 'n'): ATTR_CONDITION_POURING,
+    (5, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
+    (5, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
+    (6, 'd'): ATTR_CONDITION_POURING,
+    (6, 'n'): ATTR_CONDITION_POURING,
+    (7, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
+    (7, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
+    (8, 'd'): ATTR_CONDITION_SNOWY_RAINY,
+    (8, 'n'): ATTR_CONDITION_SNOWY_RAINY,
+    (9, 'd'): ATTR_CONDITION_SNOWY_RAINY,
+    (9, 'n'): ATTR_CONDITION_SNOWY_RAINY,
+    (10, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
+    (10, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
+    (11, 'd'): ATTR_CONDITION_SNOWY,
+    (11, 'n'): ATTR_CONDITION_SNOWY,
+    (12, 'd'): ATTR_CONDITION_SNOWY,
+    (12, 'n'): ATTR_CONDITION_SNOWY,
+    (13, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
+    (13, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
+    (14, 'd'): ATTR_CONDITION_CLOUDY,
+    (14, 'n'): ATTR_CONDITION_CLOUDY,
+    (15, 'd'): ATTR_CONDITION_CLOUDY,
+    (15, 'n'): ATTR_CONDITION_CLOUDY,
+    (16, 'd'): ATTR_CONDITION_POURING,
+    (16, 'n'): ATTR_CONDITION_POURING,
+    (17, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
+    (17, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
+    (18, 'd'): ATTR_CONDITION_RAINY,
+    (18, 'n'): ATTR_CONDITION_RAINY,
+    (19, 'd'): ATTR_CONDITION_POURING,
+    (19, 'n'): ATTR_CONDITION_POURING,
+    (20, 'd'): ATTR_CONDITION_SNOWY_RAINY,
+    (20, 'n'): ATTR_CONDITION_SNOWY_RAINY,
+    (21, 'd'): ATTR_CONDITION_EXCEPTIONAL,
+    (21, 'n'): ATTR_CONDITION_EXCEPTIONAL,
+    (22, 'd'): ATTR_CONDITION_SNOWY,
+    (22, 'n'): ATTR_CONDITION_SNOWY,
+    (23, 'd'): ATTR_CONDITION_SNOWY,
+    (23, 'n'): ATTR_CONDITION_SNOWY,
+    (24, 'd'): ATTR_CONDITION_FOG,
+    (24, 'n'): ATTR_CONDITION_FOG,
+    (25, 'd'): ATTR_CONDITION_FOG,
+    (25, 'n'): ATTR_CONDITION_FOG,
+    (26, 'd'): ATTR_CONDITION_FOG,
+    (26, 'n'): ATTR_CONDITION_FOG,
+    (27, 'd'): ATTR_CONDITION_EXCEPTIONAL,
+    (27, 'n'): ATTR_CONDITION_EXCEPTIONAL
+}

custom_components/irm_kmi/coordinator.py

@@ -0,0 +1,50 @@
+"""Example integration using DataUpdateCoordinator."""
+
+from datetime import timedelta
+import logging
+
+import async_timeout
+
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.update_coordinator import (
+    DataUpdateCoordinator,
+    UpdateFailed,
+)
+
+from .api import IrmKmiApiClient, IrmKmiApiError
+
+_LOGGER = logging.getLogger(__name__)
+
+class IrmKmiCoordinator(DataUpdateCoordinator):
+    """Coordinator to update data from IRM KMI"""
+
+    def __init__(self, hass, city_id):
+        """Initialize my coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            # Name of the data. For logging purposes.
+            name="IRM KMI weather",
+            # Polling interval. Will only be polled if there are subscribers.
+            update_interval=timedelta(seconds=30),
+        )
+        self._api_client = IrmKmiApiClient(session=async_get_clientsession(hass))
+        self._city_id = city_id
+
+    async def _async_update_data(self):
+        """Fetch data from API endpoint.
+
+        This is the place to pre-process the data to lookup tables
+        so entities can quickly look up their data.
+        """
+        try:
+            # Note: asyncio.TimeoutError and aiohttp.ClientError are already
+            # handled by the data update coordinator.
+            async with async_timeout.timeout(10):
+                # Grab active context variables to limit data required to be fetched from API
+                # Note: using context is not required if there is no need or ability to limit
+                # data retrieved from API.
+                data = await self._api_client.get_forecasts_city(city_id=self._city_id)
+                return data
+        except IrmKmiApiError as err:
+            raise UpdateFailed(f"Error communicating with API: {err}")

custom_components/irm_kmi/weather.py

@@ -1,29 +1,47 @@
 import logging
+
 from homeassistant.components.weather import WeatherEntity
-from homeassistant.components.weather import ATTR_CONDITION_PARTLYCLOUDY
 from homeassistant.const import UnitOfTemperature
+from homeassistant.helpers.update_coordinator import (
+    CoordinatorEntity,
+)
+
+from .const import IRM_KMI_TO_HA_CONDITION_MAP as CDT_MAP
+from .coordinator import IrmKmiCoordinator
 
 _LOGGER = logging.getLogger(__name__)
 
 
-def setup_platform(hass, config, add_devices, discovery_info=None):
-    add_devices([IrmKmiWeather()])
-    _LOGGER.warning("Irm KMI setup")
+async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
+    _LOGGER.debug(f"IRM KMI setup.  Config: {config}")
+    coordinator = IrmKmiCoordinator(hass, city_id=config.get("city_id"))
+    await coordinator.async_request_refresh()
+
+    async_add_entities([IrmKmiWeather(
+        coordinator,
+        config.get("name", "IRM KMI Weather")
+    )])
+
 
+class IrmKmiWeather(CoordinatorEntity, WeatherEntity):
 
-class IrmKmiWeather(WeatherEntity):
+    def __init__(self, coordinator: IrmKmiCoordinator, name: str) -> None:
+        super().__init__(coordinator)
+        self._name = name
 
     @property
     def name(self) -> str:
-        return "IRM KMI Weather"
+        return self._name
 
     @property
     def condition(self) -> str | None:
-        return ATTR_CONDITION_PARTLYCLOUDY
+        irm_condition = (self.coordinator.data.get('obs', {}).get('ww'),
+                         self.coordinator.data.get('obs', {}).get('dayNight'))
+        return CDT_MAP.get(irm_condition, None)
 
     @property
     def native_temperature(self) -> float | None:
-        return 20.2
+        return self.coordinator.data.get('obs', {}).get('temp')
 
     @property
     def native_temperature_unit(self) -> str: