-
-
Notifications
You must be signed in to change notification settings - Fork 29.9k
/
config_flow.py
165 lines (127 loc) · 4.43 KB
/
config_flow.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
"""Config flow for Monoprice 6-Zone Amplifier integration."""
from __future__ import annotations
import logging
from typing import Any
from pymonoprice import get_monoprice
from serial import SerialException
import voluptuous as vol
from homeassistant.config_entries import (
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
OptionsFlow,
)
from homeassistant.const import CONF_PORT
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.typing import VolDictType
from .const import (
CONF_SOURCE_1,
CONF_SOURCE_2,
CONF_SOURCE_3,
CONF_SOURCE_4,
CONF_SOURCE_5,
CONF_SOURCE_6,
CONF_SOURCES,
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
SOURCES = [
CONF_SOURCE_1,
CONF_SOURCE_2,
CONF_SOURCE_3,
CONF_SOURCE_4,
CONF_SOURCE_5,
CONF_SOURCE_6,
]
OPTIONS_FOR_DATA: VolDictType = {vol.Optional(source): str for source in SOURCES}
DATA_SCHEMA = vol.Schema({vol.Required(CONF_PORT): str, **OPTIONS_FOR_DATA})
@callback
def _sources_from_config(data):
sources_config = {
str(idx + 1): data.get(source) for idx, source in enumerate(SOURCES)
}
return {
index: name.strip()
for index, name in sources_config.items()
if (name is not None and name.strip() != "")
}
async def validate_input(hass: HomeAssistant, data):
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user.
"""
try:
await hass.async_add_executor_job(get_monoprice, data[CONF_PORT])
except SerialException as err:
_LOGGER.error("Error connecting to Monoprice controller")
raise CannotConnect from err
sources = _sources_from_config(data)
# Return info that you want to store in the config entry.
return {CONF_PORT: data[CONF_PORT], CONF_SOURCES: sources}
class MonoPriceConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Monoprice 6-Zone Amplifier."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors = {}
if user_input is not None:
try:
info = await validate_input(self.hass, user_input)
return self.async_create_entry(title=user_input[CONF_PORT], data=info)
except CannotConnect:
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
return self.async_show_form(
step_id="user", data_schema=DATA_SCHEMA, errors=errors
)
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
) -> MonopriceOptionsFlowHandler:
"""Define the config flow to handle options."""
return MonopriceOptionsFlowHandler(config_entry)
@callback
def _key_for_source(index, source, previous_sources):
if str(index) in previous_sources:
key = vol.Optional(
source, description={"suggested_value": previous_sources[str(index)]}
)
else:
key = vol.Optional(source)
return key
class MonopriceOptionsFlowHandler(OptionsFlow):
"""Handle a Monoprice options flow."""
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize."""
self.config_entry = config_entry
@callback
def _previous_sources(self):
if CONF_SOURCES in self.config_entry.options:
previous = self.config_entry.options[CONF_SOURCES]
else:
previous = self.config_entry.data[CONF_SOURCES]
return previous
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(
title="", data={CONF_SOURCES: _sources_from_config(user_input)}
)
previous_sources = self._previous_sources()
options = {
_key_for_source(idx + 1, source, previous_sources): str
for idx, source in enumerate(SOURCES)
}
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(options),
)
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect."""