Add Ukraine Alarm integration (home-assistant#71501)

Co-authored-by: J. Nick Koston <[email protected]>
Co-authored-by: Paulus Schoutsen <[email protected]>
Co-authored-by: Martin Hjelmare <[email protected]>
Co-authored-by: Paulus Schoutsen <[email protected]>

.coveragerc

@@ -1313,6 +1313,9 @@ omit =
     homeassistant/components/twitter/notify.py
     homeassistant/components/ubus/device_tracker.py
     homeassistant/components/ue_smart_radio/media_player.py
+    homeassistant/components/ukraine_alarm/__init__.py
+    homeassistant/components/ukraine_alarm/const.py
+    homeassistant/components/ukraine_alarm/binary_sensor.py
     homeassistant/components/unifiled/*
     homeassistant/components/upb/__init__.py
     homeassistant/components/upb/const.py

CODEOWNERS

@@ -1071,6 +1071,8 @@ build.json @home-assistant/supervisor
 /tests/components/twentemilieu/ @frenck
 /homeassistant/components/twinkly/ @dr1rrb @Robbie1221
 /tests/components/twinkly/ @dr1rrb @Robbie1221
+/homeassistant/components/ukraine_alarm/ @PaulAnnekov
+/tests/components/ukraine_alarm/ @PaulAnnekov
 /homeassistant/components/unifi/ @Kane610
 /tests/components/unifi/ @Kane610
 /homeassistant/components/unifiled/ @florisvdk

homeassistant/components/ukraine_alarm/__init__.py

@@ -0,0 +1,79 @@
+"""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 ukrainealarm.client import Client
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_API_KEY, CONF_REGION
+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 ALERT_TYPES, DOMAIN, PLATFORMS
+
+_LOGGER = logging.getLogger(__name__)
+
+UPDATE_INTERVAL = timedelta(seconds=10)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Ukraine Alarm as config entry."""
+    api_key = entry.data[CONF_API_KEY]
+    region_id = entry.data[CONF_REGION]
+
+    websession = async_get_clientsession(hass)
+
+    coordinator = UkraineAlarmDataUpdateCoordinator(
+        hass, websession, api_key, region_id
+    )
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
+
+    hass.config_entries.async_setup_platforms(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok
+
+
+class UkraineAlarmDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
+    """Class to manage fetching Ukraine Alarm API."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        session: ClientSession,
+        api_key: str,
+        region_id: str,
+    ) -> None:
+        """Initialize."""
+        self.region_id = region_id
+        self.ukrainealarm = Client(session, api_key)
+
+        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.ukrainealarm.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

homeassistant/components/ukraine_alarm/binary_sensor.py

@@ -0,0 +1,106 @@
+"""binary sensors for Ukraine Alarm integration."""
+from __future__ import annotations
+
+from homeassistant.components.binary_sensor import (
+    BinarySensorDeviceClass,
+    BinarySensorEntity,
+    BinarySensorEntityDescription,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_NAME
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.device_registry import DeviceEntryType
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from . import UkraineAlarmDataUpdateCoordinator
+from .const import (
+    ALERT_TYPE_AIR,
+    ALERT_TYPE_ARTILLERY,
+    ALERT_TYPE_UNKNOWN,
+    ALERT_TYPE_URBAN_FIGHTS,
+    ATTRIBUTION,
+    DOMAIN,
+    MANUFACTURER,
+)
+
+BINARY_SENSOR_TYPES: tuple[BinarySensorEntityDescription, ...] = (
+    BinarySensorEntityDescription(
+        key=ALERT_TYPE_UNKNOWN,
+        name="Unknown",
+        device_class=BinarySensorDeviceClass.SAFETY,
+    ),
+    BinarySensorEntityDescription(
+        key=ALERT_TYPE_AIR,
+        name="Air",
+        device_class=BinarySensorDeviceClass.SAFETY,
+        icon="mdi:cloud",
+    ),
+    BinarySensorEntityDescription(
+        key=ALERT_TYPE_URBAN_FIGHTS,
+        name="Urban Fights",
+        device_class=BinarySensorDeviceClass.SAFETY,
+        icon="mdi:pistol",
+    ),
+    BinarySensorEntityDescription(
+        key=ALERT_TYPE_ARTILLERY,
+        name="Artillery",
+        device_class=BinarySensorDeviceClass.SAFETY,
+        icon="mdi:tank",
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up Ukraine Alarm binary sensor entities based on a config entry."""
+    name = config_entry.data[CONF_NAME]
+    coordinator = hass.data[DOMAIN][config_entry.entry_id]
+
+    async_add_entities(
+        UkraineAlarmSensor(
+            name,
+            config_entry.unique_id,
+            description,
+            coordinator,
+        )
+        for description in BINARY_SENSOR_TYPES
+    )
+
+
+class UkraineAlarmSensor(
+    CoordinatorEntity[UkraineAlarmDataUpdateCoordinator], BinarySensorEntity
+):
+    """Class for a Ukraine Alarm binary sensor."""
+
+    _attr_attribution = ATTRIBUTION
+
+    def __init__(
+        self,
+        name,
+        unique_id,
+        description: BinarySensorEntityDescription,
+        coordinator: UkraineAlarmDataUpdateCoordinator,
+    ) -> None:
+        """Initialize the sensor."""
+        super().__init__(coordinator)
+
+        self.entity_description = description
+
+        self._attr_name = f"{name} {description.name}"
+        self._attr_unique_id = f"{unique_id}-{description.key}".lower()
+        self._attr_device_info = DeviceInfo(
+            entry_type=DeviceEntryType.SERVICE,
+            identifiers={(DOMAIN, unique_id)},
+            manufacturer=MANUFACTURER,
+            name=name,
+        )
+
+    @property
+    def is_on(self) -> bool | None:
+        """Return true if the binary sensor is on."""
+        return self.coordinator.data.get(self.entity_description.key, None)

homeassistant/components/ukraine_alarm/config_flow.py

@@ -0,0 +1,154 @@
+"""Config flow for Ukraine Alarm."""
+from __future__ import annotations
+
+import asyncio
+
+import aiohttp
+from ukrainealarm.client import Client
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_API_KEY, CONF_NAME, CONF_REGION
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import DOMAIN
+
+
+class UkraineAlarmConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Config flow for Ukraine Alarm."""
+
+    VERSION = 1
+
+    def __init__(self):
+        """Initialize a new UkraineAlarmConfigFlow."""
+        self.api_key = None
+        self.states = None
+        self.selected_region = None
+
+    async def async_step_user(self, user_input=None):
+        """Handle a flow initialized by the user."""
+        errors = {}
+
+        if user_input is not None:
+            websession = async_get_clientsession(self.hass)
+            try:
+                regions = await Client(
+                    websession, user_input[CONF_API_KEY]
+                ).get_regions()
+            except aiohttp.ClientResponseError as ex:
+                errors["base"] = "invalid_api_key" if ex.status == 401 else "unknown"
+            except aiohttp.ClientConnectionError:
+                errors["base"] = "cannot_connect"
+            except aiohttp.ClientError:
+                errors["base"] = "unknown"
+            except asyncio.TimeoutError:
+                errors["base"] = "timeout"
+
+            if not errors and not regions:
+                errors["base"] = "unknown"
+
+            if not errors:
+                self.api_key = user_input[CONF_API_KEY]
+                self.states = regions["states"]
+                return await self.async_step_state()
+
+        schema = vol.Schema(
+            {
+                vol.Required(CONF_API_KEY): str,
+            }
+        )
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=schema,
+            description_placeholders={"api_url": "https://api.ukrainealarm.com/"},
+            errors=errors,
+            last_step=False,
+        )
+
+    async def async_step_state(self, user_input=None):
+        """Handle user-chosen state."""
+        return await self._handle_pick_region("state", "district", user_input)
+
+    async def async_step_district(self, user_input=None):
+        """Handle user-chosen district."""
+        return await self._handle_pick_region("district", "community", user_input)
+
+    async def async_step_community(self, user_input=None):
+        """Handle user-chosen community."""
+        return await self._handle_pick_region("community", None, user_input, True)
+
+    async def _handle_pick_region(
+        self, step_id: str, next_step: str | None, user_input, last_step=False
+    ):
+        """Handle picking a (sub)region."""
+        if self.selected_region:
+            source = self.selected_region["regionChildIds"]
+        else:
+            source = self.states
+
+        if user_input is not None:
+            # Only offer to browse subchildren if picked region wasn't the previously picked one
+            if (
+                not self.selected_region
+                or user_input[CONF_REGION] != self.selected_region["regionId"]
+            ):
+                self.selected_region = _find(source, user_input[CONF_REGION])
+
+                if next_step and self.selected_region["regionChildIds"]:
+                    return await getattr(self, f"async_step_{next_step}")()
+
+            return await self._async_finish_flow()
+
+        regions = {}
+        if self.selected_region:
+            regions[self.selected_region["regionId"]] = self.selected_region[
+                "regionName"
+            ]
+
+        regions.update(_make_regions_object(source))
+
+        schema = vol.Schema(
+            {
+                vol.Required(CONF_REGION): vol.In(regions),
+            }
+        )
+
+        return self.async_show_form(
+            step_id=step_id, data_schema=schema, last_step=last_step
+        )
+
+    async def _async_finish_flow(self):
+        """Finish the setup."""
+        await self.async_set_unique_id(self.selected_region["regionId"])
+        self._abort_if_unique_id_configured()
+
+        return self.async_create_entry(
+            title=self.selected_region["regionName"],
+            data={
+                CONF_API_KEY: self.api_key,
+                CONF_REGION: self.selected_region["regionId"],
+                CONF_NAME: self.selected_region["regionName"],
+            },
+        )
+
+
+def _find(regions, region_id):
+    return next((region for region in regions if region["regionId"] == region_id), None)
+
+
+def _make_regions_object(regions):
+    regions_list = []
+    for region in regions:
+        regions_list.append(
+            {
+                "id": region["regionId"],
+                "name": region["regionName"],
+            }
+        )
+    regions_list = sorted(regions_list, key=lambda region: region["name"].lower())
+    regions_object = {}
+    for region in regions_list:
+        regions_object[region["id"]] = region["name"]
+
+    return regions_object

homeassistant/components/ukraine_alarm/const.py

@@ -0,0 +1,19 @@
+"""Consts for the Ukraine Alarm."""
+from __future__ import annotations
+
+from homeassistant.const import Platform
+
+DOMAIN = "ukraine_alarm"
+ATTRIBUTION = "Data provided by Ukraine Alarm"
+MANUFACTURER = "Ukraine Alarm"
+ALERT_TYPE_UNKNOWN = "UNKNOWN"
+ALERT_TYPE_AIR = "AIR"
+ALERT_TYPE_ARTILLERY = "ARTILLERY"
+ALERT_TYPE_URBAN_FIGHTS = "URBAN_FIGHTS"
+ALERT_TYPES = {
+    ALERT_TYPE_UNKNOWN,
+    ALERT_TYPE_AIR,
+    ALERT_TYPE_ARTILLERY,
+    ALERT_TYPE_URBAN_FIGHTS,
+}
+PLATFORMS = [Platform.BINARY_SENSOR]

homeassistant/components/ukraine_alarm/manifest.json

@@ -0,0 +1,9 @@
+{
+  "domain": "ukraine_alarm",
+  "name": "Ukraine Alarm",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/ukraine_alarm",
+  "requirements": ["ukrainealarm==0.0.1"],
+  "codeowners": ["@PaulAnnekov"],
+  "iot_class": "cloud_polling"
+}