core/homeassistant/components/group/config_flow.py

"""Config flow for Group integration."""
from __future__ import annotations

from collections.abc import Callable, Mapping
from functools import partial
from typing import Any, cast

import voluptuous as vol

from homeassistant.const import CONF_ENTITIES
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er, selector
from homeassistant.helpers.schema_config_entry_flow import (
    SchemaConfigFlowHandler,
    SchemaFlowFormStep,
    SchemaFlowMenuStep,
    SchemaOptionsFlowHandler,
    entity_selector_without_own_entities,
)

from . import DOMAIN
from .binary_sensor import CONF_ALL
from .const import CONF_HIDE_MEMBERS


def basic_group_options_schema(
    domain: str,
    handler: SchemaConfigFlowHandler | SchemaOptionsFlowHandler,
    options: dict[str, Any],
) -> vol.Schema:
    """Generate options schema."""
    handler = cast(SchemaOptionsFlowHandler, handler)
    return vol.Schema(
        {
            vol.Required(CONF_ENTITIES): entity_selector_without_own_entities(
                handler, {"domain": domain, "multiple": True}
            ),
            vol.Required(CONF_HIDE_MEMBERS, default=False): selector.selector(
                {"boolean": {}}
            ),
        }
    )


def basic_group_config_schema(domain: str) -> vol.Schema:
    """Generate config schema."""
    return vol.Schema(
        {
            vol.Required("name"): selector.selector({"text": {}}),
            vol.Required(CONF_ENTITIES): selector.selector(
                {"entity": {"domain": domain, "multiple": True}}
            ),
            vol.Required(CONF_HIDE_MEMBERS, default=False): selector.selector(
                {"boolean": {}}
            ),
        }
    )


def binary_sensor_options_schema(
    handler: SchemaConfigFlowHandler | SchemaOptionsFlowHandler,
    options: dict[str, Any],
) -> vol.Schema:
    """Generate options schema."""
    return basic_group_options_schema("binary_sensor", handler, options).extend(
        {
            vol.Required(CONF_ALL, default=False): selector.selector({"boolean": {}}),
        }
    )


BINARY_SENSOR_CONFIG_SCHEMA = basic_group_config_schema("binary_sensor").extend(
    {
        vol.Required(CONF_ALL, default=False): selector.selector({"boolean": {}}),
    }
)


def light_switch_options_schema(
    domain: str,
    handler: SchemaConfigFlowHandler | SchemaOptionsFlowHandler,
    options: dict[str, Any],
) -> vol.Schema:
    """Generate options schema."""
    return basic_group_options_schema(domain, handler, options).extend(
        {
            vol.Required(
                CONF_ALL, default=False, description={"advanced": True}
            ): selector.selector({"boolean": {}}),
        }
    )


GROUP_TYPES = [
    "binary_sensor",
    "cover",
    "fan",
    "light",
    "lock",
    "media_player",
    "switch",
]


@callback
def choose_options_step(options: dict[str, Any]) -> str:
    """Return next step_id for options flow according to group_type."""
    return cast(str, options["group_type"])


def set_group_type(group_type: str) -> Callable[[dict[str, Any]], dict[str, Any]]:
    """Set group type."""

    @callback
    def _set_group_type(user_input: dict[str, Any]) -> dict[str, Any]:
        """Add group type to user input."""
        return {"group_type": group_type, **user_input}

    return _set_group_type


CONFIG_FLOW: dict[str, SchemaFlowFormStep | SchemaFlowMenuStep] = {
    "user": SchemaFlowMenuStep(GROUP_TYPES),
    "binary_sensor": SchemaFlowFormStep(
        BINARY_SENSOR_CONFIG_SCHEMA, set_group_type("binary_sensor")
    ),
    "cover": SchemaFlowFormStep(
        basic_group_config_schema("cover"), set_group_type("cover")
    ),
    "fan": SchemaFlowFormStep(basic_group_config_schema("fan"), set_group_type("fan")),
    "light": SchemaFlowFormStep(
        basic_group_config_schema("light"), set_group_type("light")
    ),
    "lock": SchemaFlowFormStep(
        basic_group_config_schema("lock"), set_group_type("lock")
    ),
    "media_player": SchemaFlowFormStep(
        basic_group_config_schema("media_player"), set_group_type("media_player")
    ),
    "switch": SchemaFlowFormStep(
        basic_group_config_schema("switch"), set_group_type("switch")
    ),
}


OPTIONS_FLOW: dict[str, SchemaFlowFormStep | SchemaFlowMenuStep] = {
    "init": SchemaFlowFormStep(None, next_step=choose_options_step),
    "binary_sensor": SchemaFlowFormStep(binary_sensor_options_schema),
    "cover": SchemaFlowFormStep(partial(basic_group_options_schema, "cover")),
    "fan": SchemaFlowFormStep(partial(basic_group_options_schema, "fan")),
    "light": SchemaFlowFormStep(partial(light_switch_options_schema, "light")),
    "lock": SchemaFlowFormStep(partial(basic_group_options_schema, "lock")),
    "media_player": SchemaFlowFormStep(
        partial(basic_group_options_schema, "media_player")
    ),
    "switch": SchemaFlowFormStep(partial(light_switch_options_schema, "switch")),
}


class GroupConfigFlowHandler(SchemaConfigFlowHandler, domain=DOMAIN):
    """Handle a config or options flow for groups."""

    config_flow = CONFIG_FLOW
    options_flow = OPTIONS_FLOW

    @callback
    def async_config_entry_title(self, options: Mapping[str, Any]) -> str:
        """Return config entry title.

        The options parameter contains config entry options, which is the union of user
        input from the config flow steps.
        """
        return cast(str, options["name"]) if "name" in options else ""

    @callback
    def async_config_flow_finished(self, options: Mapping[str, Any]) -> None:
        """Hide the group members if requested."""
        if options[CONF_HIDE_MEMBERS]:
            _async_hide_members(
                self.hass, options[CONF_ENTITIES], er.RegistryEntryHider.INTEGRATION
            )

    @callback
    @staticmethod
    def async_options_flow_finished(
        hass: HomeAssistant, options: Mapping[str, Any]
    ) -> None:
        """Hide or unhide the group members as requested."""
        hidden_by = (
            er.RegistryEntryHider.INTEGRATION if options[CONF_HIDE_MEMBERS] else None
        )
        _async_hide_members(hass, options[CONF_ENTITIES], hidden_by)


def _async_hide_members(
    hass: HomeAssistant, members: list[str], hidden_by: er.RegistryEntryHider | None
) -> None:
    """Hide or unhide group members."""
    registry = er.async_get(hass)
    for member in members:
        if not (entity_id := er.async_resolve_entity_id(registry, member)):
            continue
        if entity_id not in registry.entities:
            continue
        registry.async_update_entity(entity_id, hidden_by=hidden_by)