721 lines
27 KiB
Python
721 lines
27 KiB
Python
"""Config flow fglair: облако / ручной ввод / импорт config_*.json + пробная
|
|
сессия (start → ONLINE ≤ 10 c → базовые свойства)."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from functools import partial
|
|
from typing import Any
|
|
|
|
import voluptuous as vol
|
|
from homeassistant.config_entries import (
|
|
ConfigEntry,
|
|
ConfigFlow,
|
|
ConfigFlowResult,
|
|
OptionsFlow,
|
|
)
|
|
from homeassistant.helpers import selector
|
|
|
|
import pyfglair
|
|
from pyfglair import Prop, Template
|
|
from pyfglair.provision import Device
|
|
from pyfglair.templates import detect
|
|
|
|
from .const import (
|
|
CONF_CONFIG_JSON,
|
|
CONF_DEVICE_PORT,
|
|
CONF_DSN,
|
|
CONF_EMAIL,
|
|
CONF_FEATURES,
|
|
CONF_HOST,
|
|
CONF_LANIP_KEY,
|
|
CONF_LANIP_KEY_ID,
|
|
CONF_LISTEN_PORT,
|
|
CONF_MODEL,
|
|
CONF_NAME,
|
|
CONF_OVERRIDES,
|
|
CONF_PASSWORD,
|
|
CONF_REGION,
|
|
CONF_TEMP_DEN,
|
|
CONF_TEMP_MAX,
|
|
CONF_TEMP_MIN,
|
|
CONF_TEMP_NUM,
|
|
CONF_TEMP_OFFSET,
|
|
CONF_TEMP_STEP,
|
|
CONF_TEMPLATE,
|
|
DEFAULT_DEVICE_PORT,
|
|
DEFAULT_LISTEN_PORT,
|
|
DOMAIN,
|
|
FAN_VALUE_TO_MODE,
|
|
FEATURES,
|
|
REGIONS,
|
|
TEMP_MAX_TENTHS,
|
|
TEMP_MIN_TENTHS,
|
|
TEMP_STEP_TENTHS,
|
|
TRIAL_ERRORS,
|
|
TRIAL_TIMEOUT,
|
|
)
|
|
from .features import resolve_features
|
|
from .overrides import ConversionSet, LinearOverride
|
|
from .trial import TrialResult, probe_template, trial_connect
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
def _password_selector() -> selector.TextSelector:
|
|
return selector.TextSelector(
|
|
selector.TextSelectorConfig(type=selector.TextSelectorType.PASSWORD)
|
|
)
|
|
|
|
|
|
def _cloud_schema() -> vol.Schema:
|
|
return vol.Schema(
|
|
{
|
|
vol.Required(CONF_EMAIL): str,
|
|
vol.Required(CONF_PASSWORD): _password_selector(),
|
|
vol.Required(CONF_REGION, default="eu"): vol.In(REGIONS),
|
|
}
|
|
)
|
|
|
|
|
|
def _manual_schema() -> vol.Schema:
|
|
return vol.Schema(
|
|
{
|
|
vol.Required(CONF_HOST): str,
|
|
vol.Optional(
|
|
CONF_DEVICE_PORT, default=DEFAULT_DEVICE_PORT
|
|
): vol.All(int, vol.Range(min=1, max=65535)),
|
|
vol.Required(CONF_DSN): str,
|
|
vol.Required(CONF_LANIP_KEY): _password_selector(),
|
|
vol.Required(CONF_LANIP_KEY_ID): vol.All(int, vol.Range(min=0)),
|
|
vol.Optional(CONF_NAME): str,
|
|
vol.Optional(CONF_MODEL): str,
|
|
}
|
|
)
|
|
|
|
|
|
def _import_schema() -> vol.Schema:
|
|
return vol.Schema(
|
|
{
|
|
vol.Required(CONF_CONFIG_JSON): selector.TextSelector(
|
|
selector.TextSelectorConfig(
|
|
type=selector.TextSelectorType.TEXT, multiline=True
|
|
)
|
|
)
|
|
}
|
|
)
|
|
|
|
|
|
def _spec_int(spec: dict[str, Any], key: str, default: int) -> int:
|
|
try:
|
|
return int(spec.get(key, default))
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def _template_schema(template: Template) -> vol.Schema:
|
|
return vol.Schema(
|
|
{
|
|
vol.Required(CONF_TEMPLATE, default=template.name): vol.In(
|
|
["A", "B", "F"]
|
|
)
|
|
}
|
|
)
|
|
|
|
|
|
def _capabilities_schema(defaults: dict[str, bool]) -> vol.Schema:
|
|
fields = {}
|
|
for key in FEATURES:
|
|
fields[vol.Optional(key, default=bool(defaults.get(key)))] = bool
|
|
return vol.Schema(fields)
|
|
|
|
|
|
def _limits_schema(data: dict[str, Any]) -> vol.Schema:
|
|
template = Template[str(data.get(CONF_TEMPLATE, "A"))]
|
|
overrides = data.get(CONF_OVERRIDES)
|
|
spec = (
|
|
overrides.get(Prop.ADJUST_TEMPERATURE.name.lower(), {})
|
|
if isinstance(overrides, dict)
|
|
else {}
|
|
)
|
|
if not isinstance(spec, dict):
|
|
spec = {}
|
|
temp_min = _spec_int(spec, "min", TEMP_MIN_TENTHS)
|
|
temp_max = _spec_int(spec, "max", TEMP_MAX_TENTHS)
|
|
try:
|
|
step = int(data.get(CONF_TEMP_STEP) or TEMP_STEP_TENTHS[template])
|
|
except (TypeError, ValueError):
|
|
step = TEMP_STEP_TENTHS[template]
|
|
return vol.Schema(
|
|
{
|
|
vol.Required(CONF_TEMP_MIN, default=temp_min / 10): vol.All(
|
|
vol.Coerce(float), vol.Range(min=-10.0, max=45.0)
|
|
),
|
|
vol.Required(CONF_TEMP_MAX, default=temp_max / 10): vol.All(
|
|
vol.Coerce(float), vol.Range(min=-10.0, max=45.0)
|
|
),
|
|
vol.Required(CONF_TEMP_STEP, default=step / 10): vol.All(
|
|
vol.Coerce(float), vol.Range(min=0.1, max=5.0)
|
|
),
|
|
vol.Required(
|
|
CONF_TEMP_NUM, default=_spec_int(spec, "num", 1)
|
|
): vol.All(int, vol.Range(min=1, max=1000)),
|
|
vol.Required(
|
|
CONF_TEMP_DEN, default=_spec_int(spec, "den", 1)
|
|
): vol.All(int, vol.Range(min=1, max=1000)),
|
|
vol.Required(
|
|
CONF_TEMP_OFFSET, default=_spec_int(spec, "offset", 0)
|
|
): vol.All(int, vol.Range(min=-1000, max=1000)),
|
|
}
|
|
)
|
|
|
|
|
|
def _device_values_text(properties: dict[Any, Any]) -> str:
|
|
"""Что прочитано с устройства (человекочитаемо) для описаний шагов."""
|
|
|
|
def raw(prop) -> int | None:
|
|
value = properties.get(prop)
|
|
return value.int_value if value is not None else None
|
|
|
|
lines: list[str] = []
|
|
mode = raw(Prop.OPERATION_MODE)
|
|
if mode is not None:
|
|
names = {0: "off", 1: "on", 2: "auto", 3: "cool", 4: "dry",
|
|
5: "fan", 6: "heat"}
|
|
lines.append(f"- режим: {names.get(mode, '?')} ({mode})")
|
|
fan = raw(Prop.FAN_SPEED)
|
|
if fan is not None:
|
|
lines.append(f"- скорость: {FAN_VALUE_TO_MODE.get(fan, '?')} ({fan})")
|
|
current = raw(Prop.DISPLAY_TEMPERATURE)
|
|
if current is not None:
|
|
lines.append(f"- температура в помещении: {current / 10:.1f} °C")
|
|
caps = raw(Prop.DEVICE_CAPABILITIES)
|
|
if caps is not None:
|
|
lines.append(f"- device_capabilities: 0x{caps:08X}")
|
|
vnum = raw(Prop.AF_VERTICAL_NUM_DIR)
|
|
hnum = raw(Prop.AF_HORIZONTAL_NUM_DIR)
|
|
if vnum is not None or hnum is not None:
|
|
lines.append(f"- направлений заслонок: верт. {vnum or 0}, гориз. {hnum or 0}")
|
|
return "\n".join(lines) or "- данные не получены"
|
|
|
|
|
|
def _template_name(model: str | None) -> str:
|
|
if model:
|
|
found = detect(model)
|
|
if found is not None:
|
|
return found.name
|
|
return Template.A.name
|
|
|
|
|
|
def _manual_data(user_input: dict[str, Any]) -> dict[str, Any]:
|
|
model = str(user_input.get(CONF_MODEL) or "").strip()
|
|
return {
|
|
CONF_HOST: str(user_input[CONF_HOST]).strip(),
|
|
CONF_DEVICE_PORT: int(
|
|
user_input.get(CONF_DEVICE_PORT, DEFAULT_DEVICE_PORT)
|
|
),
|
|
CONF_DSN: str(user_input[CONF_DSN]).strip(),
|
|
CONF_LANIP_KEY: str(user_input[CONF_LANIP_KEY]).strip(),
|
|
CONF_LANIP_KEY_ID: int(user_input[CONF_LANIP_KEY_ID]),
|
|
CONF_NAME: str(user_input.get(CONF_NAME) or "").strip(),
|
|
CONF_MODEL: model,
|
|
CONF_TEMPLATE: _template_name(model),
|
|
}
|
|
|
|
|
|
def _parse_import(source: str | dict[str, Any]) -> dict[str, Any] | None:
|
|
"""config_*.json (fglair-discover/fglctl) → данные записи."""
|
|
if isinstance(source, str):
|
|
try:
|
|
obj = json.loads(source)
|
|
except ValueError:
|
|
return None
|
|
else:
|
|
obj = source
|
|
if not isinstance(obj, dict):
|
|
return None
|
|
host = obj.get("ip_address") or obj.get("host")
|
|
dsn = obj.get("dsn")
|
|
lanip_key = obj.get("lanip_key")
|
|
lanip_key_id = obj.get("lanip_key_id")
|
|
if not host or not dsn or not lanip_key or lanip_key_id is None:
|
|
return None
|
|
|
|
def to_int(value: Any) -> int | None:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
key_id = to_int(lanip_key_id)
|
|
if key_id is None or key_id < 0:
|
|
return None
|
|
port = to_int(obj.get("device_port", DEFAULT_DEVICE_PORT))
|
|
if port is None or not 1 <= port <= 65535:
|
|
port = DEFAULT_DEVICE_PORT
|
|
model = str(obj.get("model") or "").strip()
|
|
return {
|
|
CONF_HOST: str(host).strip(),
|
|
CONF_DEVICE_PORT: port,
|
|
CONF_DSN: str(dsn).strip(),
|
|
CONF_LANIP_KEY: str(lanip_key).strip(),
|
|
CONF_LANIP_KEY_ID: key_id,
|
|
CONF_NAME: str(obj.get("name") or "").strip(),
|
|
CONF_MODEL: model,
|
|
CONF_TEMPLATE: _template_name(model),
|
|
}
|
|
|
|
|
|
def _run_trial(data: dict[str, Any]) -> tuple[TrialResult, Template, bool]:
|
|
"""Пробная сессия; для неизвестной модели — предварительный пробинг A/B/F.
|
|
|
|
Возвращает (результат, шаблон, был_ли_пробинг).
|
|
"""
|
|
template = Template[str(data.get(CONF_TEMPLATE, "A"))]
|
|
model = str(data.get(CONF_MODEL) or "")
|
|
probed = False
|
|
if not model or detect(model) is None:
|
|
detected, _probe_props, fatal = probe_template(
|
|
data[CONF_HOST],
|
|
data[CONF_DSN],
|
|
data[CONF_LANIP_KEY],
|
|
int(data[CONF_LANIP_KEY_ID]),
|
|
device_port=int(data.get(CONF_DEVICE_PORT, DEFAULT_DEVICE_PORT)),
|
|
listen_port=int(data.get(CONF_LISTEN_PORT, DEFAULT_LISTEN_PORT)),
|
|
timeout=TRIAL_TIMEOUT,
|
|
)
|
|
probed = True
|
|
if fatal is not None:
|
|
# Модуль недоступен/нет слотов/ключ — вторую сессию не открываем.
|
|
return TrialResult(ok=False, reason=fatal), template, True
|
|
if detected is not None:
|
|
template = detected
|
|
result = trial_connect(
|
|
host=data[CONF_HOST],
|
|
dsn=data[CONF_DSN],
|
|
lanip_key=data[CONF_LANIP_KEY],
|
|
lanip_key_id=int(data[CONF_LANIP_KEY_ID]),
|
|
device_port=int(data.get(CONF_DEVICE_PORT, DEFAULT_DEVICE_PORT)),
|
|
template=template,
|
|
listen_port=int(data.get(CONF_LISTEN_PORT, DEFAULT_LISTEN_PORT)),
|
|
timeout=TRIAL_TIMEOUT,
|
|
)
|
|
return result, template, probed
|
|
|
|
|
|
class FglairConfigFlow(ConfigFlow, domain=DOMAIN):
|
|
"""Настройка интеграции: облако, ручной ввод или импорт."""
|
|
|
|
VERSION = 1
|
|
|
|
def __init__(self) -> None:
|
|
self._devices: list[Device] = []
|
|
self._pending_data: dict[str, Any] = {}
|
|
self._trial_properties: dict[Any, Any] = {}
|
|
self._trial_answered: set[Prop] = set()
|
|
self._resolved_template: Template = Template.A
|
|
self._probed: bool = False
|
|
self._selected_features: dict[str, bool] = {}
|
|
self._reconfigure_entry: ConfigEntry | None = None
|
|
|
|
async def async_step_reconfigure(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
"""Повторная настройка (в т.ч. смена LAN-ключа после key_error)."""
|
|
self._reconfigure_entry = self._get_reconfigure_entry()
|
|
return await self.async_step_user()
|
|
|
|
def _reconfigure_suggestions(self) -> dict[str, Any] | None:
|
|
entry = self._reconfigure_entry
|
|
if entry is None:
|
|
return None
|
|
fields = (
|
|
CONF_HOST,
|
|
CONF_DEVICE_PORT,
|
|
CONF_DSN,
|
|
CONF_LANIP_KEY,
|
|
CONF_LANIP_KEY_ID,
|
|
CONF_NAME,
|
|
CONF_MODEL,
|
|
)
|
|
return {key: entry.data[key] for key in fields if key in entry.data}
|
|
|
|
def _show_form(
|
|
self,
|
|
step_id: str,
|
|
schema: vol.Schema,
|
|
errors: dict[str, str],
|
|
user_input: dict[str, Any] | None = None,
|
|
*,
|
|
exclude: tuple[str, ...] = (),
|
|
) -> ConfigFlowResult:
|
|
"""Показывает форму, возвращая введённые значения (кроме секретов)."""
|
|
if user_input:
|
|
suggested = {
|
|
key: value
|
|
for key, value in user_input.items()
|
|
if key not in exclude
|
|
}
|
|
schema = self.add_suggested_values_to_schema(schema, suggested)
|
|
return self.async_show_form(
|
|
step_id=step_id, data_schema=schema, errors=errors
|
|
)
|
|
|
|
async def async_step_user(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
return self.async_show_menu(
|
|
step_id="user", menu_options=["cloud", "manual", "import_json"]
|
|
)
|
|
|
|
async def async_step_cloud(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
errors: dict[str, str] = {}
|
|
if user_input is not None:
|
|
try:
|
|
self._devices = await pyfglair.discover(
|
|
user_input[CONF_EMAIL],
|
|
user_input[CONF_PASSWORD],
|
|
user_input[CONF_REGION],
|
|
)
|
|
except pyfglair.ProvisionError as err:
|
|
_LOGGER.debug("discover не удался: %s", err)
|
|
errors["base"] = {
|
|
"auth": "invalid_auth",
|
|
"no_devices": "no_devices",
|
|
}.get(err.kind, "cannot_connect")
|
|
except Exception: # pragma: no cover - защитный путь
|
|
_LOGGER.exception("discover: неожиданная ошибка")
|
|
errors["base"] = "cannot_connect"
|
|
else:
|
|
if not self._devices:
|
|
errors["base"] = "no_devices"
|
|
else:
|
|
return await self.async_step_cloud_device()
|
|
return self._show_form(
|
|
"cloud", _cloud_schema(), errors, user_input,
|
|
exclude=(CONF_PASSWORD,),
|
|
)
|
|
|
|
async def async_step_cloud_device(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
errors: dict[str, str] = {}
|
|
if user_input is not None:
|
|
device = next(
|
|
(d for d in self._devices if d.dsn == user_input[CONF_DSN]), None
|
|
)
|
|
if device is None:
|
|
errors["base"] = "unknown"
|
|
else:
|
|
data = {
|
|
CONF_HOST: device.lan_ip or device.dsn,
|
|
CONF_DEVICE_PORT: DEFAULT_DEVICE_PORT,
|
|
CONF_DSN: device.dsn,
|
|
CONF_LANIP_KEY: device.lanip_key,
|
|
CONF_LANIP_KEY_ID: device.lanip_key_id,
|
|
CONF_NAME: device.name,
|
|
CONF_MODEL: device.model,
|
|
CONF_TEMPLATE: _template_name(device.model),
|
|
CONF_REGION: device.region,
|
|
}
|
|
error = await self._async_trial(data)
|
|
if error is None:
|
|
return await self.async_step_template()
|
|
errors["base"] = error
|
|
schema = vol.Schema(
|
|
{
|
|
vol.Required(CONF_DSN): vol.In(
|
|
{
|
|
d.dsn: f"{d.name} ({d.model or '?'}, {d.lan_ip or '?'})"
|
|
for d in self._devices
|
|
}
|
|
)
|
|
}
|
|
)
|
|
return self._show_form("cloud_device", schema, errors, user_input)
|
|
|
|
async def async_step_manual(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
errors: dict[str, str] = {}
|
|
if user_input is not None:
|
|
data = _manual_data(user_input)
|
|
error = await self._async_trial(data)
|
|
if error is None:
|
|
return await self.async_step_template()
|
|
errors["base"] = error
|
|
if user_input is None and self._reconfigure_entry is not None:
|
|
user_input = self._reconfigure_suggestions()
|
|
return self._show_form(
|
|
"manual", _manual_schema(), errors, user_input,
|
|
exclude=(CONF_LANIP_KEY,),
|
|
)
|
|
|
|
async def async_step_import_json(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
errors: dict[str, str] = {}
|
|
if user_input is not None:
|
|
data = _parse_import(user_input[CONF_CONFIG_JSON])
|
|
if data is None:
|
|
errors["base"] = "invalid_import"
|
|
else:
|
|
error = await self._async_trial(data)
|
|
if error is None:
|
|
return await self.async_step_template()
|
|
errors["base"] = error
|
|
return self._show_form("import_json", _import_schema(), errors, user_input)
|
|
|
|
async def async_step_import(
|
|
self, import_data: dict[str, Any]
|
|
) -> ConfigFlowResult:
|
|
"""Импорт из configuration.yaml / discovery-конфига."""
|
|
data = _parse_import(import_data) if isinstance(import_data, dict) else None
|
|
if data is None:
|
|
return self.async_abort(reason="invalid_import")
|
|
error = await self._async_trial(data)
|
|
if error is not None:
|
|
return self.async_abort(reason=error)
|
|
entry_data = dict(self._pending_data)
|
|
entry_data[CONF_FEATURES] = resolve_features(
|
|
self._trial_properties, self._trial_answered
|
|
)
|
|
return self._create_entry(entry_data)
|
|
|
|
async def async_step_template(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
"""Страница 1: шаблон (таблица свойств модуля) с пояснениями."""
|
|
data = dict(self._pending_data)
|
|
if user_input is not None:
|
|
template = Template[str(user_input[CONF_TEMPLATE])]
|
|
data[CONF_TEMPLATE] = template.name
|
|
self._pending_data = data
|
|
return await self.async_step_capabilities()
|
|
default = self._resolved_template.name
|
|
if self._reconfigure_entry is not None:
|
|
default = str(data.get(CONF_TEMPLATE, default))
|
|
placeholders = {
|
|
"device_values": _device_values_text(self._trial_properties),
|
|
}
|
|
if self._probed:
|
|
placeholders["probe_note"] = (
|
|
"Модель модуля неизвестна — шаблон определён проверкой "
|
|
"свойств."
|
|
)
|
|
else:
|
|
placeholders["probe_note"] = (
|
|
"Шаблон определён по модели модуля."
|
|
)
|
|
return self.async_show_form(
|
|
step_id="template",
|
|
data_schema=self.add_suggested_values_to_schema(
|
|
_template_schema(Template[default]),
|
|
{CONF_TEMPLATE: default},
|
|
),
|
|
description_placeholders=placeholders,
|
|
)
|
|
|
|
async def async_step_capabilities(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
"""Страница 2: возможности прибора (тумблеры, дефолты с устройства)."""
|
|
if user_input is not None:
|
|
self._selected_features = {
|
|
key: bool(user_input.get(key, False)) for key in FEATURES
|
|
}
|
|
return await self.async_step_limits()
|
|
defaults = resolve_features(
|
|
self._trial_properties,
|
|
self._trial_answered,
|
|
self._reconfigure_defaults(),
|
|
)
|
|
schema = self.add_suggested_values_to_schema(
|
|
_capabilities_schema(defaults),
|
|
{key: value for key, value in defaults.items()},
|
|
)
|
|
return self.async_show_form(
|
|
step_id="capabilities",
|
|
data_schema=schema,
|
|
description_placeholders={
|
|
"device_values": _device_values_text(self._trial_properties),
|
|
},
|
|
)
|
|
|
|
async def async_step_limits(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
"""Страница 3: диапазон и шаг уставки (при необходимости — конверсия)."""
|
|
data = dict(self._pending_data)
|
|
errors: dict[str, str] = {}
|
|
if user_input is not None:
|
|
template = Template[str(data.get(CONF_TEMPLATE, "A"))]
|
|
temp_min = int(round(float(user_input[CONF_TEMP_MIN]) * 10))
|
|
temp_max = int(round(float(user_input[CONF_TEMP_MAX]) * 10))
|
|
step = int(round(float(user_input[CONF_TEMP_STEP]) * 10))
|
|
num = int(user_input[CONF_TEMP_NUM])
|
|
den = int(user_input[CONF_TEMP_DEN])
|
|
offset = int(user_input[CONF_TEMP_OFFSET])
|
|
conversions = ConversionSet(
|
|
template,
|
|
{
|
|
Prop.ADJUST_TEMPERATURE: LinearOverride(
|
|
num=num,
|
|
den=den,
|
|
offset=offset,
|
|
min_value=temp_min,
|
|
max_value=temp_max,
|
|
)
|
|
},
|
|
step,
|
|
)
|
|
bounds = conversions.range(Prop.ADJUST_TEMPERATURE)
|
|
if (
|
|
temp_min >= temp_max
|
|
or bounds is None
|
|
or bounds[0] >= bounds[1]
|
|
):
|
|
errors["base"] = "invalid_range"
|
|
else:
|
|
data[CONF_TEMP_STEP] = step
|
|
data[CONF_OVERRIDES] = {
|
|
Prop.ADJUST_TEMPERATURE.name.lower(): {
|
|
"num": num,
|
|
"den": den,
|
|
"offset": offset,
|
|
"min": temp_min,
|
|
"max": temp_max,
|
|
}
|
|
}
|
|
data[CONF_FEATURES] = dict(self._selected_features)
|
|
_LOGGER.debug(
|
|
"limits: template=%s range=%d..%d step=%d features=%d",
|
|
template.name, temp_min, temp_max, step,
|
|
len(self._selected_features),
|
|
)
|
|
return self._create_entry(data)
|
|
if user_input is None and self._reconfigure_entry is not None:
|
|
user_input = self._reconfigure_limits_suggestions()
|
|
return self._show_form(
|
|
"limits",
|
|
_limits_schema(data),
|
|
errors,
|
|
user_input,
|
|
)
|
|
|
|
def _reconfigure_defaults(self) -> dict[str, bool] | None:
|
|
entry = self._reconfigure_entry
|
|
if entry is None:
|
|
return None
|
|
features = entry.data.get(CONF_FEATURES)
|
|
return features if isinstance(features, dict) else None
|
|
|
|
def _reconfigure_limits_suggestions(self) -> dict[str, Any] | None:
|
|
entry = self._reconfigure_entry
|
|
if entry is None:
|
|
return None
|
|
spec = (entry.data.get(CONF_OVERRIDES) or {}).get(
|
|
Prop.ADJUST_TEMPERATURE.name.lower(), {}
|
|
)
|
|
if not isinstance(spec, dict):
|
|
return None
|
|
return {
|
|
CONF_TEMP_MIN: _spec_int(spec, "min", TEMP_MIN_TENTHS) / 10,
|
|
CONF_TEMP_MAX: _spec_int(spec, "max", TEMP_MAX_TENTHS) / 10,
|
|
CONF_TEMP_STEP: int(
|
|
entry.data.get(CONF_TEMP_STEP) or TEMP_STEP_TENTHS[
|
|
Template[str(entry.data.get(CONF_TEMPLATE, "A"))]
|
|
]
|
|
)
|
|
/ 10,
|
|
CONF_TEMP_NUM: _spec_int(spec, "num", 1),
|
|
CONF_TEMP_DEN: _spec_int(spec, "den", 1),
|
|
CONF_TEMP_OFFSET: _spec_int(spec, "offset", 0),
|
|
}
|
|
|
|
async def _async_trial(self, data: dict[str, Any]) -> str | None:
|
|
"""None — сессия поднялась; иначе ключ ошибки формы."""
|
|
if self._reconfigure_entry is not None:
|
|
# Сохраняем служебные поля записи (listen_port/keepalive_ms) и
|
|
# текущие конверсии, если путь их не задаёт.
|
|
for key, value in self._reconfigure_entry.data.items():
|
|
data.setdefault(key, value)
|
|
# Сохранённый шаблон важнее пересчёта по модели: пользователь
|
|
# мог выбрать его вручную (страница подтвердит/изменит).
|
|
if self._reconfigure_entry.data.get(CONF_TEMPLATE):
|
|
data[CONF_TEMPLATE] = self._reconfigure_entry.data[
|
|
CONF_TEMPLATE
|
|
]
|
|
await self.async_set_unique_id(data[CONF_DSN], raise_on_progress=False)
|
|
if self._reconfigure_entry is not None:
|
|
self._abort_if_unique_id_mismatch()
|
|
else:
|
|
self._abort_if_unique_id_configured()
|
|
result, template, probed = await self.hass.async_add_executor_job(
|
|
partial(_run_trial, data)
|
|
)
|
|
if result.ok:
|
|
self._pending_data = data
|
|
if (
|
|
self._reconfigure_entry is None
|
|
or not self._reconfigure_entry.data.get(CONF_TEMPLATE)
|
|
):
|
|
self._pending_data[CONF_TEMPLATE] = template.name
|
|
self._trial_properties = result.properties
|
|
self._trial_answered = result.answered
|
|
self._resolved_template = template
|
|
self._probed = probed
|
|
return None
|
|
_LOGGER.debug(
|
|
"пробное подключение не удалось: reason=%s state=%s error=%s",
|
|
result.reason, result.state, result.error,
|
|
)
|
|
return TRIAL_ERRORS.get(result.reason, "cannot_connect")
|
|
|
|
def _create_entry(self, data: dict[str, Any]) -> ConfigFlowResult:
|
|
title = data.get(CONF_NAME) or data[CONF_DSN]
|
|
if self._reconfigure_entry is not None:
|
|
return self.async_update_reload_and_abort(
|
|
self._reconfigure_entry, title=title, data=data
|
|
)
|
|
return self.async_create_entry(title=title, data=data)
|
|
|
|
@staticmethod
|
|
def async_get_options_flow(config_entry: ConfigEntry) -> "FglairOptionsFlow":
|
|
return FglairOptionsFlow()
|
|
|
|
|
|
class FglairOptionsFlow(OptionsFlow):
|
|
"""Показывает полный LAN-ключ (для secrets.yaml ESPHome) и позволяет
|
|
обновить его без облака."""
|
|
|
|
async def async_step_init(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
entry = self.config_entry
|
|
if user_input is not None:
|
|
key = str(user_input[CONF_LANIP_KEY]).strip()
|
|
if key and key != entry.data.get(CONF_LANIP_KEY):
|
|
self.hass.config_entries.async_update_entry(
|
|
entry, data={**entry.data, CONF_LANIP_KEY: key}
|
|
)
|
|
self.hass.config_entries.async_schedule_reload(entry.entry_id)
|
|
return self.async_create_entry(title="", data={})
|
|
schema = vol.Schema(
|
|
{vol.Required(CONF_LANIP_KEY): selector.TextSelector()}
|
|
)
|
|
schema = self.add_suggested_values_to_schema(
|
|
schema, {CONF_LANIP_KEY: entry.data.get(CONF_LANIP_KEY, "")}
|
|
)
|
|
return self.async_show_form(
|
|
step_id="init",
|
|
data_schema=schema,
|
|
description_placeholders={
|
|
"dsn": str(entry.data.get(CONF_DSN, "")),
|
|
"lanip_key_id": str(entry.data.get(CONF_LANIP_KEY_ID, "")),
|
|
"host": str(entry.data.get(CONF_HOST, "")),
|
|
},
|
|
)
|