mirror of
https://github.com/home-assistant/core.git
synced 2025-07-25 06:07:17 +00:00
Move ukraine_alarm coordinator to separate module (#126549)
This commit is contained in:
parent
380019dd56
commit
02ab2c1433
@ -2,25 +2,13 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from datetime import timedelta
|
|
||||||
import logging
|
|
||||||
from typing import Any
|
|
||||||
|
|
||||||
import aiohttp
|
|
||||||
from aiohttp import ClientSession
|
|
||||||
from uasiren.client import Client
|
|
||||||
|
|
||||||
from homeassistant.config_entries import ConfigEntry
|
from homeassistant.config_entries import ConfigEntry
|
||||||
from homeassistant.const import CONF_REGION
|
from homeassistant.const import CONF_REGION
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
|
||||||
|
|
||||||
from .const import ALERT_TYPES, DOMAIN, PLATFORMS
|
from .const import DOMAIN, PLATFORMS
|
||||||
|
from .coordinator import UkraineAlarmDataUpdateCoordinator
|
||||||
_LOGGER = logging.getLogger(__name__)
|
|
||||||
|
|
||||||
UPDATE_INTERVAL = timedelta(seconds=10)
|
|
||||||
|
|
||||||
|
|
||||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||||
@ -45,32 +33,3 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||||||
hass.data[DOMAIN].pop(entry.entry_id)
|
hass.data[DOMAIN].pop(entry.entry_id)
|
||||||
|
|
||||||
return unload_ok
|
return unload_ok
|
||||||
|
|
||||||
|
|
||||||
class UkraineAlarmDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): # pylint: disable=hass-enforce-class-module
|
|
||||||
"""Class to manage fetching Ukraine Alarm API."""
|
|
||||||
|
|
||||||
def __init__(
|
|
||||||
self,
|
|
||||||
hass: HomeAssistant,
|
|
||||||
session: ClientSession,
|
|
||||||
region_id: str,
|
|
||||||
) -> None:
|
|
||||||
"""Initialize."""
|
|
||||||
self.region_id = region_id
|
|
||||||
self.uasiren = Client(session)
|
|
||||||
|
|
||||||
super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=UPDATE_INTERVAL)
|
|
||||||
|
|
||||||
async def _async_update_data(self) -> dict[str, Any]:
|
|
||||||
"""Update data via library."""
|
|
||||||
try:
|
|
||||||
res = await self.uasiren.get_alerts(self.region_id)
|
|
||||||
except aiohttp.ClientError as error:
|
|
||||||
raise UpdateFailed(f"Error fetching alerts from API: {error}") from error
|
|
||||||
|
|
||||||
current = {alert_type: False for alert_type in ALERT_TYPES}
|
|
||||||
for alert in res[0]["activeAlerts"]:
|
|
||||||
current[alert["type"]] = True
|
|
||||||
|
|
||||||
return current
|
|
||||||
|
@ -14,7 +14,6 @@ from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
|
|||||||
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 . import UkraineAlarmDataUpdateCoordinator
|
|
||||||
from .const import (
|
from .const import (
|
||||||
ALERT_TYPE_AIR,
|
ALERT_TYPE_AIR,
|
||||||
ALERT_TYPE_ARTILLERY,
|
ALERT_TYPE_ARTILLERY,
|
||||||
@ -26,6 +25,7 @@ from .const import (
|
|||||||
DOMAIN,
|
DOMAIN,
|
||||||
MANUFACTURER,
|
MANUFACTURER,
|
||||||
)
|
)
|
||||||
|
from .coordinator import UkraineAlarmDataUpdateCoordinator
|
||||||
|
|
||||||
BINARY_SENSOR_TYPES: tuple[BinarySensorEntityDescription, ...] = (
|
BINARY_SENSOR_TYPES: tuple[BinarySensorEntityDescription, ...] = (
|
||||||
BinarySensorEntityDescription(
|
BinarySensorEntityDescription(
|
||||||
|
49
homeassistant/components/ukraine_alarm/coordinator.py
Normal file
49
homeassistant/components/ukraine_alarm/coordinator.py
Normal file
@ -0,0 +1,49 @@
|
|||||||
|
"""The ukraine_alarm component."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from datetime import timedelta
|
||||||
|
import logging
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
import aiohttp
|
||||||
|
from aiohttp import ClientSession
|
||||||
|
from uasiren.client import Client
|
||||||
|
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||||
|
|
||||||
|
from .const import ALERT_TYPES, DOMAIN
|
||||||
|
|
||||||
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
UPDATE_INTERVAL = timedelta(seconds=10)
|
||||||
|
|
||||||
|
|
||||||
|
class UkraineAlarmDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
|
||||||
|
"""Class to manage fetching Ukraine Alarm API."""
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
hass: HomeAssistant,
|
||||||
|
session: ClientSession,
|
||||||
|
region_id: str,
|
||||||
|
) -> None:
|
||||||
|
"""Initialize."""
|
||||||
|
self.region_id = region_id
|
||||||
|
self.uasiren = Client(session)
|
||||||
|
|
||||||
|
super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=UPDATE_INTERVAL)
|
||||||
|
|
||||||
|
async def _async_update_data(self) -> dict[str, Any]:
|
||||||
|
"""Update data via library."""
|
||||||
|
try:
|
||||||
|
res = await self.uasiren.get_alerts(self.region_id)
|
||||||
|
except aiohttp.ClientError as error:
|
||||||
|
raise UpdateFailed(f"Error fetching alerts from API: {error}") from error
|
||||||
|
|
||||||
|
current = {alert_type: False for alert_type in ALERT_TYPES}
|
||||||
|
for alert in res[0]["activeAlerts"]:
|
||||||
|
current[alert["type"]] = True
|
||||||
|
|
||||||
|
return current
|
Loading…
x
Reference in New Issue
Block a user