Home Assistant Unofficial Reference 2024.12.1
config_flow.py
Go to the documentation of this file.
1 """Config flow for Flexit Nordic (BACnet) integration."""
2 
3 from __future__ import annotations
4 
5 import asyncio.exceptions
6 import logging
7 from typing import Any
8 
9 from flexit_bacnet import FlexitBACnet
10 from flexit_bacnet.bacnet import DecodingError
11 import voluptuous as vol
12 
13 from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
14 from homeassistant.const import CONF_DEVICE_ID, CONF_IP_ADDRESS
15 
16 from .const import DOMAIN
17 
18 _LOGGER = logging.getLogger(__name__)
19 
20 DEFAULT_DEVICE_ID = 2
21 
22 STEP_USER_DATA_SCHEMA = vol.Schema(
23  {
24  vol.Required(CONF_IP_ADDRESS): str,
25  vol.Required(CONF_DEVICE_ID, default=DEFAULT_DEVICE_ID): int,
26  }
27 )
28 
29 
30 class FlexitBacnetConfigFlow(ConfigFlow, domain=DOMAIN):
31  """Handle a config flow for Flexit Nordic (BACnet)."""
32 
33  VERSION = 1
34 
35  async def async_step_user(
36  self, user_input: dict[str, Any] | None = None
37  ) -> ConfigFlowResult:
38  """Handle the initial step."""
39  errors: dict[str, str] = {}
40 
41  if user_input is not None:
42  device = FlexitBACnet(
43  user_input[CONF_IP_ADDRESS], user_input[CONF_DEVICE_ID]
44  )
45  try:
46  await device.update()
47  except (asyncio.exceptions.TimeoutError, ConnectionError, DecodingError):
48  errors["base"] = "cannot_connect"
49  except Exception:
50  _LOGGER.exception("Unexpected exception")
51  errors["base"] = "unknown"
52  else:
53  await self.async_set_unique_idasync_set_unique_id(device.serial_number)
54  self._abort_if_unique_id_configured_abort_if_unique_id_configured()
55 
56  return self.async_create_entryasync_create_entryasync_create_entry(
57  title=device.device_name, data=user_input
58  )
59 
60  return self.async_show_formasync_show_formasync_show_form(
61  step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
62  )
ConfigFlowResult async_step_user(self, dict[str, Any]|None user_input=None)
Definition: config_flow.py:37
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)