"""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 = 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 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), "detected": default, "model": str(data.get(CONF_MODEL) or "-"), } 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) 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 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, "")), }, )