mirror of
https://github.com/jdejaegh/irm-kmi-ha.git
synced 2025-06-27 03:35:56 +02:00
Merge pull request #41 from jdejaegh/blocking_calls
Use non-blocking calls in the main loop
This commit is contained in:
GitHub
commit
0812ef5eef
13 changed files with 107 additions and 105 deletions
|
|
@ -1,8 +1,6 @@
|
|||
"""Sensor to signal weather warning from the IRM KMI"""
|
||||
import datetime
|
||||
import logging
|
||||
|
||||
import pytz
|
||||
from homeassistant.components import binary_sensor
|
||||
from homeassistant.components.binary_sensor import (BinarySensorDeviceClass,
|
||||
BinarySensorEntity)
|
||||
|
|
@ -10,6 +8,7 @@ from homeassistant.config_entries import ConfigEntry
|
|||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
from homeassistant.util import dt
|
||||
|
||||
from custom_components.irm_kmi import DOMAIN, IrmKmiCoordinator
|
||||
|
||||
|
|
@ -44,7 +43,7 @@ class IrmKmiWarning(CoordinatorEntity, BinarySensorEntity):
|
|||
if self.coordinator.data.get('warnings') is None:
|
||||
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'):
|
||||
if item.get('starts_at') < now < item.get('ends_at'):
|
||||
return True
|
||||
|
|
@ -56,7 +55,7 @@ class IrmKmiWarning(CoordinatorEntity, BinarySensorEntity):
|
|||
"""Return the warning sensor attributes."""
|
||||
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']:
|
||||
warning['is_active'] = warning.get('starts_at') < now < warning.get('ends_at')
|
||||
|
||||
|
|
|
|||
|
|
@ -5,7 +5,6 @@ from datetime import datetime, timedelta
|
|||
from typing import Any, List, Tuple
|
||||
|
||||
import async_timeout
|
||||
import pytz
|
||||
from homeassistant.components.weather import Forecast
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
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.update_coordinator import (
|
||||
TimestampDataUpdateCoordinator, UpdateFailed)
|
||||
from homeassistant.util import dt
|
||||
from homeassistant.util.dt import utcnow
|
||||
|
||||
from .api import IrmKmiApiClient, IrmKmiApiError
|
||||
|
|
@ -133,7 +133,8 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
|
|||
unit=api_data.get('animation', {}).get('unit', {}).get(lang),
|
||||
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_still'] = rain_graph.get_svg_string(still_image=True)
|
||||
return radar_animation
|
||||
|
|
@ -164,9 +165,9 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
|
|||
async def process_api_data(self, api_data: dict) -> ProcessedCoordinatorData:
|
||||
"""From the API data, create the object that will be used in the entities"""
|
||||
return ProcessedCoordinatorData(
|
||||
current_weather=IrmKmiCoordinator.current_weather_from_data(api_data),
|
||||
daily_forecast=self.daily_list_to_forecast(api_data.get('for', {}).get('daily')),
|
||||
hourly_forecast=IrmKmiCoordinator.hourly_list_to_forecast(api_data.get('for', {}).get('hourly')),
|
||||
current_weather=await IrmKmiCoordinator.current_weather_from_data(api_data),
|
||||
daily_forecast=await self.daily_list_to_forecast(api_data.get('for', {}).get('daily')),
|
||||
hourly_forecast=await IrmKmiCoordinator.hourly_list_to_forecast(api_data.get('for', {}).get('hourly')),
|
||||
radar_forecast=IrmKmiCoordinator.radar_list_to_forecast(api_data.get('animation', {})),
|
||||
animation=await self._async_animation_data(api_data=api_data),
|
||||
warnings=self.warnings_from_data(api_data.get('for', {}).get('warning')),
|
||||
|
|
@ -194,17 +195,19 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
|
|||
return images_from_api
|
||||
|
||||
@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."""
|
||||
# Process data to get current hour forecast
|
||||
now_hourly = None
|
||||
hourly_forecast_data = api_data.get('for', {}).get('hourly')
|
||||
tz = await dt.async_get_time_zone('Europe/Brussels')
|
||||
now = dt.now(time_zone=tz)
|
||||
if not (hourly_forecast_data is None
|
||||
or not isinstance(hourly_forecast_data, list)
|
||||
or len(hourly_forecast_data) == 0):
|
||||
|
||||
for current in hourly_forecast_data[:2]:
|
||||
if datetime.now().strftime('%H') == current['hour']:
|
||||
if now.strftime('%H') == current['hour']:
|
||||
now_hourly = current
|
||||
break
|
||||
# Get UV index
|
||||
|
|
@ -267,13 +270,14 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
|
|||
return current_weather
|
||||
|
||||
@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"""
|
||||
if data is None or not isinstance(data, list) or len(data) == 0:
|
||||
return None
|
||||
|
||||
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 idx, f in enumerate(data):
|
||||
if 'dateShow' in f and idx > 0:
|
||||
|
|
@ -332,7 +336,7 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
|
|||
)
|
||||
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"""
|
||||
if data is None or not isinstance(data, list) or len(data) == 0:
|
||||
return None
|
||||
|
|
@ -340,6 +344,8 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
|
|||
forecasts = list()
|
||||
n_days = 0
|
||||
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):
|
||||
precipitation = None
|
||||
|
|
@ -364,10 +370,9 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
|
|||
pass
|
||||
|
||||
is_daytime = f.get('dayNight', None) == 'd'
|
||||
now = datetime.now(pytz.timezone('Europe/Brussels'))
|
||||
forecast = IrmKmiForecast(
|
||||
datetime=(now + timedelta(days=n_days)).strftime('%Y-%m-%d')
|
||||
if is_daytime else now.strftime('%Y-%m-%d'),
|
||||
datetime=(now + timedelta(days=n_days)).strftime('%Y-%m-%d') if is_daytime else now.strftime(
|
||||
'%Y-%m-%d'),
|
||||
condition=CDT_MAP.get((f.get('ww1', None), f.get('dayNight', None)), None),
|
||||
native_precipitation=precipitation,
|
||||
native_temperature=f.get('tempMax', None),
|
||||
|
|
@ -392,16 +397,17 @@ class IrmKmiCoordinator(TimestampDataUpdateCoordinator):
|
|||
|
||||
return forecasts
|
||||
|
||||
def create_rain_graph(self,
|
||||
radar_animation: RadarAnimationData,
|
||||
api_animation_data: List[dict],
|
||||
country: str,
|
||||
images_from_api: Tuple[bytes],
|
||||
) -> RainGraph:
|
||||
async def create_rain_graph(self,
|
||||
radar_animation: RadarAnimationData,
|
||||
api_animation_data: List[dict],
|
||||
country: str,
|
||||
images_from_api: Tuple[bytes],
|
||||
) -> RainGraph:
|
||||
"""Create a RainGraph object that is ready to output animated and still SVG images"""
|
||||
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
|
||||
|
||||
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")
|
||||
bg_size = (640, 490)
|
||||
|
||||
return RainGraph(radar_animation, image_path, bg_size,
|
||||
config_dir=self.hass.config.config_dir,
|
||||
dark_mode=self._dark_mode,
|
||||
tz=self.hass.config.time_zone)
|
||||
return await RainGraph(radar_animation, image_path, bg_size, tz=tz, config_dir=self.hass.config.config_dir,
|
||||
dark_mode=self._dark_mode).build()
|
||||
|
||||
def warnings_from_data(self, warning_data: list | None) -> List[WarningData]:
|
||||
"""Create a list of warning data instances based on the api data"""
|
||||
|
|
|
|||
|
|
@ -9,8 +9,8 @@
|
|||
"iot_class": "cloud_polling",
|
||||
"issue_tracker": "https://github.com/jdejaegh/irm-kmi-ha/issues",
|
||||
"requirements": [
|
||||
"pytz==2024.1",
|
||||
"svgwrite==1.4.3"
|
||||
"svgwrite==1.4.3",
|
||||
"aiofile==3.8.8"
|
||||
],
|
||||
"version": "0.2.14"
|
||||
}
|
||||
|
|
@ -2,13 +2,16 @@
|
|||
|
||||
import base64
|
||||
import copy
|
||||
import datetime
|
||||
import logging
|
||||
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.animate import Animate
|
||||
from svgwrite.utils import font_mimetype
|
||||
|
||||
from custom_components.irm_kmi.data import (AnimationFrameData,
|
||||
RadarAnimationData)
|
||||
|
|
@ -23,7 +26,7 @@ class RainGraph:
|
|||
background_size: (int, int),
|
||||
config_dir: str = '.',
|
||||
dark_mode: bool = False,
|
||||
tz: str = 'UTC',
|
||||
tz: datetime.tzinfo = dt.get_default_time_zone(),
|
||||
svg_width: float = 640,
|
||||
inset: float = 20,
|
||||
graph_height: float = 150,
|
||||
|
|
@ -31,7 +34,6 @@ class RainGraph:
|
|||
top_text_y_pos: float = 20,
|
||||
bottom_text_space: float = 50,
|
||||
bottom_text_y_pos: float = 218,
|
||||
auto=True
|
||||
):
|
||||
|
||||
self._animation_data: RadarAnimationData = animation_data
|
||||
|
|
@ -39,7 +41,7 @@ class RainGraph:
|
|||
self._background_size: (int, int) = background_size
|
||||
self._config_dir: str = config_dir
|
||||
self._dark_mode: bool = dark_mode
|
||||
self._tz = pytz.timezone(tz)
|
||||
self._tz = tz
|
||||
self._svg_width: float = svg_width
|
||||
self._inset: float = inset
|
||||
self._graph_height: float = graph_height
|
||||
|
|
@ -62,38 +64,45 @@ class RainGraph:
|
|||
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_save: Drawing
|
||||
self._dwg_animated: Drawing
|
||||
self._dwg_still: Drawing
|
||||
self._dwg_save: Drawing = Drawing()
|
||||
self._dwg_animated: Drawing = Drawing()
|
||||
self._dwg_still: Drawing = Drawing()
|
||||
|
||||
if auto:
|
||||
self.draw_svg_frame()
|
||||
self.draw_hour_bars()
|
||||
self.draw_chances_path()
|
||||
self.draw_data_line()
|
||||
self.write_hint()
|
||||
self.insert_background()
|
||||
self._dwg_save = copy.deepcopy(self._dwg)
|
||||
async def build(self) -> Self:
|
||||
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)
|
||||
|
||||
self.draw_current_fame_line()
|
||||
self.draw_description_text()
|
||||
self.insert_cloud_layer()
|
||||
self.draw_location()
|
||||
self._dwg_animated = self._dwg
|
||||
self.draw_current_fame_line()
|
||||
self.draw_description_text()
|
||||
self.insert_cloud_layer()
|
||||
self.draw_location()
|
||||
self._dwg_animated = self._dwg
|
||||
|
||||
self._dwg = self._dwg_save
|
||||
idx = self._animation_data['most_recent_image_idx']
|
||||
self.draw_current_fame_line(idx)
|
||||
self.draw_description_text(idx)
|
||||
self.insert_cloud_layer(idx)
|
||||
self.draw_location()
|
||||
self._dwg_still = self._dwg
|
||||
self._dwg = self._dwg_save
|
||||
idx = self._animation_data['most_recent_image_idx']
|
||||
self.draw_current_fame_line(idx)
|
||||
self.draw_description_text(idx)
|
||||
self.insert_cloud_layer(idx)
|
||||
self.draw_location()
|
||||
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"""
|
||||
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}")
|
||||
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("""
|
||||
.roboto {
|
||||
font-family: "Roboto Medium";
|
||||
|
|
@ -299,10 +308,10 @@ class RainGraph:
|
|||
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()
|
||||
|
||||
def insert_background(self):
|
||||
async def insert_background(self):
|
||||
bg_image_path = os.path.join(self._config_dir, self._background_image_path)
|
||||
with open(bg_image_path, 'rb') as f:
|
||||
png_data = base64.b64encode(f.read()).decode('utf-8')
|
||||
async with async_open(bg_image_path, 'rb') as f:
|
||||
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)
|
||||
self._dwg.add(image)
|
||||
|
||||
|
|
|
|||
|
|
@ -2,13 +2,13 @@
|
|||
import datetime
|
||||
import logging
|
||||
|
||||
import pytz
|
||||
from homeassistant.components import sensor
|
||||
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
from homeassistant.util import dt
|
||||
|
||||
from custom_components.irm_kmi import DOMAIN, IrmKmiCoordinator
|
||||
from custom_components.irm_kmi.const import POLLEN_NAMES, POLLEN_TO_ICON_MAP
|
||||
|
|
@ -75,7 +75,7 @@ class IrmKmiNextWarning(CoordinatorEntity, SensorEntity):
|
|||
if self.coordinator.data.get('warnings') is None:
|
||||
return None
|
||||
|
||||
now = datetime.datetime.now(tz=pytz.timezone(self.hass.config.time_zone))
|
||||
now = dt.now()
|
||||
earliest_next = None
|
||||
for item in self.coordinator.data.get('warnings'):
|
||||
if now < item.get('starts_at'):
|
||||
|
|
@ -89,7 +89,7 @@ class IrmKmiNextWarning(CoordinatorEntity, SensorEntity):
|
|||
@property
|
||||
def extra_state_attributes(self) -> dict:
|
||||
"""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_friendly_names"] = ", ".join(
|
||||
|
|
|
|||
|
|
@ -159,7 +159,6 @@ class IrmKmiWeather(CoordinatorEntity, WeatherEntity):
|
|||
:param include_past_forecasts: whether to include data points that are in the past
|
||||
:return: ordered list of forecasts
|
||||
"""
|
||||
# now = datetime.now(tz=pytz.timezone(self.hass.config.time_zone))
|
||||
now = dt.now()
|
||||
now = now.replace(minute=(now.minute // 10) * 10, second=0, microsecond=0)
|
||||
|
||||
|
|
|
|||
|
|
@ -2,5 +2,5 @@ aiohttp==3.9.5
|
|||
async-timeout==4.0.3
|
||||
homeassistant==2024.6.0b4
|
||||
voluptuous==0.13.1
|
||||
pytz==2024.1
|
||||
svgwrite==1.4.3
|
||||
svgwrite==1.4.3
|
||||
aiofile==3.8.8
|
||||
|
|
@ -1,5 +1,5 @@
|
|||
homeassistant==2024.6.0b4
|
||||
pytest
|
||||
pytest_homeassistant_custom_component==0.13.128
|
||||
pytest
|
||||
freezegun
|
||||
isort
|
||||
|
|
@ -13,9 +13,9 @@ from pytest_homeassistant_custom_component.common import (MockConfigEntry,
|
|||
from custom_components.irm_kmi.api import (IrmKmiApiError,
|
||||
IrmKmiApiParametersError)
|
||||
from custom_components.irm_kmi.const import (
|
||||
CONF_DARK_MODE, CONF_STYLE, CONF_USE_DEPRECATED_FORECAST, DOMAIN,
|
||||
OPTION_DEPRECATED_FORECAST_NOT_USED,
|
||||
OPTION_DEPRECATED_FORECAST_TWICE_DAILY, OPTION_STYLE_STD, CONF_LANGUAGE_OVERRIDE)
|
||||
CONF_DARK_MODE, CONF_LANGUAGE_OVERRIDE, CONF_STYLE,
|
||||
CONF_USE_DEPRECATED_FORECAST, DOMAIN, OPTION_DEPRECATED_FORECAST_NOT_USED,
|
||||
OPTION_DEPRECATED_FORECAST_TWICE_DAILY, OPTION_STYLE_STD)
|
||||
|
||||
|
||||
def get_api_data(fixture: str) -> dict:
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ async def test_jules_forgot_to_revert_update_interval_before_pushing(
|
|||
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(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry
|
||||
|
|
@ -49,10 +49,10 @@ async def test_warning_data(
|
|||
assert first.get('level') == 1
|
||||
|
||||
|
||||
@freeze_time(datetime.fromisoformat('2023-12-26T18:30:00'))
|
||||
def test_current_weather_be() -> None:
|
||||
@freeze_time(datetime.fromisoformat('2023-12-26T17:30:00+00:00'))
|
||||
async def test_current_weather_be() -> None:
|
||||
api_data = get_api_data("forecast.json")
|
||||
result = IrmKmiCoordinator.current_weather_from_data(api_data)
|
||||
result = await IrmKmiCoordinator.current_weather_from_data(api_data)
|
||||
|
||||
expected = CurrentWeatherData(
|
||||
condition=ATTR_CONDITION_CLOUDY,
|
||||
|
|
@ -68,9 +68,9 @@ def test_current_weather_be() -> None:
|
|||
|
||||
|
||||
@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")
|
||||
result = IrmKmiCoordinator.current_weather_from_data(api_data)
|
||||
result = await IrmKmiCoordinator.current_weather_from_data(api_data)
|
||||
|
||||
expected = CurrentWeatherData(
|
||||
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'}
|
||||
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 len(result) == 8
|
||||
|
|
@ -117,9 +117,9 @@ async def test_daily_forecast(
|
|||
|
||||
|
||||
@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')
|
||||
result = IrmKmiCoordinator.hourly_list_to_forecast(api_data)
|
||||
result = await IrmKmiCoordinator.hourly_list_to_forecast(api_data)
|
||||
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 49
|
||||
|
|
@ -142,9 +142,9 @@ def test_hourly_forecast() -> None:
|
|||
|
||||
|
||||
@freeze_time(datetime.fromisoformat('2024-05-31T01:50:00+02:00'))
|
||||
def test_hourly_forecast_bis() -> None:
|
||||
async def test_hourly_forecast_bis() -> None:
|
||||
api_data = get_api_data("no-midnight-bug-31-05-2024T01-55.json").get('for', {}).get('hourly')
|
||||
result = IrmKmiCoordinator.hourly_list_to_forecast(api_data)
|
||||
result = await IrmKmiCoordinator.hourly_list_to_forecast(api_data)
|
||||
|
||||
assert isinstance(result, list)
|
||||
|
||||
|
|
@ -158,10 +158,10 @@ def test_hourly_forecast_bis() -> None:
|
|||
|
||||
|
||||
@freeze_time(datetime.fromisoformat('2024-05-31T00:10:00+02:00'))
|
||||
def test_hourly_forecast_midnight_bug() -> None:
|
||||
async def test_hourly_forecast_midnight_bug() -> None:
|
||||
# Related to https://github.com/jdejaegh/irm-kmi-ha/issues/38
|
||||
api_data = get_api_data("midnight-bug-31-05-2024T00-13.json").get('for', {}).get('hourly')
|
||||
result = IrmKmiCoordinator.hourly_list_to_forecast(api_data)
|
||||
result = await IrmKmiCoordinator.hourly_list_to_forecast(api_data)
|
||||
|
||||
assert isinstance(result, list)
|
||||
|
||||
|
|
|
|||
|
|
@ -9,8 +9,10 @@ from homeassistant.core import HomeAssistant
|
|||
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
||||
|
||||
from custom_components.irm_kmi import async_migrate_entry
|
||||
from custom_components.irm_kmi.const import DOMAIN, CONFIG_FLOW_VERSION, CONF_LANGUAGE_OVERRIDE, \
|
||||
CONF_USE_DEPRECATED_FORECAST, OPTION_DEPRECATED_FORECAST_NOT_USED, CONF_DARK_MODE, CONF_STYLE, OPTION_STYLE_STD
|
||||
from custom_components.irm_kmi.const import (
|
||||
CONF_DARK_MODE, CONF_LANGUAGE_OVERRIDE, CONF_STYLE,
|
||||
CONF_USE_DEPRECATED_FORECAST, CONFIG_FLOW_VERSION, DOMAIN,
|
||||
OPTION_DEPRECATED_FORECAST_NOT_USED, OPTION_STYLE_STD)
|
||||
|
||||
|
||||
async def test_load_unload_config_entry(
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
rain_graph = RainGraph(
|
||||
animation_data=data,
|
||||
background_image_path="custom_components/irm_kmi/resources/be_white.png",
|
||||
background_size=(640, 490),
|
||||
auto=False
|
||||
)
|
||||
|
||||
rain_graph.draw_svg_frame()
|
||||
await rain_graph.draw_svg_frame()
|
||||
|
||||
svg_str = rain_graph.get_dwg().tostring()
|
||||
|
||||
|
|
@ -60,7 +59,6 @@ def test_svg_hint():
|
|||
animation_data=data,
|
||||
background_image_path="custom_components/irm_kmi/resources/be_white.png",
|
||||
background_size=(640, 490),
|
||||
auto=False
|
||||
)
|
||||
|
||||
rain_graph.write_hint()
|
||||
|
|
@ -76,7 +74,6 @@ def test_svg_time_bars():
|
|||
animation_data=data,
|
||||
background_image_path="custom_components/irm_kmi/resources/be_white.png",
|
||||
background_size=(640, 490),
|
||||
auto=False
|
||||
)
|
||||
|
||||
rain_graph.draw_hour_bars()
|
||||
|
|
@ -96,7 +93,6 @@ def test_draw_chances_path():
|
|||
animation_data=data,
|
||||
background_image_path="custom_components/irm_kmi/resources/be_white.png",
|
||||
background_size=(640, 490),
|
||||
auto=False
|
||||
)
|
||||
|
||||
rain_graph.draw_chances_path()
|
||||
|
|
@ -115,7 +111,6 @@ def test_draw_data_line():
|
|||
animation_data=data,
|
||||
background_image_path="custom_components/irm_kmi/resources/be_white.png",
|
||||
background_size=(640, 490),
|
||||
auto=False
|
||||
)
|
||||
|
||||
rain_graph.draw_data_line()
|
||||
|
|
@ -128,16 +123,15 @@ def test_draw_data_line():
|
|||
assert '<path ' in svg_str
|
||||
|
||||
|
||||
def test_insert_background():
|
||||
async def test_insert_background():
|
||||
data = get_radar_animation_data()
|
||||
rain_graph = RainGraph(
|
||||
animation_data=data,
|
||||
background_image_path="custom_components/irm_kmi/resources/be_white.png",
|
||||
background_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:
|
||||
png_b64 = base64.b64encode(file.read()).decode('utf-8')
|
||||
|
|
@ -158,7 +152,6 @@ def test_draw_current_frame_line_moving():
|
|||
animation_data=data,
|
||||
background_image_path="custom_components/irm_kmi/resources/be_white.png",
|
||||
background_size=(640, 490),
|
||||
auto=False
|
||||
)
|
||||
|
||||
rain_graph.draw_current_fame_line()
|
||||
|
|
@ -187,7 +180,6 @@ def test_draw_current_frame_line_index():
|
|||
animation_data=data,
|
||||
background_image_path="custom_components/irm_kmi/resources/be_white.png",
|
||||
background_size=(640, 490),
|
||||
auto=False
|
||||
)
|
||||
|
||||
rain_graph.draw_current_fame_line(0)
|
||||
|
|
@ -216,7 +208,6 @@ def test_draw_description_text():
|
|||
animation_data=data,
|
||||
background_image_path="custom_components/irm_kmi/resources/be_white.png",
|
||||
background_size=(640, 490),
|
||||
auto=False
|
||||
)
|
||||
|
||||
rain_graph.draw_description_text()
|
||||
|
|
@ -244,7 +235,6 @@ def test_draw_cloud_layer():
|
|||
animation_data=data,
|
||||
background_image_path="custom_components/irm_kmi/resources/be_white.png",
|
||||
background_size=(640, 490),
|
||||
auto=False
|
||||
)
|
||||
|
||||
rain_graph.insert_cloud_layer()
|
||||
|
|
@ -265,7 +255,6 @@ def test_draw_location_layer():
|
|||
animation_data=data,
|
||||
background_image_path="custom_components/irm_kmi/resources/be_white.png",
|
||||
background_size=(640, 490),
|
||||
auto=False
|
||||
)
|
||||
|
||||
rain_graph.draw_location()
|
||||
|
|
|
|||
|
|
@ -35,7 +35,7 @@ async def test_warning_data(
|
|||
|
||||
|
||||
@freeze_time(datetime.fromisoformat('2024-01-12T07:55:00+01:00'))
|
||||
async def test_warning_data(
|
||||
async def test_warning_data_unknown_lang(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue