-
-
Notifications
You must be signed in to change notification settings - Fork 31.5k
/
config_flow.py
188 lines (164 loc) · 6.74 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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
"""Config flow to configure the Bluetooth integration."""
from __future__ import annotations
import platform
from typing import Any, cast
from bluetooth_adapters import (
ADAPTER_ADDRESS,
ADAPTER_MANUFACTURER,
DEFAULT_ADDRESS,
AdapterDetails,
adapter_human_name,
adapter_model,
get_adapters,
)
from habluetooth import get_manager
import voluptuous as vol
from homeassistant.components import onboarding
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult
from homeassistant.core import callback
from homeassistant.helpers.schema_config_entry_flow import (
SchemaFlowFormStep,
SchemaOptionsFlowHandler,
)
from homeassistant.helpers.typing import DiscoveryInfoType
from .const import CONF_ADAPTER, CONF_DETAILS, CONF_PASSIVE, DOMAIN
from .util import adapter_title
OPTIONS_SCHEMA = vol.Schema(
{
vol.Required(CONF_PASSIVE, default=False): bool,
}
)
OPTIONS_FLOW = {
"init": SchemaFlowFormStep(OPTIONS_SCHEMA),
}
def adapter_display_info(adapter: str, details: AdapterDetails) -> str:
"""Return the adapter display info."""
name = adapter_human_name(adapter, details[ADAPTER_ADDRESS])
model = adapter_model(details)
manufacturer = details[ADAPTER_MANUFACTURER] or "Unknown"
return f"{name} {manufacturer} {model}"
class BluetoothConfigFlow(ConfigFlow, domain=DOMAIN):
"""Config flow for Bluetooth."""
VERSION = 1
def __init__(self) -> None:
"""Initialize the config flow."""
self._adapter: str | None = None
self._details: AdapterDetails | None = None
self._adapters: dict[str, AdapterDetails] = {}
self._placeholders: dict[str, str] = {}
async def async_step_integration_discovery(
self, discovery_info: DiscoveryInfoType
) -> ConfigFlowResult:
"""Handle a flow initialized by discovery."""
self._adapter = cast(str, discovery_info[CONF_ADAPTER])
self._details = cast(AdapterDetails, discovery_info[CONF_DETAILS])
await self.async_set_unique_id(self._details[ADAPTER_ADDRESS])
self._abort_if_unique_id_configured()
details = self._details
self._async_set_adapter_info(self._adapter, details)
return await self.async_step_single_adapter()
@callback
def _async_set_adapter_info(self, adapter: str, details: AdapterDetails) -> None:
"""Set the adapter info."""
name = adapter_human_name(adapter, details[ADAPTER_ADDRESS])
model = adapter_model(details)
manufacturer = details[ADAPTER_MANUFACTURER]
self._placeholders = {
"name": name,
"model": model,
"manufacturer": manufacturer or "Unknown",
}
self.context["title_placeholders"] = self._placeholders
async def async_step_single_adapter(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Select an adapter."""
adapter = self._adapter
details = self._details
assert adapter is not None
assert details is not None
assert self._placeholders is not None
address = details[ADAPTER_ADDRESS]
if user_input is not None or not onboarding.async_is_onboarded(self.hass):
await self.async_set_unique_id(address, raise_on_progress=False)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=adapter_title(adapter, details), data={}
)
return self.async_show_form(
step_id="single_adapter",
description_placeholders=self._placeholders,
)
async def async_step_multiple_adapters(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
if user_input is not None:
assert self._adapters is not None
adapter = user_input[CONF_ADAPTER]
details = self._adapters[adapter]
address = details[ADAPTER_ADDRESS]
await self.async_set_unique_id(address, raise_on_progress=False)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=adapter_title(adapter, details), data={}
)
configured_addresses = self._async_current_ids()
bluetooth_adapters = get_adapters()
await bluetooth_adapters.refresh()
self._adapters = bluetooth_adapters.adapters
system = platform.system()
unconfigured_adapters = [
adapter
for adapter, details in self._adapters.items()
if details[ADAPTER_ADDRESS] not in configured_addresses
# DEFAULT_ADDRESS is perfectly valid on MacOS but on
# Linux it means the adapter is not yet configured
# or crashed
and not (system == "Linux" and details[ADAPTER_ADDRESS] == DEFAULT_ADDRESS)
]
if not unconfigured_adapters:
ignored_adapters = len(
self._async_current_entries(include_ignore=True)
) - len(self._async_current_entries(include_ignore=False))
return self.async_abort(
reason="no_adapters",
description_placeholders={"ignored_adapters": str(ignored_adapters)},
)
if len(unconfigured_adapters) == 1:
self._adapter = list(self._adapters)[0]
self._details = self._adapters[self._adapter]
self._async_set_adapter_info(self._adapter, self._details)
return await self.async_step_single_adapter()
return self.async_show_form(
step_id="multiple_adapters",
data_schema=vol.Schema(
{
vol.Required(CONF_ADAPTER): vol.In(
{
adapter: adapter_display_info(
adapter, self._adapters[adapter]
)
for adapter in sorted(unconfigured_adapters)
}
),
}
),
)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
return await self.async_step_multiple_adapters()
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
) -> SchemaOptionsFlowHandler:
"""Get the options flow for this handler."""
return SchemaOptionsFlowHandler(config_entry, OPTIONS_FLOW)
@classmethod
@callback
def async_supports_options_flow(cls, config_entry: ConfigEntry) -> bool:
"""Return options flow support for this handler."""
return bool((manager := get_manager()) and manager.supports_passive_scan)