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

Initial support for pollen data

Browse source
This commit is contained in:
Jules 2024-04-01 18:08:16 +02:00
parent 18737439db
commit c8b2954ef3
Signed by: jdejaegh
GPG key ID: 99D6D184CA66933A
8 changed files with 118 additions and 39 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
custom_components/irm_kmi/api.py
View file

@ -54,6 +54,11 @@ class IrmKmiApiClient:
r: ClientResponse = await self._api_wrapper(base_url=url, params={} if params is None else params)
return await r.read()
async def get_svg(self, url, params: dict | None = None) -> str:
"""Get SVG as str at the specified url with the parameters"""
r: ClientResponse = await self._api_wrapper(base_url=url, params={} if params is None else params)
return await r.text()
async def _api_wrapper(
self,
params: dict,

2
custom_components/irm_kmi/const.py
View file

@ -14,7 +14,7 @@ from homeassistant.components.weather import (ATTR_CONDITION_CLEAR_NIGHT,
from homeassistant.const import Platform
DOMAIN: Final = 'irm_kmi'
PLATFORMS: Final = [Platform.WEATHER, Platform.CAMERA, Platform.BINARY_SENSOR]
PLATFORMS: Final = [Platform.WEATHER, Platform.CAMERA, Platform.BINARY_SENSOR, Platform.SENSOR]
CONFIG_FLOW_VERSION = 3
OUT_OF_BENELUX: Final = ["außerhalb der Benelux (Brussels)",

29
custom_components/irm_kmi/coordinator.py
View file

@ -23,6 +23,7 @@ from .const import MAP_WARNING_ID_TO_SLUG as SLUG_MAP
from .const import OPTION_STYLE_SATELLITE, OUT_OF_BENELUX, STYLE_TO_PARAM_MAP
from .data import (AnimationFrameData, CurrentWeatherData, IrmKmiForecast,
ProcessedCoordinatorData, RadarAnimationData, WarningData)
from .pollen import PollenParser
from .rain_graph import RainGraph
from .utils import disable_from_config, get_config_value
@ -65,7 +66,8 @@ class IrmKmiCoordinator(DataUpdateCoordinator):
'long': zone.attributes[ATTR_LONGITUDE]}
)
_LOGGER.debug(f"Observation for {api_data.get('cityName', '')}: {api_data.get('obs', '{}')}")
_LOGGER.debug(f"Full data: {api_data}")
# TODO re-enable logging here
# _LOGGER.debug(f"Full data: {api_data}")
except IrmKmiApiError as err:
raise UpdateFailed(f"Error communicating with API: {err}")
@ -124,6 +126,28 @@ class IrmKmiCoordinator(DataUpdateCoordinator):
radar_animation['svg_still'] = rain_graph.get_svg_string(still_image=True)
return radar_animation
async def _async_pollen_data(self, api_data: dict) -> dict:
_LOGGER.debug("Getting pollen data from API")
svg_url = None
for module in api_data.get('module', []):
_LOGGER.debug(f"module: {module}")
if module.get('type', None) == 'svg':
url = module.get('data', {}).get('url', {}).get('en', '')
if 'pollen' in url:
svg_url = url
break
if svg_url is None:
return PollenParser.get_default_data()
try:
_LOGGER.debug(f"Requesting pollen SVG at url {svg_url}")
pollen_svg: str = await self._api_client.get_svg(svg_url)
except IrmKmiApiError:
_LOGGER.warning(f"Could not get pollen data from the API")
return PollenParser.get_default_data()
return PollenParser(pollen_svg).get_pollen_data()
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(
@ -131,7 +155,8 @@ class IrmKmiCoordinator(DataUpdateCoordinator):
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')),
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')),
pollen=await self._async_pollen_data(api_data=api_data)
)
async def download_images_from_api(self,

1
custom_components/irm_kmi/data.py
View file

@ -62,3 +62,4 @@ class ProcessedCoordinatorData(TypedDict, total=False):
daily_forecast: List[IrmKmiForecast] | None
animation: RadarAnimationData
warnings: List[WarningData]
pollen: dict

51
custom_components/irm_kmi/pollen.py
View file

@ -31,23 +31,31 @@ class PollenParser:
return True
@staticmethod
def extract_elements(root) -> List[ET.Element]:
def get_default_data() -> dict:
return {k.lower(): 'none' for k in POLLEN_NAMES}
@staticmethod
def get_option_values() -> List[str]:
return ['active', 'green', 'yellow', 'orange', 'red', 'purple', 'none']
@staticmethod
def _extract_elements(root) -> List[ET.Element]:
elements = []
for child in root:
elements.append(child)
elements.extend(PollenParser.extract_elements(child))
elements.extend(PollenParser._extract_elements(child))
return elements
@staticmethod
def dot_to_color_value(dot: ET.Element) -> str | None:
def _dot_to_color_value(dot: ET.Element) -> str:
try:
cx = float(dot.attrib.get('cx'))
except ValueError:
return None
return 'none'
if cx > 155:
return None
return 'none'
elif cx > 140:
return 'purple'
elif cx > 125:
@ -59,20 +67,21 @@ class PollenParser:
elif cx > 80:
return 'green'
else:
return None
return 'none'
def get_pollen_data(self):
def get_pollen_data(self) -> dict:
pollen_data = self.get_default_data()
try:
root = ET.fromstring("self._xml")
root = ET.fromstring(self._xml)
except ET.ParseError:
# TODO Handle with default case
return None
_LOGGER.warning("Could not parse SVG pollen XML")
return pollen_data
elements: List[ET.Element] = self.extract_elements(root)
elements: List[ET.Element] = self._extract_elements(root)
if not self._validate_svg(elements):
# TODO return default value
return None
_LOGGER.warning("Could not validate SVG pollen data")
return pollen_data
pollens = [e for e in elements if 'tspan' in e.tag and e.text in POLLEN_NAMES]
active = [e for e in elements if 'tspan' in e.tag and e.text == 'active']
@ -80,23 +89,19 @@ class PollenParser:
and 'fill:#ffffff' in e.attrib.get('style', '')
and 3 == float(e.attrib.get('rx', '0'))]
pollen_data = {k: None for k in POLLEN_NAMES}
for pollen in pollens:
try:
y = float(pollen.attrib.get('y'))
if y in [float(e.attrib.get('y')) for e in active]:
pollen_data[pollen.text] = 'active'
pollen_data[pollen.text.lower()] = 'active'
else:
dot = [d for d in dots if y - 3 <= float(d.attrib.get('cy', '0')) <= y + 3]
if len(dot) == 1:
dot = dot[0]
pollen_data[pollen.text] = self.dot_to_color_value(dot)
pollen_data[pollen.text.lower()] = self._dot_to_color_value(dot)
except ValueError | NameError:
pass
_LOGGER.warning("Skipped some data in the pollen SVG")
_LOGGER.debug(f"Pollen data: {pollen_data}")
return pollen_data
print(pollen_data)
for e in pollens + active:
print(e.text)
for d in dots:
print(d.attrib)

56
custom_components/irm_kmi/sensor.py Normal file
View file

@ -0,0 +1,56 @@
"""Sensor for pollen from the IRM KMI"""
import logging
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.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from custom_components.irm_kmi import DOMAIN, IrmKmiCoordinator
from custom_components.irm_kmi.const import POLLEN_NAMES
from custom_components.irm_kmi.pollen import PollenParser
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback):
"""Set up the sensor platform"""
coordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities([IrmKmiPollen(coordinator, entry, pollen.lower()) for pollen in POLLEN_NAMES])
class IrmKmiPollen(CoordinatorEntity, SensorEntity):
"""Representation of a pollen sensor"""
_attr_has_entity_name = True
_attr_device_class = SensorDeviceClass.ENUM
def __init__(self,
coordinator: IrmKmiCoordinator,
entry: ConfigEntry,
pollen: str
) -> None:
super().__init__(coordinator)
SensorEntity.__init__(self)
self._attr_unique_id = f"{entry.entry_id}-pollen-{pollen}"
self.entity_id = sensor.ENTITY_ID_FORMAT.format(f"{str(entry.title).lower()}_{pollen}_level")
self._attr_options = PollenParser.get_option_values()
self._attr_name = f"Pollen {pollen}"
self._attr_device_info = DeviceInfo(
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN, entry.entry_id)},
manufacturer="IRM KMI",
name=f"Pollen {pollen}"
)
self._pollen = pollen
# TODO add translation for name
# self._attr_translation_key = f"pollen_{pollen}"
# _LOGGER.debug(f"translation key: {self._attr_translation_key}")
@property
def native_value(self) -> str | None:
"""Return the state of the sensor."""
return self.coordinator.data.get('pollen', {}).get(self._pollen, None)

