From 6476f0e57a40471e0934fa38071d82824cb2b1b1 Mon Sep 17 00:00:00 2001 From: Jules Dejaeghere Date: Fri, 2 May 2025 19:31:45 +0200 Subject: [PATCH] Move API related code to a sub-module + update tests --- custom_components/irm_kmi/__init__.py | 4 +- custom_components/irm_kmi/api.py | 125 ---- custom_components/irm_kmi/config_flow.py | 11 +- custom_components/irm_kmi/const.py | 33 +- custom_components/irm_kmi/coordinator.py | 454 +------------- custom_components/irm_kmi/data.py | 38 +- .../irm_kmi/irm_kmi_api/__init__.py | 0 custom_components/irm_kmi/irm_kmi_api/api.py | 562 ++++++++++++++++++ .../irm_kmi/irm_kmi_api/const.py | 28 + custom_components/irm_kmi/irm_kmi_api/data.py | 95 +++ custom_components/irm_kmi/irm_kmi_api/ha.py | 2 + .../irm_kmi/{ => irm_kmi_api}/pollen.py | 2 +- .../irm_kmi/{ => irm_kmi_api}/rain_graph.py | 7 +- .../irm_kmi/irm_kmi_api/resources/__init__.py | 0 .../{ => irm_kmi_api}/resources/be_black.png | Bin .../{ => irm_kmi_api}/resources/be_black.py | 0 .../resources/be_satellite.png | Bin .../resources/be_satellite.py | 0 .../{ => irm_kmi_api}/resources/be_white.png | Bin .../{ => irm_kmi_api}/resources/be_white.py | 0 .../{ => irm_kmi_api}/resources/nl.png | Bin .../irm_kmi/{ => irm_kmi_api}/resources/nl.py | 0 .../{ => irm_kmi_api}/resources/roboto.py | 0 .../resources/roboto_medium.ttf | Bin custom_components/irm_kmi/radar_data.py | 35 -- custom_components/irm_kmi/repairs.py | 20 +- custom_components/irm_kmi/sensor.py | 10 +- tests/conftest.py | 154 +---- tests/test_config_flow.py | 6 +- tests/test_coordinator.py | 106 ++-- tests/test_current_weather_sensors.py | 25 +- tests/test_init.py | 7 +- tests/test_pollen.py | 35 +- tests/test_rain_graph.py | 30 +- tests/test_repairs.py | 18 +- tests/test_sensors.py | 58 +- tests/test_weather.py | 56 +- 37 files changed, 943 insertions(+), 978 deletions(-) delete mode 100644 custom_components/irm_kmi/api.py create mode 100644 custom_components/irm_kmi/irm_kmi_api/__init__.py create mode 100644 custom_components/irm_kmi/irm_kmi_api/api.py create mode 100644 custom_components/irm_kmi/irm_kmi_api/const.py create mode 100644 custom_components/irm_kmi/irm_kmi_api/data.py create mode 100644 custom_components/irm_kmi/irm_kmi_api/ha.py rename custom_components/irm_kmi/{ => irm_kmi_api}/pollen.py (98%) rename custom_components/irm_kmi/{ => irm_kmi_api}/rain_graph.py (98%) create mode 100644 custom_components/irm_kmi/irm_kmi_api/resources/__init__.py rename custom_components/irm_kmi/{ => irm_kmi_api}/resources/be_black.png (100%) rename custom_components/irm_kmi/{ => irm_kmi_api}/resources/be_black.py (100%) rename custom_components/irm_kmi/{ => irm_kmi_api}/resources/be_satellite.png (100%) rename custom_components/irm_kmi/{ => irm_kmi_api}/resources/be_satellite.py (100%) rename custom_components/irm_kmi/{ => irm_kmi_api}/resources/be_white.png (100%) rename custom_components/irm_kmi/{ => irm_kmi_api}/resources/be_white.py (100%) rename custom_components/irm_kmi/{ => irm_kmi_api}/resources/nl.png (100%) rename custom_components/irm_kmi/{ => irm_kmi_api}/resources/nl.py (100%) rename custom_components/irm_kmi/{ => irm_kmi_api}/resources/roboto.py (100%) rename custom_components/irm_kmi/{ => irm_kmi_api}/resources/roboto_medium.ttf (100%) delete mode 100644 custom_components/irm_kmi/radar_data.py diff --git a/custom_components/irm_kmi/__init__.py b/custom_components/irm_kmi/__init__.py index 14c05e2..196f442 100644 --- a/custom_components/irm_kmi/__init__.py +++ b/custom_components/irm_kmi/__init__.py @@ -9,9 +9,9 @@ from homeassistant.exceptions import ConfigEntryError from .const import (CONF_DARK_MODE, CONF_LANGUAGE_OVERRIDE, CONF_STYLE, CONF_USE_DEPRECATED_FORECAST, CONFIG_FLOW_VERSION, DOMAIN, - OPTION_DEPRECATED_FORECAST_NOT_USED, OPTION_STYLE_STD, - PLATFORMS) + OPTION_DEPRECATED_FORECAST_NOT_USED, PLATFORMS) from .coordinator import IrmKmiCoordinator +from .irm_kmi_api.const import OPTION_STYLE_STD from .weather import IrmKmiWeather _LOGGER = logging.getLogger(__name__) diff --git a/custom_components/irm_kmi/api.py b/custom_components/irm_kmi/api.py deleted file mode 100644 index 7851e33..0000000 --- a/custom_components/irm_kmi/api.py +++ /dev/null @@ -1,125 +0,0 @@ -"""API Client for IRM KMI weather""" -from __future__ import annotations - -import asyncio -import hashlib -import json -import logging -import socket -import time -from datetime import datetime - -import aiohttp -import async_timeout -from .const import USER_AGENT - -_LOGGER = logging.getLogger(__name__) - - -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.""" - - -def _api_key(method_name: str) -> str: - """Get API key.""" - return hashlib.md5(f"r9EnW374jkJ9acc;{method_name};{datetime.now().strftime('%d/%m/%Y')}".encode()).hexdigest() - - -class IrmKmiApiClient: - """API client for IRM KMI weather data""" - COORD_DECIMALS = 6 - cache_max_age = 60 * 60 * 2 # Remove items from the cache if they have not been hit since 2 hours - cache = {} - - def __init__(self, session: aiohttp.ClientSession) -> None: - self._session = session - self._base_url = "https://app.meteo.be/services/appv4/" - - async def get_forecasts_coord(self, coord: dict) -> dict: - """Get forecasts for given city.""" - assert 'lat' in coord - assert 'long' in coord - coord['lat'] = round(coord['lat'], self.COORD_DECIMALS) - coord['long'] = round(coord['long'], self.COORD_DECIMALS) - - response: bytes = await self._api_wrapper(params={"s": "getForecasts", "k": _api_key("getForecasts")} | coord) - return json.loads(response) - - async def get_image(self, url, params: dict | None = None) -> bytes: - """Get the image at the specified url with the parameters""" - r: bytes = await self._api_wrapper(base_url=url, params={} if params is None else params) - return r - - async def get_svg(self, url, params: dict | None = None) -> str: - """Get SVG as str at the specified url with the parameters""" - r: bytes = await self._api_wrapper(base_url=url, params={} if params is None else params) - return r.decode() - - async def _api_wrapper( - self, - params: dict, - base_url: str | None = None, - path: str = "", - method: str = "get", - data: dict | None = None, - headers: dict | None = None, - ) -> bytes: - """Get information from the API.""" - url = f"{self._base_url if base_url is None else base_url}{path}" - - if headers is None: - headers = {'User-Agent': USER_AGENT} - else: - headers['User-Agent'] = USER_AGENT - - if url in self.cache: - headers['If-None-Match'] = self.cache[url]['etag'] - - try: - async with async_timeout.timeout(60): - response = await self._session.request( - method=method, - url=url, - headers=headers, - json=data, - params=params - ) - response.raise_for_status() - - if response.status == 304: - _LOGGER.debug(f"Cache hit for {url}") - self.cache[url]['timestamp'] = time.time() - return self.cache[url]['response'] - - if 'ETag' in response.headers: - _LOGGER.debug(f"Saving in cache {url}") - r = await response.read() - self.cache[url] = {'etag': response.headers['ETag'], 'response': r, 'timestamp': time.time()} - return r - - return await response.read() - - 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(f"Something really wrong happened! {exception}") from exception - - def expire_cache(self): - now = time.time() - keys_to_delete = set() - for key, value in self.cache.items(): - if now - value['timestamp'] > self.cache_max_age: - keys_to_delete.add(key) - for key in keys_to_delete: - del self.cache[key] - _LOGGER.info(f"Expired {len(keys_to_delete)} elements from API cache") diff --git a/custom_components/irm_kmi/config_flow.py b/custom_components/irm_kmi/config_flow.py index 8e5993c..0bb2778 100644 --- a/custom_components/irm_kmi/config_flow.py +++ b/custom_components/irm_kmi/config_flow.py @@ -15,13 +15,14 @@ from homeassistant.helpers.selector import (EntitySelector, SelectSelectorConfig, SelectSelectorMode) -from .api import IrmKmiApiClient +from . import OPTION_STYLE_STD from .const import (CONF_DARK_MODE, CONF_LANGUAGE_OVERRIDE, CONF_LANGUAGE_OVERRIDE_OPTIONS, CONF_STYLE, CONF_STYLE_OPTIONS, CONF_USE_DEPRECATED_FORECAST, CONF_USE_DEPRECATED_FORECAST_OPTIONS, CONFIG_FLOW_VERSION, DOMAIN, OPTION_DEPRECATED_FORECAST_NOT_USED, - OPTION_STYLE_STD, OUT_OF_BENELUX) + OUT_OF_BENELUX, USER_AGENT) +from .irm_kmi_api.api import IrmKmiApiClient from .utils import get_config_value _LOGGER = logging.getLogger(__name__) @@ -50,9 +51,11 @@ class IrmKmiConfigFlow(ConfigFlow, domain=DOMAIN): if not errors: api_data = {} try: - async with async_timeout.timeout(60): + async with (async_timeout.timeout(60)): api_data = await IrmKmiApiClient( - session=async_get_clientsession(self.hass)).get_forecasts_coord( + session=async_get_clientsession(self.hass), + user_agent=USER_AGENT + ).get_forecasts_coord( {'lat': zone.attributes[ATTR_LATITUDE], 'long': zone.attributes[ATTR_LONGITUDE]} ) diff --git a/custom_components/irm_kmi/const.py b/custom_components/irm_kmi/const.py index 45c16f3..dd881af 100644 --- a/custom_components/irm_kmi/const.py +++ b/custom_components/irm_kmi/const.py @@ -14,6 +14,9 @@ from homeassistant.components.weather import (ATTR_CONDITION_CLEAR_NIGHT, ATTR_CONDITION_SUNNY) from homeassistant.const import Platform, UnitOfPressure, UnitOfSpeed, UnitOfTemperature, DEGREE +from custom_components.irm_kmi.irm_kmi_api.const import OPTION_STYLE_CONTRAST, OPTION_STYLE_YELLOW_RED, \ + OPTION_STYLE_SATELLITE, OPTION_STYLE_STD + DOMAIN: Final = 'irm_kmi' PLATFORMS: Final = [Platform.WEATHER, Platform.CAMERA, Platform.BINARY_SENSOR, Platform.SENSOR] CONFIG_FLOW_VERSION = 5 @@ -24,10 +27,6 @@ OUT_OF_BENELUX: Final = ["außerhalb der Benelux (Brussels)", "Buiten de Benelux (Brussel)"] LANGS: Final = ['en', 'fr', 'nl', 'de'] -OPTION_STYLE_STD: Final = 'standard_style' -OPTION_STYLE_CONTRAST: Final = 'contrast_style' -OPTION_STYLE_YELLOW_RED: Final = 'yellow_red_style' -OPTION_STYLE_SATELLITE: Final = 'satellite_style' CONF_STYLE: Final = "style" CONF_STYLE_OPTIONS: Final = [ @@ -39,13 +38,6 @@ CONF_STYLE_OPTIONS: Final = [ CONF_DARK_MODE: Final = "dark_mode" -STYLE_TO_PARAM_MAP: Final = { - OPTION_STYLE_STD: 1, - OPTION_STYLE_CONTRAST: 2, - OPTION_STYLE_YELLOW_RED: 3, - OPTION_STYLE_SATELLITE: 4 -} - CONF_USE_DEPRECATED_FORECAST: Final = 'use_deprecated_forecast_attribute' OPTION_DEPRECATED_FORECAST_NOT_USED: Final = 'do_not_use_deprecated_forecast' OPTION_DEPRECATED_FORECAST_DAILY: Final = 'daily_in_deprecated_forecast' @@ -130,23 +122,6 @@ IRM_KMI_TO_HA_CONDITION_MAP: Final = { (27, 'n'): ATTR_CONDITION_FOG } -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'} - -POLLEN_NAMES: Final = {'Alder', 'Ash', 'Birch', 'Grasses', 'Hazel', 'Mugwort', 'Oak'} -POLLEN_LEVEL_TO_COLOR = {'null': 'green', 'low': 'yellow', 'moderate': 'orange', 'high': 'red', 'very high': 'purple', - 'active': 'active'} - POLLEN_TO_ICON_MAP: Final = { 'alder': 'mdi:tree', 'ash': 'mdi:tree', 'birch': 'mdi:tree', 'grasses': 'mdi:grass', 'hazel': 'mdi:tree', 'mugwort': 'mdi:sprout', 'oak': 'mdi:tree' @@ -159,8 +134,6 @@ IRM_KMI_NAME: Final = { 'en': 'Royal Meteorological Institute of Belgium' } -WEEKDAYS = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'] - USER_AGENT: Final = 'github.com/jdejaegh/irm-kmi-ha 0.2.32' CURRENT_WEATHER_SENSORS: Final = {'temperature', 'wind_speed', 'wind_gust_speed', 'wind_bearing', 'uv_index', diff --git a/custom_components/irm_kmi/coordinator.py b/custom_components/irm_kmi/coordinator.py index 125860c..cad228b 100644 --- a/custom_components/irm_kmi/coordinator.py +++ b/custom_components/irm_kmi/coordinator.py @@ -1,12 +1,8 @@ """DataUpdateCoordinator for the IRM KMI integration.""" import logging -from datetime import datetime, timedelta -from statistics import mean -from typing import List -import urllib.parse +from datetime import timedelta import async_timeout -from homeassistant.components.weather import Forecast from homeassistant.config_entries import ConfigEntry from homeassistant.const import ATTR_LATITUDE, ATTR_LONGITUDE, CONF_ZONE from homeassistant.core import HomeAssistant @@ -18,20 +14,14 @@ from homeassistant.helpers.update_coordinator import ( from homeassistant.util import dt from homeassistant.util.dt import utcnow -from .api import IrmKmiApiClient, IrmKmiApiError -from .const import CONF_DARK_MODE, CONF_STYLE, DOMAIN, IRM_KMI_NAME +from .const import CONF_DARK_MODE, CONF_STYLE, DOMAIN, IRM_KMI_NAME, USER_AGENT from .const import IRM_KMI_TO_HA_CONDITION_MAP as CDT_MAP -from .const import MAP_WARNING_ID_TO_SLUG as SLUG_MAP -from .const import (OPTION_STYLE_SATELLITE, OUT_OF_BENELUX, STYLE_TO_PARAM_MAP, - WEEKDAYS) -from .data import (CurrentWeatherData, IrmKmiForecast, - ProcessedCoordinatorData, - WarningData) -from .radar_data import IrmKmiRadarForecast, AnimationFrameData, RadarAnimationData -from .pollen import PollenParser -from .rain_graph import RainGraph -from .utils import (disable_from_config, get_config_value, next_weekday, - preferred_language) +from .const import (OUT_OF_BENELUX) +from .data import ProcessedCoordinatorData +from .irm_kmi_api.api import IrmKmiApiClientHa, IrmKmiApiError +from .irm_kmi_api.pollen import PollenParser +from .irm_kmi_api.rain_graph import RainGraph +from .utils import (disable_from_config, get_config_value, preferred_language) _LOGGER = logging.getLogger(__name__) @@ -50,7 +40,7 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator): # Polling interval. Will only be polled if there are subscribers. update_interval=timedelta(minutes=7), ) - self._api_client = IrmKmiApiClient(session=async_get_clientsession(hass)) + self._api = IrmKmiApiClientHa(session=async_get_clientsession(hass), user_agent=USER_AGENT, cdt_map=CDT_MAP) self._zone = get_config_value(entry, CONF_ZONE) self._dark_mode = get_config_value(entry, CONF_DARK_MODE) self._style = get_config_value(entry, CONF_STYLE) @@ -67,19 +57,17 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator): This is the place to pre-process the data to lookup tables so entities can quickly look up their data. """ - self._api_client.expire_cache() + self._api.expire_cache() if (zone := self.hass.states.get(self._zone)) is None: raise UpdateFailed(f"Zone '{self._zone}' not found") try: # Note: asyncio.TimeoutError and aiohttp.ClientError are already # handled by the data update coordinator. async with async_timeout.timeout(60): - api_data = await self._api_client.get_forecasts_coord( + await self._api.refresh_forecasts_coord( {'lat': zone.attributes[ATTR_LATITUDE], 'long': zone.attributes[ATTR_LONGITUDE]} ) - _LOGGER.debug(f"Observation for {api_data.get('cityName', '')}: {api_data.get('obs', '{}')}") - _LOGGER.debug(f"Full data: {api_data}") except IrmKmiApiError as err: if self.last_update_success_time is not None \ @@ -90,7 +78,7 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator): raise UpdateFailed(f"Error communicating with API for general forecast: {err}. " f"Last success time is: {self.last_update_success_time}") - if api_data.get('cityName', None) in OUT_OF_BENELUX: + if self._api.get_city() in OUT_OF_BENELUX: _LOGGER.error(f"The zone {self._zone} is now out of Benelux and forecast is only available in Benelux. " f"Associated device is now disabled. Move the zone back in Benelux and re-enable to fix " f"this") @@ -108,414 +96,38 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator): ) return ProcessedCoordinatorData() - return await self.process_api_data(api_data) + return await self.process_api_data() async def async_refresh(self) -> None: """Refresh data and log errors.""" await self._async_refresh(log_failures=True, raise_on_entry_error=True) - async def _async_animation_data(self, api_data: dict) -> RainGraph | None: - """From the API data passed in, call the API to get all the images and create the radar animation data object. - Frames from the API are merged with the background map and the location marker to create each frame.""" - animation_data = api_data.get('animation', {}).get('sequence') - localisation_layer_url = api_data.get('animation', {}).get('localisationLayer') - country = api_data.get('country', '') - - if animation_data is None or localisation_layer_url is None or not isinstance(animation_data, list): - return None - - localisation = self.merge_url_and_params(localisation_layer_url, - {'th': 'd' if country == 'NL' or not self._dark_mode else 'n'}) - images_from_api = [self.merge_url_and_params(frame.get('uri'), {'rs': STYLE_TO_PARAM_MAP[self._style]}) - for frame in animation_data if frame is not None and frame.get('uri') is not None - ] - + async def process_api_data(self) -> ProcessedCoordinatorData: + """From the API data, create the object that will be used in the entities""" + tz = await dt.async_get_time_zone('Europe/Brussels') lang = preferred_language(self.hass, self.config_entry) - radar_animation = RadarAnimationData( - hint=api_data.get('animation', {}).get('sequenceHint', {}).get(lang), - unit=api_data.get('animation', {}).get('unit', {}).get(lang), - location=localisation - ) - rain_graph: RainGraph = await self.create_rain_graph(radar_animation, animation_data, country, images_from_api) - return rain_graph - - @staticmethod - def merge_url_and_params(url, params): - parsed_url = urllib.parse.urlparse(url) - query_params = urllib.parse.parse_qs(parsed_url.query) - query_params.update(params) - new_query = urllib.parse.urlencode(query_params, doseq=True) - new_url = parsed_url._replace(query=new_query) - return str(urllib.parse.urlunparse(new_url)) - - async def _async_pollen_data(self, api_data: dict) -> dict: - """Get SVG pollen info from the API, return the pollen data dict""" - _LOGGER.debug("Getting pollen data from API") - svg_url = None - for module in api_data.get('module', []): - _LOGGER.debug(f"module: {module}") - if module.get('type', None) == 'svg': - url = module.get('data', {}).get('url', {}).get('en', '') - if 'pollen' in url: - svg_url = url - break - if svg_url is None: - return PollenParser.get_default_data() - try: - _LOGGER.debug(f"Requesting pollen SVG at url {svg_url}") - pollen_svg: str = await self._api_client.get_svg(svg_url) + pollen = await self._api.get_pollen() except IrmKmiApiError as err: _LOGGER.warning(f"Could not get pollen data from the API: {err}. Keeping the same data.") - return self.data.get('pollen', PollenParser.get_unavailable_data()) \ + pollen = self.data.get('pollen', PollenParser.get_unavailable_data()) \ if self.data is not None else PollenParser.get_unavailable_data() - return PollenParser(pollen_svg).get_pollen_data() + try: + radar_animation, image_path, bg_size = await self._api.get_animation_data(tz, lang, self._style, + self._dark_mode) + animation = await RainGraph(radar_animation, image_path, bg_size, tz=tz, dark_mode=self._dark_mode, + api_client=self._api).build() + except ValueError: + animation = None - async def process_api_data(self, api_data: dict) -> ProcessedCoordinatorData: - """From the API data, create the object that will be used in the entities""" return ProcessedCoordinatorData( - current_weather=await IrmKmiCoordinator.current_weather_from_data(api_data), - daily_forecast=await self.daily_list_to_forecast(api_data.get('for', {}).get('daily')), - hourly_forecast=await IrmKmiCoordinator.hourly_list_to_forecast(api_data.get('for', {}).get('hourly')), - radar_forecast=IrmKmiCoordinator.radar_list_to_forecast(api_data.get('animation', {})), - animation=await self._async_animation_data(api_data=api_data), - warnings=self.warnings_from_data(api_data.get('for', {}).get('warning')), - pollen=await self._async_pollen_data(api_data=api_data), - country=api_data.get('country') + current_weather=await self._api.get_current_weather(tz), + daily_forecast=await self._api.get_daily_forecast(tz, lang), + hourly_forecast=await self._api.get_hourly_forecast(tz), + radar_forecast=self._api.get_radar_forecast(), + animation=animation, + warnings=self._api.get_warnings(lang), + pollen=pollen, + country=self._api.get_country() ) - - - @staticmethod - async def current_weather_from_data(api_data: dict) -> CurrentWeatherData: - """Parse the API data to build a CurrentWeatherData.""" - # Process data to get current hour forecast - now_hourly = None - hourly_forecast_data = api_data.get('for', {}).get('hourly') - tz = await dt.async_get_time_zone('Europe/Brussels') - now = dt.now(time_zone=tz) - if not (hourly_forecast_data is None - or not isinstance(hourly_forecast_data, list) - or len(hourly_forecast_data) == 0): - - for current in hourly_forecast_data[:4]: - if now.strftime('%H') == current['hour']: - now_hourly = current - break - # Get UV index - module_data = api_data.get('module', None) - uv_index = None - if not (module_data is None or not isinstance(module_data, list)): - for module in module_data: - if module.get('type', None) == 'uv': - uv_index = module.get('data', {}).get('levelValue') - - try: - pressure = float(now_hourly.get('pressure', None)) if now_hourly is not None else None - except (TypeError, ValueError): - pressure = None - - try: - wind_speed = float(now_hourly.get('windSpeedKm', None)) if now_hourly is not None else None - except (TypeError, ValueError): - wind_speed = None - - try: - wind_gust_speed = float(now_hourly.get('windPeakSpeedKm', None)) if now_hourly is not None else None - except (TypeError, ValueError): - wind_gust_speed = None - - try: - temperature = float(api_data.get('obs', {}).get('temp')) - except (TypeError, ValueError): - temperature = None - - try: - dir_cardinal = now_hourly.get('windDirectionText', {}).get('en') if now_hourly is not None else None - if dir_cardinal == 'VAR' or now_hourly is None: - wind_bearing = None - else: - wind_bearing = (float(now_hourly.get('windDirection')) + 180) % 360 - except (TypeError, ValueError): - wind_bearing = None - - current_weather = CurrentWeatherData( - condition=CDT_MAP.get((api_data.get('obs', {}).get('ww'), api_data.get('obs', {}).get('dayNight')), None), - temperature=temperature, - wind_speed=wind_speed, - wind_gust_speed=wind_gust_speed, - wind_bearing=wind_bearing, - pressure=pressure, - uv_index=uv_index - ) - - if api_data.get('country', '') == 'NL': - current_weather['wind_speed'] = api_data.get('obs', {}).get('windSpeedKm') - if api_data.get('obs', {}).get('windDirectionText', {}).get('en') == 'VAR': - current_weather['wind_bearing'] = None - else: - try: - current_weather['wind_bearing'] = (float(api_data.get('obs', {}).get('windDirection')) + 180) % 360 - except ValueError: - current_weather['wind_bearing'] = None - - # Since June 2024, the NL weather does not include the condition in the 'ww' key, so we get it from the current - # hourly forecast instead if it is missing - if current_weather['condition'] is None: - try: - current_weather['condition'] = CDT_MAP.get((int(now_hourly.get('ww')), now_hourly.get('dayNight'))) - except (TypeError, ValueError, AttributeError): - current_weather['condition'] = None - - return current_weather - - @staticmethod - async def hourly_list_to_forecast(data: List[dict] | None) -> List[Forecast] | None: - """Parse data from the API to create a list of hourly forecasts""" - if data is None or not isinstance(data, list) or len(data) == 0: - return None - - forecasts = list() - tz = await dt.async_get_time_zone('Europe/Brussels') - day = dt.now(time_zone=tz).replace(hour=0, minute=0, second=0, microsecond=0) - - for idx, f in enumerate(data): - if 'dateShow' in f and idx > 0: - day = day + timedelta(days=1) - - 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: - precipitation_probability = int(f.get('precipChance')) - - ww = None - if f.get('ww', None) is not None: - ww = int(f.get('ww')) - - wind_bearing = None - if f.get('windDirectionText', {}).get('en') != 'VAR': - try: - wind_bearing = (float(f.get('windDirection')) + 180) % 360 - except (TypeError, ValueError): - pass - - forecast = Forecast( - 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), - native_templow=None, - native_wind_gust_speed=f.get('windPeakSpeedKm', None), - native_wind_speed=f.get('windSpeedKm', None), - precipitation_probability=precipitation_probability, - wind_bearing=wind_bearing, - native_pressure=f.get('pressure', None), - is_daytime=f.get('dayNight', None) == 'd' - ) - - forecasts.append(forecast) - - return forecasts - - @staticmethod - def radar_list_to_forecast(data: dict | None) -> List[IrmKmiRadarForecast] | None: - """Create a list of short term forecasts for rain based on the data provided by the rain radar""" - if data is None: - return None - sequence = data.get("sequence", []) - unit = data.get("unit", {}).get("en", None) - ratios = [f['value'] / f['position'] for f in sequence if f['position'] > 0] - - if len(ratios) > 0: - ratio = mean(ratios) - else: - ratio = 0 - - forecast = list() - for f in sequence: - forecast.append( - IrmKmiRadarForecast( - datetime=f.get("time"), - native_precipitation=f.get('value'), - rain_forecast_max=round(f.get('positionHigher') * ratio, 2), - rain_forecast_min=round(f.get('positionLower') * ratio, 2), - might_rain=f.get('positionHigher') > 0, - unit=unit - ) - ) - return forecast - - async def daily_list_to_forecast(self, data: List[dict] | None) -> List[IrmKmiForecast] | None: - """Parse data from the API to create a list of daily forecasts""" - if data is None or not isinstance(data, list) or len(data) == 0: - return None - - forecasts = list() - lang = preferred_language(self.hass, self.config_entry) - tz = await dt.async_get_time_zone('Europe/Brussels') - forecast_day = dt.now(tz) - - for (idx, f) in enumerate(data): - precipitation = None - if f.get('precipQuantity', None) is not None: - try: - precipitation = float(f.get('precipQuantity')) - except (TypeError, ValueError): - pass - - native_wind_gust_speed = None - if f.get('wind', {}).get('peakSpeed') is not None: - try: - native_wind_gust_speed = int(f.get('wind', {}).get('peakSpeed')) - except (TypeError, ValueError): - pass - - wind_bearing = None - if f.get('wind', {}).get('dirText', {}).get('en') != 'VAR': - try: - wind_bearing = (float(f.get('wind', {}).get('dir')) + 180) % 360 - except (TypeError, ValueError): - pass - - is_daytime = f.get('dayNight', None) == 'd' - - day_name = f.get('dayName', {}).get('en', None) - timestamp = f.get('timestamp', None) - if timestamp is not None: - forecast_day = datetime.fromisoformat(timestamp) - elif day_name in WEEKDAYS: - forecast_day = next_weekday(forecast_day, WEEKDAYS.index(day_name)) - elif day_name in ['Today', 'Tonight']: - forecast_day = dt.now(tz) - elif day_name == 'Tomorrow': - forecast_day = dt.now(tz) + timedelta(days=1) - - sunrise_sec = f.get('dawnRiseSeconds', None) - if sunrise_sec is None: - sunrise_sec = f.get('sunRise', None) - sunrise = None - if sunrise_sec is not None: - try: - sunrise = (forecast_day.replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=tz) - + timedelta(seconds=float(sunrise_sec))) - except (TypeError, ValueError): - pass - - sunset_sec = f.get('dawnSetSeconds', None) - if sunset_sec is None: - sunset_sec = f.get('sunSet', None) - sunset = None - if sunset_sec is not None: - try: - sunset = (forecast_day.replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=tz) - + timedelta(seconds=float(sunset_sec))) - except (TypeError, ValueError): - pass - - forecast = IrmKmiForecast( - datetime=(forecast_day.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), - native_templow=f.get('tempMin', None), - native_wind_gust_speed=native_wind_gust_speed, - native_wind_speed=f.get('wind', {}).get('speed'), - precipitation_probability=f.get('precipChance', None), - wind_bearing=wind_bearing, - is_daytime=is_daytime, - text=f.get('text', {}).get(lang, ""), - sunrise=sunrise.isoformat() if sunrise is not None else None, - sunset=sunset.isoformat() if sunset is not None else None - ) - # Swap temperature and templow if needed - if (forecast['native_templow'] is not None - and forecast['native_temperature'] is not None - and forecast['native_templow'] > forecast['native_temperature']): - (forecast['native_templow'], forecast['native_temperature']) = \ - (forecast['native_temperature'], forecast['native_templow']) - - forecasts.append(forecast) - - return forecasts - - async def create_rain_graph(self, - radar_animation: RadarAnimationData, - api_animation_data: List[dict], - country: str, - images_from_api: list[str], - ) -> RainGraph: - """Create a RainGraph object that is ready to output animated and still SVG images""" - sequence: List[AnimationFrameData] = list() - - tz = await dt.async_get_time_zone(self.hass.config.time_zone) - current_time = dt.now(time_zone=tz) - most_recent_frame = None - - for idx, item in enumerate(api_animation_data): - frame = AnimationFrameData( - image=images_from_api[idx], - time=datetime.fromisoformat(item.get('time')) if item.get('time', None) is not None else None, - value=item.get('value', 0), - position=item.get('position', 0), - position_lower=item.get('positionLower', 0), - position_higher=item.get('positionHigher', 0) - ) - sequence.append(frame) - - if most_recent_frame is None and current_time < frame['time']: - most_recent_frame = idx - 1 if idx > 0 else idx - - radar_animation['sequence'] = sequence - radar_animation['most_recent_image_idx'] = most_recent_frame - - satellite_mode = self._style == OPTION_STYLE_SATELLITE - - if country == 'NL': - image_path = "custom_components/irm_kmi/resources/nl.png" - bg_size = (640, 600) - else: - image_path = (f"custom_components/irm_kmi/resources/be_" - f"{'satellite' if satellite_mode else 'black' if self._dark_mode else 'white'}.png") - bg_size = (640, 490) - - return await RainGraph(radar_animation, image_path, bg_size, tz=tz, config_dir=self.hass.config.config_dir, - dark_mode=self._dark_mode, api_client=self._api_client).build() - - def warnings_from_data(self, warning_data: list | None) -> List[WarningData]: - """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 [] - - lang = preferred_language(self.hass, self.config_entry) - 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(lang, ''), - text=data.get('text', {}).get(lang, ''), - starts_at=start, - ends_at=end - ) - ) - - return result if len(result) > 0 else [] diff --git a/custom_components/irm_kmi/data.py b/custom_components/irm_kmi/data.py index 9db8d80..03a2536 100644 --- a/custom_components/irm_kmi/data.py +++ b/custom_components/irm_kmi/data.py @@ -1,41 +1,9 @@ -"""Data classes for IRM KMI integration""" -from datetime import datetime -from typing import List, TypedDict +from typing import TypedDict, List from homeassistant.components.weather import Forecast -from .rain_graph import RainGraph - - -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? - text: str | None - sunrise: str | None - sunset: str | None - - -class CurrentWeatherData(TypedDict, total=False): - """Class to hold the currently observable weather at a given location""" - condition: str | None - temperature: float | None - wind_speed: float | None - wind_gust_speed: float | None - wind_bearing: float | str | None - uv_index: float | None - pressure: float | 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 +from .irm_kmi_api.data import CurrentWeatherData, IrmKmiForecast, WarningData +from .irm_kmi_api.rain_graph import RainGraph class ProcessedCoordinatorData(TypedDict, total=False): diff --git a/custom_components/irm_kmi/irm_kmi_api/__init__.py b/custom_components/irm_kmi/irm_kmi_api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/custom_components/irm_kmi/irm_kmi_api/api.py b/custom_components/irm_kmi/irm_kmi_api/api.py new file mode 100644 index 0000000..ef2b9a8 --- /dev/null +++ b/custom_components/irm_kmi/irm_kmi_api/api.py @@ -0,0 +1,562 @@ +"""API Client for IRM KMI weather""" +import asyncio +import hashlib +import json +import logging +import socket +import time +import urllib.parse +from datetime import datetime, timedelta +from statistics import mean +from typing import List, Tuple +from zoneinfo import ZoneInfo + +import aiohttp +import async_timeout + +from custom_components.irm_kmi.irm_kmi_api.const import WEEKDAYS, STYLE_TO_PARAM_MAP, OPTION_STYLE_SATELLITE, \ + MAP_WARNING_ID_TO_SLUG as SLUG_MAP +from custom_components.irm_kmi.irm_kmi_api.data import CurrentWeatherData, IrmKmiForecast, Forecast, \ + IrmKmiRadarForecast, RadarAnimationData, AnimationFrameData, WarningData +from custom_components.irm_kmi.irm_kmi_api.pollen import PollenParser +from custom_components.irm_kmi.utils import next_weekday + +_LOGGER = logging.getLogger(__name__) + + +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.""" + + +def _api_key(method_name: str) -> str: + """Get API key.""" + return hashlib.md5(f"r9EnW374jkJ9acc;{method_name};{datetime.now().strftime('%d/%m/%Y')}".encode()).hexdigest() + + +class IrmKmiApiClient: + """API client for IRM KMI weather data""" + COORD_DECIMALS = 6 + cache_max_age = 60 * 60 * 2 # Remove items from the cache if they have not been hit since 2 hours + cache = {} + + def __init__(self, session: aiohttp.ClientSession, user_agent: str) -> None: + self._session = session + self._base_url = "https://app.meteo.be/services/appv4/" + self._user_agent = user_agent + + async def get_forecasts_coord(self, coord: dict) -> dict: + """Get forecasts for given city.""" + assert 'lat' in coord + assert 'long' in coord + coord['lat'] = round(coord['lat'], self.COORD_DECIMALS) + coord['long'] = round(coord['long'], self.COORD_DECIMALS) + + response: bytes = await self._api_wrapper(params={"s": "getForecasts", "k": _api_key("getForecasts")} | coord) + response: dict = json.loads(response) + + _LOGGER.debug(f"Observation for {response.get('cityName', '')}: {response.get('obs', '{}')}") + _LOGGER.debug(f"Full data: {response}") + return response + + async def get_image(self, url, params: dict | None = None) -> bytes: + """Get the image at the specified url with the parameters""" + r: bytes = await self._api_wrapper(base_url=url, params={} if params is None else params) + return r + + async def get_svg(self, url, params: dict | None = None) -> str: + """Get SVG as str at the specified url with the parameters""" + r: bytes = await self._api_wrapper(base_url=url, params={} if params is None else params) + return r.decode() + + async def _api_wrapper( + self, + params: dict, + base_url: str | None = None, + path: str = "", + method: str = "get", + data: dict | None = None, + headers: dict | None = None, + ) -> bytes: + """Get information from the API.""" + url = f"{self._base_url if base_url is None else base_url}{path}" + + if headers is None: + headers = {'User-Agent': self._user_agent} + else: + headers['User-Agent'] = self._user_agent + + if url in self.cache: + headers['If-None-Match'] = self.cache[url]['etag'] + + try: + async with async_timeout.timeout(60): + response = await self._session.request( + method=method, + url=url, + headers=headers, + json=data, + params=params + ) + response.raise_for_status() + + if response.status == 304: + _LOGGER.debug(f"Cache hit for {url}") + self.cache[url]['timestamp'] = time.time() + return self.cache[url]['response'] + + if 'ETag' in response.headers: + _LOGGER.debug(f"Saving in cache {url}") + r = await response.read() + self.cache[url] = {'etag': response.headers['ETag'], 'response': r, 'timestamp': time.time()} + return r + + return await response.read() + + 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(f"Something really wrong happened! {exception}") from exception + + def expire_cache(self): + now = time.time() + keys_to_delete = set() + for key, value in self.cache.items(): + if now - value['timestamp'] > self.cache_max_age: + keys_to_delete.add(key) + for key in keys_to_delete: + del self.cache[key] + _LOGGER.info(f"Expired {len(keys_to_delete)} elements from API cache") + + +class IrmKmiApiClientHa(IrmKmiApiClient): + def __init__(self, session: aiohttp.ClientSession, user_agent: str, cdt_map: dict) -> None: + super().__init__(session, user_agent) + self._api_data = dict() + self._cdt_map = cdt_map + + async def refresh_forecasts_coord(self, coord: dict) -> None: + self._api_data = await self.get_forecasts_coord(coord) + + def get_city(self) -> str | None: + return self._api_data.get('cityName', None) + + def get_country(self) -> str | None: + return self._api_data.get('country', None) + + async def get_current_weather(self, tz: ZoneInfo) -> CurrentWeatherData: + """Parse the API data to build a CurrentWeatherData.""" + + now_hourly = await self._get_now_hourly(tz) + uv_index = await self._get_uv_index() + + try: + pressure = float(now_hourly.get('pressure', None)) if now_hourly is not None else None + except (TypeError, ValueError): + pressure = None + + try: + wind_speed = float(now_hourly.get('windSpeedKm', None)) if now_hourly is not None else None + except (TypeError, ValueError): + wind_speed = None + + try: + wind_gust_speed = float(now_hourly.get('windPeakSpeedKm', None)) if now_hourly is not None else None + except (TypeError, ValueError): + wind_gust_speed = None + + try: + temperature = float(self._api_data.get('obs', {}).get('temp')) + except (TypeError, ValueError): + temperature = None + + try: + dir_cardinal = now_hourly.get('windDirectionText', {}).get('en') if now_hourly is not None else None + if dir_cardinal == 'VAR' or now_hourly is None: + wind_bearing = None + else: + wind_bearing = (float(now_hourly.get('windDirection')) + 180) % 360 + except (TypeError, ValueError): + wind_bearing = None + + current_weather = CurrentWeatherData( + condition=self._cdt_map.get( + (self._api_data.get('obs', {}).get('ww'), self._api_data.get('obs', {}).get('dayNight')), None), + temperature=temperature, + wind_speed=wind_speed, + wind_gust_speed=wind_gust_speed, + wind_bearing=wind_bearing, + pressure=pressure, + uv_index=uv_index + ) + + if self._api_data.get('country', '') == 'NL': + current_weather['wind_speed'] = self._api_data.get('obs', {}).get('windSpeedKm') + if self._api_data.get('obs', {}).get('windDirectionText', {}).get('en') == 'VAR': + current_weather['wind_bearing'] = None + else: + try: + current_weather['wind_bearing'] = (float( + self._api_data.get('obs', {}).get('windDirection')) + 180) % 360 + except ValueError: + current_weather['wind_bearing'] = None + + # Since June 2024, the NL weather does not include the condition in the 'ww' key, so we get it from the current + # hourly forecast instead if it is missing + if current_weather['condition'] is None: + try: + current_weather['condition'] = self._cdt_map.get( + (int(now_hourly.get('ww')), now_hourly.get('dayNight')), None) + except (TypeError, ValueError, AttributeError): + current_weather['condition'] = None + + return current_weather + + async def _get_uv_index(self) -> float | None: + uv_index = None + module_data = self._api_data.get('module', None) + if not (module_data is None or not isinstance(module_data, list)): + for module in module_data: + if module.get('type', None) == 'uv': + uv_index = module.get('data', {}).get('levelValue') + return uv_index + + async def _get_now_hourly(self, tz: ZoneInfo) -> dict | None: + now_hourly = None + hourly_forecast_data = self._api_data.get('for', {}).get('hourly') + now = datetime.now(tz) + if not (hourly_forecast_data is None + or not isinstance(hourly_forecast_data, list) + or len(hourly_forecast_data) == 0): + + for current in hourly_forecast_data[:4]: + if now.strftime('%H') == current['hour']: + now_hourly = current + break + return now_hourly + + async def get_daily_forecast(self, tz: ZoneInfo, lang: str) -> List[IrmKmiForecast] | None: + """Parse data from the API to create a list of daily forecasts""" + data = self._api_data.get('for', {}).get('daily') + if data is None or not isinstance(data, list) or len(data) == 0: + return None + + forecasts = list() + forecast_day = datetime.now(tz) + + for (idx, f) in enumerate(data): + precipitation = None + if f.get('precipQuantity', None) is not None: + try: + precipitation = float(f.get('precipQuantity')) + except (TypeError, ValueError): + pass + + native_wind_gust_speed = None + if f.get('wind', {}).get('peakSpeed') is not None: + try: + native_wind_gust_speed = int(f.get('wind', {}).get('peakSpeed')) + except (TypeError, ValueError): + pass + + wind_bearing = None + if f.get('wind', {}).get('dirText', {}).get('en') != 'VAR': + try: + wind_bearing = (float(f.get('wind', {}).get('dir')) + 180) % 360 + except (TypeError, ValueError): + pass + + is_daytime = f.get('dayNight', None) == 'd' + + day_name = f.get('dayName', {}).get('en', None) + timestamp = f.get('timestamp', None) + if timestamp is not None: + forecast_day = datetime.fromisoformat(timestamp) + elif day_name in WEEKDAYS: + forecast_day = next_weekday(forecast_day, WEEKDAYS.index(day_name)) + elif day_name in ['Today', 'Tonight']: + forecast_day = datetime.now(tz) + elif day_name == 'Tomorrow': + forecast_day = datetime.now(tz) + timedelta(days=1) + + sunrise_sec = f.get('dawnRiseSeconds', None) + if sunrise_sec is None: + sunrise_sec = f.get('sunRise', None) + sunrise = None + if sunrise_sec is not None: + try: + sunrise = (forecast_day.replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=tz) + + timedelta(seconds=float(sunrise_sec))) + except (TypeError, ValueError): + pass + + sunset_sec = f.get('dawnSetSeconds', None) + if sunset_sec is None: + sunset_sec = f.get('sunSet', None) + sunset = None + if sunset_sec is not None: + try: + sunset = (forecast_day.replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=tz) + + timedelta(seconds=float(sunset_sec))) + except (TypeError, ValueError): + pass + + forecast = IrmKmiForecast( + datetime=(forecast_day.strftime('%Y-%m-%d')), + condition=self._cdt_map.get((f.get('ww1', None), f.get('dayNight', None)), None), + native_precipitation=precipitation, + native_temperature=f.get('tempMax', None), + native_templow=f.get('tempMin', None), + native_wind_gust_speed=native_wind_gust_speed, + native_wind_speed=f.get('wind', {}).get('speed'), + precipitation_probability=f.get('precipChance', None), + wind_bearing=wind_bearing, + is_daytime=is_daytime, + text=f.get('text', {}).get(lang, ""), + sunrise=sunrise.isoformat() if sunrise is not None else None, + sunset=sunset.isoformat() if sunset is not None else None + ) + # Swap temperature and templow if needed + if (forecast['native_templow'] is not None + and forecast['native_temperature'] is not None + and forecast['native_templow'] > forecast['native_temperature']): + (forecast['native_templow'], forecast['native_temperature']) = \ + (forecast['native_temperature'], forecast['native_templow']) + + forecasts.append(forecast) + + return forecasts + + async def get_hourly_forecast(self, tz: ZoneInfo) -> List[Forecast] | None: + """Parse data from the API to create a list of hourly forecasts""" + data = self._api_data.get('for', {}).get('hourly') + + if data is None or not isinstance(data, list) or len(data) == 0: + return None + + forecasts = list() + day = datetime.now(tz).replace(hour=0, minute=0, second=0, microsecond=0) + + for idx, f in enumerate(data): + if 'dateShow' in f and idx > 0: + day = day + timedelta(days=1) + + 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: + precipitation_probability = int(f.get('precipChance')) + + ww = None + if f.get('ww', None) is not None: + ww = int(f.get('ww')) + + wind_bearing = None + if f.get('windDirectionText', {}).get('en') != 'VAR': + try: + wind_bearing = (float(f.get('windDirection')) + 180) % 360 + except (TypeError, ValueError): + pass + + forecast = Forecast( + datetime=day.isoformat(), + condition=self._cdt_map.get((ww, f.get('dayNight', None)), None), + native_precipitation=f.get('precipQuantity', None), + native_temperature=f.get('temp', None), + native_templow=None, + native_wind_gust_speed=f.get('windPeakSpeedKm', None), + native_wind_speed=f.get('windSpeedKm', None), + precipitation_probability=precipitation_probability, + wind_bearing=wind_bearing, + native_pressure=f.get('pressure', None), + is_daytime=f.get('dayNight', None) == 'd' + ) + + forecasts.append(forecast) + + return forecasts + + def get_radar_forecast(self) -> List[IrmKmiRadarForecast] | None: + """Create a list of short term forecasts for rain based on the data provided by the rain radar""" + data = self._api_data.get('animation', {}) + + if data is None: + return None + sequence = data.get("sequence", []) + unit = data.get("unit", {}).get("en", None) + ratios = [f['value'] / f['position'] for f in sequence if f['position'] > 0] + + if len(ratios) > 0: + ratio = mean(ratios) + else: + ratio = 0 + + forecast = list() + for f in sequence: + forecast.append( + IrmKmiRadarForecast( + datetime=f.get("time"), + native_precipitation=f.get('value'), + rain_forecast_max=round(f.get('positionHigher') * ratio, 2), + rain_forecast_min=round(f.get('positionLower') * ratio, 2), + might_rain=f.get('positionHigher') > 0, + unit=unit + ) + ) + return forecast + + async def get_animation_data(self, tz: ZoneInfo, lang: str, style: str, dark_mode: bool) -> (RadarAnimationData, + str, Tuple[int, int]): + """From the API data passed in, call the API to get all the images and create the radar animation data object. + Frames from the API are merged with the background map and the location marker to create each frame.""" + animation_data = self._api_data.get('animation', {}).get('sequence') + localisation_layer_url = self._api_data.get('animation', {}).get('localisationLayer') + country = self.get_country() + + if animation_data is None or localisation_layer_url is None or not isinstance(animation_data, list): + raise ValueError("Cannot create animation data") + + localisation = self.merge_url_and_params(localisation_layer_url, + {'th': 'd' if country == 'NL' or not dark_mode else 'n'}) + images_from_api = [self.merge_url_and_params(frame.get('uri'), {'rs': STYLE_TO_PARAM_MAP[style]}) + for frame in animation_data if frame is not None and frame.get('uri') is not None + ] + + radar_animation = RadarAnimationData( + hint=self._api_data.get('animation', {}).get('sequenceHint', {}).get(lang), + unit=self._api_data.get('animation', {}).get('unit', {}).get(lang), + location=localisation + ) + + r = self._get_rain_graph_data( + radar_animation, + animation_data, + country, + images_from_api, + tz, + style, + dark_mode) + + return r + + def get_warnings(self, lang: str) -> List[WarningData]: + """Create a list of warning data instances based on the api data""" + warning_data = self._api_data.get('for', {}).get('warning') + if warning_data is None or not isinstance(warning_data, list) or len(warning_data) == 0: + return [] + + 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(lang, ''), + text=data.get('text', {}).get(lang, ''), + starts_at=start, + ends_at=end + ) + ) + + return result if len(result) > 0 else [] + + async def get_pollen(self) -> dict: + """Get SVG pollen info from the API, return the pollen data dict""" + _LOGGER.debug("Getting pollen data from API") + svg_url = None + for module in self._api_data.get('module', []): + _LOGGER.debug(f"module: {module}") + if module.get('type', None) == 'svg': + url = module.get('data', {}).get('url', {}).get('en', '') + if 'pollen' in url: + svg_url = url + break + if svg_url is None: + return PollenParser.get_default_data() + + try: + _LOGGER.debug(f"Requesting pollen SVG at url {svg_url}") + pollen_svg: str = await self.get_svg(svg_url) + except IrmKmiApiError as err: + raise err + + return PollenParser(pollen_svg).get_pollen_data() + + @staticmethod + def merge_url_and_params(url, params): + parsed_url = urllib.parse.urlparse(url) + query_params = urllib.parse.parse_qs(parsed_url.query) + query_params.update(params) + new_query = urllib.parse.urlencode(query_params, doseq=True) + new_url = parsed_url._replace(query=new_query) + return str(urllib.parse.urlunparse(new_url)) + + @staticmethod + def _get_rain_graph_data(radar_animation: RadarAnimationData, + api_animation_data: List[dict], + country: str | None, + images_from_api: list[str], + tz: ZoneInfo, + style: str, + dark_mode: bool + ) -> (RadarAnimationData, str, Tuple[int, int]): + """Create a RainGraph object that is ready to output animated and still SVG images""" + sequence: List[AnimationFrameData] = list() + + current_time = datetime.now(tz) + most_recent_frame = None + + for idx, item in enumerate(api_animation_data): + frame = AnimationFrameData( + image=images_from_api[idx], + time=datetime.fromisoformat(item.get('time')) if item.get('time', None) is not None else None, + value=item.get('value', 0), + position=item.get('position', 0), + position_lower=item.get('positionLower', 0), + position_higher=item.get('positionHigher', 0) + ) + sequence.append(frame) + + if most_recent_frame is None and current_time < frame['time']: + most_recent_frame = idx - 1 if idx > 0 else idx + + radar_animation['sequence'] = sequence + radar_animation['most_recent_image_idx'] = most_recent_frame + + satellite_mode = style == OPTION_STYLE_SATELLITE + + if country == 'NL': + image_path = "custom_components/irm_kmi/resources/nl.png" + bg_size = (640, 600) + else: + image_path = (f"custom_components/irm_kmi/resources/be_" + f"{'satellite' if satellite_mode else 'black' if dark_mode else 'white'}.png") + bg_size = (640, 490) + + return radar_animation, image_path, bg_size diff --git a/custom_components/irm_kmi/irm_kmi_api/const.py b/custom_components/irm_kmi/irm_kmi_api/const.py new file mode 100644 index 0000000..8060220 --- /dev/null +++ b/custom_components/irm_kmi/irm_kmi_api/const.py @@ -0,0 +1,28 @@ +from typing import Final + +POLLEN_NAMES: Final = {'Alder', 'Ash', 'Birch', 'Grasses', 'Hazel', 'Mugwort', 'Oak'} +POLLEN_LEVEL_TO_COLOR = {'null': 'green', 'low': 'yellow', 'moderate': 'orange', 'high': 'red', 'very high': 'purple', + 'active': 'active'} +WEEKDAYS = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'] +OPTION_STYLE_STD: Final = 'standard_style' +OPTION_STYLE_CONTRAST: Final = 'contrast_style' +OPTION_STYLE_YELLOW_RED: Final = 'yellow_red_style' +OPTION_STYLE_SATELLITE: Final = 'satellite_style' +STYLE_TO_PARAM_MAP: Final = { + OPTION_STYLE_STD: 1, + OPTION_STYLE_CONTRAST: 2, + OPTION_STYLE_YELLOW_RED: 3, + OPTION_STYLE_SATELLITE: 4 +} +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'} diff --git a/custom_components/irm_kmi/irm_kmi_api/data.py b/custom_components/irm_kmi/irm_kmi_api/data.py new file mode 100644 index 0000000..270bd52 --- /dev/null +++ b/custom_components/irm_kmi/irm_kmi_api/data.py @@ -0,0 +1,95 @@ +"""Data classes for IRM KMI integration""" +from datetime import datetime +from typing import TypedDict, Required, List + + +class Forecast(TypedDict, total=False): + """Typed weather forecast dict. + + All attributes are in native units and old attributes kept + for backwards compatibility. + + Data from Home Assistant to avoid to depend on Home Assistant for this + """ + + condition: str | None + datetime: Required[str] + humidity: float | None + precipitation_probability: int | None + cloud_coverage: int | None + native_precipitation: float | None + precipitation: None + native_pressure: float | None + pressure: None + native_temperature: float | None + temperature: None + native_templow: float | None + templow: None + native_apparent_temperature: float | None + wind_bearing: float | str | None + native_wind_gust_speed: float | None + native_wind_speed: float | None + wind_speed: None + native_dew_point: float | None + uv_index: float | None + is_daytime: bool | None # Mandatory to use with forecast_twice_daily + + +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? + text: str | None + sunrise: str | None + sunset: str | None + + +class CurrentWeatherData(TypedDict, total=False): + """Class to hold the currently observable weather at a given location""" + condition: str | None + temperature: float | None + wind_speed: float | None + wind_gust_speed: float | None + wind_bearing: float | str | None + uv_index: float | None + pressure: float | 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 IrmKmiRadarForecast(Forecast): + """Forecast class to handle rain forecast from the IRM KMI rain radar""" + rain_forecast_max: float + rain_forecast_min: float + might_rain: bool + unit: str | None + + +class AnimationFrameData(TypedDict, total=False): + """Holds one single frame of the radar camera, along with the timestamp of the frame""" + time: datetime | None + image: bytes | str | None + value: float | None + position: float | None + position_higher: float | None + position_lower: float | None + + +class RadarAnimationData(TypedDict, total=False): + """Holds frames and additional data for the animation to be rendered""" + sequence: List[AnimationFrameData] | None + most_recent_image_idx: int | None + hint: str | None + unit: str | None + location: bytes | str | None + svg_still: bytes | None + svg_animated: bytes | None diff --git a/custom_components/irm_kmi/irm_kmi_api/ha.py b/custom_components/irm_kmi/irm_kmi_api/ha.py new file mode 100644 index 0000000..139597f --- /dev/null +++ b/custom_components/irm_kmi/irm_kmi_api/ha.py @@ -0,0 +1,2 @@ + + diff --git a/custom_components/irm_kmi/pollen.py b/custom_components/irm_kmi/irm_kmi_api/pollen.py similarity index 98% rename from custom_components/irm_kmi/pollen.py rename to custom_components/irm_kmi/irm_kmi_api/pollen.py index 601bf35..a3f1f09 100644 --- a/custom_components/irm_kmi/pollen.py +++ b/custom_components/irm_kmi/irm_kmi_api/pollen.py @@ -3,7 +3,7 @@ import logging import xml.etree.ElementTree as ET from typing import List -from .const import POLLEN_LEVEL_TO_COLOR, POLLEN_NAMES +from .const import POLLEN_NAMES, POLLEN_LEVEL_TO_COLOR _LOGGER = logging.getLogger(__name__) diff --git a/custom_components/irm_kmi/rain_graph.py b/custom_components/irm_kmi/irm_kmi_api/rain_graph.py similarity index 98% rename from custom_components/irm_kmi/rain_graph.py rename to custom_components/irm_kmi/irm_kmi_api/rain_graph.py index d80545d..8efcc19 100644 --- a/custom_components/irm_kmi/rain_graph.py +++ b/custom_components/irm_kmi/irm_kmi_api/rain_graph.py @@ -13,8 +13,8 @@ from svgwrite.animate import Animate from svgwrite.container import FONT_TEMPLATE from .api import IrmKmiApiClient -from .radar_data import AnimationFrameData, RadarAnimationData -from custom_components.irm_kmi.resources import roboto, be_black, be_satellite, be_white, nl +from .data import AnimationFrameData, RadarAnimationData +from .resources import be_black, be_satellite, be_white, nl, roboto _LOGGER = logging.getLogger(__name__) @@ -24,7 +24,7 @@ class RainGraph: animation_data: RadarAnimationData, background_image_path: str, background_size: (int, int), - config_dir: str = '.', + config_dir: str = '.', # TODO remove ununsed dark_mode: bool = False, tz: datetime.tzinfo = dt.get_default_time_zone(), svg_width: float = 640, @@ -40,7 +40,6 @@ class RainGraph: self._animation_data: RadarAnimationData = animation_data self._background_image_path: str = background_image_path self._background_size: (int, int) = background_size - self._config_dir: str = config_dir self._dark_mode: bool = dark_mode self._tz = tz self._svg_width: float = svg_width diff --git a/custom_components/irm_kmi/irm_kmi_api/resources/__init__.py b/custom_components/irm_kmi/irm_kmi_api/resources/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/custom_components/irm_kmi/resources/be_black.png b/custom_components/irm_kmi/irm_kmi_api/resources/be_black.png similarity index 100% rename from custom_components/irm_kmi/resources/be_black.png rename to custom_components/irm_kmi/irm_kmi_api/resources/be_black.png diff --git a/custom_components/irm_kmi/resources/be_black.py b/custom_components/irm_kmi/irm_kmi_api/resources/be_black.py similarity index 100% rename from custom_components/irm_kmi/resources/be_black.py rename to custom_components/irm_kmi/irm_kmi_api/resources/be_black.py diff --git a/custom_components/irm_kmi/resources/be_satellite.png b/custom_components/irm_kmi/irm_kmi_api/resources/be_satellite.png similarity index 100% rename from custom_components/irm_kmi/resources/be_satellite.png rename to custom_components/irm_kmi/irm_kmi_api/resources/be_satellite.png diff --git a/custom_components/irm_kmi/resources/be_satellite.py b/custom_components/irm_kmi/irm_kmi_api/resources/be_satellite.py similarity index 100% rename from custom_components/irm_kmi/resources/be_satellite.py rename to custom_components/irm_kmi/irm_kmi_api/resources/be_satellite.py diff --git a/custom_components/irm_kmi/resources/be_white.png b/custom_components/irm_kmi/irm_kmi_api/resources/be_white.png similarity index 100% rename from custom_components/irm_kmi/resources/be_white.png rename to custom_components/irm_kmi/irm_kmi_api/resources/be_white.png diff --git a/custom_components/irm_kmi/resources/be_white.py b/custom_components/irm_kmi/irm_kmi_api/resources/be_white.py similarity index 100% rename from custom_components/irm_kmi/resources/be_white.py rename to custom_components/irm_kmi/irm_kmi_api/resources/be_white.py diff --git a/custom_components/irm_kmi/resources/nl.png b/custom_components/irm_kmi/irm_kmi_api/resources/nl.png similarity index 100% rename from custom_components/irm_kmi/resources/nl.png rename to custom_components/irm_kmi/irm_kmi_api/resources/nl.png diff --git a/custom_components/irm_kmi/resources/nl.py b/custom_components/irm_kmi/irm_kmi_api/resources/nl.py similarity index 100% rename from custom_components/irm_kmi/resources/nl.py rename to custom_components/irm_kmi/irm_kmi_api/resources/nl.py diff --git a/custom_components/irm_kmi/resources/roboto.py b/custom_components/irm_kmi/irm_kmi_api/resources/roboto.py similarity index 100% rename from custom_components/irm_kmi/resources/roboto.py rename to custom_components/irm_kmi/irm_kmi_api/resources/roboto.py diff --git a/custom_components/irm_kmi/resources/roboto_medium.ttf b/custom_components/irm_kmi/irm_kmi_api/resources/roboto_medium.ttf similarity index 100% rename from custom_components/irm_kmi/resources/roboto_medium.ttf rename to custom_components/irm_kmi/irm_kmi_api/resources/roboto_medium.ttf diff --git a/custom_components/irm_kmi/radar_data.py b/custom_components/irm_kmi/radar_data.py deleted file mode 100644 index c6f7aad..0000000 --- a/custom_components/irm_kmi/radar_data.py +++ /dev/null @@ -1,35 +0,0 @@ -"""Data classes related to radar forecast for IRM KMI integration""" -# This file was needed to avoid circular import with rain_graph.py and data.py -from datetime import datetime -from typing import TypedDict, List - -from homeassistant.components.weather import Forecast - - -class IrmKmiRadarForecast(Forecast): - """Forecast class to handle rain forecast from the IRM KMI rain radar""" - rain_forecast_max: float - rain_forecast_min: float - might_rain: bool - unit: str | None - - -class AnimationFrameData(TypedDict, total=False): - """Holds one single frame of the radar camera, along with the timestamp of the frame""" - time: datetime | None - image: bytes | str | None - value: float | None - position: float | None - position_higher: float | None - position_lower: float | None - - -class RadarAnimationData(TypedDict, total=False): - """Holds frames and additional data for the animation to be rendered""" - sequence: List[AnimationFrameData] | None - most_recent_image_idx: int | None - hint: str | None - unit: str | None - location: bytes | str | None - svg_still: bytes | None - svg_animated: bytes | None diff --git a/custom_components/irm_kmi/repairs.py b/custom_components/irm_kmi/repairs.py index 99fbaa2..9df4dfd 100644 --- a/custom_components/irm_kmi/repairs.py +++ b/custom_components/irm_kmi/repairs.py @@ -9,12 +9,12 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.selector import SelectSelector, SelectSelectorConfig -from custom_components.irm_kmi import async_reload_entry -from custom_components.irm_kmi.api import IrmKmiApiClient -from custom_components.irm_kmi.const import (OUT_OF_BENELUX, REPAIR_OPT_DELETE, - REPAIR_OPT_MOVE, REPAIR_OPTIONS, - REPAIR_SOLUTION) -from custom_components.irm_kmi.utils import modify_from_config +from . import async_reload_entry +from .const import (OUT_OF_BENELUX, REPAIR_OPT_DELETE, + REPAIR_OPT_MOVE, REPAIR_OPTIONS, + REPAIR_SOLUTION, USER_AGENT) +from .irm_kmi_api.api import IrmKmiApiClient +from .utils import modify_from_config _LOGGER = logging.getLogger(__name__) @@ -50,7 +50,9 @@ class OutOfBeneluxRepairFlow(RepairsFlow): try: async with async_timeout.timeout(10): api_data = await IrmKmiApiClient( - session=async_get_clientsession(self.hass)).get_forecasts_coord( + session=async_get_clientsession(self.hass), + user_agent=USER_AGENT + ).get_forecasts_coord( {'lat': zone.attributes[ATTR_LATITUDE], 'long': zone.attributes[ATTR_LONGITUDE]} ) @@ -84,8 +86,8 @@ class OutOfBeneluxRepairFlow(RepairsFlow): async def async_create_fix_flow( - hass: HomeAssistant, - issue_id: str, + _hass: HomeAssistant, + _issue_id: str, data: dict[str, str | int | float | None] | None, ) -> OutOfBeneluxRepairFlow: """Create flow.""" diff --git a/custom_components/irm_kmi/sensor.py b/custom_components/irm_kmi/sensor.py index fcb11b8..7d994d5 100644 --- a/custom_components/irm_kmi/sensor.py +++ b/custom_components/irm_kmi/sensor.py @@ -10,12 +10,12 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.util import dt -from custom_components.irm_kmi import DOMAIN, IrmKmiCoordinator -from custom_components.irm_kmi.const import POLLEN_NAMES, POLLEN_TO_ICON_MAP, CURRENT_WEATHER_SENSOR_UNITS, \ +from . import DOMAIN, IrmKmiCoordinator +from .const import POLLEN_TO_ICON_MAP, CURRENT_WEATHER_SENSOR_UNITS, \ CURRENT_WEATHER_SENSOR_CLASS, CURRENT_WEATHER_SENSORS, CURRENT_WEATHER_SENSOR_ICON -from custom_components.irm_kmi.data import IrmKmiForecast -from custom_components.irm_kmi.pollen import PollenParser -from custom_components.irm_kmi.radar_data import IrmKmiRadarForecast +from .irm_kmi_api.const import POLLEN_NAMES +from .irm_kmi_api.data import IrmKmiForecast, IrmKmiRadarForecast +from .irm_kmi_api.pollen import PollenParser _LOGGER = logging.getLogger(__name__) diff --git a/tests/conftest.py b/tests/conftest.py index f619cba..35b8806 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,7 +2,7 @@ from __future__ import annotations import json -from collections.abc import Generator +from typing import Generator from unittest.mock import MagicMock, patch import pytest @@ -10,32 +10,22 @@ from homeassistant.const import CONF_ZONE from pytest_homeassistant_custom_component.common import (MockConfigEntry, load_fixture) -from custom_components.irm_kmi.api import (IrmKmiApiError, - IrmKmiApiParametersError) -from custom_components.irm_kmi.const import ( - CONF_DARK_MODE, CONF_LANGUAGE_OVERRIDE, CONF_STYLE, - CONF_USE_DEPRECATED_FORECAST, DOMAIN, OPTION_DEPRECATED_FORECAST_NOT_USED, - OPTION_DEPRECATED_FORECAST_TWICE_DAILY, OPTION_STYLE_STD) +from custom_components.irm_kmi import OPTION_STYLE_STD +from custom_components.irm_kmi.const import (CONF_DARK_MODE, CONF_LANGUAGE_OVERRIDE, CONF_STYLE, + CONF_USE_DEPRECATED_FORECAST, DOMAIN, OPTION_DEPRECATED_FORECAST_NOT_USED, + OPTION_DEPRECATED_FORECAST_TWICE_DAILY, IRM_KMI_TO_HA_CONDITION_MAP) +from custom_components.irm_kmi.irm_kmi_api.api import (IrmKmiApiError, + IrmKmiApiParametersError, IrmKmiApiClientHa) 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" - elif "app.meteo.be/services/appv4/?s=getIncaImage" in url: - file_name = "tests/fixtures/clouds_be.png" - elif "getLocalizationLayerBE" in url: - file_name = "tests/fixtures/loc_layer_be_n.png" - elif "getLocalizationLayerNL" in url: - file_name = "tests/fixtures/loc_layer_nl.png" - else: - raise ValueError(f"Not a valid parameter for the mock: {url}") - - with open(file_name, "rb") as file: - return file.read() +def get_api_with_data(fixture: str) -> IrmKmiApiClientHa: + api = IrmKmiApiClientHa(session=MagicMock(), user_agent='', cdt_map=IRM_KMI_TO_HA_CONDITION_MAP) + api._api_data = get_api_data(fixture) + return api @pytest.fixture(autouse=True) @@ -121,21 +111,7 @@ def mock_irm_kmi_api(request: pytest.FixtureRequest) -> Generator[None, MagicMoc forecast = json.loads(load_fixture(fixture)) with patch( - "custom_components.irm_kmi.coordinator.IrmKmiApiClient", autospec=True - ) as irm_kmi_api_mock: - irm_kmi = irm_kmi_api_mock.return_value - irm_kmi.get_forecasts_coord.return_value = forecast - yield irm_kmi - - -@pytest.fixture() -def mock_irm_kmi_api_coordinator_out_benelux(request: pytest.FixtureRequest) -> Generator[None, MagicMock, None]: - """Return a mocked IrmKmi api client.""" - fixture: str = "forecast_out_of_benelux.json" - - forecast = json.loads(load_fixture(fixture)) - with patch( - "custom_components.irm_kmi.coordinator.IrmKmiApiClient", autospec=True + "custom_components.irm_kmi.coordinator.IrmKmiApiClientHa", autospec=True ) as irm_kmi_api_mock: irm_kmi = irm_kmi_api_mock.return_value irm_kmi.get_forecasts_coord.return_value = forecast @@ -174,111 +150,9 @@ def mock_irm_kmi_api_repair_out_of_benelux(request: pytest.FixtureRequest) -> Ge def mock_exception_irm_kmi_api(request: pytest.FixtureRequest) -> Generator[None, MagicMock, None]: """Return a mocked IrmKmi api client.""" with patch( - "custom_components.irm_kmi.coordinator.IrmKmiApiClient", autospec=True + "custom_components.irm_kmi.coordinator.IrmKmiApiClientHa", autospec=True ) as irm_kmi_api_mock: irm_kmi = irm_kmi_api_mock.return_value - irm_kmi.get_forecasts_coord.side_effect = IrmKmiApiParametersError + irm_kmi.refresh_forecasts_coord.side_effect = IrmKmiApiParametersError yield irm_kmi - -@pytest.fixture() -def mock_image_and_nl_forecast_irm_kmi_api(request: pytest.FixtureRequest) -> Generator[None, MagicMock, None]: - """Return a mocked IrmKmi api client.""" - fixture: str = "forecast_nl.json" - - forecast = json.loads(load_fixture(fixture)) - - with patch( - "custom_components.irm_kmi.coordinator.IrmKmiApiClient", autospec=True - ) as irm_kmi_api_mock: - irm_kmi = irm_kmi_api_mock.return_value - irm_kmi.get_image.side_effect = patched - irm_kmi.get_forecasts_coord.return_value = forecast - yield irm_kmi - - -@pytest.fixture() -def mock_image_and_high_temp_irm_kmi_api(request: pytest.FixtureRequest) -> Generator[None, MagicMock, None]: - """Return a mocked IrmKmi api client.""" - fixture: str = "high_low_temp.json" - - forecast = json.loads(load_fixture(fixture)) - - with patch( - "custom_components.irm_kmi.coordinator.IrmKmiApiClient", autospec=True - ) as irm_kmi_api_mock: - irm_kmi = irm_kmi_api_mock.return_value - irm_kmi.get_image.side_effect = patched - irm_kmi.get_forecasts_coord.return_value = forecast - yield irm_kmi - - -@pytest.fixture() -def mock_image_and_simple_forecast_irm_kmi_api(request: pytest.FixtureRequest) -> Generator[None, MagicMock, None]: - """Return a mocked IrmKmi api client.""" - fixture: str = "forecast.json" - - forecast = json.loads(load_fixture(fixture)) - - with patch( - "custom_components.irm_kmi.coordinator.IrmKmiApiClient", autospec=True - ) as irm_kmi_api_mock: - irm_kmi = irm_kmi_api_mock.return_value - irm_kmi.get_image.side_effect = patched - irm_kmi.get_svg.return_value = "" - irm_kmi.get_forecasts_coord.return_value = forecast - yield irm_kmi - - -@pytest.fixture() -def mock_svg_pollen(request: pytest.FixtureRequest) -> Generator[None, MagicMock, None]: - """Return a mocked IrmKmi api client.""" - fixture: str = "pollen.svg" - - svg_str = load_fixture(fixture) - - with patch( - "custom_components.irm_kmi.coordinator.IrmKmiApiClient", autospec=True - ) as irm_kmi_api_mock: - irm_kmi = irm_kmi_api_mock.return_value - irm_kmi.get_svg.return_value = svg_str - yield irm_kmi - - -@pytest.fixture() -def mock_exception_irm_kmi_api_svg_pollen(request: pytest.FixtureRequest) -> Generator[None, MagicMock, None]: - """Return a mocked IrmKmi api client.""" - with patch( - "custom_components.irm_kmi.coordinator.IrmKmiApiClient", autospec=True - ) as irm_kmi_api_mock: - irm_kmi = irm_kmi_api_mock.return_value - irm_kmi.get_svg.side_effect = IrmKmiApiParametersError - yield irm_kmi - - -@pytest.fixture() -def mock_coordinator(request: pytest.FixtureRequest) -> Generator[None, MagicMock, None]: - """Return a mocked coordinator.""" - with patch( - "custom_components.irm_kmi.IrmKmiCoordinator", autospec=True - ) as coordinator_mock: - coord = coordinator_mock.return_value - coord._async_animation_data.return_value = {'animation': None} - yield coord - - -@pytest.fixture() -def mock_irm_kmi_api_works_but_pollen_and_radar_fail(request: pytest.FixtureRequest) -> Generator[ - None, MagicMock, None]: - """Return a mocked IrmKmi api client.""" - fixture: str = "forecast.json" - - forecast = json.loads(load_fixture(fixture)) - with patch( - "custom_components.irm_kmi.coordinator.IrmKmiApiClient", autospec=True - ) as irm_kmi_api_mock: - irm_kmi = irm_kmi_api_mock.return_value - irm_kmi.get_forecasts_coord.return_value = forecast - irm_kmi.get_svg.side_effect = IrmKmiApiError - irm_kmi.get_image.side_effect = IrmKmiApiError - yield irm_kmi diff --git a/tests/test_config_flow.py b/tests/test_config_flow.py index 549148c..ed66c74 100644 --- a/tests/test_config_flow.py +++ b/tests/test_config_flow.py @@ -9,12 +9,12 @@ from homeassistant.core import HomeAssistant from homeassistant.data_entry_flow import FlowResultType from pytest_homeassistant_custom_component.common import MockConfigEntry -from custom_components.irm_kmi import async_migrate_entry +from custom_components.irm_kmi import async_migrate_entry, OPTION_STYLE_STD from custom_components.irm_kmi.const import ( CONF_DARK_MODE, CONF_LANGUAGE_OVERRIDE, CONF_STYLE, CONF_USE_DEPRECATED_FORECAST, CONFIG_FLOW_VERSION, DOMAIN, - OPTION_DEPRECATED_FORECAST_NOT_USED, OPTION_STYLE_SATELLITE, - OPTION_STYLE_STD) + OPTION_DEPRECATED_FORECAST_NOT_USED) +from custom_components.irm_kmi.irm_kmi_api.const import OPTION_STYLE_SATELLITE async def test_full_user_flow( diff --git a/tests/test_coordinator.py b/tests/test_coordinator.py index 6768794..67087a0 100644 --- a/tests/test_coordinator.py +++ b/tests/test_coordinator.py @@ -1,4 +1,5 @@ from datetime import datetime, timedelta +from zoneinfo import ZoneInfo from freezegun import freeze_time from homeassistant.components.weather import (ATTR_CONDITION_CLOUDY, @@ -9,11 +10,11 @@ from pytest_homeassistant_custom_component.common import MockConfigEntry from custom_components.irm_kmi.const import CONF_LANGUAGE_OVERRIDE from custom_components.irm_kmi.coordinator import IrmKmiCoordinator -from custom_components.irm_kmi.data import (CurrentWeatherData, IrmKmiForecast, - ProcessedCoordinatorData) -from custom_components.irm_kmi.pollen import PollenParser -from custom_components.irm_kmi.radar_data import IrmKmiRadarForecast -from tests.conftest import get_api_data +from custom_components.irm_kmi.irm_kmi_api.api import IrmKmiApiClientHa +from custom_components.irm_kmi.irm_kmi_api.data import (CurrentWeatherData, IrmKmiForecast, IrmKmiRadarForecast) +from custom_components.irm_kmi.data import ProcessedCoordinatorData +from custom_components.irm_kmi.irm_kmi_api.pollen import PollenParser +from tests.conftest import get_api_data, get_api_with_data async def test_jules_forgot_to_revert_update_interval_before_pushing( @@ -27,19 +28,16 @@ async def test_jules_forgot_to_revert_update_interval_before_pushing( @freeze_time(datetime.fromisoformat('2024-01-12T07:10:00+00: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) + api = get_api_with_data("be_forecast_warning.json") - result = coordinator.warnings_from_data(api_data.get('for', {}).get('warning')) + result = api.get_warnings(lang='en') 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() @@ -51,8 +49,9 @@ async def test_warning_data( @freeze_time(datetime.fromisoformat('2023-12-26T17:30:00+00:00')) async def test_current_weather_be() -> None: - api_data = get_api_data("forecast.json") - result = await IrmKmiCoordinator.current_weather_from_data(api_data) + api = get_api_with_data("forecast.json") + tz = ZoneInfo("Europe/Brussels") + result = await api.get_current_weather(tz) expected = CurrentWeatherData( condition=ATTR_CONDITION_CLOUDY, @@ -69,8 +68,9 @@ async def test_current_weather_be() -> None: @freeze_time(datetime.fromisoformat("2023-12-28T15:30:00")) async def test_current_weather_nl() -> None: - api_data = get_api_data("forecast_nl.json") - result = await IrmKmiCoordinator.current_weather_from_data(api_data) + api = get_api_with_data("forecast_nl.json") + tz = ZoneInfo("Europe/Brussels") + result = await api.get_current_weather(tz) expected = CurrentWeatherData( condition=ATTR_CONDITION_CLOUDY, @@ -90,11 +90,10 @@ async def test_daily_forecast( hass: HomeAssistant, mock_config_entry: MockConfigEntry ) -> None: - api_data = get_api_data("forecast.json").get('for', {}).get('daily') - await hass.config_entries.async_add(mock_config_entry) - hass.config_entries.async_update_entry(mock_config_entry, data=mock_config_entry.data | {CONF_LANGUAGE_OVERRIDE: 'fr'}) - coordinator = IrmKmiCoordinator(hass, mock_config_entry) - result = await coordinator.daily_list_to_forecast(api_data) + api = get_api_with_data("forecast.json") + tz = ZoneInfo("Europe/Brussels") + + result = await api.get_daily_forecast(tz, 'fr') assert isinstance(result, list) assert len(result) == 8 @@ -121,8 +120,9 @@ async def test_daily_forecast( @freeze_time(datetime.fromisoformat('2023-12-26T18:30:00+01:00')) async def test_hourly_forecast() -> None: - api_data = get_api_data("forecast.json").get('for', {}).get('hourly') - result = await IrmKmiCoordinator.hourly_list_to_forecast(api_data) + api = get_api_with_data("forecast.json") + tz = ZoneInfo("Europe/Brussels") + result = await api.get_hourly_forecast(tz) assert isinstance(result, list) assert len(result) == 49 @@ -146,8 +146,10 @@ async def test_hourly_forecast() -> None: @freeze_time(datetime.fromisoformat('2024-05-31T01:50:00+02:00')) async def test_hourly_forecast_bis() -> None: - api_data = get_api_data("no-midnight-bug-31-05-2024T01-55.json").get('for', {}).get('hourly') - result = await IrmKmiCoordinator.hourly_list_to_forecast(api_data) + api = get_api_with_data("no-midnight-bug-31-05-2024T01-55.json") + tz = ZoneInfo("Europe/Brussels") + + result = await api.get_hourly_forecast(tz) assert isinstance(result, list) @@ -163,8 +165,10 @@ async def test_hourly_forecast_bis() -> None: @freeze_time(datetime.fromisoformat('2024-05-31T00:10:00+02:00')) async def test_hourly_forecast_midnight_bug() -> None: # Related to https://github.com/jdejaegh/irm-kmi-ha/issues/38 - api_data = get_api_data("midnight-bug-31-05-2024T00-13.json").get('for', {}).get('hourly') - result = await IrmKmiCoordinator.hourly_list_to_forecast(api_data) + api = get_api_with_data("midnight-bug-31-05-2024T00-13.json") + tz = ZoneInfo("Europe/Brussels") + + result = await api.get_hourly_forecast(tz) assert isinstance(result, list) @@ -200,10 +204,10 @@ async def test_daily_forecast_midnight_bug( hass: HomeAssistant, mock_config_entry: MockConfigEntry ) -> None: - coordinator = IrmKmiCoordinator(hass, mock_config_entry) + api = get_api_with_data("midnight-bug-31-05-2024T00-13.json") + tz = ZoneInfo("Europe/Brussels") - api_data = get_api_data("midnight-bug-31-05-2024T00-13.json").get('for', {}).get('daily') - result = await coordinator.daily_list_to_forecast(api_data) + result = await api.get_daily_forecast(tz, 'en') assert result[0]['datetime'] == '2024-05-31' assert not result[0]['is_daytime'] @@ -221,19 +225,11 @@ async def test_daily_forecast_midnight_bug( async def test_refresh_succeed_even_when_pollen_and_radar_fail( hass: HomeAssistant, mock_config_entry: MockConfigEntry, - mock_irm_kmi_api_works_but_pollen_and_radar_fail ): - hass.states.async_set( - "zone.home", - 0, - {"latitude": 50.738681639, "longitude": 4.054077148}, - ) - hass.config.config_dir = "." - mock_config_entry.add_to_hass(hass) - coordinator = IrmKmiCoordinator(hass, mock_config_entry) + coordinator._api._api_data = get_api_data("forecast.json") - result = await coordinator._async_update_data() + result = await coordinator.process_api_data() assert result.get('current_weather').get('condition') == ATTR_CONDITION_CLOUDY @@ -250,7 +246,7 @@ async def test_refresh_succeed_even_when_pollen_and_radar_fail( pollen={'foo': 'bar'} ) coordinator.data = existing_data - result = await coordinator._async_update_data() + result = await coordinator.process_api_data() assert result.get('current_weather').get('condition') == ATTR_CONDITION_CLOUDY @@ -260,8 +256,8 @@ async def test_refresh_succeed_even_when_pollen_and_radar_fail( def test_radar_forecast() -> None: - api_data = get_api_data("forecast.json") - result = IrmKmiCoordinator.radar_list_to_forecast(api_data.get('animation')) + api = get_api_with_data("forecast.json") + result = api.get_radar_forecast() expected = [ IrmKmiRadarForecast(datetime="2023-12-26T17:00:00+01:00", native_precipitation=0, might_rain=False, @@ -292,8 +288,8 @@ def test_radar_forecast() -> None: def test_radar_forecast_rain_interval() -> None: - api_data = get_api_data('forecast_with_rain_on_radar.json') - result = IrmKmiCoordinator.radar_list_to_forecast(api_data.get('animation')) + api = get_api_with_data('forecast_with_rain_on_radar.json') + result = api.get_radar_forecast() _12 = IrmKmiRadarForecast( datetime='2024-05-30T18:00:00+02:00', @@ -322,10 +318,10 @@ async def test_datetime_daily_forecast_nl( hass: HomeAssistant, mock_config_entry: MockConfigEntry ) -> None: - api_data = get_api_data("forecast_ams_no_ww.json").get('for', {}).get('daily') + api = get_api_with_data("forecast_ams_no_ww.json") + tz = ZoneInfo("Europe/Brussels") - coordinator = IrmKmiCoordinator(hass, mock_config_entry) - result = await coordinator.daily_list_to_forecast(api_data) + result = await api.get_daily_forecast(tz, 'en') assert result[0]['datetime'] == '2024-06-09' assert result[0]['is_daytime'] @@ -339,8 +335,10 @@ async def test_datetime_daily_forecast_nl( @freeze_time("2024-06-09T13:40:00+00:00") async def test_current_condition_forecast_nl() -> None: - api_data = get_api_data("forecast_ams_no_ww.json") - result = await IrmKmiCoordinator.current_weather_from_data(api_data) + api = get_api_with_data("forecast_ams_no_ww.json") + tz = ZoneInfo("Europe/Brussels") + + result = await api.get_current_weather(tz) expected = CurrentWeatherData( condition=ATTR_CONDITION_PARTLYCLOUDY, @@ -359,10 +357,10 @@ async def test_sunrise_sunset_nl( hass: HomeAssistant, mock_config_entry: MockConfigEntry ) -> None: - api_data = get_api_data("forecast_ams_no_ww.json").get('for', {}).get('daily') + api = get_api_with_data("forecast_ams_no_ww.json") + tz = ZoneInfo("Europe/Brussels") - coordinator = IrmKmiCoordinator(hass, mock_config_entry) - result = await coordinator.daily_list_to_forecast(api_data) + result = await api.get_daily_forecast(tz, 'en') assert result[0]['sunrise'] == '2024-06-09T05:19:28+02:00' assert result[0]['sunset'] == '2024-06-09T22:01:09+02:00' @@ -379,10 +377,10 @@ async def test_sunrise_sunset_be( hass: HomeAssistant, mock_config_entry: MockConfigEntry ) -> None: - api_data = get_api_data("forecast.json").get('for', {}).get('daily') + api = get_api_with_data("forecast.json") + tz = ZoneInfo("Europe/Brussels") - coordinator = IrmKmiCoordinator(hass, mock_config_entry) - result = await coordinator.daily_list_to_forecast(api_data) + result = await api.get_daily_forecast(tz, 'en') assert result[1]['sunrise'] == '2023-12-27T08:44:00+01:00' assert result[1]['sunset'] == '2023-12-27T16:43:00+01:00' diff --git a/tests/test_current_weather_sensors.py b/tests/test_current_weather_sensors.py index ef667ef..a9e2e73 100644 --- a/tests/test_current_weather_sensors.py +++ b/tests/test_current_weather_sensors.py @@ -1,5 +1,6 @@ import inspect from datetime import datetime, timedelta +from zoneinfo import ZoneInfo import pytest from freezegun import freeze_time @@ -9,9 +10,10 @@ from pytest_homeassistant_custom_component.common import MockConfigEntry from custom_components.irm_kmi import IrmKmiCoordinator from custom_components.irm_kmi.const import CURRENT_WEATHER_SENSORS, CURRENT_WEATHER_SENSOR_UNITS, \ CURRENT_WEATHER_SENSOR_CLASS -from custom_components.irm_kmi.data import CurrentWeatherData, ProcessedCoordinatorData +from custom_components.irm_kmi.irm_kmi_api.data import CurrentWeatherData +from custom_components.irm_kmi.data import ProcessedCoordinatorData from custom_components.irm_kmi.sensor import IrmKmiCurrentWeather, IrmKmiCurrentRainfall -from tests.conftest import get_api_data +from tests.conftest import get_api_data, get_api_with_data def test_sensors_in_current_weather_data(): @@ -110,14 +112,16 @@ async def test_current_weather_sensors( api_data = get_api_data(filename) time = api_data.get('obs').get('timestamp') + api = get_api_with_data(filename) + tz = ZoneInfo("Europe/Brussels") @freeze_time(datetime.fromisoformat(time) + timedelta(seconds=45, minutes=1)) async def run(mock_config_entry_, sensor_, expected_): coordinator = IrmKmiCoordinator(hass, mock_config_entry_) coordinator.data = ProcessedCoordinatorData( - current_weather=await IrmKmiCoordinator.current_weather_from_data(api_data), - hourly_forecast=await IrmKmiCoordinator.hourly_list_to_forecast(api_data.get('for', {}).get('hourly')), - radar_forecast=IrmKmiCoordinator.radar_list_to_forecast(api_data.get('animation', {})), + current_weather=await api.get_current_weather(tz), + hourly_forecast=await api.get_hourly_forecast(tz), + radar_forecast=api.get_radar_forecast(), country=api_data.get('country') ) @@ -145,13 +149,14 @@ async def test_current_rainfall_unit( ) -> None: hass.config.time_zone = 'Europe/Brussels' coordinator = IrmKmiCoordinator(hass, mock_config_entry) - api_data = get_api_data(filename) + api = get_api_with_data(filename) + tz = ZoneInfo("Europe/Brussels") coordinator.data = ProcessedCoordinatorData( - current_weather=await IrmKmiCoordinator.current_weather_from_data(api_data), - hourly_forecast=await IrmKmiCoordinator.hourly_list_to_forecast(api_data.get('for', {}).get('hourly')), - radar_forecast=IrmKmiCoordinator.radar_list_to_forecast(api_data.get('animation', {})), - country=api_data.get('country') + current_weather=await api.get_current_weather(tz), + hourly_forecast=await api.get_hourly_forecast(tz), + radar_forecast=api.get_radar_forecast(), + country=api.get_country() ) s = IrmKmiCurrentRainfall(coordinator, mock_config_entry) diff --git a/tests/test_init.py b/tests/test_init.py index 1947ae0..e78de77 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -8,18 +8,17 @@ from homeassistant.const import CONF_ZONE from homeassistant.core import HomeAssistant from pytest_homeassistant_custom_component.common import MockConfigEntry -from custom_components.irm_kmi import async_migrate_entry +from custom_components.irm_kmi import async_migrate_entry, OPTION_STYLE_STD from custom_components.irm_kmi.const import ( CONF_DARK_MODE, CONF_LANGUAGE_OVERRIDE, CONF_STYLE, CONF_USE_DEPRECATED_FORECAST, CONFIG_FLOW_VERSION, DOMAIN, - OPTION_DEPRECATED_FORECAST_NOT_USED, OPTION_STYLE_STD) + OPTION_DEPRECATED_FORECAST_NOT_USED) async def test_load_unload_config_entry( hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_irm_kmi_api: AsyncMock, - mock_coordinator: AsyncMock ) -> None: """Test the IRM KMI configuration entry loading/unloading.""" hass.states.async_set( @@ -57,7 +56,7 @@ async def test_config_entry_not_ready( await hass.config_entries.async_setup(mock_config_entry.entry_id) await hass.async_block_till_done() - assert mock_exception_irm_kmi_api.get_forecasts_coord.call_count == 1 + assert mock_exception_irm_kmi_api.refresh_forecasts_coord.call_count == 1 assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY diff --git a/tests/test_pollen.py b/tests/test_pollen.py index c705da0..f48239f 100644 --- a/tests/test_pollen.py +++ b/tests/test_pollen.py @@ -1,11 +1,12 @@ -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, MagicMock from homeassistant.core import HomeAssistant -from pytest_homeassistant_custom_component.common import MockConfigEntry +from pytest_homeassistant_custom_component.common import MockConfigEntry, load_fixture from custom_components.irm_kmi import IrmKmiCoordinator -from custom_components.irm_kmi.pollen import PollenParser -from tests.conftest import get_api_data +from custom_components.irm_kmi.irm_kmi_api.api import IrmKmiApiError +from custom_components.irm_kmi.irm_kmi_api.pollen import PollenParser +from tests.conftest import get_api_with_data def test_svg_pollen_parsing(): @@ -38,15 +39,13 @@ def test_pollen_default_values(): 'alder': 'none', 'grasses': 'none', 'ash': 'none'} -async def test_pollen_data_from_api( - hass: HomeAssistant, - mock_svg_pollen: AsyncMock, - mock_config_entry: MockConfigEntry -) -> None: - coordinator = IrmKmiCoordinator(hass, mock_config_entry) - api_data = get_api_data("be_forecast_warning.json") +async def test_pollen_data_from_api() -> None: + api = get_api_with_data("be_forecast_warning.json") - result = await coordinator._async_pollen_data(api_data) + # Mock get_svg function + api.get_svg = AsyncMock(return_value=load_fixture("pollen.svg")) + + result = await api.get_pollen() expected = {'mugwort': 'none', 'birch': 'none', 'alder': 'none', 'ash': 'none', 'oak': 'none', 'grasses': 'purple', 'hazel': 'none'} assert result == expected @@ -55,11 +54,15 @@ async def test_pollen_data_from_api( async def test_pollen_error_leads_to_unavailable_on_first_call( hass: HomeAssistant, mock_config_entry: MockConfigEntry, - mock_exception_irm_kmi_api_svg_pollen: AsyncMock ) -> None: coordinator = IrmKmiCoordinator(hass, mock_config_entry) - api_data = get_api_data("be_forecast_warning.json") + api = get_api_with_data("be_forecast_warning.json") - result = await coordinator._async_pollen_data(api_data) + api.get_svg = AsyncMock() + api.get_svg.side_effect = IrmKmiApiError + + coordinator._api = api + + result = await coordinator.process_api_data() expected = PollenParser.get_unavailable_data() - assert result == expected + assert result['pollen'] == expected diff --git a/tests/test_rain_graph.py b/tests/test_rain_graph.py index 0758ae3..fbf0512 100644 --- a/tests/test_rain_graph.py +++ b/tests/test_rain_graph.py @@ -1,8 +1,8 @@ import base64 from datetime import datetime, timedelta -from custom_components.irm_kmi.radar_data import AnimationFrameData, RadarAnimationData -from custom_components.irm_kmi.rain_graph import RainGraph +from custom_components.irm_kmi.irm_kmi_api.data import AnimationFrameData, RadarAnimationData +from custom_components.irm_kmi.irm_kmi_api.rain_graph import RainGraph def get_radar_animation_data() -> RadarAnimationData: @@ -36,7 +36,7 @@ async def test_svg_frame_setup(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="custom_components/irm_kmi/resources/be_white.png", + background_image_path="custom_components/irm_kmi/irm_kmi_api/resources/be_white.png", background_size=(640, 490), ) @@ -44,7 +44,7 @@ async def test_svg_frame_setup(): svg_str = rain_graph.get_dwg().tostring() - with open("custom_components/irm_kmi/resources/roboto_medium.ttf", "rb") as file: + with open("custom_components/irm_kmi/irm_kmi_api/resources/roboto_medium.ttf", "rb") as file: font_b64 = base64.b64encode(file.read()).decode('utf-8') assert '#385E95' in svg_str @@ -56,7 +56,7 @@ def test_svg_hint(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="custom_components/irm_kmi/resources/be_white.png", + background_image_path="custom_components/irm_kmi/irm_kmi_api/resources/be_white.png", background_size=(640, 490), ) @@ -71,7 +71,7 @@ def test_svg_time_bars(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="custom_components/irm_kmi/resources/be_white.png", + background_image_path="custom_components/irm_kmi/irm_kmi_api/resources/be_white.png", background_size=(640, 490), ) @@ -90,7 +90,7 @@ def test_draw_chances_path(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="custom_components/irm_kmi/resources/be_white.png", + background_image_path="custom_components/irm_kmi/irm_kmi_api/resources/be_white.png", background_size=(640, 490), ) @@ -108,7 +108,7 @@ def test_draw_data_line(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="custom_components/irm_kmi/resources/be_white.png", + background_image_path="custom_components/irm_kmi/irm_kmi_api/resources/be_white.png", background_size=(640, 490), ) @@ -126,13 +126,13 @@ async def test_insert_background(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="custom_components/irm_kmi/resources/be_white.png", + background_image_path="custom_components/irm_kmi/irm_kmi_api/resources/be_white.png", background_size=(640, 490), ) await rain_graph.insert_background() - with open("custom_components/irm_kmi/resources/be_white.png", "rb") as file: + with open("custom_components/irm_kmi/irm_kmi_api/resources/be_white.png", "rb") as file: png_b64 = base64.b64encode(file.read()).decode('utf-8') svg_str = rain_graph.get_dwg().tostring() @@ -149,7 +149,7 @@ def test_draw_current_frame_line_moving(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="custom_components/irm_kmi/resources/be_white.png", + background_image_path="custom_components/irm_kmi/irm_kmi_api/resources/be_white.png", background_size=(640, 490), ) @@ -177,7 +177,7 @@ def test_draw_current_frame_line_index(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="custom_components/irm_kmi/resources/be_white.png", + background_image_path="custom_components/irm_kmi/irm_kmi_api/resources/be_white.png", background_size=(640, 490), ) @@ -205,7 +205,7 @@ def test_draw_description_text(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="custom_components/irm_kmi/resources/be_white.png", + background_image_path="custom_components/irm_kmi/irm_kmi_api/resources/be_white.png", background_size=(640, 490), ) @@ -232,7 +232,7 @@ def test_draw_cloud_layer(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="custom_components/irm_kmi/resources/be_white.png", + background_image_path="custom_components/irm_kmi/irm_kmi_api/resources/be_white.png", background_size=(640, 490), ) @@ -252,7 +252,7 @@ async def test_draw_location_layer(): data = get_radar_animation_data() rain_graph = RainGraph( animation_data=data, - background_image_path="custom_components/irm_kmi/resources/be_white.png", + background_image_path="custom_components/irm_kmi/irm_kmi_api/resources/be_white.png", background_size=(640, 490), ) diff --git a/tests/test_repairs.py b/tests/test_repairs.py index 504c189..64c154d 100644 --- a/tests/test_repairs.py +++ b/tests/test_repairs.py @@ -1,10 +1,11 @@ +import json import logging -from unittest.mock import MagicMock +from unittest.mock import MagicMock, AsyncMock from homeassistant.core import HomeAssistant from homeassistant.data_entry_flow import FlowResultType from homeassistant.helpers import issue_registry -from pytest_homeassistant_custom_component.common import MockConfigEntry +from pytest_homeassistant_custom_component.common import MockConfigEntry, load_fixture from custom_components.irm_kmi import DOMAIN, IrmKmiCoordinator from custom_components.irm_kmi.const import (REPAIR_OPT_DELETE, @@ -28,6 +29,11 @@ async def get_repair_flow( await hass.config_entries.async_setup(mock_config_entry.entry_id) await hass.async_block_till_done() coordinator = IrmKmiCoordinator(hass, mock_config_entry) + + fixture: str = "forecast_out_of_benelux.json" + forecast = json.loads(load_fixture(fixture)) + coordinator._api.get_forecasts_coord = AsyncMock(return_value=forecast) + await coordinator._async_update_data() ir = issue_registry.async_get(hass) issue = ir.async_get_issue(DOMAIN, "zone_moved") @@ -38,7 +44,6 @@ async def get_repair_flow( async def test_repair_triggers_when_out_of_benelux( hass: HomeAssistant, - mock_irm_kmi_api_coordinator_out_benelux: MagicMock, mock_config_entry: MockConfigEntry ) -> None: hass.states.async_set( @@ -50,6 +55,8 @@ async def test_repair_triggers_when_out_of_benelux( mock_config_entry.add_to_hass(hass) coordinator = IrmKmiCoordinator(hass, mock_config_entry) + coordinator._api.get_forecasts_coord = AsyncMock(return_value=json.loads(load_fixture("forecast_out_of_benelux.json"))) + await coordinator._async_update_data() ir = issue_registry.async_get(hass) @@ -65,7 +72,6 @@ async def test_repair_triggers_when_out_of_benelux( async def test_repair_flow( hass: HomeAssistant, - mock_irm_kmi_api_coordinator_out_benelux: MagicMock, mock_irm_kmi_api_repair_in_benelux: MagicMock, mock_config_entry: MockConfigEntry ) -> None: @@ -87,7 +93,6 @@ async def test_repair_flow( async def test_repair_flow_invalid_choice( hass: HomeAssistant, - mock_irm_kmi_api_coordinator_out_benelux: MagicMock, mock_irm_kmi_api_repair_in_benelux: MagicMock, mock_config_entry: MockConfigEntry ) -> None: @@ -106,7 +111,6 @@ async def test_repair_flow_invalid_choice( async def test_repair_flow_api_error( hass: HomeAssistant, - mock_irm_kmi_api_coordinator_out_benelux: MagicMock, mock_get_forecast_api_error_repair: MagicMock, mock_config_entry: MockConfigEntry ) -> None: @@ -125,7 +129,6 @@ async def test_repair_flow_api_error( async def test_repair_flow_out_of_benelux( hass: HomeAssistant, - mock_irm_kmi_api_coordinator_out_benelux: MagicMock, mock_irm_kmi_api_repair_out_of_benelux: MagicMock, mock_config_entry: MockConfigEntry ) -> None: @@ -144,7 +147,6 @@ async def test_repair_flow_out_of_benelux( async def test_repair_flow_delete_entry( hass: HomeAssistant, - mock_irm_kmi_api_coordinator_out_benelux: MagicMock, mock_config_entry: MockConfigEntry ) -> None: repair_flow = await get_repair_flow(hass, mock_config_entry) diff --git a/tests/test_sensors.py b/tests/test_sensors.py index e221897..847581c 100644 --- a/tests/test_sensors.py +++ b/tests/test_sensors.py @@ -1,4 +1,5 @@ from datetime import datetime +from unittest.mock import AsyncMock from freezegun import freeze_time from homeassistant.core import HomeAssistant @@ -8,7 +9,7 @@ from custom_components.irm_kmi import IrmKmiCoordinator from custom_components.irm_kmi.binary_sensor import IrmKmiWarning from custom_components.irm_kmi.const import CONF_LANGUAGE_OVERRIDE from custom_components.irm_kmi.sensor import IrmKmiNextSunMove, IrmKmiNextWarning -from tests.conftest import get_api_data +from tests.conftest import get_api_data, get_api_with_data @freeze_time(datetime.fromisoformat('2024-01-12T07:55:00+01:00')) @@ -16,10 +17,10 @@ async def test_warning_data( hass: HomeAssistant, mock_config_entry: MockConfigEntry ) -> None: - api_data = get_api_data("be_forecast_warning.json") + api = get_api_with_data("be_forecast_warning.json") coordinator = IrmKmiCoordinator(hass, mock_config_entry) - result = coordinator.warnings_from_data(api_data.get('for', {}).get('warning')) + result = api.get_warnings('en') coordinator.data = {'warnings': result} warning = IrmKmiWarning(coordinator, mock_config_entry) @@ -39,15 +40,18 @@ async def test_warning_data_unknown_lang( 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") + api = get_api_with_data("be_forecast_warning.json") coordinator = IrmKmiCoordinator(hass, mock_config_entry) - result = coordinator.warnings_from_data(api_data.get('for', {}).get('warning')) + api.get_pollen = AsyncMock() + api.get_animation_data = AsyncMock() + coordinator._api = api - coordinator.data = {'warnings': result} + + result = await coordinator.process_api_data() + + coordinator.data = {'warnings': result['warnings']} warning = IrmKmiWarning(coordinator, mock_config_entry) warning.hass = hass @@ -65,15 +69,19 @@ async def test_next_warning_when_data_available( hass: HomeAssistant, mock_config_entry: MockConfigEntry ) -> None: - api_data = get_api_data("be_forecast_warning.json") + api = get_api_with_data("be_forecast_warning.json") await hass.config_entries.async_add(mock_config_entry) hass.config_entries.async_update_entry(mock_config_entry, data=mock_config_entry.data | {CONF_LANGUAGE_OVERRIDE: 'de'}) coordinator = IrmKmiCoordinator(hass, mock_config_entry) - result = coordinator.warnings_from_data(api_data.get('for', {}).get('warning')) + api.get_pollen = AsyncMock() + api.get_animation_data = AsyncMock() + coordinator._api = api - coordinator.data = {'warnings': result} + result = await coordinator.process_api_data() + + coordinator.data = {'warnings': result['warnings']} warning = IrmKmiNextWarning(coordinator, mock_config_entry) warning.hass = hass @@ -93,12 +101,16 @@ 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") + api = get_api_with_data("be_forecast_warning.json") coordinator = IrmKmiCoordinator(hass, mock_config_entry) - result = coordinator.warnings_from_data(api_data.get('for', {}).get('warning')) + api.get_pollen = AsyncMock() + api.get_animation_data = AsyncMock() + coordinator._api = api - coordinator.data = {'warnings': result} + result = await coordinator.process_api_data() + + coordinator.data = {'warnings': result['warnings']} warning = IrmKmiNextWarning(coordinator, mock_config_entry) warning.hass = hass @@ -154,13 +166,16 @@ async def test_next_sunrise_sunset( hass: HomeAssistant, mock_config_entry: MockConfigEntry ) -> None: - api_data = get_api_data("forecast.json") + api = get_api_with_data("forecast.json") coordinator = IrmKmiCoordinator(hass, mock_config_entry) + api.get_pollen = AsyncMock() + api.get_animation_data = AsyncMock() + coordinator._api = api - result = await coordinator.daily_list_to_forecast(api_data.get('for', {}).get('daily')) + result = await coordinator.process_api_data() - coordinator.data = {'daily_forecast': result} + coordinator.data = {'daily_forecast': result['daily_forecast']} sunset = IrmKmiNextSunMove(coordinator, mock_config_entry, 'sunset') sunrise = IrmKmiNextSunMove(coordinator, mock_config_entry, 'sunrise') @@ -180,13 +195,16 @@ async def test_next_sunrise_sunset_bis( hass: HomeAssistant, mock_config_entry: MockConfigEntry ) -> None: - api_data = get_api_data("forecast.json") + api = get_api_with_data("forecast.json") coordinator = IrmKmiCoordinator(hass, mock_config_entry) + api.get_pollen = AsyncMock() + api.get_animation_data = AsyncMock() + coordinator._api = api - result = await coordinator.daily_list_to_forecast(api_data.get('for', {}).get('daily')) + result = await coordinator.process_api_data() - coordinator.data = {'daily_forecast': result} + coordinator.data = {'daily_forecast': result['daily_forecast']} sunset = IrmKmiNextSunMove(coordinator, mock_config_entry, 'sunset') sunrise = IrmKmiNextSunMove(coordinator, mock_config_entry, 'sunrise') diff --git a/tests/test_weather.py b/tests/test_weather.py index 6cebe02..039a718 100644 --- a/tests/test_weather.py +++ b/tests/test_weather.py @@ -1,35 +1,28 @@ -import os +import json from datetime import datetime from typing import List -from unittest.mock import AsyncMock from freezegun import freeze_time from homeassistant.components.weather import Forecast from homeassistant.core import HomeAssistant -from pytest_homeassistant_custom_component.common import MockConfigEntry +from pytest_homeassistant_custom_component.common import MockConfigEntry, load_fixture from custom_components.irm_kmi import IrmKmiCoordinator, IrmKmiWeather -from custom_components.irm_kmi.data import (ProcessedCoordinatorData) -from custom_components.irm_kmi.radar_data import IrmKmiRadarForecast -from tests.conftest import get_api_data +from custom_components.irm_kmi.data import ProcessedCoordinatorData +from custom_components.irm_kmi.irm_kmi_api.data import IrmKmiRadarForecast +from tests.conftest import get_api_with_data @freeze_time(datetime.fromisoformat("2023-12-28T15:30:00+01:00")) async def test_weather_nl( hass: HomeAssistant, - mock_image_and_nl_forecast_irm_kmi_api: AsyncMock, mock_config_entry: MockConfigEntry ) -> None: - hass.states.async_set( - "zone.home", - 0, - {"latitude": 50.738681639, "longitude": 4.054077148}, - ) - hass.config.config_dir = os.getcwd() - coordinator = IrmKmiCoordinator(hass, mock_config_entry) - await coordinator.async_refresh() - print(coordinator.data) + forecast = json.loads(load_fixture("forecast_nl.json")) + coordinator._api._api_data = forecast + + coordinator.data = await coordinator.process_api_data() weather = IrmKmiWeather(coordinator, mock_config_entry) result = await weather.async_forecast_daily() @@ -44,19 +37,14 @@ async def test_weather_nl( @freeze_time(datetime.fromisoformat("2024-01-21T14:15:00+01:00")) async def test_weather_higher_temp_at_night( hass: HomeAssistant, - mock_image_and_high_temp_irm_kmi_api: AsyncMock, mock_config_entry: MockConfigEntry ) -> None: # Test case for https://github.com/jdejaegh/irm-kmi-ha/issues/8 - hass.states.async_set( - "zone.home", - 0, - {"latitude": 50.738681639, "longitude": 4.054077148}, - ) - hass.config.config_dir = os.getcwd() - coordinator = IrmKmiCoordinator(hass, mock_config_entry) - await coordinator.async_refresh() + forecast = json.loads(load_fixture("high_low_temp.json")) + coordinator._api._api_data = forecast + + coordinator.data = await coordinator.process_api_data() weather = IrmKmiWeather(coordinator, mock_config_entry) result: List[Forecast] = await weather.async_forecast_daily() @@ -75,18 +63,13 @@ async def test_weather_higher_temp_at_night( @freeze_time(datetime.fromisoformat("2023-12-26T18:30:00+01:00")) async def test_forecast_attribute_same_as_service_call( hass: HomeAssistant, - mock_image_and_simple_forecast_irm_kmi_api: AsyncMock, mock_config_entry_with_deprecated: MockConfigEntry ) -> None: - hass.states.async_set( - "zone.home", - 0, - {"latitude": 50.738681639, "longitude": 4.054077148}, - ) - hass.config.config_dir = os.getcwd() - coordinator = IrmKmiCoordinator(hass, mock_config_entry_with_deprecated) - await coordinator.async_refresh() + forecast = json.loads(load_fixture("forecast.json")) + coordinator._api._api_data = forecast + + coordinator.data = await coordinator.process_api_data() weather = IrmKmiWeather(coordinator, mock_config_entry_with_deprecated) @@ -104,11 +87,10 @@ async def test_radar_forecast_service( hass.config.time_zone = 'Europe/Brussels' coordinator = IrmKmiCoordinator(hass, mock_config_entry) - api_data = get_api_data("forecast.json") - data = IrmKmiCoordinator.radar_list_to_forecast(api_data.get('animation')) + coordinator._api = get_api_with_data("forecast.json") coordinator.data = ProcessedCoordinatorData( - radar_forecast=data + radar_forecast=coordinator._api.get_radar_forecast() ) weather = IrmKmiWeather(coordinator, mock_config_entry)