jdejaegh/irm-kmi-api
1
0
Fork 0
mirror of https://github.com/jdejaegh/irm-kmi-api.git synced 2025-06-27 12:09:26 +02:00
Code Issues Projects Releases Packages Wiki Activity

Compare commits

base: jdejaegh:main
Branches Tags
jdejaegh:main
jdejaegh:heat-v0
jdejaegh:0.2.0
jdejaegh:1.1.0
jdejaegh:1.0.3
jdejaegh:1.0.2
jdejaegh:1.0.1
jdejaegh:1.0.0
jdejaegh:0.1.6
jdejaegh:0.1.5
jdejaegh:0.1.4
jdejaegh:0.1.3
jdejaegh:0.1.2
jdejaegh:0.1.1
..
compare: jdejaegh:0.1.3
Branches Tags
jdejaegh:heat-v0
jdejaegh:main
jdejaegh:0.2.0
jdejaegh:1.1.0
jdejaegh:1.0.3
jdejaegh:1.0.2
jdejaegh:1.0.1
jdejaegh:1.0.0
jdejaegh:0.1.6
jdejaegh:0.1.5
jdejaegh:0.1.4
jdejaegh:0.1.3
jdejaegh:0.1.2
jdejaegh:0.1.1

No commits in common. "main" and "0.1.3" have entirely different histories.
main ... 0.1.3

26 changed files with 595 additions and 2682 deletions
Show stats Expand all files Collapse all files

2
.github/workflows/pytest.yml vendored
View file

@ -11,7 +11,7 @@ jobs:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.12", "3.13"]
python-version: ["3.11", "3.13"]
steps:
- uses: MathRobin/timezone-action@v1.1

67
README.md
View file

