mirror of https://github.com/home-assistant/core
148 lines
4.5 KiB
Python
148 lines
4.5 KiB
Python
"""Config flow for Threshold integration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.components import websocket_api
|
|
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
|
from homeassistant.const import CONF_ENTITY_ID, CONF_NAME
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.helpers import selector
|
|
from homeassistant.helpers.schema_config_entry_flow import (
|
|
SchemaCommonFlowHandler,
|
|
SchemaConfigFlowHandler,
|
|
SchemaFlowError,
|
|
SchemaFlowFormStep,
|
|
)
|
|
|
|
from .binary_sensor import ThresholdSensor
|
|
from .const import CONF_HYSTERESIS, CONF_LOWER, CONF_UPPER, DEFAULT_HYSTERESIS, DOMAIN
|
|
|
|
|
|
async def _validate_mode(
|
|
handler: SchemaCommonFlowHandler, user_input: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
"""Validate the threshold mode, and set limits to None if not set."""
|
|
if CONF_LOWER not in user_input and CONF_UPPER not in user_input:
|
|
raise SchemaFlowError("need_lower_upper")
|
|
return {CONF_LOWER: None, CONF_UPPER: None, **user_input}
|
|
|
|
|
|
OPTIONS_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Required(
|
|
CONF_HYSTERESIS, default=DEFAULT_HYSTERESIS
|
|
): selector.NumberSelector(
|
|
selector.NumberSelectorConfig(
|
|
mode=selector.NumberSelectorMode.BOX, step="any"
|
|
),
|
|
),
|
|
vol.Optional(CONF_LOWER): selector.NumberSelector(
|
|
selector.NumberSelectorConfig(
|
|
mode=selector.NumberSelectorMode.BOX, step="any"
|
|
),
|
|
),
|
|
vol.Optional(CONF_UPPER): selector.NumberSelector(
|
|
selector.NumberSelectorConfig(
|
|
mode=selector.NumberSelectorMode.BOX, step="any"
|
|
),
|
|
),
|
|
vol.Required(CONF_ENTITY_ID): selector.EntitySelector(
|
|
selector.EntitySelectorConfig(domain=SENSOR_DOMAIN)
|
|
),
|
|
}
|
|
)
|
|
|
|
CONFIG_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Required(CONF_NAME): selector.TextSelector(),
|
|
}
|
|
).extend(OPTIONS_SCHEMA.schema)
|
|
|
|
CONFIG_FLOW = {
|
|
"user": SchemaFlowFormStep(
|
|
CONFIG_SCHEMA, preview="threshold", validate_user_input=_validate_mode
|
|
)
|
|
}
|
|
|
|
OPTIONS_FLOW = {
|
|
"init": SchemaFlowFormStep(
|
|
OPTIONS_SCHEMA, preview="threshold", validate_user_input=_validate_mode
|
|
)
|
|
}
|
|
|
|
|
|
class ConfigFlowHandler(SchemaConfigFlowHandler, domain=DOMAIN):
|
|
"""Handle a config or options flow for Threshold."""
|
|
|
|
config_flow = CONFIG_FLOW
|
|
options_flow = OPTIONS_FLOW
|
|
|
|
def async_config_entry_title(self, options: Mapping[str, Any]) -> str:
|
|
"""Return config entry title."""
|
|
name: str = options[CONF_NAME]
|
|
return name
|
|
|
|
@staticmethod
|
|
async def async_setup_preview(hass: HomeAssistant) -> None:
|
|
"""Set up preview WS API."""
|
|
websocket_api.async_register_command(hass, ws_start_preview)
|
|
|
|
|
|
@websocket_api.websocket_command(
|
|
{
|
|
vol.Required("type"): "threshold/start_preview",
|
|
vol.Required("flow_id"): str,
|
|
vol.Required("flow_type"): vol.Any("config_flow", "options_flow"),
|
|
vol.Required("user_input"): dict,
|
|
}
|
|
)
|
|
@callback
|
|
def ws_start_preview(
|
|
hass: HomeAssistant,
|
|
connection: websocket_api.ActiveConnection,
|
|
msg: dict[str, Any],
|
|
) -> None:
|
|
"""Generate a preview."""
|
|
|
|
if msg["flow_type"] == "config_flow":
|
|
entity_id = msg["user_input"][CONF_ENTITY_ID]
|
|
name = msg["user_input"][CONF_NAME]
|
|
else:
|
|
flow_status = hass.config_entries.options.async_get(msg["flow_id"])
|
|
config_entry = hass.config_entries.async_get_entry(flow_status["handler"])
|
|
if not config_entry:
|
|
raise HomeAssistantError("Config entry not found")
|
|
entity_id = config_entry.options[CONF_ENTITY_ID]
|
|
name = config_entry.options[CONF_NAME]
|
|
|
|
@callback
|
|
def async_preview_updated(state: str, attributes: Mapping[str, Any]) -> None:
|
|
"""Forward config entry state events to websocket."""
|
|
connection.send_message(
|
|
websocket_api.event_message(
|
|
msg["id"], {"attributes": attributes, "state": state}
|
|
)
|
|
)
|
|
|
|
preview_entity = ThresholdSensor(
|
|
entity_id,
|
|
name,
|
|
msg["user_input"].get(CONF_LOWER),
|
|
msg["user_input"].get(CONF_UPPER),
|
|
msg["user_input"].get(CONF_HYSTERESIS),
|
|
None,
|
|
None,
|
|
)
|
|
preview_entity.hass = hass
|
|
|
|
connection.send_result(msg["id"])
|
|
connection.subscriptions[msg["id"]] = preview_entity.async_start_preview(
|
|
async_preview_updated
|
|
)
|