Home Assistant Unofficial Reference 2024.12.1
config_flow.py
Go to the documentation of this file.
1 """Config Flow for Advantage Air integration."""
2 
3 from __future__ import annotations
4 
5 from typing import Any
6 
7 from advantage_air import ApiError, advantage_air
8 import voluptuous as vol
9 
10 from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
11 from homeassistant.const import CONF_IP_ADDRESS, CONF_PORT
12 from homeassistant.helpers.aiohttp_client import async_get_clientsession
13 
14 from .const import ADVANTAGE_AIR_RETRY, DOMAIN
15 
16 ADVANTAGE_AIR_DEFAULT_PORT = 2025
17 
18 ADVANTAGE_AIR_SCHEMA = vol.Schema(
19  {
20  vol.Required(CONF_IP_ADDRESS): str,
21  vol.Optional(CONF_PORT, default=ADVANTAGE_AIR_DEFAULT_PORT): int,
22  }
23 )
24 
25 
26 class AdvantageAirConfigFlow(ConfigFlow, domain=DOMAIN):
27  """Config Advantage Air API connection."""
28 
29  VERSION = 1
30 
31  DOMAIN = DOMAIN
32 
33  async def async_step_user(
34  self, user_input: dict[str, Any] | None = None
35  ) -> ConfigFlowResult:
36  """Get configuration from the user."""
37  errors = {}
38  if user_input:
39  ip_address = user_input[CONF_IP_ADDRESS]
40  port = user_input[CONF_PORT]
41 
42  try:
43  data = await advantage_air(
44  ip_address,
45  port=port,
46  session=async_get_clientsession(self.hass),
47  retry=ADVANTAGE_AIR_RETRY,
48  ).async_get()
49  except ApiError:
50  errors["base"] = "cannot_connect"
51  else:
52  await self.async_set_unique_idasync_set_unique_id(data["system"]["rid"])
53  self._abort_if_unique_id_configured_abort_if_unique_id_configured()
54 
55  return self.async_create_entryasync_create_entryasync_create_entry(
56  title=data["system"]["name"],
57  data=user_input,
58  )
59 
60  return self.async_show_formasync_show_formasync_show_form(
61  step_id="user",
62  data_schema=ADVANTAGE_AIR_SCHEMA,
63  errors=errors,
64  )
ConfigFlowResult async_step_user(self, dict[str, Any]|None user_input=None)
Definition: config_flow.py:35
None _abort_if_unique_id_configured(self, dict[str, Any]|None updates=None, bool reload_on_update=True, *str error="already_configured")
ConfigEntry|None async_set_unique_id(self, str|None unique_id=None, *bool raise_on_progress=True)
ConfigFlowResult async_create_entry(self, *str title, Mapping[str, Any] data, str|None description=None, Mapping[str, str]|None description_placeholders=None, Mapping[str, Any]|None options=None)
ConfigFlowResult async_show_form(self, *str|None step_id=None, vol.Schema|None data_schema=None, dict[str, str]|None errors=None, Mapping[str, str]|None description_placeholders=None, bool|None last_step=None, str|None preview=None)
_FlowResultT async_show_form(self, *str|None step_id=None, vol.Schema|None data_schema=None, dict[str, str]|None errors=None, Mapping[str, str]|None description_placeholders=None, bool|None last_step=None, str|None preview=None)
_FlowResultT async_create_entry(self, *str|None title=None, Mapping[str, Any] data, str|None description=None, Mapping[str, str]|None description_placeholders=None)
aiohttp.ClientSession async_get_clientsession(HomeAssistant hass, bool verify_ssl=True, socket.AddressFamily family=socket.AF_UNSPEC, ssl_util.SSLCipherList ssl_cipher=ssl_util.SSLCipherList.PYTHON_DEFAULT)
AreaRegistry async_get(HomeAssistant hass)