@ -1,69 +1,6 @@
# Async API to retrieve data from the Belgian IRM KMI in Python
The package exposes the data from the [mobile application of the Belgian IRM KMI](https://www.meteo.be/en/info/faq/products-services/the-rmi-weather-app) as a Python module.
See more information in the wiki: https://github.com/jdejaegh/irm-kmi-api/wiki
## Quick start example
```python
import aiohttp
import asyncio
from zoneinfo import ZoneInfo
from irm_kmi_api import IrmKmiApiClientHa
async def print_weather():
session = aiohttp.ClientSession()
client = IrmKmiApiClientHa(session=session, user_agent="jdejaegh/irm-kmi-api README example")
await client.refresh_forecasts_coord({'lat': 50.47, 'long': 4.87})
await session.close()
weather = client.get_current_weather(tz=ZoneInfo('Europe/Brussels'))
city = client.get_city()
print(f"{weather['temperature']}°C with wind of {weather['wind_speed']} km/h in {city}")
if __name__ == '__main__':
asyncio.run(print_weather())
```
## Features
The package provides the following:
1. Current weather
2. Hourly and daily forecast
3. Rain radar forecast and animation
4. Warning data (for extreme weather condition such as storm, thunder, floods)
5. Pollen data
<details>
<summary>Screenshots of the rain radar animation</summary>
<img src="https://github.com/jdejaegh/irm-kmi-api/raw/main/img/camera_light.png"/> <br>
<img src="https://github.com/jdejaegh/irm-kmi-api/raw/main/img/camera_dark.png"/> <br>
<img src="https://github.com/jdejaegh/irm-kmi-api/raw/main/img/camera_sat.png"/>
</details>
## Limitations
The package does not provide the 14-days forcast as in the application.
This package will not implement any feature that is not available via the API (e.g., humidity and dew point data is not
provided by the API and thus is not available in this package).
## Usage considerations
The API is not publicly documented and has been reversed engineered: it can change at any time without notice and break this package.
Be mindful when using the API: put a meaningful User-Agent string when creating an `IrmKmiApiClient` and apply rate-limiting for your queries.
## Contributions
Contributions are welcome. Please discuss major changes in an issue before submitting a pull request.
# API to retrieve data from the Belgian IRM KMI in Python
The data is collected via their non-public mobile application API.
## Disclaimer

BIN
img/camera_dark.png
View file

Binary file not shown.
Side by side

Before

Width:  |  Height:  |  Size: 141 KiB

BIN
img/camera_light.png
View file

Binary file not shown.
Side by side

Before

Width:  |  Height:  |  Size: 97 KiB

BIN
img/camera_sat.png
View file

Binary file not shown.
Side by side

Before

Width:  |  Height:  |  Size: 770 KiB

46
irm_kmi_api/__init__.py
View file

@ -1,45 +1 @@
from .api import (
IrmKmiApiClient,
IrmKmiApiClientHa,
IrmKmiApiCommunicationError,
IrmKmiApiError,
)
from .data import (
AnimationFrameData,
ConditionEvol,
CurrentWeatherData,
ExtendedForecast,
Forecast,
PollenLevel,
PollenName,
RadarAnimationData,
RadarForecast,
RadarStyle,
WarningData,
WarningType,
)
from .pollen import PollenParser
from .rain_graph import RainGraph
__all__ = [
"IrmKmiApiClient",
"IrmKmiApiClientHa",
"IrmKmiApiCommunicationError",
"IrmKmiApiError",
"AnimationFrameData",
"ConditionEvol",
"CurrentWeatherData",
"ExtendedForecast",
"Forecast",
"PollenLevel",
"PollenName",
"RadarAnimationData",
"RadarForecast",
"RadarStyle",
"WarningData",
"WarningType",
"PollenParser",
"RainGraph",
]
__version__ = '1.1.0'
__version__ = '0.1.3'

508
irm_kmi_api/api.py
View file

@ -8,30 +8,19 @@ import time
import urllib.parse
from datetime import datetime, timedelta
from statistics import mean
from typing import Dict, List, Tuple
from typing import List, Tuple
from zoneinfo import ZoneInfo
import aiohttp
import async_timeout
from .const import (
IRM_KMI_TO_HA_CONDITION_MAP,
STYLE_TO_PARAM_MAP,
WEEKDAYS,
WWEVOL_TO_ENUM_MAP,
)
from .const import MAP_WARNING_ID_TO_SLUG as SLUG_MAP
from .data import (
AnimationFrameData,
CurrentWeatherData,
ExtendedForecast,
Forecast,
RadarAnimationData,
RadarForecast,
RadarStyle,
WarningData,
WarningType,
)
from .pollen import PollenLevel, PollenName, PollenParser
from .const import STYLE_TO_PARAM_MAP, WEEKDAYS
from .data import (AnimationFrameData, CurrentWeatherData, Forecast,
IrmKmiForecast, IrmKmiRadarForecast, RadarAnimationData,
WarningData)
from .pollen import PollenParser
from .utils import next_weekday
_LOGGER = logging.getLogger(__name__)
@ -44,83 +33,50 @@ 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 = {}
_base_url = "https://app.meteo.be/services/appv4/"
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:
"""
Create a new instance of the API client
:param session: aiohttp.ClientSession to use for the request
:param user_agent: string that will indentify your application in the User-Agent header of the HTTP requests
"""
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[str, float | int]) -> dict:
"""
Get forecasts for given location.
:param coord: dict with the following keys: 'lat', 'long' (both float or int)
:return: raw forecast as python dict
:raise: IrmKmiApiError when communication with the API fails
"""
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": self._api_key("getForecasts")} | coord
)
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[str, str] | None = None) -> bytes:
"""
Get the image at the specified url with the parameters
:param url: URL to fetch
:param params: query parameters to add to the request
:return: response body as bytes
:raise: IrmKmiApiError when communication with the API fails
"""
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[str, str] | None = None) -> str:
"""
Get SVG as str at the specified url with the parameters
:param url: URL to fetch
:param params: query parameters to add to the request
:return: request body decoded as utf-8 str
:raise: IrmKmiApiError when communication with the API fails
"""
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()
def expire_cache(self) -> None:
"""
Expire items from the cache which have not been accessed since self._cache_max_age (default 2h).
Must be called regularly to clear the cache.
"""
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")
async def _api_wrapper(
self,
params: dict,
@ -138,11 +94,11 @@ class IrmKmiApiClient:
else:
headers['User-Agent'] = self._user_agent
if url in self._cache:
headers['If-None-Match'] = self._cache[url]['etag']
if url in self.cache:
headers['If-None-Match'] = self.cache[url]['etag']
try:
async with asyncio.timeout(60):
async with async_timeout.timeout(60):
response = await self._session.request(
method=method,
url=url,
@ -154,13 +110,13 @@ class IrmKmiApiClient:
if response.status == 304:
_LOGGER.debug(f"Cache hit for {url}")
self._cache[url]['timestamp'] = time.time()
return self._cache[url]['response']
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()}
self.cache[url] = {'etag': response.headers['ETag'], 'response': r, 'timestamp': time.time()}
return r
return await response.read()
@ -172,46 +128,34 @@ class IrmKmiApiClient:
except Exception as exception: # pylint: disable=broad-except
raise IrmKmiApiError(f"Something really wrong happened! {exception}") from exception
@staticmethod
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()
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):
"""API client for IRM KMI weather data with additional methods to integrate easily with Home Assistant"""
def __init__(self, session: aiohttp.ClientSession, user_agent: str, cdt_map: Dict[Tuple[int, str], str] | None = None) -> None:
"""
Create a new instance of the API client. This client has more methods to integrate easily with Home Assistant
:param session: aiohttp.ClientSession to use for the request
:param user_agent: string that will indentify your application in the User-Agent header of the HTTP requests
:param cdt_map: mapping of weather conditions returned by the API and string that should be used when calling the
methods. See the wiki for more information on what conditions are possible:
https://github.com/jdejaegh/irm-kmi-api/wiki/API-documentation#obs-key (Table with icons matching).
Example: cdt_map = { (0, 'd'): 'sunny', (0, 'n'): 'clear_night' }
"""
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 if cdt_map is not None else IRM_KMI_TO_HA_CONDITION_MAP
self._cdt_map = cdt_map
async def refresh_forecasts_coord(self, coord: Dict[str, float | int]) -> None:
"""
Update the weather data by contacting the remote API. Keep the data in memory for future methods calls.
:param coord: dict with the following keys: 'lat', 'long' (both float or int)
:raise: IrmKmiApiError when communication with the API fails
"""
async def refresh_forecasts_coord(self, coord: dict) -> None:
self._api_data = await self.get_forecasts_coord(coord)
def get_current_weather(self, tz: ZoneInfo) -> CurrentWeatherData:
"""
Parse the API data we currently have to build a CurrentWeatherData.
def get_city(self) -> str | None:
return self._api_data.get('cityName', None)
:param tz: time zone to use to interpret the timestamps in the forecast (generally is Europe/Brussels)
:return: current weather
"""
def get_country(self) -> str | None:
return self._api_data.get('country', None)
def get_current_weather(self, tz: ZoneInfo) -> CurrentWeatherData:
"""Parse the API data to build a CurrentWeatherData."""
now_hourly = self._get_now_hourly(tz)
uv_index = self._get_uv_index()
@ -278,46 +222,123 @@ class IrmKmiApiClientHa(IrmKmiApiClient):
return current_weather
def get_radar_forecast(self) -> List[RadarForecast]:
"""
Create a list of short term forecasts for rain based on the data provided by the rain radar
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
:return: chronologically ordered list of 'few'-minutes radar forecasts
"""
data = self._api_data.get('animation', {})
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):
if not isinstance(data, dict):
for current in hourly_forecast_data[:4]:
if now.strftime('%H') == current['hour']:
now_hourly = current
break
return now_hourly
def get_daily_forecast(self, tz: ZoneInfo, lang: str) -> List[IrmKmiForecast]:
"""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 []
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
forecasts = list()
forecast_day = datetime.now(tz)
forecast = list()
for f in sequence:
forecast.append(
RadarForecast(
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
)
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
)
return forecast
# 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
def get_hourly_forecast(self, tz: ZoneInfo) -> List[Forecast]:
"""
Parse data from the API to create a list of hourly forecasts
:param tz: time zone to use to interpret the timestamps in the forecast (generally is Europe/Brussels)
:return: chronologically ordered list of hourly forecasts
"""
"""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:
@ -368,116 +389,43 @@ class IrmKmiApiClientHa(IrmKmiApiClient):
return forecasts
def get_daily_forecast(self, tz: ZoneInfo, lang: str) -> List[ExtendedForecast]:
"""
Parse the API data we currently have to build the daily forecast list.
def get_radar_forecast(self) -> List[IrmKmiRadarForecast]:
"""Create a list of short term forecasts for rain based on the data provided by the rain radar"""
data = self._api_data.get('animation', {})
:param tz: time zone to use to interpret the timestamps in the forecast (generally is Europe/Brussels)
:param lang: langage to get data for (must be 'fr', 'nl', 'de' or 'en')
:return: chronologically ordered 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:
if not isinstance(data, dict):
return []
sequence = data.get("sequence", [])
unit = data.get("unit", {}).get("en", None)
ratios = [f['value'] / f['position'] for f in sequence if f['position'] > 0]
forecasts = list()
forecast_day = datetime.now(tz)
if len(ratios) > 0:
ratio = mean(ratios)
else:
ratio = 0
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 = self._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 = ExtendedForecast(
datetime=(forecast_day.strftime('%Y-%m-%d')),
condition=self._cdt_map.get((f.get('ww1', None), f.get('dayNight', None)), None),
condition_2=self._cdt_map.get((f.get('ww2', None), f.get('dayNight', None)), None),
condition_evol=WWEVOL_TO_ENUM_MAP.get(f.get('wwevol'), 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
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
)
)
# 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'])
return forecast
forecasts.append(forecast)
return forecasts
def get_animation_data(self, tz: ZoneInfo, lang: str, style: RadarStyle, dark_mode: bool) -> RadarAnimationData:
"""
Get all the image URLs and create the radar animation data object.
:param tz: time zone to use to interpret the timestamps in the forecast (generally is Europe/Brussels)
:param lang: langage to get data for (must be 'fr', 'nl', 'de' or 'en')
:param style: style of the radar (key of STYLE_TO_PARAM_MAP)
:param dark_mode: true if dark mode
:return: animation data that can be used to download the images and build the rain graph animation
"""
def get_animation_data(self,
tz: ZoneInfo,
lang: str,
style: str,
dark_mode: bool
) -> RadarAnimationData:
"""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()
@ -485,9 +433,9 @@ class IrmKmiApiClientHa(IrmKmiApiClient):
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]})
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
]
@ -522,12 +470,7 @@ class IrmKmiApiClientHa(IrmKmiApiClient):
return radar_animation
def get_warnings(self, lang: str) -> List[WarningData]:
"""
Parse the API data we currently have to build the list of warnings.
:param lang: langage to get data for (must be 'fr', 'nl', 'de' or 'en')
:return: unordered list of warnings
"""
"""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 []
@ -549,7 +492,7 @@ class IrmKmiApiClientHa(IrmKmiApiClient):
result.append(
WarningData(
slug=SLUG_MAP.get(warning_id, WarningType.UNKNOWN),
slug=SLUG_MAP.get(warning_id, 'unknown'),
id=warning_id,
level=level,
friendly_name=data.get('warningType', {}).get('name', {}).get(lang, ''),
@ -561,13 +504,8 @@ class IrmKmiApiClientHa(IrmKmiApiClient):
return result if len(result) > 0 else []
async def get_pollen(self) -> Dict[PollenName, PollenLevel | None]:
"""
Get SVG pollen info from the API, return the pollen data dict
:return: pollen data as dict mapping from pollen name to pollen level as a color
:raise: IrmKmiApiError when communication with the API fails
"""
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', []):
@ -588,48 +526,8 @@ class IrmKmiApiClientHa(IrmKmiApiClient):
return PollenParser(pollen_svg).get_pollen_data()
def get_city(self) -> str | None:
"""
Get the city for which we currently have the forecast
:return: city name as str or None if unavailable
"""
return self._api_data.get('cityName', None)
def get_country(self) -> str | None:
"""
Get the two-letters country code for which we currently have the forecast
:return: country code as str or None if unavailable
"""
return self._api_data.get('country', None)
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
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
@staticmethod
def _merge_url_and_params(url: str, params: dict) -> str:
"""Merge query string params in the URL"""
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)
@ -637,9 +535,3 @@ class IrmKmiApiClientHa(IrmKmiApiClient):
new_url = parsed_url._replace(query=new_query)
return str(urllib.parse.urlunparse(new_url))
@staticmethod
def _next_weekday(current, weekday):
days_ahead = weekday - current.weekday()
if days_ahead < 0:
days_ahead += 7
return current + timedelta(days_ahead)

133
irm_kmi_api/const.py
View file

@ -1,117 +1,28 @@
from typing import Final
from .data import ConditionEvol, PollenLevel, RadarStyle, WarningType
POLLEN_LEVEL_TO_COLOR = {
'null': PollenLevel.GREEN,
'low': PollenLevel.YELLOW,
'moderate': PollenLevel.ORANGE,
'high': PollenLevel.RED,
'very high': PollenLevel.PURPLE,
'active': PollenLevel.ACTIVE
}
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 = {
RadarStyle.OPTION_STYLE_STD: 1,
RadarStyle.OPTION_STYLE_CONTRAST: 2,
RadarStyle.OPTION_STYLE_YELLOW_RED: 3,
RadarStyle.OPTION_STYLE_SATELLITE: 4
OPTION_STYLE_STD: 1,
OPTION_STYLE_CONTRAST: 2,
OPTION_STYLE_YELLOW_RED: 3,
OPTION_STYLE_SATELLITE: 4
}
MAP_WARNING_ID_TO_SLUG: Final = {
0: WarningType.WIND,
1: WarningType.RAIN,
2: WarningType.ICE_OR_SNOW,
3: WarningType.THUNDER,
7: WarningType.FOG,
9: WarningType.COLD,
10: WarningType.HEAT,
12: WarningType.THUNDER_WIND_RAIN,
13: WarningType.THUNDERSTORM_STRONG_GUSTS,
14: WarningType.THUNDERSTORM_LARGE_RAINFALL,
15: WarningType.STORM_SURGE,
17: WarningType.COLDSPELL
}
WWEVOL_TO_ENUM_MAP: Final = {
None: ConditionEvol.STABLE,
0: ConditionEvol.ONE_WAY,
1: ConditionEvol.TWO_WAYS
}
ATTR_CONDITION_CLEAR_NIGHT = "clear-night"
ATTR_CONDITION_CLOUDY = "cloudy"
ATTR_CONDITION_EXCEPTIONAL = "exceptional"
ATTR_CONDITION_FOG = "fog"
ATTR_CONDITION_HAIL = "hail"
ATTR_CONDITION_LIGHTNING = "lightning"
ATTR_CONDITION_LIGHTNING_RAINY = "lightning-rainy"
ATTR_CONDITION_PARTLYCLOUDY = "partlycloudy"
ATTR_CONDITION_POURING = "pouring"
ATTR_CONDITION_RAINY = "rainy"
ATTR_CONDITION_SNOWY = "snowy"
ATTR_CONDITION_SNOWY_RAINY = "snowy-rainy"
ATTR_CONDITION_SUNNY = "sunny"
ATTR_CONDITION_WINDY = "windy"
ATTR_CONDITION_WINDY_VARIANT = "windy-variant"
IRM_KMI_TO_HA_CONDITION_MAP: Final = {
(0, 'd'): ATTR_CONDITION_SUNNY,
(0, 'n'): ATTR_CONDITION_CLEAR_NIGHT,
(1, 'd'): ATTR_CONDITION_SUNNY,
(1, 'n'): ATTR_CONDITION_CLEAR_NIGHT,
(2, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
(2, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
(3, 'd'): ATTR_CONDITION_PARTLYCLOUDY,
(3, 'n'): ATTR_CONDITION_PARTLYCLOUDY,
(4, 'd'): ATTR_CONDITION_POURING,
(4, 'n'): ATTR_CONDITION_POURING,
(5, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
(5, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
(6, 'd'): ATTR_CONDITION_POURING,
(6, 'n'): ATTR_CONDITION_POURING,
(7, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
(7, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
(8, 'd'): ATTR_CONDITION_SNOWY_RAINY,
(8, 'n'): ATTR_CONDITION_SNOWY_RAINY,
(9, 'd'): ATTR_CONDITION_SNOWY_RAINY,
(9, 'n'): ATTR_CONDITION_SNOWY_RAINY,
(10, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
(10, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
(11, 'd'): ATTR_CONDITION_SNOWY,
(11, 'n'): ATTR_CONDITION_SNOWY,
(12, 'd'): ATTR_CONDITION_SNOWY,
(12, 'n'): ATTR_CONDITION_SNOWY,
(13, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
(13, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
(14, 'd'): ATTR_CONDITION_CLOUDY,
(14, 'n'): ATTR_CONDITION_CLOUDY,
(15, 'd'): ATTR_CONDITION_CLOUDY,
(15, 'n'): ATTR_CONDITION_CLOUDY,
(16, 'd'): ATTR_CONDITION_POURING,
(16, 'n'): ATTR_CONDITION_POURING,
(17, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
(17, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
(18, 'd'): ATTR_CONDITION_RAINY,
(18, 'n'): ATTR_CONDITION_RAINY,
(19, 'd'): ATTR_CONDITION_POURING,
(19, 'n'): ATTR_CONDITION_POURING,
(20, 'd'): ATTR_CONDITION_SNOWY_RAINY,
(20, 'n'): ATTR_CONDITION_SNOWY_RAINY,
(21, 'd'): ATTR_CONDITION_RAINY,
(21, 'n'): ATTR_CONDITION_RAINY,
(22, 'd'): ATTR_CONDITION_SNOWY,
(22, 'n'): ATTR_CONDITION_SNOWY,
(23, 'd'): ATTR_CONDITION_SNOWY,
(23, 'n'): ATTR_CONDITION_SNOWY,
(24, 'd'): ATTR_CONDITION_FOG,
(24, 'n'): ATTR_CONDITION_FOG,
(25, 'd'): ATTR_CONDITION_FOG,
(25, 'n'): ATTR_CONDITION_FOG,
(26, 'd'): ATTR_CONDITION_FOG,
(26, 'n'): ATTR_CONDITION_FOG,
(27, 'd'): ATTR_CONDITION_FOG,
(27, 'n'): ATTR_CONDITION_FOG
}
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'}

70
irm_kmi_api/data.py
View file

@ -1,6 +1,5 @@
"""Data classes for IRM KMI integration"""
from datetime import datetime
from enum import StrEnum
from typing import List, Required, TypedDict
@ -36,66 +35,10 @@ class Forecast(TypedDict, total=False):
is_daytime: bool | None # Mandatory to use with forecast_twice_daily
class ConditionEvol(StrEnum):
"""Possible state for evolution between weather conditions"""
ONE_WAY = 'one_way'
TWO_WAYS = 'two_ways'
STABLE = 'stable'
class RadarStyle(StrEnum):
"""Possible style for the rain radar"""
OPTION_STYLE_STD = 'standard_style'
OPTION_STYLE_CONTRAST = 'contrast_style'
OPTION_STYLE_YELLOW_RED = 'yellow_red_style'
OPTION_STYLE_SATELLITE = 'satellite_style'
class PollenName(StrEnum):
ALDER = 'alder'
ASH = 'ash'
BIRCH = 'birch'
GRASSES = 'grasses'
HAZEL = 'hazel'
MUGWORT = 'mugwort'
OAK = 'oak'
class PollenLevel(StrEnum):
"""Possible pollen levels"""
NONE = 'none'
ACTIVE = 'active'
GREEN = 'green'
YELLOW = 'yellow'
ORANGE = 'orange'
RED = 'red'
PURPLE = 'purple'
class WarningType(StrEnum):
"""Possible warning types"""
COLD = 'cold'
COLDSPELL = 'coldspell'
FOG = 'fog'
HEAT = 'heat'
ICE_OR_SNOW = 'ice_or_snow'
RAIN = 'rain'
STORM_SURGE = 'storm_surge'
THUNDER = 'thunder'
THUNDERSTORM_LARGE_RAINFALL = 'thunderstorm_large_rainfall'
THUNDERSTORM_STRONG_GUSTS = 'thunderstorm_strong_gusts'
THUNDER_WIND_RAIN = 'thunder_wind_rain'
WIND = 'wind'
UNKNOWN = 'unknown'
class ExtendedForecast(Forecast, total=False):
class IrmKmiForecast(Forecast):
"""Forecast class with additional attributes for IRM KMI"""
condition_2: str | None
condition_evol: ConditionEvol | None
# TODO: add condition_2 as well and evolution to match data from the API?
text: str | None
sunrise: str | None
sunset: str | None
@ -103,7 +46,6 @@ class ExtendedForecast(Forecast, total=False):
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
@ -115,8 +57,7 @@ class CurrentWeatherData(TypedDict, total=False):
class WarningData(TypedDict, total=False):
"""Holds data about a specific warning"""
slug: WarningType
slug: str
id: int
level: int
friendly_name: str
@ -125,9 +66,8 @@ class WarningData(TypedDict, total=False):
ends_at: datetime
class RadarForecast(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
@ -136,7 +76,6 @@ class RadarForecast(Forecast):
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
@ -147,7 +86,6 @@ class AnimationFrameData(TypedDict, total=False):
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

141
irm_kmi_api/pollen.py
View file

@ -1,10 +1,9 @@
"""Parse pollen info from SVG from IRM KMI api"""
import logging
import xml.etree.ElementTree as ET
from typing import Dict, List
from typing import List
from .const import POLLEN_LEVEL_TO_COLOR
from .data import PollenLevel, PollenName
from .const import POLLEN_LEVEL_TO_COLOR, POLLEN_NAMES
_LOGGER = logging.getLogger(__name__)
@ -22,85 +21,20 @@ class PollenParser:
):
self._xml = xml_string
def get_pollen_data(self) -> Dict[PollenName, PollenLevel | None]:
"""
Parse the SVG and extract the pollen data from the image.
If an error occurs, return the default value.
:return: pollen dict
"""
pollen_data = self.get_default_data()
try:
_LOGGER.debug(f"Full SVG: {self._xml}")
root = ET.fromstring(self._xml)
except ET.ParseError as e:
_LOGGER.warning(f"Could not parse SVG pollen XML: {e}")
return pollen_data
elements: List[ET.Element] = self._extract_elements(root)
pollens = {e.attrib.get('x', None): self._get_txt(e).lower()
for e in elements if 'tspan' in e.tag and str(self._get_txt(e)).lower() in PollenName}
pollen_levels = {e.attrib.get('x', None): POLLEN_LEVEL_TO_COLOR[self._get_txt(e)]
for e in elements if 'tspan' in e.tag and self._get_txt(e) in POLLEN_LEVEL_TO_COLOR}
level_dots = {e.attrib.get('cx', None) for e in elements if 'circle' in e.tag}
# For each pollen name found, check the text just below.
# As of January 2025, the text is always 'active' and the dot shows the real level
# If text says 'active', check the dot; else trust the text
for position, pollen in pollens.items():
# Check if pollen is a known one
try:
pollen: PollenName = PollenName(pollen)
except ValueError:
_LOGGER.warning(f'Unknown pollen name {pollen}')
continue
# Determine pollen level based on text
if position is not None and position in pollen_levels:
pollen_data[pollen] = pollen_levels[position]
_LOGGER.debug(f"{pollen.value} is {pollen_data[pollen]} according to text")
# If text is 'active' or if there is no text, check the dot as a fallback
if pollen_data[pollen] not in {PollenLevel.NONE, PollenLevel.ACTIVE}:
_LOGGER.debug(f"{pollen} trusting text")
else:
for dot in level_dots:
try:
relative_x_position = float(position) - float(dot)
except TypeError:
pass
else:
if 24 <= relative_x_position <= 34:
pollen_data[pollen] = PollenLevel.GREEN
elif 13 <= relative_x_position <= 23:
pollen_data[pollen] = PollenLevel.YELLOW
elif -5 <= relative_x_position <= 5:
pollen_data[pollen] = PollenLevel.ORANGE
elif -23 <= relative_x_position <= -13:
pollen_data[pollen] = PollenLevel.RED
elif -34 <= relative_x_position <= -24:
pollen_data[pollen] = PollenLevel.PURPLE
_LOGGER.debug(f"{pollen.value} is {pollen_data[pollen]} according to dot")
_LOGGER.debug(f"Pollen data: {pollen_data}")
return pollen_data
@staticmethod
def get_default_data() -> Dict[PollenName, PollenLevel | None]:
def get_default_data() -> dict:
"""Return all the known pollen with 'none' value"""
return {k: PollenLevel.NONE for k in PollenName}
return {k.lower(): 'none' for k in POLLEN_NAMES}
@staticmethod
def get_unavailable_data() -> Dict[PollenName, PollenLevel | None]:
"""Return all the known pollen with None value"""
return {k: None for k in PollenName}
def get_unavailable_data() -> dict:
"""Return all the known pollen with 'none' value"""
return {k.lower(): None for k in POLLEN_NAMES}
@staticmethod
def get_option_values() -> List[PollenLevel]:
def get_option_values() -> List[str]:
"""List all the values that the pollen can have"""
return list(POLLEN_LEVEL_TO_COLOR.values()) + [PollenLevel.NONE]
return list(POLLEN_LEVEL_TO_COLOR.values()) + ['none']
@staticmethod
def _extract_elements(root) -> List[ET.Element]:
@ -112,8 +46,63 @@ class PollenParser:
return elements
@staticmethod
def _get_txt(e) -> str | None:
def _get_elem_text(e) -> str | None:
if e.text is not None:
return e.text.strip()
return None
def get_pollen_data(self) -> dict:
"""From the XML string, parse the SVG and extract the pollen data from the image.
If an error occurs, return the default value"""
pollen_data = self.get_default_data()
try:
_LOGGER.debug(f"Full SVG: {self._xml}")
root = ET.fromstring(self._xml)
except ET.ParseError as e:
_LOGGER.warning(f"Could not parse SVG pollen XML: {e}")
return pollen_data
elements: List[ET.Element] = self._extract_elements(root)
pollens = {e.attrib.get('x', None): self._get_elem_text(e).lower()
for e in elements if 'tspan' in e.tag and self._get_elem_text(e) in POLLEN_NAMES}
pollen_levels = {e.attrib.get('x', None): POLLEN_LEVEL_TO_COLOR[self._get_elem_text(e)]
for e in elements if 'tspan' in e.tag and self._get_elem_text(e) in POLLEN_LEVEL_TO_COLOR}
level_dots = {e.attrib.get('cx', None) for e in elements if 'circle' in e.tag}
# For each pollen name found, check the text just below.
# As of January 2025, the text is always 'active' and the dot shows the real level
# If text says 'active', check the dot; else trust the text
for position, pollen in pollens.items():
# Determine pollen level based on text
if position is not None and position in pollen_levels:
pollen_data[pollen] = pollen_levels[position]
_LOGGER.debug(f"{pollen} is {pollen_data[pollen]} according to text")
# If text is 'active' or if there is no text, check the dot as a fallback
if pollen_data[pollen] not in {'none', 'active'}:
_LOGGER.debug(f"{pollen} trusting text")
else:
for dot in level_dots:
try:
relative_x_position = float(position) - float(dot)
except TypeError:
pass
else:
if 24 <= relative_x_position <= 34:
pollen_data[pollen] = 'green'
elif 13 <= relative_x_position <= 23:
pollen_data[pollen] = 'yellow'
elif -5 <= relative_x_position <= 5:
pollen_data[pollen] = 'orange'
elif -23 <= relative_x_position <= -13:
pollen_data[pollen] = 'red'
elif -34 <= relative_x_position <= -24:
pollen_data[pollen] = 'purple'
_LOGGER.debug(f"{pollen} is {pollen_data[pollen]} according to dot")
_LOGGER.debug(f"Pollen data: {pollen_data}")
return pollen_data

222
irm_kmi_api/rain_graph.py
View file

@ -4,26 +4,26 @@ import base64
import copy
import datetime
import logging
from typing import List, Self
from typing import Any, List, Self
import async_timeout
from svgwrite import Drawing
from svgwrite.animate import Animate
from svgwrite.container import FONT_TEMPLATE
from .api import IrmKmiApiClient, IrmKmiApiError
from .data import AnimationFrameData, RadarAnimationData, RadarStyle
from .api import IrmKmiApiClient
from .const import OPTION_STYLE_SATELLITE
from .data import AnimationFrameData, RadarAnimationData
from .resources import be_black, be_satellite, be_white, nl, roboto
_LOGGER = logging.getLogger(__name__)
class RainGraph:
"""Create and get rain radar animated SVG"""
def __init__(self,
animation_data: RadarAnimationData,
country: str,
style: RadarStyle,
style: str,
dark_mode: bool = False,
tz: datetime.tzinfo = None,
svg_width: float = 640,
@ -77,121 +77,82 @@ class RainGraph:
async def build(self) -> Self:
"""Build the rain graph by calling all the method in the right order. Returns self when done"""
await self._draw_svg_frame()
self._draw_hour_bars()
self._draw_chances_path()
self._draw_data_line()
self._write_hint()
await self._insert_background()
await self.draw_svg_frame()
self.draw_hour_bars()
self.draw_chances_path()
self.draw_data_line()
self.write_hint()
await self.insert_background()
self._dwg_save = copy.deepcopy(self._dwg)
return self
async def get_animated(self) -> bytes:
"""
Get the animated SVG. If called for the first time, downloads the cloud images to build the file.
:return: utf-8 encoded animated SVG string
:raises: ValueError if build() was not called before
"""
if self._dwg_save is None:
raise ValueError("You need to call .build() before getting the SVG")
"""Get the animated SVG. If called for the first time since refresh, downloads the images to build the file."""
_LOGGER.info(f"Get animated with _dwg_animated {self._dwg_animated}")
if self._dwg_animated is None:
clouds = self._download_clouds()
clouds = self.download_clouds()
self._dwg = copy.deepcopy(self._dwg_save)
self._draw_current_fame_line()
self._draw_description_text()
try:
await clouds
self._insert_cloud_layer()
except IrmKmiApiError as err:
_LOGGER.warning(f"Could not download clouds from API: {err}")
await self._draw_location()
self.draw_current_fame_line()
self.draw_description_text()
await clouds
self.insert_cloud_layer()
await self.draw_location()
self._dwg_animated = self._dwg
return self._get_svg_string(still_image=False)
return self.get_svg_string(still_image=False)
async def get_still(self) -> bytes:
"""
Get the still SVG. If called for the first time, downloads the cloud images to build the file.
:return: utf-8 encoded SVG string
:raises: ValueError if build() was not called before
"""
if self._dwg_save is None:
raise ValueError("You need to call .build() before getting the SVG")
"""Get the animated SVG. If called for the first time since refresh, downloads the images to build the file."""
_LOGGER.info(f"Get still with _dwg_still {self._dwg_still}")
if self._dwg_still is None:
idx = self._animation_data['most_recent_image_idx']
cloud = self._download_clouds(idx)
cloud = self.download_clouds(idx)
self._dwg = copy.deepcopy(self._dwg_save)
self._draw_current_fame_line(idx)
self._draw_description_text(idx)
try:
await cloud
self._insert_cloud_layer(idx)
except IrmKmiApiError as err:
_LOGGER.warning(f"Could not download clouds from API: {err}")
await self._draw_location()
self.draw_current_fame_line(idx)
self.draw_description_text(idx)
await cloud
self.insert_cloud_layer(idx)
await self.draw_location()
self._dwg_still = self._dwg
return self._get_svg_string(still_image=True)
return self.get_svg_string(still_image=True)
def get_hint(self) -> str:
"""
Get hint to display on the rain graph
:return: hint sentence as str
"""
return self._animation_data.get('hint', '')
async def _download_clouds(self, idx: int | None = None):
"""
Download cloud images and save the result in the internal state.
:param idx: index of the image to download (if not specified, downloads all the images)
:raises: IrmKmiApiError if communication with the API fails
"""
async def download_clouds(self, idx = None):
imgs = [e['image'] for e in self._animation_data['sequence']]
if idx is not None and type(imgs[idx]) is str:
_LOGGER.info("Download single cloud image")
print("Download single cloud image")
result = await self._download_images_from_api([imgs[idx]])
result = await self.download_images_from_api([imgs[idx]])
self._animation_data['sequence'][idx]['image'] = result[0]
else:
_LOGGER.info("Download many cloud images")
result = await self._download_images_from_api([img for img in imgs if type(img) is str])
result = await self.download_images_from_api([img for img in imgs if type(img) is str])
for i in range(len(self._animation_data['sequence'])):
if type(self._animation_data['sequence'][i]['image']) is str:
self._animation_data['sequence'][i]['image'] = result[0]
result = result[1:]
async def _download_images_from_api(self, urls: list[str]) -> list[bytes]:
"""
Download a batch of images to create the radar frames.
:param urls: list of urls to download
:return: list images downloaded as bytes
:raises: IrmKmiApiError if communication with the API fails
"""
async def download_images_from_api(self, urls: list[str]) -> list[Any]:
"""Download a batch of images to create the radar frames."""
coroutines = list()
for url in urls:
coroutines.append(self._api_client.get_image(url))
async with asyncio.timeout(60):
async with async_timeout.timeout(60):
images_from_api = await asyncio.gather(*coroutines)
_LOGGER.info(f"Just downloaded {len(images_from_api)} images")
return images_from_api
async def _draw_svg_frame(self):
def get_hint(self) -> str:
return self._animation_data.get('hint', None)
async def draw_svg_frame(self):
"""Create the global area to draw the other items"""
mimetype = "application/x-font-ttf"
@ -200,7 +161,9 @@ class RainGraph:
self._dwg.embed_stylesheet(content)
self._dwg.embed_stylesheet("""
.roboto { font-family: "Roboto Medium"; }
.roboto {
font-family: "Roboto Medium";
}
""")
fill_color = '#393C40' if self._dark_mode else '#385E95'
@ -209,7 +172,7 @@ class RainGraph:
rx=None, ry=None,
fill=fill_color, stroke='none'))
def _draw_description_text(self, idx: int | None = None):
def draw_description_text(self, idx: int | None = None):
"""For every frame write the amount of precipitation and the time at the top of the graph.
If idx is set, only do it for the given idx"""
@ -223,7 +186,7 @@ class RainGraph:
paragraph = self._dwg.add(self._dwg.g(class_="roboto", ))
self._write_time_and_rain(paragraph, rain_level, time)
self.write_time_and_rain(paragraph, rain_level, time)
return
for i in range(self._frame_count):
@ -243,9 +206,9 @@ class RainGraph:
repeatCount="indefinite"
))
self._write_time_and_rain(paragraph, rain_level, time)
self.write_time_and_rain(paragraph, rain_level, time)
def _write_time_and_rain(self, paragraph, rain_level, time):
def write_time_and_rain(self, paragraph, rain_level, time):
"""Using the paragraph object, write the time and rain level data"""
paragraph.add(self._dwg.text(f"{time}", insert=(self._offset, self._top_text_y_pos),
text_anchor="start",
@ -258,11 +221,11 @@ class RainGraph:
fill="white",
stroke='none'))
def _write_hint(self):
def write_hint(self):
"""Add the hint text at the bottom of the graph"""
paragraph = self._dwg.add(self._dwg.g(class_="roboto", ))
hint = self.get_hint()
hint = self._animation_data['hint']
paragraph.add(self._dwg.text(f"{hint}", insert=(self._svg_width / 2, self._bottom_text_y_pos),
text_anchor="middle",
@ -270,7 +233,7 @@ class RainGraph:
fill="white",
stroke='none'))
def _draw_chances_path(self):
def draw_chances_path(self):
"""Draw the prevision margin area around the main forecast line"""
list_lower_points = []
list_higher_points = []
@ -295,9 +258,36 @@ class RainGraph:
graph_rect_right -= self._interval_width
if list_higher_points and list_lower_points:
self._draw_chance_precip(list_higher_points, list_lower_points)
self.draw_chance_precip(list_higher_points, list_lower_points)
def _draw_data_line(self):
def draw_chance_precip(self, list_higher_points: List, list_lower_points: List):
"""Draw the blue solid line representing the actual rain forecast"""
precip_higher_chance_path = self._dwg.path(fill='#63c8fa', stroke='none', opacity=.3)
list_higher_points[-1] = tuple(list(list_higher_points[-1]) + ['last'])
self.set_curved_path(precip_higher_chance_path, list_higher_points + list_lower_points)
self._dwg.add(precip_higher_chance_path)
@staticmethod
def set_curved_path(path, points):
"""Pushes points on the path by creating a nice curve between them"""
if len(points) < 2:
return
path.push('M', *points[0])
for i in range(1, len(points)):
x_mid = (points[i - 1][0] + points[i][0]) / 2
y_mid = (points[i - 1][1] + points[i][1]) / 2
path.push('Q', points[i - 1][0], points[i - 1][1], x_mid, y_mid)
if points[i][-1] == 'last' or points[i - 1][-1] == 'last':
path.push('Q', points[i][0], points[i][1], points[i][0], points[i][1])
path.push('Q', points[-1][0], points[-1][1], points[-1][0], points[-1][1])
def draw_data_line(self):
"""Draw the main data line for the rain forecast"""
rain_list: List[AnimationFrameData] = self._animation_data['sequence']
graph_rect_left = self._offset
@ -312,10 +302,10 @@ class RainGraph:
graph_rect_top + (1.0 - position) * self._graph_height))
graph_rect_left += self._interval_width
data_line_path = self._dwg.path(fill='none', stroke='#63c8fa', stroke_width=2)
self._set_curved_path(data_line_path, entry_list)
self.set_curved_path(data_line_path, entry_list)
self._dwg.add(data_line_path)
def _draw_hour_bars(self):
def draw_hour_bars(self):
"""Draw the small bars at the bottom to represent the time"""
hour_bar_height = 8
horizontal_inset = self._offset
@ -354,7 +344,7 @@ class RainGraph:
end=(self._graph_width + self._interval_width / 2, self._graph_bottom),
stroke='white'))
def _draw_current_fame_line(self, idx: int | None = None):
def draw_current_fame_line(self, idx: int | None = None):
"""Draw a solid white line on the timeline at the position of the given frame index"""
x_position = self._offset if idx is None else self._offset + idx * self._interval_width
now = self._dwg.add(self._dwg.line(start=(x_position, self._top_text_space),
@ -372,21 +362,15 @@ class RainGraph:
dur=f"{self._frame_count * 0.3}s",
repeatCount="indefinite"))
def _get_svg_string(self, still_image: bool = False) -> bytes:
"""
Get the utf-8 encoded string representing the SVG
:param still_image: if true the non-animated version is returned
:return: utf-8 encoded string
"""
def get_svg_string(self, still_image: bool = False) -> bytes:
return self._dwg_still.tostring().encode() if still_image else self._dwg_animated.tostring().encode()
async def _insert_background(self):
png_data = self._get_background_png_b64()
async def insert_background(self):
png_data = self.get_background_png_b64()
image = self._dwg.image("data:image/png;base64," + png_data, insert=(0, 0), size=self._background_size)
self._dwg.add(image)
def _insert_cloud_layer(self, idx: int | None = None):
def insert_cloud_layer(self, idx: int | None = None):
imgs = [e['image'] for e in self._animation_data['sequence']]
if idx is not None:
@ -412,54 +396,28 @@ class RainGraph:
repeatCount="indefinite"
))
async def _draw_location(self):
async def draw_location(self):
img = self._animation_data['location']
_LOGGER.info(f"Draw location layer with img of type {type(img)}")
if type(img) is str:
result = await self._download_images_from_api([img])
result = await self.download_images_from_api([img])
img = result[0]
self._animation_data['location'] = img
png_data = base64.b64encode(img).decode('utf-8')
image = self._dwg.image("data:image/png;base64," + png_data, insert=(0, 0), size=self._background_size)
self._dwg.add(image)
def _get_dwg(self) -> Drawing:
def get_dwg(self):
return copy.deepcopy(self._dwg)
def _get_background_png_b64(self) -> str:
def get_background_png_b64(self):
_LOGGER.debug(f"Get b64 for {self._country} {self._style} {'dark' if self._dark_mode else 'light'} mode")
if self._country == 'NL':
return nl.nl_b64
elif self._style == RadarStyle.OPTION_STYLE_SATELLITE:
elif self._style == OPTION_STYLE_SATELLITE:
return be_satellite.be_satelitte_b64
elif self._dark_mode:
return be_black.be_black_b64
else:
return be_white.be_white_b64
def _draw_chance_precip(self, list_higher_points: List, list_lower_points: List):
"""Draw the blue solid line representing the actual rain forecast"""
precip_higher_chance_path = self._dwg.path(fill='#63c8fa', stroke='none', opacity=.3)
list_higher_points[-1] = tuple(list(list_higher_points[-1]) + ['last'])
self._set_curved_path(precip_higher_chance_path, list_higher_points + list_lower_points)
self._dwg.add(precip_higher_chance_path)
@staticmethod
def _set_curved_path(path, points):
"""Pushes points on the path by creating a nice curve between them"""
if len(points) < 2:
return
path.push('M', *points[0])
for i in range(1, len(points)):
x_mid = (points[i - 1][0] + points[i][0]) / 2
y_mid = (points[i - 1][1] + points[i][1]) / 2
path.push('Q', points[i - 1][0], points[i - 1][1], x_mid, y_mid)
if points[i][-1] == 'last' or points[i - 1][-1] == 'last':
path.push('Q', points[i][0], points[i][1], points[i][0], points[i][1])
path.push('Q', points[-1][0], points[-1][1], points[-1][0], points[-1][1])

8
irm_kmi_api/utils.py Normal file
View file

@ -0,0 +1,8 @@
from datetime import timedelta
def next_weekday(current, weekday):
days_ahead = weekday - current.weekday()
if days_ahead < 0:
days_ahead += 7
return current + timedelta(days_ahead)

14
pyproject.toml
View file

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "irm-kmi-api"
version = "1.1.0"
version = "0.1.3"
description = "Retrieve data from the Belgian IRM KMI in Python"
readme = "README.md"
authors = [{ name = "Jules Dejaeghere", email = "curable.grass491@mailer.me" }]
@ -16,19 +16,17 @@ classifiers = [
]
keywords = ["weather", "weather-api", "netherlands", "weather-forecast", "pollen", "belgium", "luxembourg", "rain-radar"]
dependencies = [
"aiohttp>=3.11.0,<4.0.0",
"svgwrite>=1.4.3,<2.0.0",
"aiohttp>=3.11.13",
"async-timeout>=4.0.3",
"svgwrite>=1.4.3",
]
requires-python = ">=3.12"
requires-python = ">=3.11"
[project.urls]
Homepage = "https://github.com/jdejaegh/irm-kmi-api"
[tool.setuptools]
packages = ["irm_kmi_api", "irm_kmi_api.resources"]
[tool.bumpver]
current_version = "1.1.0"
current_version = "0.1.3"
version_pattern = "MAJOR.MINOR.PATCH"
commit_message = "bump version {old_version} -> {new_version}"
tag_message = "{new_version}"

5
requirements.txt
View file

@ -1,2 +1,3 @@
aiohttp>=3.11.0,<4.0.0
svgwrite>=1.4.3,<2.0.0
aiohttp>=3.11.13
async-timeout>=4.0.3
svgwrite>=1.4.3

16
tests/conftest.py
View file

@ -4,8 +4,8 @@ from __future__ import annotations
import json
from unittest.mock import MagicMock
from irm_kmi_api import IrmKmiApiClientHa
from irm_kmi_api.const import IRM_KMI_TO_HA_CONDITION_MAP
from irm_kmi_api.api import IrmKmiApiClientHa
from tests.const import IRM_KMI_TO_HA_CONDITION_MAP
def load_fixture(fixture):
@ -21,15 +21,3 @@ 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
def is_serializable(x):
try:
json.dumps(x)
return True
except (TypeError, OverflowError):
return False
def assert_all_serializable(elements: list):
for element in elements:
for v in element.values():
assert is_serializable(v)

77
tests/const.py Normal file
View file

@ -0,0 +1,77 @@
from typing import Final
ATTR_CONDITION_CLASS = "condition_class"
ATTR_CONDITION_CLEAR_NIGHT = "clear-night"
ATTR_CONDITION_CLOUDY = "cloudy"
ATTR_CONDITION_EXCEPTIONAL = "exceptional"
ATTR_CONDITION_FOG = "fog"
ATTR_CONDITION_HAIL = "hail"
ATTR_CONDITION_LIGHTNING = "lightning"
ATTR_CONDITION_LIGHTNING_RAINY = "lightning-rainy"
ATTR_CONDITION_PARTLYCLOUDY = "partlycloudy"
ATTR_CONDITION_POURING = "pouring"
ATTR_CONDITION_RAINY = "rainy"
ATTR_CONDITION_SNOWY = "snowy"
ATTR_CONDITION_SNOWY_RAINY = "snowy-rainy"
ATTR_CONDITION_SUNNY = "sunny"
ATTR_CONDITION_WINDY = "windy"
ATTR_CONDITION_WINDY_VARIANT = "windy-variant"
IRM_KMI_TO_HA_CONDITION_MAP: Final = {
(0, 'd'): ATTR_CONDITION_SUNNY,
(0, 'n'): ATTR_CONDITION_CLEAR_NIGHT,
(1, 'd'): ATTR_CONDITION_SUNNY,
(1, 'n'): ATTR_CONDITION_CLEAR_NIGHT,
(2, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
(2, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
(3, 'd'): ATTR_CONDITION_PARTLYCLOUDY,
(3, 'n'): ATTR_CONDITION_PARTLYCLOUDY,
(4, 'd'): ATTR_CONDITION_POURING,
(4, 'n'): ATTR_CONDITION_POURING,
(5, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
(5, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
(6, 'd'): ATTR_CONDITION_POURING,
(6, 'n'): ATTR_CONDITION_POURING,
(7, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
(7, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
(8, 'd'): ATTR_CONDITION_SNOWY_RAINY,
(8, 'n'): ATTR_CONDITION_SNOWY_RAINY,
(9, 'd'): ATTR_CONDITION_SNOWY_RAINY,
(9, 'n'): ATTR_CONDITION_SNOWY_RAINY,
(10, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
(10, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
(11, 'd'): ATTR_CONDITION_SNOWY,
(11, 'n'): ATTR_CONDITION_SNOWY,
(12, 'd'): ATTR_CONDITION_SNOWY,
(12, 'n'): ATTR_CONDITION_SNOWY,
(13, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
(13, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
(14, 'd'): ATTR_CONDITION_CLOUDY,
(14, 'n'): ATTR_CONDITION_CLOUDY,
(15, 'd'): ATTR_CONDITION_CLOUDY,
(15, 'n'): ATTR_CONDITION_CLOUDY,
(16, 'd'): ATTR_CONDITION_POURING,
(16, 'n'): ATTR_CONDITION_POURING,
(17, 'd'): ATTR_CONDITION_LIGHTNING_RAINY,
(17, 'n'): ATTR_CONDITION_LIGHTNING_RAINY,
(18, 'd'): ATTR_CONDITION_RAINY,
(18, 'n'): ATTR_CONDITION_RAINY,
(19, 'd'): ATTR_CONDITION_POURING,
(19, 'n'): ATTR_CONDITION_POURING,
(20, 'd'): ATTR_CONDITION_SNOWY_RAINY,
(20, 'n'): ATTR_CONDITION_SNOWY_RAINY,
(21, 'd'): ATTR_CONDITION_RAINY,
(21, 'n'): ATTR_CONDITION_RAINY,
(22, 'd'): ATTR_CONDITION_SNOWY,
(22, 'n'): ATTR_CONDITION_SNOWY,
(23, 'd'): ATTR_CONDITION_SNOWY,
(23, 'n'): ATTR_CONDITION_SNOWY,
(24, 'd'): ATTR_CONDITION_FOG,
(24, 'n'): ATTR_CONDITION_FOG,
(25, 'd'): ATTR_CONDITION_FOG,
(25, 'n'): ATTR_CONDITION_FOG,
(26, 'd'): ATTR_CONDITION_FOG,
(26, 'n'): ATTR_CONDITION_FOG,
(27, 'd'): ATTR_CONDITION_FOG,
(27, 'n'): ATTR_CONDITION_FOG
}

1647
tests/fixtures/antwerp_with_heat_warning.json vendored
View file

File diff suppressed because it is too large Load diff

2
tests/fixtures/forecast.json vendored
View file

@ -75,7 +75,7 @@
"tempMax": 9,
"ww1": 3,
"ww2": null,
"wwevol": 1,
"wwevol": null,
"ff1": 4,
"ff2": null,
"ffevol": null,

29
tests/test_api.py
View file

@ -1,16 +1,15 @@
import json
import time
from datetime import datetime as dt
from datetime import timedelta
from datetime import datetime as dt, timedelta
from unittest.mock import AsyncMock, MagicMock
from zoneinfo import ZoneInfo
import freezegun
import pytest
from irm_kmi_api import IrmKmiApiClient, IrmKmiApiClientHa
from irm_kmi_api import CurrentWeatherData
from irm_kmi_api import PollenParser
from irm_kmi_api.api import IrmKmiApiClient, IrmKmiApiClientHa
from irm_kmi_api.data import CurrentWeatherData
from irm_kmi_api.pollen import PollenParser
@freezegun.freeze_time(dt.fromisoformat('2025-05-03T17:30:00+00:00'))
@ -70,9 +69,9 @@ async def test_get_image_api_called() -> None:
def test_expire_cache_clears_items() -> None:
api = IrmKmiApiClient(session=MagicMock(), user_agent="test-user-agent")
assert api._cache_max_age == 60 * 60 * 2
assert api.cache_max_age == 60 * 60 * 2
api._cache = {
api.cache = {
'first-url': {
'timestamp': time.time() - timedelta(hours=3).seconds,
'response': 'wowo',
@ -85,12 +84,12 @@ def test_expire_cache_clears_items() -> None:
}
}
assert len(api._cache) == 2
assert len(api.cache) == 2
api.expire_cache()
assert len(api._cache) == 1
assert 'second-url' in api._cache
assert len(api.cache) == 1
assert 'second-url' in api.cache
async def test_api_wrapper_puts_response_in_cache() -> None:
@ -108,8 +107,8 @@ async def test_api_wrapper_puts_response_in_cache() -> None:
r = await api._api_wrapper(params={}, base_url='test-url')
assert r == b"response value"
assert len(api._cache) == 1
assert 'test-url' in api._cache
assert len(api.cache) == 1
assert 'test-url' in api.cache
session.request.assert_awaited_once_with(
method='get', url='test-url', headers={'User-Agent': 'test-user-agent'}, json=None, params={}
@ -127,7 +126,7 @@ async def test_api_wrapper_gets_response_from_cache() -> None:
session.request = AsyncMock(return_value=response)
api = IrmKmiApiClient(session=session, user_agent="test-user-agent")
api._cache = {
api.cache = {
'test-url': {
'timestamp': time.time(),
'response': b"response value",
@ -138,8 +137,8 @@ async def test_api_wrapper_gets_response_from_cache() -> None:
r = await api._api_wrapper(params={}, base_url='test-url')
assert r == b"response value"
assert len(api._cache) == 1
assert 'test-url' in api._cache
assert len(api.cache) == 1
assert 'test-url' in api.cache
session.request.assert_awaited_once_with(
method='get',

13
tests/test_current_weather.py
View file

@ -4,9 +4,9 @@ from zoneinfo import ZoneInfo
import pytest
from freezegun import freeze_time
from irm_kmi_api import CurrentWeatherData
from tests.conftest import get_api_data, get_api_with_data, is_serializable
from irm_kmi_api.const import ATTR_CONDITION_CLOUDY, ATTR_CONDITION_PARTLYCLOUDY
from irm_kmi_api.data import CurrentWeatherData
from tests.conftest import get_api_data, get_api_with_data
from tests.const import ATTR_CONDITION_CLOUDY, ATTR_CONDITION_PARTLYCLOUDY
@freeze_time(datetime.fromisoformat('2023-12-26T17:30:00+00:00'))
@ -140,10 +140,3 @@ def test_current_weather_attributes(
assert r == expected_
run(sensor, expected)
@freeze_time(datetime.fromisoformat('2023-12-26T17:30:00+00:00'))
def test_current_weather_is_serializable() -> None:
api = get_api_with_data("forecast.json")
tz = ZoneInfo("Europe/Brussels")
result = api.get_current_weather(tz)
assert is_serializable(result)

18
tests/test_daily_forecast.py
View file

@ -3,9 +3,9 @@ from zoneinfo import ZoneInfo
from freezegun import freeze_time
from irm_kmi_api import ConditionEvol, ExtendedForecast
from tests.conftest import get_api_with_data, assert_all_serializable
from irm_kmi_api.const import ATTR_CONDITION_PARTLYCLOUDY
from irm_kmi_api.data import IrmKmiForecast
from tests.conftest import get_api_with_data
from tests.const import ATTR_CONDITION_PARTLYCLOUDY
@freeze_time(datetime.fromisoformat('2023-12-26T18:30:00+01:00'))
@ -19,11 +19,9 @@ async def test_daily_forecast() -> None:
assert len(result) == 8
assert result[0]['datetime'] == '2023-12-26'
assert not result[0]['is_daytime']
expected = ExtendedForecast(
expected = IrmKmiForecast(
datetime='2023-12-27',
condition=ATTR_CONDITION_PARTLYCLOUDY,
condition_2=None,
condition_evol=ConditionEvol.TWO_WAYS,
native_precipitation=0,
native_temperature=9,
native_templow=4,
@ -106,11 +104,3 @@ async def test_sunrise_sunset_be() -> None:
assert result[2]['sunrise'] == '2023-12-28T08:45:00+01:00'
assert result[2]['sunset'] == '2023-12-28T16:43:00+01:00'
def test_daily_serializable() -> None:
api = get_api_with_data("forecast.json")
tz = ZoneInfo("Europe/Brussels")
result = api.get_daily_forecast(tz, 'fr')
assert_all_serializable(result)

12
tests/test_hourly_forecast.py
View file

@ -3,9 +3,9 @@ from zoneinfo import ZoneInfo
from freezegun import freeze_time
from irm_kmi_api import Forecast
from tests.conftest import get_api_with_data, assert_all_serializable
from irm_kmi_api.const import ATTR_CONDITION_CLOUDY, ATTR_CONDITION_RAINY
from irm_kmi_api.data import Forecast
from tests.conftest import get_api_with_data
from tests.const import ATTR_CONDITION_CLOUDY, ATTR_CONDITION_RAINY
@freeze_time(datetime.fromisoformat('2023-12-26T18:30:00+01:00'))
@ -88,10 +88,4 @@ def test_hourly_forecast_midnight_bug() -> None:
assert result[24]['datetime'] == '2024-06-01T00:00:00+02:00'
def test_hourly_serializable() -> None:
api = get_api_with_data("forecast.json")
tz = ZoneInfo("Europe/Brussels")
result = api.get_hourly_forecast(tz)
assert_all_serializable(result)

63
tests/test_pollen.py
View file

@ -1,63 +1,37 @@
from unittest.mock import AsyncMock
from irm_kmi_api import PollenLevel, PollenName, PollenParser
from tests.conftest import get_api_with_data, load_fixture, is_serializable
from irm_kmi_api.pollen import PollenParser
from tests.conftest import get_api_with_data, load_fixture
def test_svg_pollen_parsing():
with open("tests/fixtures/pollen.svg", "r") as file:
svg_data = file.read()
data = PollenParser(svg_data).get_pollen_data()
assert data == {PollenName.BIRCH: PollenLevel.NONE,
PollenName.OAK: PollenLevel.NONE,
PollenName.HAZEL: PollenLevel.NONE,
PollenName.MUGWORT: PollenLevel.NONE,
PollenName.ALDER: PollenLevel.NONE,
PollenName.GRASSES: PollenLevel.PURPLE,
PollenName.ASH: PollenLevel.NONE}
assert data == {'birch': 'none', 'oak': 'none', 'hazel': 'none', 'mugwort': 'none', 'alder': 'none',
'grasses': 'purple', 'ash': 'none'}
def test_svg_two_pollen_parsing():
with open("tests/fixtures/new_two_pollens.svg", "r") as file:
svg_data = file.read()
data = PollenParser(svg_data).get_pollen_data()
assert data == {PollenName.BIRCH: PollenLevel.NONE,
PollenName.OAK: PollenLevel.NONE,
PollenName.HAZEL: PollenLevel.NONE,
PollenName.MUGWORT: PollenLevel.ACTIVE,
PollenName.ALDER: PollenLevel.NONE,
PollenName.GRASSES: PollenLevel.RED,
PollenName.ASH: PollenLevel.NONE}
assert data == {'birch': 'none', 'oak': 'none', 'hazel': 'none', 'mugwort': 'active', 'alder': 'none',
'grasses': 'red', 'ash': 'none'}
def test_svg_two_pollen_parsing_2025_update():
with open("tests/fixtures/pollens-2025.svg", "r") as file:
svg_data = file.read()
data = PollenParser(svg_data).get_pollen_data()
assert data == {PollenName.BIRCH: PollenLevel.NONE,
PollenName.OAK: PollenLevel.NONE,
PollenName.HAZEL: PollenLevel.ACTIVE,
PollenName.MUGWORT: PollenLevel.NONE,
PollenName.ALDER: PollenLevel.GREEN,
PollenName.GRASSES: PollenLevel.NONE,
PollenName.ASH: PollenLevel.NONE}
assert data == {'birch': 'none', 'oak': 'none', 'hazel': 'active', 'mugwort': 'none', 'alder': 'green',
'grasses': 'none', 'ash': 'none'}
def test_pollen_options():
assert set(PollenParser.get_option_values()) == {PollenLevel.GREEN,
PollenLevel.YELLOW,
PollenLevel.ORANGE,
PollenLevel.RED,
PollenLevel.PURPLE,
PollenLevel.ACTIVE,
PollenLevel.NONE}
assert set(PollenParser.get_option_values()) == {'green', 'yellow', 'orange', 'red', 'purple', 'active', 'none'}
def test_pollen_default_values():
assert PollenParser.get_default_data() == {PollenName.BIRCH: PollenLevel.NONE,
PollenName.OAK: PollenLevel.NONE,
PollenName.HAZEL: PollenLevel.NONE,
PollenName.MUGWORT: PollenLevel.NONE,
PollenName.ALDER: PollenLevel.NONE,
PollenName.GRASSES: PollenLevel.NONE,
PollenName.ASH: PollenLevel.NONE}
assert PollenParser.get_default_data() == {'birch': 'none', 'oak': 'none', 'hazel': 'none', 'mugwort': 'none',
'alder': 'none', 'grasses': 'none', 'ash': 'none'}
async def test_pollen_data_from_api() -> None:
@ -67,18 +41,7 @@ async def test_pollen_data_from_api() -> None:
api.get_svg = AsyncMock(return_value=load_fixture("pollen.svg"))
result = await api.get_pollen()
expected = {PollenName.MUGWORT: PollenLevel.NONE,
PollenName.BIRCH: PollenLevel.NONE,
PollenName.ALDER: PollenLevel.NONE,
PollenName.ASH: PollenLevel.NONE,
PollenName.OAK: PollenLevel.NONE,
PollenName.GRASSES: PollenLevel.PURPLE,
PollenName.HAZEL: PollenLevel.NONE}
expected = {'mugwort': 'none', 'birch': 'none', 'alder': 'none', 'ash': 'none', 'oak': 'none',
'grasses': 'purple', 'hazel': 'none'}
assert result == expected
def test_pollen_is_serializable():
with open("tests/fixtures/pollens-2025.svg", "r") as file:
svg_data = file.read()
data = PollenParser(svg_data).get_pollen_data()
assert is_serializable(data)

59
tests/test_radar_forecast.py
View file

@ -1,7 +1,7 @@
import pytest
from irm_kmi_api import RadarForecast
from tests.conftest import get_api_with_data, assert_all_serializable
from irm_kmi_api.data import IrmKmiRadarForecast
from tests.conftest import get_api_with_data
def test_radar_forecast() -> None:
@ -9,28 +9,28 @@ def test_radar_forecast() -> None:
result = api.get_radar_forecast()
expected = [
RadarForecast(datetime="2023-12-26T17:00:00+01:00", native_precipitation=0, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
RadarForecast(datetime="2023-12-26T17:10:00+01:00", native_precipitation=0, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
RadarForecast(datetime="2023-12-26T17:20:00+01:00", native_precipitation=0, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
RadarForecast(datetime="2023-12-26T17:30:00+01:00", native_precipitation=0, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
RadarForecast(datetime="2023-12-26T17:40:00+01:00", native_precipitation=0.1, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
RadarForecast(datetime="2023-12-26T17:50:00+01:00", native_precipitation=0.01, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
RadarForecast(datetime="2023-12-26T18:00:00+01:00", native_precipitation=0.12, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
RadarForecast(datetime="2023-12-26T18:10:00+01:00", native_precipitation=1.2, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
RadarForecast(datetime="2023-12-26T18:20:00+01:00", native_precipitation=2, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
RadarForecast(datetime="2023-12-26T18:30:00+01:00", native_precipitation=0, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
RadarForecast(datetime="2023-12-26T18:40:00+01:00", native_precipitation=0, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min')
IrmKmiRadarForecast(datetime="2023-12-26T17:00:00+01:00", native_precipitation=0, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
IrmKmiRadarForecast(datetime="2023-12-26T17:10:00+01:00", native_precipitation=0, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
IrmKmiRadarForecast(datetime="2023-12-26T17:20:00+01:00", native_precipitation=0, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
IrmKmiRadarForecast(datetime="2023-12-26T17:30:00+01:00", native_precipitation=0, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
IrmKmiRadarForecast(datetime="2023-12-26T17:40:00+01:00", native_precipitation=0.1, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
IrmKmiRadarForecast(datetime="2023-12-26T17:50:00+01:00", native_precipitation=0.01, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
IrmKmiRadarForecast(datetime="2023-12-26T18:00:00+01:00", native_precipitation=0.12, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
IrmKmiRadarForecast(datetime="2023-12-26T18:10:00+01:00", native_precipitation=1.2, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
IrmKmiRadarForecast(datetime="2023-12-26T18:20:00+01:00", native_precipitation=2, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
IrmKmiRadarForecast(datetime="2023-12-26T18:30:00+01:00", native_precipitation=0, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min'),
IrmKmiRadarForecast(datetime="2023-12-26T18:40:00+01:00", native_precipitation=0, might_rain=False,
rain_forecast_max=0, rain_forecast_min=0, unit='mm/10min')
]
assert expected == result
@ -40,7 +40,7 @@ def test_radar_forecast_rain_interval() -> None:
api = get_api_with_data('forecast_with_rain_on_radar.json')
result = api.get_radar_forecast()
_12 = RadarForecast(
_12 = IrmKmiRadarForecast(
datetime='2024-05-30T18:00:00+02:00',
native_precipitation=0.89,
might_rain=True,
@ -49,7 +49,7 @@ def test_radar_forecast_rain_interval() -> None:
unit='mm/10min'
)
_13 = RadarForecast(
_13 = IrmKmiRadarForecast(
datetime="2024-05-30T18:10:00+02:00",
native_precipitation=0.83,
might_rain=True,
@ -78,10 +78,3 @@ async def test_current_rainfall_unit(
for r in radar_forecast:
assert r.get('unit') == expected
def test_radar_serializable() -> None:
api = get_api_with_data("forecast.json")
result = api.get_radar_forecast()
assert_all_serializable(result)

95
tests/test_rain_graph.py
View file

@ -3,12 +3,13 @@ import datetime
import json
from datetime import datetime as dt
from datetime import timedelta
from unittest.mock import AsyncMock, MagicMock
from unittest.mock import MagicMock, AsyncMock
from zoneinfo import ZoneInfo
from irm_kmi_api import IrmKmiApiClientHa
from irm_kmi_api import AnimationFrameData, RadarAnimationData, RadarStyle
from irm_kmi_api import RainGraph
from irm_kmi_api.api import IrmKmiApiClientHa
from irm_kmi_api.const import OPTION_STYLE_SATELLITE
from irm_kmi_api.data import AnimationFrameData, RadarAnimationData
from irm_kmi_api.rain_graph import RainGraph
from tests.conftest import load_fixture
@ -44,12 +45,12 @@ async def test_svg_frame_setup():
rain_graph = RainGraph(
animation_data=data,
country='BE',
style=RadarStyle.OPTION_STYLE_STD,
style='STD',
)
await rain_graph._draw_svg_frame()
await rain_graph.draw_svg_frame()
svg_str = rain_graph._get_dwg().tostring()
svg_str = rain_graph.get_dwg().tostring()
with open("irm_kmi_api/resources/roboto_medium.ttf", "rb") as file:
font_b64 = base64.b64encode(file.read()).decode('utf-8')
@ -64,12 +65,12 @@ def test_svg_hint():
rain_graph = RainGraph(
animation_data=data,
country='BE',
style=RadarStyle.OPTION_STYLE_STD,
style='STD',
)
rain_graph._write_hint()
rain_graph.write_hint()
svg_str = rain_graph._get_dwg().tostring()
svg_str = rain_graph.get_dwg().tostring()
assert "Testing SVG camera" in svg_str
@ -80,12 +81,12 @@ def test_svg_time_bars():
tz = datetime.UTC,
animation_data=data,
country='BE',
style=RadarStyle.OPTION_STYLE_STD,
style='STD',
)
rain_graph._draw_hour_bars()
rain_graph.draw_hour_bars()
svg_str = rain_graph._get_dwg().tostring()
svg_str = rain_graph.get_dwg().tostring()
assert "19h" in svg_str
assert "20h" in svg_str
@ -99,12 +100,12 @@ def test_draw_chances_path():
rain_graph = RainGraph(
animation_data=data,
country='BE',
style=RadarStyle.OPTION_STYLE_STD,
style='STD',
)
rain_graph._draw_chances_path()
rain_graph.draw_chances_path()
svg_str = rain_graph._get_dwg().tostring()
svg_str = rain_graph.get_dwg().tostring()
assert 'fill="#63c8fa"' in svg_str
assert 'opacity="0.3"' in svg_str
@ -117,12 +118,12 @@ def test_draw_data_line():
rain_graph = RainGraph(
animation_data=data,
country='BE',
style=RadarStyle.OPTION_STYLE_STD,
style='STD',
)
rain_graph._draw_data_line()
rain_graph.draw_data_line()
svg_str = rain_graph._get_dwg().tostring()
svg_str = rain_graph.get_dwg().tostring()
assert 'fill="none"' in svg_str
assert 'stroke-width="2"' in svg_str
@ -135,15 +136,15 @@ async def test_insert_background():
rain_graph = RainGraph(
animation_data=data,
country='BE',
style=RadarStyle.OPTION_STYLE_STD,
style='STD',
)
await rain_graph._insert_background()
await rain_graph.insert_background()
with open("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()
svg_str = rain_graph.get_dwg().tostring()
assert png_b64 in svg_str
assert "<image " in svg_str
@ -158,12 +159,12 @@ def test_draw_current_frame_line_moving():
rain_graph = RainGraph(
animation_data=data,
country='BE',
style=RadarStyle.OPTION_STYLE_STD,
style='STD',
)
rain_graph._draw_current_fame_line()
rain_graph.draw_current_fame_line()
str_svg = rain_graph._get_dwg().tostring()
str_svg = rain_graph.get_dwg().tostring()
assert '<line' in str_svg
assert 'id="now"' in str_svg
@ -186,12 +187,12 @@ def test_draw_current_frame_line_index():
rain_graph = RainGraph(
animation_data=data,
country='BE',
style=RadarStyle.OPTION_STYLE_STD,
style='STD',
)
rain_graph._draw_current_fame_line(0)
rain_graph.draw_current_fame_line(0)
str_svg = rain_graph._get_dwg().tostring()
str_svg = rain_graph.get_dwg().tostring()
assert '<line' in str_svg
assert 'id="now"' in str_svg
@ -215,12 +216,12 @@ def test_draw_description_text():
tz=datetime.UTC,
animation_data=data,
country='BE',
style=RadarStyle.OPTION_STYLE_STD,
style='STD',
)
rain_graph._draw_description_text()
rain_graph.draw_description_text()
str_svg = rain_graph._get_dwg().tostring()
str_svg = rain_graph.get_dwg().tostring()
assert "18:30" in str_svg
assert "18:40" in str_svg
@ -242,12 +243,12 @@ def test_draw_cloud_layer():
rain_graph = RainGraph(
animation_data=data,
country='BE',
style=RadarStyle.OPTION_STYLE_STD,
style='STD',
)
rain_graph._insert_cloud_layer()
rain_graph.insert_cloud_layer()
str_svg = rain_graph._get_dwg().tostring()
str_svg = rain_graph.get_dwg().tostring()
with open("tests/fixtures/clouds_be.png", "rb") as file:
png_b64 = base64.b64encode(file.read()).decode('utf-8')
@ -262,12 +263,12 @@ async def test_draw_location_layer():
rain_graph = RainGraph(
animation_data=data,
country='BE',
style=RadarStyle.OPTION_STYLE_STD,
style='STD',
)
await rain_graph._draw_location()
await rain_graph.draw_location()
str_svg = rain_graph._get_dwg().tostring()
str_svg = rain_graph.get_dwg().tostring()
with open("tests/fixtures/loc_layer_be_n.png", "rb") as file:
png_b64 = base64.b64encode(file.read()).decode('utf-8')
@ -280,7 +281,7 @@ def test_get_animation_data():
tz = ZoneInfo('Europe/Brussels')
lang = 'en'
style = RadarStyle.OPTION_STYLE_SATELLITE
style = OPTION_STYLE_SATELLITE
dark_mode = False
api._api_data = json.loads(load_fixture("forecast.json"))
@ -305,13 +306,13 @@ async def test_download_single_cloud():
rain_graph = RainGraph(
animation_data=data,
country='BE',
style=RadarStyle.OPTION_STYLE_STD,
style='STD',
)
rain_graph._api_client = MagicMock()
rain_graph._api_client.get_image = AsyncMock()
await rain_graph._download_clouds(2)
await rain_graph.download_clouds(2)
rain_graph._api_client.get_image.assert_called_once_with('image-url-2')
@ -323,26 +324,26 @@ async def test_download_many_clouds():
rain_graph = RainGraph(
animation_data=data,
country='BE',
style=RadarStyle.OPTION_STYLE_STD,
style='STD',
)
rain_graph._api_client = MagicMock()
rain_graph._api_client.get_image = AsyncMock()
await rain_graph._download_clouds()
await rain_graph.download_clouds()
for i in range(10):
rain_graph._api_client.get_image.assert_any_call(f'image-url-{i}')
def test_can_build_rain_graph_with_empty_sequence():
RainGraph(
RadarAnimationData(sequence=None),
'BE', RadarStyle.OPTION_STYLE_STD
)
# RainGraph(
# RadarAnimationData(sequence=None),
# 'en', 'style'
# )
RainGraph(
RadarAnimationData(sequence=[]),
'BE', RadarStyle.OPTION_STYLE_STD
'en', 'style'
)

30
tests/test_warning.py
View file

@ -2,8 +2,7 @@ from datetime import datetime
from freezegun import freeze_time
from irm_kmi_api import WarningType
from tests.conftest import get_api_with_data, is_serializable
from tests.conftest import get_api_with_data
@freeze_time(datetime.fromisoformat('2024-01-12T07:10:00+00:00'))
@ -19,32 +18,7 @@ async def test_warning_data() -> None:
assert first.get('starts_at').replace(tzinfo=None) < datetime.now()
assert first.get('ends_at').replace(tzinfo=None) > datetime.now()
assert first.get('slug') == WarningType.FOG
assert first.get('slug') == 'fog'
assert first.get('friendly_name') == 'Fog'
assert first.get('id') == 7
assert first.get('level') == 1
async def test_warning_heat() -> None:
api = get_api_with_data("antwerp_with_heat_warning.json")
result = api.get_warnings(lang='en')
assert isinstance(result, list)
assert len(result) == 1
first = result[0]
assert first.get('slug') == WarningType.HEAT
assert first.get('friendly_name') == 'Heat'
assert first.get('id') == 10
assert first.get('level') == 1
async def test_warning_data_is_serializable() -> None:
api = get_api_with_data("be_forecast_warning.json")
result = api.get_warnings(lang='en')
for r in result:
del r["starts_at"]
del r["ends_at"]
assert is_serializable(r)