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

Use non-blocking calls in the main loop

Browse source
This commit is contained in:
Jules 2024-06-01 18:30:12 +02:00
parent 2e90931996
commit 59a3a3f07a
Signed by: jdejaegh
GPG key ID: 99D6D184CA66933A
12 changed files with 104 additions and 102 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
custom_components/irm_kmi/binary_sensor.py
View file

@ -1,8 +1,6 @@
"""Sensor to signal weather warning from the IRM KMI""" """Sensor to signal weather warning from the IRM KMI"""
import datetime
import logging import logging
import pytz
from homeassistant.components import binary_sensor from homeassistant.components import binary_sensor
from homeassistant.components.binary_sensor import (BinarySensorDeviceClass, from homeassistant.components.binary_sensor import (BinarySensorDeviceClass,
BinarySensorEntity) BinarySensorEntity)
@ -10,6 +8,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity 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 import DOMAIN, IrmKmiCoordinator
@ -44,7 +43,7 @@ class IrmKmiWarning(CoordinatorEntity, BinarySensorEntity):
if self.coordinator.data.get('warnings') is None: if self.coordinator.data.get('warnings') is None:
return False return False
now = datetime.datetime.now(tz=pytz.timezone(self.hass.config.time_zone)) now = dt.now()
for item in self.coordinator.data.get('warnings'): for item in self.coordinator.data.get('warnings'):
if item.get('starts_at') < now < item.get('ends_at'): if item.get('starts_at') < now < item.get('ends_at'):
return True return True
@ -56,7 +55,7 @@ class IrmKmiWarning(CoordinatorEntity, BinarySensorEntity):
"""Return the warning sensor attributes.""" """Return the warning sensor attributes."""
attrs = {"warnings": self.coordinator.data.get('warnings', [])} attrs = {"warnings": self.coordinator.data.get('warnings', [])}
now = datetime.datetime.now(tz=pytz.timezone(self.hass.config.time_zone)) now = dt.now()
for warning in attrs['warnings']: for warning in attrs['warnings']:
warning['is_active'] = warning.get('starts_at') < now < warning.get('ends_at') warning['is_active'] = warning.get('starts_at') < now < warning.get('ends_at')

54
custom_components/irm_kmi/coordinator.py
View file

