From 78535b99df0b6a13d85ee9befc5a17fcb30f2084 Mon Sep 17 00:00:00 2001 From: Jan-Philipp Benecke Date: Sun, 8 Oct 2023 20:07:04 +0200 Subject: [PATCH] Move nina coordinator and entity to their own file (#101610) --- homeassistant/components/nina/__init__.py | 135 +---------------- .../components/nina/binary_sensor.py | 2 +- homeassistant/components/nina/coordinator.py | 138 ++++++++++++++++++ 3 files changed, 140 insertions(+), 135 deletions(-) create mode 100644 homeassistant/components/nina/coordinator.py diff --git a/homeassistant/components/nina/__init__.py b/homeassistant/components/nina/__init__.py index 4ac2518ffb6..435ea288aa7 100644 --- a/homeassistant/components/nina/__init__.py +++ b/homeassistant/components/nina/__init__.py @@ -1,21 +1,11 @@ """The Nina integration.""" from __future__ import annotations -import asyncio -from dataclasses import dataclass -import re -from typing import Any - -from pynina import ApiError, Nina - from homeassistant.config_entries import ConfigEntry from homeassistant.const import Platform from homeassistant.core import HomeAssistant -from homeassistant.helpers.aiohttp_client import async_get_clientsession -from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from .const import ( - _LOGGER, ALL_MATCH_REGEX, CONF_AREA_FILTER, CONF_FILTER_CORONA, @@ -23,8 +13,8 @@ from .const import ( CONF_REGIONS, DOMAIN, NO_MATCH_REGEX, - SCAN_INTERVAL, ) +from .coordinator import NINADataUpdateCoordinator PLATFORMS: list[str] = [Platform.BINARY_SENSOR] @@ -74,126 +64,3 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None: """Handle options update.""" await hass.config_entries.async_reload(entry.entry_id) - - -@dataclass -class NinaWarningData: - """Class to hold the warning data.""" - - id: str - headline: str - description: str - sender: str - severity: str - recommended_actions: str - affected_areas: str - sent: str - start: str - expires: str - is_valid: bool - - -class NINADataUpdateCoordinator( - DataUpdateCoordinator[dict[str, list[NinaWarningData]]] -): - """Class to manage fetching NINA data API.""" - - def __init__( - self, - hass: HomeAssistant, - regions: dict[str, str], - headline_filter: str, - area_filter: str, - ) -> None: - """Initialize.""" - self._regions: dict[str, str] = regions - self._nina: Nina = Nina(async_get_clientsession(hass)) - self.headline_filter: str = headline_filter - self.area_filter: str = area_filter - - for region in regions: - self._nina.addRegion(region) - - super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=SCAN_INTERVAL) - - async def _async_update_data(self) -> dict[str, list[NinaWarningData]]: - """Update data.""" - async with asyncio.timeout(10): - try: - await self._nina.update() - except ApiError as err: - raise UpdateFailed(err) from err - return self._parse_data() - - @staticmethod - def _remove_duplicate_warnings( - warnings: dict[str, list[Any]] - ) -> dict[str, list[Any]]: - """Remove warnings with the same title and expires timestamp in a region.""" - all_filtered_warnings: dict[str, list[Any]] = {} - - for region_id, raw_warnings in warnings.items(): - filtered_warnings: list[Any] = [] - processed_details: list[tuple[str, str]] = [] - - for raw_warn in raw_warnings: - if (raw_warn.headline, raw_warn.expires) in processed_details: - continue - - processed_details.append((raw_warn.headline, raw_warn.expires)) - - filtered_warnings.append(raw_warn) - - all_filtered_warnings[region_id] = filtered_warnings - - return all_filtered_warnings - - def _parse_data(self) -> dict[str, list[NinaWarningData]]: - """Parse warning data.""" - - return_data: dict[str, list[NinaWarningData]] = {} - - for region_id, raw_warnings in self._remove_duplicate_warnings( - self._nina.warnings - ).items(): - warnings_for_regions: list[NinaWarningData] = [] - - for raw_warn in raw_warnings: - if re.search( - self.headline_filter, raw_warn.headline, flags=re.IGNORECASE - ): - _LOGGER.debug( - f"Ignore warning ({raw_warn.id}) by headline filter ({self.headline_filter}) with headline: {raw_warn.headline}" - ) - continue - - affected_areas_string: str = ", ".join( - [str(area) for area in raw_warn.affected_areas] - ) - - if not re.search( - self.area_filter, affected_areas_string, flags=re.IGNORECASE - ): - _LOGGER.debug( - f"Ignore warning ({raw_warn.id}) by area filter ({self.area_filter}) with area: {affected_areas_string}" - ) - continue - - warning_data: NinaWarningData = NinaWarningData( - raw_warn.id, - raw_warn.headline, - raw_warn.description, - raw_warn.sender, - raw_warn.severity, - " ".join([str(action) for action in raw_warn.recommended_actions]), - affected_areas_string, - raw_warn.sent or "", - raw_warn.start or "", - raw_warn.expires or "", - raw_warn.isValid(), - ) - warnings_for_regions.append(warning_data) - - return_data[region_id] = warnings_for_regions - - return return_data diff --git a/homeassistant/components/nina/binary_sensor.py b/homeassistant/components/nina/binary_sensor.py index 19f802f1cec..568869ca402 100644 --- a/homeassistant/components/nina/binary_sensor.py +++ b/homeassistant/components/nina/binary_sensor.py @@ -12,7 +12,6 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.update_coordinator import CoordinatorEntity -from . import NINADataUpdateCoordinator from .const import ( ATTR_AFFECTED_AREAS, ATTR_DESCRIPTION, @@ -28,6 +27,7 @@ from .const import ( CONF_REGIONS, DOMAIN, ) +from .coordinator import NINADataUpdateCoordinator async def async_setup_entry( diff --git a/homeassistant/components/nina/coordinator.py b/homeassistant/components/nina/coordinator.py new file mode 100644 index 00000000000..eb5c7a7e506 --- /dev/null +++ b/homeassistant/components/nina/coordinator.py @@ -0,0 +1,138 @@ +"""DataUpdateCoordinator for the nina integration.""" +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +import re +from typing import Any + +from pynina import ApiError, Nina + +from homeassistant.core import HomeAssistant +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed + +from .const import _LOGGER, DOMAIN, SCAN_INTERVAL + + +@dataclass +class NinaWarningData: + """Class to hold the warning data.""" + + id: str + headline: str + description: str + sender: str + severity: str + recommended_actions: str + affected_areas: str + sent: str + start: str + expires: str + is_valid: bool + + +class NINADataUpdateCoordinator( + DataUpdateCoordinator[dict[str, list[NinaWarningData]]] +): + """Class to manage fetching NINA data API.""" + + def __init__( + self, + hass: HomeAssistant, + regions: dict[str, str], + headline_filter: str, + area_filter: str, + ) -> None: + """Initialize.""" + self._regions: dict[str, str] = regions + self._nina: Nina = Nina(async_get_clientsession(hass)) + self.headline_filter: str = headline_filter + self.area_filter: str = area_filter + + for region in regions: + self._nina.addRegion(region) + + super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=SCAN_INTERVAL) + + async def _async_update_data(self) -> dict[str, list[NinaWarningData]]: + """Update data.""" + async with asyncio.timeout(10): + try: + await self._nina.update() + except ApiError as err: + raise UpdateFailed(err) from err + return self._parse_data() + + @staticmethod + def _remove_duplicate_warnings( + warnings: dict[str, list[Any]] + ) -> dict[str, list[Any]]: + """Remove warnings with the same title and expires timestamp in a region.""" + all_filtered_warnings: dict[str, list[Any]] = {} + + for region_id, raw_warnings in warnings.items(): + filtered_warnings: list[Any] = [] + processed_details: list[tuple[str, str]] = [] + + for raw_warn in raw_warnings: + if (raw_warn.headline, raw_warn.expires) in processed_details: + continue + + processed_details.append((raw_warn.headline, raw_warn.expires)) + + filtered_warnings.append(raw_warn) + + all_filtered_warnings[region_id] = filtered_warnings + + return all_filtered_warnings + + def _parse_data(self) -> dict[str, list[NinaWarningData]]: + """Parse warning data.""" + + return_data: dict[str, list[NinaWarningData]] = {} + + for region_id, raw_warnings in self._remove_duplicate_warnings( + self._nina.warnings + ).items(): + warnings_for_regions: list[NinaWarningData] = [] + + for raw_warn in raw_warnings: + if re.search( + self.headline_filter, raw_warn.headline, flags=re.IGNORECASE + ): + _LOGGER.debug( + f"Ignore warning ({raw_warn.id}) by headline filter ({self.headline_filter}) with headline: {raw_warn.headline}" + ) + continue + + affected_areas_string: str = ", ".join( + [str(area) for area in raw_warn.affected_areas] + ) + + if not re.search( + self.area_filter, affected_areas_string, flags=re.IGNORECASE + ): + _LOGGER.debug( + f"Ignore warning ({raw_warn.id}) by area filter ({self.area_filter}) with area: {affected_areas_string}" + ) + continue + + warning_data: NinaWarningData = NinaWarningData( + raw_warn.id, + raw_warn.headline, + raw_warn.description, + raw_warn.sender, + raw_warn.severity, + " ".join([str(action) for action in raw_warn.recommended_actions]), + affected_areas_string, + raw_warn.sent or "", + raw_warn.start or "", + raw_warn.expires or "", + raw_warn.isValid(), + ) + warnings_for_regions.append(warning_data) + + return_data[region_id] = warnings_for_regions + + return return_data