-
-
Notifications
You must be signed in to change notification settings - Fork 29k
/
config_flow.py
147 lines (115 loc) · 4.98 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
"""Config flow for Gardena Bluetooth integration."""
from __future__ import annotations
import logging
from typing import Any
from gardena_bluetooth.client import Client
from gardena_bluetooth.const import PRODUCT_NAMES, DeviceInformation, ScanService
from gardena_bluetooth.exceptions import CharacteristicNotFound, CommunicationFailure
from gardena_bluetooth.parse import ManufacturerData, ProductType
import voluptuous as vol
from homeassistant.components.bluetooth import (
BluetoothServiceInfo,
async_discovered_service_info,
)
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_ADDRESS
from homeassistant.data_entry_flow import AbortFlow
from . import get_connection
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
def _is_supported(discovery_info: BluetoothServiceInfo):
"""Check if device is supported."""
if ScanService not in discovery_info.service_uuids:
return False
if not (data := discovery_info.manufacturer_data.get(ManufacturerData.company)):
_LOGGER.debug("Missing manufacturer data: %s", discovery_info)
return False
manufacturer_data = ManufacturerData.decode(data)
product_type = ProductType.from_manufacturer_data(manufacturer_data)
if product_type not in (
ProductType.PUMP,
ProductType.VALVE,
ProductType.WATER_COMPUTER,
):
_LOGGER.debug("Unsupported device: %s", manufacturer_data)
return False
return True
def _get_name(discovery_info: BluetoothServiceInfo):
data = discovery_info.manufacturer_data[ManufacturerData.company]
manufacturer_data = ManufacturerData.decode(data)
product_type = ProductType.from_manufacturer_data(manufacturer_data)
return PRODUCT_NAMES.get(product_type, "Gardena Device")
class GardenaBluetoothConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Gardena Bluetooth."""
VERSION = 1
def __init__(self) -> None:
"""Initialize the config flow."""
self.devices: dict[str, str] = {}
self.address: str | None
async def async_read_data(self):
"""Try to connect to device and extract information."""
client = Client(get_connection(self.hass, self.address))
try:
model = await client.read_char(DeviceInformation.model_number)
_LOGGER.debug("Found device with model: %s", model)
except (CharacteristicNotFound, CommunicationFailure) as exception:
raise AbortFlow(
"cannot_connect", description_placeholders={"error": str(exception)}
) from exception
finally:
await client.disconnect()
return {CONF_ADDRESS: self.address}
async def async_step_bluetooth(
self, discovery_info: BluetoothServiceInfo
) -> ConfigFlowResult:
"""Handle the bluetooth discovery step."""
_LOGGER.debug("Discovered device: %s", discovery_info)
if not _is_supported(discovery_info):
return self.async_abort(reason="no_devices_found")
self.address = discovery_info.address
self.devices = {discovery_info.address: _get_name(discovery_info)}
await self.async_set_unique_id(self.address)
self._abort_if_unique_id_configured()
return await self.async_step_confirm()
async def async_step_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm discovery."""
assert self.address
title = self.devices[self.address]
if user_input is not None:
data = await self.async_read_data()
return self.async_create_entry(title=title, data=data)
self.context["title_placeholders"] = {
"name": title,
}
self._set_confirm_only()
return self.async_show_form(
step_id="confirm",
description_placeholders=self.context["title_placeholders"],
)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
if user_input is not None:
self.address = user_input[CONF_ADDRESS]
await self.async_set_unique_id(self.address, raise_on_progress=False)
self._abort_if_unique_id_configured()
return await self.async_step_confirm()
current_addresses = self._async_current_ids()
for discovery_info in async_discovered_service_info(self.hass):
address = discovery_info.address
if address in current_addresses or not _is_supported(discovery_info):
continue
self.devices[address] = _get_name(discovery_info)
if not self.devices:
return self.async_abort(reason="no_devices_found")
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_ADDRESS): vol.In(self.devices),
},
),
)