12
tests/fixtures/high_low_temp.json vendored
View file

@ -1325,18 +1325,6 @@
]
},
"module": [
{
"type": "svg",
"data": {
"url": {
"nl": "https:\/\/app.meteo.be\/services\/appv4\/?s=getSvg&ins=92094&e=pollen&l=nl&k=32003c7eac2900f3d73c50f9e27330ab",
"fr": "https:\/\/app.meteo.be\/services\/appv4\/?s=getSvg&ins=92094&e=pollen&l=fr&k=32003c7eac2900f3d73c50f9e27330ab",
"en": "https:\/\/app.meteo.be\/services\/appv4\/?s=getSvg&ins=92094&e=pollen&l=en&k=32003c7eac2900f3d73c50f9e27330ab",
"de": "https:\/\/app.meteo.be\/services\/appv4\/?s=getSvg&ins=92094&e=pollen&l=de&k=32003c7eac2900f3d73c50f9e27330ab"
},
"ratio": 3.0458333333333334
}
},
{
"type": "uv",
"data": {

1
tests/test_pollen.py
View file

@ -1,6 +1,5 @@
from custom_components.irm_kmi.pollen import PollenParser
def test_svg_pollen_parsing():
# TODO make it an actual test
with open("tests/fixtures/pollen.svg", "r") as file: