jdejaegh/irm-kmi-ha
1
0
Fork 0
mirror of https://github.com/jdejaegh/irm-kmi-ha.git synced 2025-06-27 03:35:56 +02:00
Code Issues Projects Releases Packages Wiki Activity

Move API related code to a sub-module + update tests

Browse source
This commit is contained in:
Jules 2025-05-02 19:31:45 +02:00
parent 5932884c7a
commit 6476f0e57a
Signed by: jdejaegh
GPG key ID: 99D6D184CA66933A
37 changed files with 943 additions and 978 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
custom_components/irm_kmi/__init__.py
View file

@ -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__)

125
custom_components/irm_kmi/api.py
View file

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

11
custom_components/irm_kmi/config_flow.py
View file

@ -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]}
)

33
custom_components/irm_kmi/const.py
View file

@ -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',

454
custom_components/irm_kmi/coordinator.py
View file

@ -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 []

38
custom_components/irm_kmi/data.py
View file

@ -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):

0
custom_components/irm_kmi/irm_kmi_api/__init__.py Normal file
View file

562
custom_components/irm_kmi/irm_kmi_api/api.py Normal file
View file

@ -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

28
custom_components/irm_kmi/irm_kmi_api/const.py Normal file
View file

@ -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'}

95
custom_components/irm_kmi/irm_kmi_api/data.py Normal file
View file

@ -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

2
custom_components/irm_kmi/irm_kmi_api/ha.py Normal file
View file

@ -0,0 +1,2 @@

2
custom_components/irm_kmi/pollen.py → custom_components/irm_kmi/irm_kmi_api/pollen.py
View file

@ -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__)

7
custom_components/irm_kmi/rain_graph.py → custom_components/irm_kmi/irm_kmi_api/rain_graph.py
View file

@ -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

0
custom_components/irm_kmi/irm_kmi_api/resources/__init__.py Normal file
View file

0
custom_components/irm_kmi/resources/be_black.png → custom_components/irm_kmi/irm_kmi_api/resources/be_black.png
View file

Side by side Swipe Overlay

Before

Width:  |  Height:  |  Size: 68 KiB

After

Width:  |  Height:  |  Size: 68 KiB

0
custom_components/irm_kmi/resources/be_black.py → custom_components/irm_kmi/irm_kmi_api/resources/be_black.py
View file

0
custom_components/irm_kmi/resources/be_satellite.png → custom_components/irm_kmi/irm_kmi_api/resources/be_satellite.png
View file

Side by side Swipe Overlay

Before

Width:  |  Height:  |  Size: 666 KiB

After

Width:  |  Height:  |  Size: 666 KiB

0
custom_components/irm_kmi/resources/be_satellite.py → custom_components/irm_kmi/irm_kmi_api/resources/be_satellite.py
View file

0
custom_components/irm_kmi/resources/be_white.png → custom_components/irm_kmi/irm_kmi_api/resources/be_white.png
View file

Side by side Swipe Overlay

Before

Width:  |  Height:  |  Size: 38 KiB

After

Width:  |  Height:  |  Size: 38 KiB

0
custom_components/irm_kmi/resources/be_white.py → custom_components/irm_kmi/irm_kmi_api/resources/be_white.py
View file

0
custom_components/irm_kmi/resources/nl.png → custom_components/irm_kmi/irm_kmi_api/resources/nl.png
View file

Side by side Swipe Overlay

Before

Width:  |  Height:  |  Size: 60 KiB

After

Width:  |  Height:  |  Size: 60 KiB

0
custom_components/irm_kmi/resources/nl.py → custom_components/irm_kmi/irm_kmi_api/resources/nl.py
View file

0
custom_components/irm_kmi/resources/roboto.py → custom_components/irm_kmi/irm_kmi_api/resources/roboto.py
View file

0
custom_components/irm_kmi/resources/roboto_medium.ttf → custom_components/irm_kmi/irm_kmi_api/resources/roboto_medium.ttf
View file

35
custom_components/irm_kmi/radar_data.py
View file

@ -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

20
custom_components/irm_kmi/repairs.py
View file

@ -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."""

10
custom_components/irm_kmi/sensor.py
View file

@ -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__)

154
tests/conftest.py
View file

@ -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

6
tests/test_config_flow.py
View file

@ -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(

106
tests/test_coordinator.py
View file

@ -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'

25
tests/test_current_weather_sensors.py
View file

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

7
tests/test_init.py
View file

@ -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

35
tests/test_pollen.py
View file

@ -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

30
tests/test_rain_graph.py
View file

@ -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),
)

18
tests/test_repairs.py
View file

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

58
tests/test_sensors.py
View file

@ -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')

56
tests/test_weather.py
View file

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