class MyConfigFlow(ConfigFlow, domain=DOMAIN):
"""My config flow."""
def __init__(self) -> None:
"""Initialize the config flow."""
self.data: dict[str, Any] = {}
async def async_step_zeroconf(
self, discovery_info: zeroconf.ZeroconfServiceInfo
) -> ConfigFlowResult:
"""Handle zeroconf discovery."""
self.data[CONF_HOST] = host = discovery_info.host
await self.async_set_unique_id(discovery_info.properties["serialno"])
self._abort_if_unique_id_configured(updates={CONF_HOST: host})
client = MyClient(host)
try:
await client.get_data()
except MyClientError:
return self.async_abort(reason="cannot_connect")
return await self.async_step_discovery_confirm()
async def async_step_discovery_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm discovery."""
if user_input is not None:
return self.async_create_entry(
title="MyIntegration",
data={CONF_HOST: self.data[CONF_HOST]},
)
self._set_confirm_only()
return self.async_show_form(step_id="discovery_confirm")
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
errors: dict[str, str] = {}
if user_input:
client = MyClient(user_input[CONF_HOST])
try:
serial_number = await client.check_connection()
except MyException as exception:
errors["base"] = "cannot_connect"
else:
await self.async_set_unique_id(
serial_number, raise_on_progress=False
)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title="MyIntegration",
data=user_input,
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_HOST): TextSelector(),
}
),
errors=errors,
)