@ -5,7 +5,6 @@ from datetime import datetime, timedelta
from typing import Any, List, Tuple from typing import Any, List, Tuple
import async_timeout import async_timeout
import pytz
from homeassistant.components.weather import Forecast from homeassistant.components.weather import Forecast
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_LATITUDE, ATTR_LONGITUDE, CONF_ZONE from homeassistant.const import ATTR_LATITUDE, ATTR_LONGITUDE, CONF_ZONE
@ -15,6 +14,7 @@ from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.update_coordinator import ( from homeassistant.helpers.update_coordinator import (
TimestampDataUpdateCoordinator, UpdateFailed) TimestampDataUpdateCoordinator, UpdateFailed)
from homeassistant.util import dt
from homeassistant.util.dt import utcnow from homeassistant.util.dt import utcnow
from .api import IrmKmiApiClient, IrmKmiApiError from .api import IrmKmiApiClient, IrmKmiApiError
@ -133,7 +133,8 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
unit=api_data.get('animation', {}).get('unit', {}).get(lang), unit=api_data.get('animation', {}).get('unit', {}).get(lang),
location=localisation location=localisation
) )
rain_graph = self.create_rain_graph(radar_animation, animation_data, country, images_from_api) rain_graph = await self.create_rain_graph(radar_animation, animation_data, country, images_from_api)
print(rain_graph)
radar_animation['svg_animated'] = rain_graph.get_svg_string() radar_animation['svg_animated'] = rain_graph.get_svg_string()
radar_animation['svg_still'] = rain_graph.get_svg_string(still_image=True) radar_animation['svg_still'] = rain_graph.get_svg_string(still_image=True)
return radar_animation return radar_animation
@ -164,9 +165,9 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
async def process_api_data(self, api_data: dict) -> ProcessedCoordinatorData: async def process_api_data(self, api_data: dict) -> ProcessedCoordinatorData:
"""From the API data, create the object that will be used in the entities""" """From the API data, create the object that will be used in the entities"""
return ProcessedCoordinatorData( return ProcessedCoordinatorData(
current_weather=IrmKmiCoordinator.current_weather_from_data(api_data), current_weather=await IrmKmiCoordinator.current_weather_from_data(api_data),
daily_forecast=self.daily_list_to_forecast(api_data.get('for', {}).get('daily')), daily_forecast=await self.daily_list_to_forecast(api_data.get('for', {}).get('daily')),
hourly_forecast=IrmKmiCoordinator.hourly_list_to_forecast(api_data.get('for', {}).get('hourly')), 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', {})), radar_forecast=IrmKmiCoordinator.radar_list_to_forecast(api_data.get('animation', {})),
animation=await self._async_animation_data(api_data=api_data), animation=await self._async_animation_data(api_data=api_data),
warnings=self.warnings_from_data(api_data.get('for', {}).get('warning')), warnings=self.warnings_from_data(api_data.get('for', {}).get('warning')),
@ -194,17 +195,19 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
return images_from_api return images_from_api
@staticmethod @staticmethod
def current_weather_from_data(api_data: dict) -> CurrentWeatherData: async def current_weather_from_data(api_data: dict) -> CurrentWeatherData:
"""Parse the API data to build a CurrentWeatherData.""" """Parse the API data to build a CurrentWeatherData."""
# Process data to get current hour forecast # Process data to get current hour forecast
now_hourly = None now_hourly = None
hourly_forecast_data = api_data.get('for', {}).get('hourly') 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 if not (hourly_forecast_data is None
or not isinstance(hourly_forecast_data, list) or not isinstance(hourly_forecast_data, list)
or len(hourly_forecast_data) == 0): or len(hourly_forecast_data) == 0):
for current in hourly_forecast_data[:2]: for current in hourly_forecast_data[:2]:
if datetime.now().strftime('%H') == current['hour']: if now.strftime('%H') == current['hour']:
now_hourly = current now_hourly = current
break break
# Get UV index # Get UV index
@ -267,13 +270,14 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
return current_weather return current_weather
@staticmethod @staticmethod
def hourly_list_to_forecast(data: List[dict] | None) -> List[Forecast] | None: 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""" """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: if data is None or not isinstance(data, list) or len(data) == 0:
return None return None
forecasts = list() forecasts = list()
day = datetime.now(tz=pytz.timezone('Europe/Brussels')).replace(hour=0, minute=0, second=0, microsecond=0) 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 f in data: for f in data:
if 'dateShow' in f: if 'dateShow' in f:
@ -332,7 +336,7 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
) )
return forecast return forecast
def daily_list_to_forecast(self, data: List[dict] | None) -> List[Forecast] | None: async def daily_list_to_forecast(self, data: List[dict] | None) -> List[Forecast] | None:
"""Parse data from the API to create a list of daily forecasts""" """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: if data is None or not isinstance(data, list) or len(data) == 0:
return None return None
@ -340,6 +344,8 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
forecasts = list() forecasts = list()
n_days = 0 n_days = 0
lang = preferred_language(self.hass, self._config_entry) lang = preferred_language(self.hass, self._config_entry)
tz = await dt.async_get_time_zone('Europe/Brussels')
now = dt.now(tz)
for (idx, f) in enumerate(data): for (idx, f) in enumerate(data):
precipitation = None precipitation = None
@ -364,10 +370,9 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
pass pass
is_daytime = f.get('dayNight', None) == 'd' is_daytime = f.get('dayNight', None) == 'd'
now = datetime.now(pytz.timezone('Europe/Brussels'))
forecast = IrmKmiForecast( forecast = IrmKmiForecast(
datetime=(now + timedelta(days=n_days)).strftime('%Y-%m-%d') datetime=(now + timedelta(days=n_days)).strftime('%Y-%m-%d') if is_daytime else now.strftime(
if is_daytime else now.strftime('%Y-%m-%d'), '%Y-%m-%d'),
condition=CDT_MAP.get((f.get('ww1', None), f.get('dayNight', None)), None), condition=CDT_MAP.get((f.get('ww1', None), f.get('dayNight', None)), None),
native_precipitation=precipitation, native_precipitation=precipitation,
native_temperature=f.get('tempMax', None), native_temperature=f.get('tempMax', None),
@ -392,16 +397,17 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
return forecasts return forecasts
def create_rain_graph(self, async def create_rain_graph(self,
radar_animation: RadarAnimationData, radar_animation: RadarAnimationData,
api_animation_data: List[dict], api_animation_data: List[dict],
country: str, country: str,
images_from_api: Tuple[bytes], images_from_api: Tuple[bytes],
) -> RainGraph: ) -> RainGraph:
"""Create a RainGraph object that is ready to output animated and still SVG images""" """Create a RainGraph object that is ready to output animated and still SVG images"""
sequence: List[AnimationFrameData] = list() sequence: List[AnimationFrameData] = list()
tz = pytz.timezone(self.hass.config.time_zone)
current_time = datetime.now(tz=tz) tz = await dt.async_get_time_zone(self.hass.config.time_zone)
current_time = dt.now(time_zone=tz)
most_recent_frame = None most_recent_frame = None
for idx, item in enumerate(api_animation_data): for idx, item in enumerate(api_animation_data):
@ -431,10 +437,8 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
f"{'satellite' if satellite_mode else 'black' if self._dark_mode else 'white'}.png") f"{'satellite' if satellite_mode else 'black' if self._dark_mode else 'white'}.png")
bg_size = (640, 490) bg_size = (640, 490)
return RainGraph(radar_animation, image_path, bg_size, return await RainGraph(radar_animation, image_path, bg_size, tz=tz, config_dir=self.hass.config.config_dir,
config_dir=self.hass.config.config_dir, dark_mode=self._dark_mode).build()
dark_mode=self._dark_mode,
tz=self.hass.config.time_zone)
def warnings_from_data(self, warning_data: list | None) -> List[WarningData]: def warnings_from_data(self, warning_data: list | None) -> List[WarningData]:
"""Create a list of warning data instances based on the api data""" """Create a list of warning data instances based on the api data"""

4
custom_components/irm_kmi/manifest.json
View file

@ -9,8 +9,8 @@
"iot_class": "cloud_polling", "iot_class": "cloud_polling",
"issue_tracker": "https://github.com/jdejaegh/irm-kmi-ha/issues", "issue_tracker": "https://github.com/jdejaegh/irm-kmi-ha/issues",
"requirements": [ "requirements": [
"pytz==2024.1", "svgwrite==1.4.3",
"svgwrite==1.4.3" "aiofile==3.8.8"
], ],
"version": "0.2.14" "version": "0.2.14"
} }

75
custom_components/irm_kmi/rain_graph.py
View file

@ -2,13 +2,16 @@
import base64 import base64
import copy import copy
import datetime
import logging import logging
import os import os
from typing import List from typing import List, Self
import pytz from aiofile import async_open
from homeassistant.util import dt
from svgwrite import Drawing from svgwrite import Drawing
from svgwrite.animate import Animate from svgwrite.animate import Animate
from svgwrite.utils import font_mimetype
from custom_components.irm_kmi.data import (AnimationFrameData, from custom_components.irm_kmi.data import (AnimationFrameData,
RadarAnimationData) RadarAnimationData)
@ -23,7 +26,7 @@ class RainGraph:
background_size: (int, int), background_size: (int, int),
config_dir: str = '.', config_dir: str = '.',
dark_mode: bool = False, dark_mode: bool = False,
tz: str = 'UTC', tz: datetime.tzinfo = dt.get_default_time_zone(),
svg_width: float = 640, svg_width: float = 640,
inset: float = 20, inset: float = 20,
graph_height: float = 150, graph_height: float = 150,
@ -31,7 +34,6 @@ class RainGraph:
top_text_y_pos: float = 20, top_text_y_pos: float = 20,
bottom_text_space: float = 50, bottom_text_space: float = 50,
bottom_text_y_pos: float = 218, bottom_text_y_pos: float = 218,
auto=True
): ):
self._animation_data: RadarAnimationData = animation_data self._animation_data: RadarAnimationData = animation_data
@ -39,7 +41,7 @@ class RainGraph:
self._background_size: (int, int) = background_size self._background_size: (int, int) = background_size
self._config_dir: str = config_dir self._config_dir: str = config_dir
self._dark_mode: bool = dark_mode self._dark_mode: bool = dark_mode
self._tz = pytz.timezone(tz) self._tz = tz
self._svg_width: float = svg_width self._svg_width: float = svg_width
self._inset: float = inset self._inset: float = inset
self._graph_height: float = graph_height self._graph_height: float = graph_height
@ -62,38 +64,45 @@ class RainGraph:
raise ValueError("bottom_text_y_pos must be below the graph") raise ValueError("bottom_text_y_pos must be below the graph")
self._dwg: Drawing = Drawing(size=(self._svg_width, self._svg_height), profile='full') self._dwg: Drawing = Drawing(size=(self._svg_width, self._svg_height), profile='full')
self._dwg_save: Drawing self._dwg_save: Drawing = Drawing()
self._dwg_animated: Drawing self._dwg_animated: Drawing = Drawing()
self._dwg_still: Drawing self._dwg_still: Drawing = Drawing()
if auto: async def build(self) -> Self:
self.draw_svg_frame() await self.draw_svg_frame()
self.draw_hour_bars() self.draw_hour_bars()
self.draw_chances_path() self.draw_chances_path()
self.draw_data_line() self.draw_data_line()
self.write_hint() self.write_hint()
self.insert_background() await self.insert_background()
self._dwg_save = copy.deepcopy(self._dwg) self._dwg_save = copy.deepcopy(self._dwg)
self.draw_current_fame_line() self.draw_current_fame_line()
self.draw_description_text() self.draw_description_text()
self.insert_cloud_layer() self.insert_cloud_layer()
self.draw_location() self.draw_location()
self._dwg_animated = self._dwg self._dwg_animated = self._dwg
self._dwg = self._dwg_save self._dwg = self._dwg_save
idx = self._animation_data['most_recent_image_idx'] idx = self._animation_data['most_recent_image_idx']
self.draw_current_fame_line(idx) self.draw_current_fame_line(idx)
self.draw_description_text(idx) self.draw_description_text(idx)
self.insert_cloud_layer(idx) self.insert_cloud_layer(idx)
self.draw_location() self.draw_location()
self._dwg_still = self._dwg self._dwg_still = self._dwg
return self
def draw_svg_frame(self): async def draw_svg_frame(self):
"""Create the global area to draw the other items""" """Create the global area to draw the other items"""
font_file = os.path.join(self._config_dir, 'custom_components/irm_kmi/resources/roboto_medium.ttf') font_file = os.path.join(self._config_dir, 'custom_components/irm_kmi/resources/roboto_medium.ttf')
_LOGGER.debug(f"Opening font file at {font_file}") _LOGGER.debug(f"Opening font file at {font_file}")
self._dwg.embed_font(name="Roboto Medium", filename=font_file)
async with async_open(font_file, 'rb') as font:
data = await font.read()
# Need to use the private class method as the public one does not offer an async call
# As this is run in the main loop, we cannot afford a blocking open() call
self._dwg._embed_font_data("Roboto Medium", data, font_mimetype(font_file))
self._dwg.embed_stylesheet(""" self._dwg.embed_stylesheet("""
.roboto { .roboto {
font-family: "Roboto Medium"; font-family: "Roboto Medium";
@ -299,10 +308,10 @@ class RainGraph:
def get_svg_string(self, still_image: bool = False) -> bytes: 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() return self._dwg_still.tostring().encode() if still_image else self._dwg_animated.tostring().encode()
def insert_background(self): async def insert_background(self):
bg_image_path = os.path.join(self._config_dir, self._background_image_path) bg_image_path = os.path.join(self._config_dir, self._background_image_path)
with open(bg_image_path, 'rb') as f: async with async_open(bg_image_path, 'rb') as f:
png_data = base64.b64encode(f.read()).decode('utf-8') png_data = base64.b64encode(await f.read()).decode('utf-8')
image = self._dwg.image("data:image/png;base64," + png_data, insert=(0, 0), size=self._background_size) image = self._dwg.image("data:image/png;base64," + png_data, insert=(0, 0), size=self._background_size)
self._dwg.add(image) self._dwg.add(image)

6
custom_components/irm_kmi/sensor.py
View file

@ -2,13 +2,13 @@
import datetime import datetime
import logging import logging
import pytz
from homeassistant.components import sensor from homeassistant.components import sensor
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity 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 import DOMAIN, IrmKmiCoordinator
from custom_components.irm_kmi.const import POLLEN_NAMES, POLLEN_TO_ICON_MAP from custom_components.irm_kmi.const import POLLEN_NAMES, POLLEN_TO_ICON_MAP
@ -75,7 +75,7 @@ class IrmKmiNextWarning(CoordinatorEntity, SensorEntity):
if self.coordinator.data.get('warnings') is None: if self.coordinator.data.get('warnings') is None:
return None return None
now = datetime.datetime.now(tz=pytz.timezone(self.hass.config.time_zone)) now = dt.now()
earliest_next = None earliest_next = None
for item in self.coordinator.data.get('warnings'): for item in self.coordinator.data.get('warnings'):
if now < item.get('starts_at'): if now < item.get('starts_at'):
@ -89,7 +89,7 @@ class IrmKmiNextWarning(CoordinatorEntity, SensorEntity):
@property @property
def extra_state_attributes(self) -> dict: def extra_state_attributes(self) -> dict:
"""Return the attributes related to all the future warnings.""" """Return the attributes related to all the future warnings."""
now = datetime.datetime.now(tz=pytz.timezone(self.hass.config.time_zone)) now = dt.now()
attrs = {"next_warnings": [w for w in self.coordinator.data.get('warnings', []) if now < w.get('starts_at')]} attrs = {"next_warnings": [w for w in self.coordinator.data.get('warnings', []) if now < w.get('starts_at')]}
attrs["next_warnings_friendly_names"] = ", ".join( attrs["next_warnings_friendly_names"] = ", ".join(

1
custom_components/irm_kmi/weather.py
View file

@ -159,7 +159,6 @@ class IrmKmiWeather(CoordinatorEntity, WeatherEntity):
:param include_past_forecasts: whether to include data points that are in the past :param include_past_forecasts: whether to include data points that are in the past
:return: ordered list of forecasts :return: ordered list of forecasts
""" """
# now = datetime.now(tz=pytz.timezone(self.hass.config.time_zone))
now = dt.now() now = dt.now()
now = now.replace(minute=(now.minute // 10) * 10, second=0, microsecond=0) now = now.replace(minute=(now.minute // 10) * 10, second=0, microsecond=0)

6
requirements.txt
View file

@ -1,6 +1,6 @@
aiohttp==3.9.5 aiohttp==3.9.5
async-timeout==4.0.3 async-timeout==4.0.3
homeassistant==2024.5.5 homeassistant==2024.6.0b4
voluptuous==0.13.1 voluptuous==0.13.1
pytz==2024.1 svgwrite==1.4.3
svgwrite==1.4.3 aiofile==3.8.8

4
requirements_tests.txt
View file

@ -1,5 +1,5 @@
homeassistant==2024.5.5 homeassistant==2024.6.0b4
pytest_homeassistant_custom_component==0.13.128
pytest pytest
pytest_homeassistant_custom_component==0.13.125
freezegun freezegun
isort isort

6
tests/conftest.py
View file

@ -13,9 +13,9 @@ from pytest_homeassistant_custom_component.common import (MockConfigEntry,
from custom_components.irm_kmi.api import (IrmKmiApiError, from custom_components.irm_kmi.api import (IrmKmiApiError,
IrmKmiApiParametersError) IrmKmiApiParametersError)
from custom_components.irm_kmi.const import ( from custom_components.irm_kmi.const import (
CONF_DARK_MODE, CONF_STYLE, CONF_USE_DEPRECATED_FORECAST, DOMAIN, CONF_DARK_MODE, CONF_LANGUAGE_OVERRIDE, CONF_STYLE,
OPTION_DEPRECATED_FORECAST_NOT_USED, CONF_USE_DEPRECATED_FORECAST, DOMAIN, OPTION_DEPRECATED_FORECAST_NOT_USED,
OPTION_DEPRECATED_FORECAST_TWICE_DAILY, OPTION_STYLE_STD, CONF_LANGUAGE_OVERRIDE) OPTION_DEPRECATED_FORECAST_TWICE_DAILY, OPTION_STYLE_STD)
def get_api_data(fixture: str) -> dict: def get_api_data(fixture: str) -> dict:

18
tests/test_coordinator.py
View file

@ -25,7 +25,7 @@ async def test_jules_forgot_to_revert_update_interval_before_pushing(
assert timedelta(minutes=5) <= coordinator.update_interval assert timedelta(minutes=5) <= coordinator.update_interval
@freeze_time(datetime.fromisoformat('2024-01-12T07:10:00')) @freeze_time(datetime.fromisoformat('2024-01-12T07:10:00+00:00'))
async def test_warning_data( async def test_warning_data(
hass: HomeAssistant, hass: HomeAssistant,
mock_config_entry: MockConfigEntry mock_config_entry: MockConfigEntry
@ -49,10 +49,10 @@ async def test_warning_data(
assert first.get('level') == 1 assert first.get('level') == 1
@freeze_time(datetime.fromisoformat('2023-12-26T18:30:00')) @freeze_time(datetime.fromisoformat('2023-12-26T17:30:00+00:00'))
def test_current_weather_be() -> None: async def test_current_weather_be() -> None:
api_data = get_api_data("forecast.json") api_data = get_api_data("forecast.json")
result = IrmKmiCoordinator.current_weather_from_data(api_data) result = await IrmKmiCoordinator.current_weather_from_data(api_data)
expected = CurrentWeatherData( expected = CurrentWeatherData(
condition=ATTR_CONDITION_CLOUDY, condition=ATTR_CONDITION_CLOUDY,
@ -68,9 +68,9 @@ def test_current_weather_be() -> None:
@freeze_time(datetime.fromisoformat("2023-12-28T15:30:00")) @freeze_time(datetime.fromisoformat("2023-12-28T15:30:00"))
def test_current_weather_nl() -> None: async def test_current_weather_nl() -> None:
api_data = get_api_data("forecast_nl.json") api_data = get_api_data("forecast_nl.json")
result = IrmKmiCoordinator.current_weather_from_data(api_data) result = await IrmKmiCoordinator.current_weather_from_data(api_data)
expected = CurrentWeatherData( expected = CurrentWeatherData(
condition=ATTR_CONDITION_CLOUDY, condition=ATTR_CONDITION_CLOUDY,
@ -94,7 +94,7 @@ async def test_daily_forecast(
mock_config_entry.data = mock_config_entry.data | {CONF_LANGUAGE_OVERRIDE: 'fr'} mock_config_entry.data = mock_config_entry.data | {CONF_LANGUAGE_OVERRIDE: 'fr'}
coordinator = IrmKmiCoordinator(hass, mock_config_entry) coordinator = IrmKmiCoordinator(hass, mock_config_entry)
result = coordinator.daily_list_to_forecast(api_data) result = await coordinator.daily_list_to_forecast(api_data)
assert isinstance(result, list) assert isinstance(result, list)
assert len(result) == 8 assert len(result) == 8
@ -117,9 +117,9 @@ async def test_daily_forecast(
@freeze_time(datetime.fromisoformat('2023-12-26T18:30:00+01:00')) @freeze_time(datetime.fromisoformat('2023-12-26T18:30:00+01:00'))
def test_hourly_forecast() -> None: async def test_hourly_forecast() -> None:
api_data = get_api_data("forecast.json").get('for', {}).get('hourly') api_data = get_api_data("forecast.json").get('for', {}).get('hourly')
result = IrmKmiCoordinator.hourly_list_to_forecast(api_data) result = await IrmKmiCoordinator.hourly_list_to_forecast(api_data)
assert isinstance(result, list) assert isinstance(result, list)
assert len(result) == 49 assert len(result) == 49

6
tests/test_init.py
View file

@ -9,8 +9,10 @@ from homeassistant.core import HomeAssistant
from pytest_homeassistant_custom_component.common import MockConfigEntry 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
from custom_components.irm_kmi.const import DOMAIN, CONFIG_FLOW_VERSION, CONF_LANGUAGE_OVERRIDE, \ from custom_components.irm_kmi.const import (
CONF_USE_DEPRECATED_FORECAST, OPTION_DEPRECATED_FORECAST_NOT_USED, CONF_DARK_MODE, CONF_STYLE, OPTION_STYLE_STD CONF_DARK_MODE, CONF_LANGUAGE_OVERRIDE, CONF_STYLE,
CONF_USE_DEPRECATED_FORECAST, CONFIG_FLOW_VERSION, DOMAIN,
OPTION_DEPRECATED_FORECAST_NOT_USED, OPTION_STYLE_STD)
async def test_load_unload_config_entry( async def test_load_unload_config_entry(

19
tests/test_rain_graph.py
View file

@ -33,16 +33,15 @@ def get_radar_animation_data() -> RadarAnimationData:
) )
def test_svg_frame_setup(): async def test_svg_frame_setup():
data = get_radar_animation_data() data = get_radar_animation_data()
rain_graph = RainGraph( rain_graph = RainGraph(
animation_data=data, animation_data=data,
background_image_path="custom_components/irm_kmi/resources/be_white.png", background_image_path="custom_components/irm_kmi/resources/be_white.png",
background_size=(640, 490), background_size=(640, 490),
auto=False
) )
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()
@ -60,7 +59,6 @@ def test_svg_hint():
animation_data=data, animation_data=data,
background_image_path="custom_components/irm_kmi/resources/be_white.png", background_image_path="custom_components/irm_kmi/resources/be_white.png",
background_size=(640, 490), background_size=(640, 490),
auto=False
) )
rain_graph.write_hint() rain_graph.write_hint()
@ -76,7 +74,6 @@ def test_svg_time_bars():
animation_data=data, animation_data=data,
background_image_path="custom_components/irm_kmi/resources/be_white.png", background_image_path="custom_components/irm_kmi/resources/be_white.png",
background_size=(640, 490), background_size=(640, 490),
auto=False
) )
rain_graph.draw_hour_bars() rain_graph.draw_hour_bars()
@ -96,7 +93,6 @@ def test_draw_chances_path():
animation_data=data, animation_data=data,
background_image_path="custom_components/irm_kmi/resources/be_white.png", background_image_path="custom_components/irm_kmi/resources/be_white.png",
background_size=(640, 490), background_size=(640, 490),
auto=False
) )
rain_graph.draw_chances_path() rain_graph.draw_chances_path()
@ -115,7 +111,6 @@ def test_draw_data_line():
animation_data=data, animation_data=data,
background_image_path="custom_components/irm_kmi/resources/be_white.png", background_image_path="custom_components/irm_kmi/resources/be_white.png",
background_size=(640, 490), background_size=(640, 490),
auto=False
) )
rain_graph.draw_data_line() rain_graph.draw_data_line()
@ -128,16 +123,15 @@ def test_draw_data_line():
assert '<path ' in svg_str assert '<path ' in svg_str
def test_insert_background(): async def test_insert_background():
data = get_radar_animation_data() data = get_radar_animation_data()
rain_graph = RainGraph( rain_graph = RainGraph(
animation_data=data, animation_data=data,
background_image_path="custom_components/irm_kmi/resources/be_white.png", background_image_path="custom_components/irm_kmi/resources/be_white.png",
background_size=(640, 490), background_size=(640, 490),
auto=False
) )
rain_graph.insert_background() await rain_graph.insert_background()
with open("custom_components/irm_kmi/resources/be_white.png", "rb") as file: with open("custom_components/irm_kmi/resources/be_white.png", "rb") as file:
png_b64 = base64.b64encode(file.read()).decode('utf-8') png_b64 = base64.b64encode(file.read()).decode('utf-8')
@ -158,7 +152,6 @@ def test_draw_current_frame_line_moving():
animation_data=data, animation_data=data,
background_image_path="custom_components/irm_kmi/resources/be_white.png", background_image_path="custom_components/irm_kmi/resources/be_white.png",
background_size=(640, 490), background_size=(640, 490),
auto=False
) )
rain_graph.draw_current_fame_line() rain_graph.draw_current_fame_line()
@ -187,7 +180,6 @@ def test_draw_current_frame_line_index():
animation_data=data, animation_data=data,
background_image_path="custom_components/irm_kmi/resources/be_white.png", background_image_path="custom_components/irm_kmi/resources/be_white.png",
background_size=(640, 490), background_size=(640, 490),
auto=False
) )
rain_graph.draw_current_fame_line(0) rain_graph.draw_current_fame_line(0)
@ -216,7 +208,6 @@ def test_draw_description_text():
animation_data=data, animation_data=data,
background_image_path="custom_components/irm_kmi/resources/be_white.png", background_image_path="custom_components/irm_kmi/resources/be_white.png",
background_size=(640, 490), background_size=(640, 490),
auto=False
) )
rain_graph.draw_description_text() rain_graph.draw_description_text()
@ -244,7 +235,6 @@ def test_draw_cloud_layer():
animation_data=data, animation_data=data,
background_image_path="custom_components/irm_kmi/resources/be_white.png", background_image_path="custom_components/irm_kmi/resources/be_white.png",
background_size=(640, 490), background_size=(640, 490),
auto=False
) )
rain_graph.insert_cloud_layer() rain_graph.insert_cloud_layer()
@ -265,7 +255,6 @@ def test_draw_location_layer():
animation_data=data, animation_data=data,
background_image_path="custom_components/irm_kmi/resources/be_white.png", background_image_path="custom_components/irm_kmi/resources/be_white.png",
background_size=(640, 490), background_size=(640, 490),
auto=False
) )
rain_graph.draw_location() rain_graph.draw_location()