104 lines
3.8 KiB
Python
104 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
import asyncio
|
|
from aiohttp import ClientError, ClientResponseError
|
|
|
|
import voluptuous as vol
|
|
from homeassistant import config_entries
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.data_entry_flow import FlowResult
|
|
|
|
from .const import (
|
|
DOMAIN,
|
|
CONF_API_KEY,
|
|
CONF_NAME,
|
|
CONF_LATITUDE,
|
|
CONF_LONGITUDE,
|
|
CONF_LANGUAGE,
|
|
CONF_UPDATE_INTERVAL,
|
|
)
|
|
|
|
|
|
def _default_lat_lon(hass: HomeAssistant) -> tuple[float | None, float | None]:
|
|
return hass.config.latitude, hass.config.longitude
|
|
|
|
|
|
from .api import TwoGisWeatherClient
|
|
|
|
|
|
class TwoGisWeatherConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|
VERSION = 1
|
|
|
|
async def async_step_user(self, user_input: dict[str, Any] | None = None) -> FlowResult:
|
|
errors: dict[str, str] = {}
|
|
if user_input is not None:
|
|
api_key = user_input[CONF_API_KEY]
|
|
name = user_input.get(CONF_NAME) or "2GIS Weather"
|
|
lat = float(user_input.get(CONF_LATITUDE) or self.hass.config.latitude)
|
|
lon = float(user_input.get(CONF_LONGITUDE) or self.hass.config.longitude)
|
|
lang = user_input.get(CONF_LANGUAGE) or "en-US"
|
|
|
|
# Prevent duplicates for the same coordinates
|
|
await self.async_set_unique_id(f"{lat:.4f},{lon:.4f}")
|
|
self._abort_if_unique_id_configured()
|
|
|
|
client = TwoGisWeatherClient(self.hass, api_key, lat, lon, lang)
|
|
try:
|
|
await client.async_get_current()
|
|
except ClientResponseError as e: # HTTP error from server
|
|
errors["base"] = "invalid_auth" if e.status in (401, 403) else "cannot_connect"
|
|
except (asyncio.TimeoutError, ClientError):
|
|
errors["base"] = "cannot_connect"
|
|
except Exception: # noqa: BLE001
|
|
errors["base"] = "unknown"
|
|
else:
|
|
return self.async_create_entry(
|
|
title=name,
|
|
data={
|
|
CONF_API_KEY: api_key,
|
|
CONF_NAME: name,
|
|
CONF_LATITUDE: lat,
|
|
CONF_LONGITUDE: lon,
|
|
CONF_LANGUAGE: lang,
|
|
},
|
|
)
|
|
|
|
lat, lon = _default_lat_lon(self.hass)
|
|
schema = vol.Schema(
|
|
{
|
|
vol.Required(CONF_API_KEY): str,
|
|
vol.Optional(CONF_NAME, default="2GIS Weather"): str,
|
|
vol.Optional(CONF_LATITUDE, default=lat): float,
|
|
vol.Optional(CONF_LONGITUDE, default=lon): float,
|
|
vol.Optional(CONF_LANGUAGE, default="en-US"): str,
|
|
}
|
|
)
|
|
return self.async_show_form(step_id="user", data_schema=schema)
|
|
|
|
async def async_step_import(self, import_config: dict[str, Any]) -> FlowResult:
|
|
return await self.async_step_user(import_config)
|
|
|
|
|
|
class TwoGisWeatherOptionsFlowHandler(config_entries.OptionsFlow):
|
|
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
|
|
self.config_entry = config_entry
|
|
|
|
async def async_step_init(self, user_input: dict[str, Any] | None = None) -> FlowResult:
|
|
if user_input is not None:
|
|
return self.async_create_entry(title="Options", data=user_input)
|
|
|
|
current = {**self.config_entry.options}
|
|
schema = vol.Schema(
|
|
{
|
|
vol.Optional(CONF_UPDATE_INTERVAL, default=current.get(CONF_UPDATE_INTERVAL, 10)): int,
|
|
vol.Optional(CONF_LANGUAGE, default=current.get(CONF_LANGUAGE, "en-US")): str,
|
|
}
|
|
)
|
|
return self.async_show_form(step_id="init", data_schema=schema)
|
|
|
|
|
|
async def async_get_options_flow(config_entry: config_entries.ConfigEntry) -> TwoGisWeatherOptionsFlowHandler: # type: ignore[override]
|
|
return TwoGisWeatherOptionsFlowHandler(config_entry)
|
|
|
|
|