evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add typing to deCONZ init and config flow (#59999)

Browse source
This commit is contained in:
Robert Svensson 2021-12-01 18:59:52 +01:00 • committed by GitHub
parent a053c0a106
commit 8ddfa424c0
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 106 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

84
homeassistant/components/deconz/config_flow.py
View file

@ -1,6 +1,10 @@
"""Config flow to configure deCONZ component."""
from __future__ import annotations
import asyncio
from pprint import pformat
from typing import Any, cast
from urllib.parse import urlparse
import async_timeout
@ -15,8 +19,11 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import ssdp
from homeassistant.components.deconz.gateway import DeconzGateway
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
from homeassistant.const import CONF_API_KEY, CONF_HOST, CONF_PORT
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import aiohttp_client
from .const import (
@ -36,33 +43,36 @@ CONF_MANUAL_INPUT = "Manually define gateway"
@callback
def get_master_gateway(hass):
def get_master_gateway(hass: HomeAssistant) -> DeconzGateway:
"""Return the gateway which is marked as master."""
for gateway in hass.data[DOMAIN].values():
if gateway.master:
return gateway
return cast(DeconzGateway, gateway)
raise ValueError
class DeconzFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
class DeconzFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a deCONZ config flow."""
VERSION = 1
_hassio_discovery = None
_hassio_discovery: dict[str, Any]
@staticmethod
@callback
def async_get_options_flow(config_entry):
def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
"""Get the options flow for this handler."""
return DeconzOptionsFlowHandler(config_entry)
def __init__(self):
def __init__(self) -> None:
"""Initialize the deCONZ config flow."""
self.bridge_id = None
self.bridges = []
self.deconz_config = {}
self.bridge_id = ""
self.bridges: list[dict[str, int | str]] = []
self.deconz_config: dict[str, int | str] = {}
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a deCONZ config flow start.
Let user choose between discovered bridges and manual configuration.
@ -75,7 +85,7 @@ class DeconzFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
for bridge in self.bridges:
if bridge[CONF_HOST] == user_input[CONF_HOST]:
self.bridge_id = bridge[CONF_BRIDGE_ID]
self.bridge_id = cast(str, bridge[CONF_BRIDGE_ID])
self.deconz_config = {
CONF_HOST: bridge[CONF_HOST],
CONF_PORT: bridge[CONF_PORT],
@ -108,7 +118,9 @@ class DeconzFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_manual_input()
async def async_step_manual_input(self, user_input=None):
async def async_step_manual_input(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manual configuration."""
if user_input:
self.deconz_config = user_input
@ -124,9 +136,11 @@ class DeconzFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
),
)
async def async_step_link(self, user_input=None):
async def async_step_link(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Attempt to link with the deCONZ bridge."""
errors = {}
errors: dict[str, str] = {}
LOGGER.debug(
"Preparing linking with deCONZ gateway %s", pformat(self.deconz_config)
@ -153,7 +167,7 @@ class DeconzFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_show_form(step_id="link", errors=errors)
async def _create_entry(self):
async def _create_entry(self) -> FlowResult:
"""Create entry for gateway."""
if not self.bridge_id:
session = aiohttp_client.async_get_clientsession(self.hass)
@ -178,7 +192,7 @@ class DeconzFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_create_entry(title=self.bridge_id, data=self.deconz_config)
async def async_step_reauth(self, config: dict):
async def async_step_reauth(self, config: dict[str, Any]) -> FlowResult:
"""Trigger a reauthentication flow."""
self.context["title_placeholders"] = {CONF_HOST: config[CONF_HOST]}
@ -189,7 +203,7 @@ class DeconzFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_link()
async def async_step_ssdp(self, discovery_info):
async def async_step_ssdp(self, discovery_info: dict[str, str]) -> FlowResult:
"""Handle a discovered deCONZ bridge."""
if (
discovery_info.get(ssdp.ATTR_UPNP_MANUFACTURER_URL)
@ -206,20 +220,20 @@ class DeconzFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
if entry and entry.source == config_entries.SOURCE_HASSIO:
return self.async_abort(reason="already_configured")
hostname = cast(str, parsed_url.hostname)
port = cast(int, parsed_url.port)
self._abort_if_unique_id_configured(
updates={CONF_HOST: parsed_url.hostname, CONF_PORT: parsed_url.port}
updates={CONF_HOST: hostname, CONF_PORT: port}
)
self.context["title_placeholders"] = {"host": parsed_url.hostname}
self.context["title_placeholders"] = {"host": hostname}
self.deconz_config = {
CONF_HOST: parsed_url.hostname,
CONF_PORT: parsed_url.port,
}
self.deconz_config = {CONF_HOST: hostname, CONF_PORT: port}
return await self.async_step_link()
async def async_step_hassio(self, discovery_info):
async def async_step_hassio(self, discovery_info: dict[str, Any]) -> FlowResult:
"""Prepare configuration for a Hass.io deCONZ bridge.
This flow is triggered by the discovery component.
@ -241,8 +255,11 @@ class DeconzFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_hassio_confirm()
async def async_step_hassio_confirm(self, user_input=None):
async def async_step_hassio_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm a Hass.io discovery."""
if user_input is not None:
self.deconz_config = {
CONF_HOST: self._hassio_discovery[CONF_HOST],
@ -258,21 +275,26 @@ class DeconzFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
)
class DeconzOptionsFlowHandler(config_entries.OptionsFlow):
class DeconzOptionsFlowHandler(OptionsFlow):
"""Handle deCONZ options."""
def __init__(self, config_entry):
gateway: DeconzGateway
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize deCONZ options flow."""
self.config_entry = config_entry
self.options = dict(config_entry.options)
self.gateway = None
async def async_step_init(self, user_input=None):
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the deCONZ options."""
self.gateway = get_gateway_from_config_entry(self.hass, self.config_entry)
return await self.async_step_deconz_devices()
async def async_step_deconz_devices(self, user_input=None):
async def async_step_deconz_devices(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the deconz devices options."""
if user_input is not None:
self.options.update(user_input)