config_flow.py

Go to the documentation of this file.

"""Config flow for Ukraine Alarm."""
 
from __future__ import annotations
 
import logging
from typing import TYPE_CHECKING, Any
 
import aiohttp
from uasiren.client import Client
import voluptuous as vol
 
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_NAME, CONF_REGION
from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
from .const import DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
 
class UkraineAlarmConfigFlow(ConfigFlow, domain=DOMAIN):
    """Config flow for Ukraine Alarm."""
 
    VERSION = 1
 
    def __init__(self) -> None:
        """Initialize a new UkraineAlarmConfigFlow."""
        self.statesstates: list[dict[str, Any]] | None = None
        self.selected_regionselected_region: dict[str, Any] | None = None
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle a flow initialized by the user."""
 
        if len(self._async_current_entries_async_current_entries()) == 5:
            return self.async_abortasync_abortasync_abort(reason="max_regions")
 
        if not self.statesstates:
            websession = async_get_clientsession(self.hass)
            reason = None
            unknown_err_msg = None
            try:
                regions = await Client(websession).get_regions()
            except aiohttp.ClientResponseError as ex:
                if ex.status == 429:
                    reason = "rate_limit"
                else:
                    reason = "unknown"
                    unknown_err_msg = str(ex)
            except aiohttp.ClientConnectionError:
                reason = "cannot_connect"
            except aiohttp.ClientError as ex:
                reason = "unknown"
                unknown_err_msg = str(ex)
            except TimeoutError:
                reason = "timeout"
 
            if not reason and not regions:
                reason = "unknown"
                unknown_err_msg = "no regions returned"
 
            if unknown_err_msg:
                _LOGGER.error("Failed to connect to the service: %s", unknown_err_msg)
 
            if reason:
                return self.async_abortasync_abortasync_abort(reason=reason)
            self.statesstates = regions["states"]
 
        return await self._handle_pick_region_handle_pick_region("user", "district", user_input)
 
    async def async_step_district(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Handle user-chosen district."""
        return await self._handle_pick_region_handle_pick_region("district", "community", user_input)
 
    async def async_step_community(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Handle user-chosen community."""
        return await self._handle_pick_region_handle_pick_region("community", None, user_input, True)
 
    async def _handle_pick_region(
        self,
        step_id: str,
        next_step: str | None,
        user_input: dict[str, str] | None,
        last_step: bool = False,
    ) -> ConfigFlowResult:
        """Handle picking a (sub)region."""
        if self.selected_regionselected_region:
            source = self.selected_regionselected_region["regionChildIds"]
        else:
            source = self.statesstates
 
        if user_input is not None:
            # Only offer to browse subchildren if picked region wasn't the previously picked one
            if (
                not self.selected_regionselected_region
                or user_input[CONF_REGION] != self.selected_regionselected_region["regionId"]
            ):
                self.selected_regionselected_region = _find(source, user_input[CONF_REGION])
 
                if (
                    next_step
                    and self.selected_regionselected_region
                    and self.selected_regionselected_region["regionChildIds"]
                ):
                    return await getattr(self, f"async_step_{next_step}")()
 
            return await self._async_finish_flow_async_finish_flow()
 
        regions = {}
        if self.selected_regionselected_region:
            regions[self.selected_regionselected_region["regionId"]] = self.selected_regionselected_region[
                "regionName"
            ]
 
        regions.update(_make_regions_object(source))
 
        schema = vol.Schema(
            {
                vol.Required(CONF_REGION): vol.In(regions),
            }
        )
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id=step_id, data_schema=schema, last_step=last_step
        )
 
    async def _async_finish_flow(self) -> ConfigFlowResult:
        """Finish the setup."""
        if TYPE_CHECKING:
            assert self.selected_regionselected_region is not None
        await self.async_set_unique_idasync_set_unique_id(self.selected_regionselected_region["regionId"])
        self._abort_if_unique_id_configured_abort_if_unique_id_configured()
 
        return self.async_create_entryasync_create_entryasync_create_entry(
            title=self.selected_regionselected_region["regionName"],
            data={
                CONF_REGION: self.selected_regionselected_region["regionId"],
                CONF_NAME: self.selected_regionselected_region["regionName"],
            },
        )
 
 
def _find(regions: list[dict[str, Any]], region_id):
    return next((region for region in regions if region["regionId"] == region_id), None)
 
 
def _make_regions_object(regions: list[dict[str, Any]]) -> dict[str, str]:
    regions = sorted(regions, key=lambda region: region["regionName"].lower())
    return {region["regionId"]: region["regionName"] for region in regions}