jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 08:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add config flow to filter helper (#121522)

Browse Source 
Co-authored-by: Robert Resch <robert@resch.dev>
This commit is contained in:
G Johansson 2025-01-29 10:35:01 +01:00 committed by GitHub
parent 417003ad35
commit b93c2382ce
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 938 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
homeassistant/components/filter/__init__.py
View File

@ -1,6 +1,25 @@
"""The filter component."""
from homeassistant.const import Platform
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
DOMAIN = "filter"
PLATFORMS = [Platform.SENSOR]
from .const import PLATFORMS
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Filter from a config entry."""
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(update_listener))
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload Filter config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
await hass.config_entries.async_reload(entry.entry_id)

243
homeassistant/components/filter/config_flow.py Normal file
View File

@ -0,0 +1,243 @@
"""Config flow for filter."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any, cast
import voluptuous as vol
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.const import CONF_ENTITY_ID, CONF_NAME
from homeassistant.helpers.schema_config_entry_flow import (
SchemaCommonFlowHandler,
SchemaConfigFlowHandler,
SchemaFlowFormStep,
)
from homeassistant.helpers.selector import (
DurationSelector,
DurationSelectorConfig,
EntitySelector,
EntitySelectorConfig,
NumberSelector,
NumberSelectorConfig,
NumberSelectorMode,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
TextSelector,
)
from .const import (
CONF_FILTER_LOWER_BOUND,
CONF_FILTER_NAME,
CONF_FILTER_PRECISION,
CONF_FILTER_RADIUS,
CONF_FILTER_TIME_CONSTANT,
CONF_FILTER_UPPER_BOUND,
CONF_FILTER_WINDOW_SIZE,
CONF_TIME_SMA_TYPE,
DEFAULT_FILTER_RADIUS,
DEFAULT_FILTER_TIME_CONSTANT,
DEFAULT_NAME,
DEFAULT_PRECISION,
DEFAULT_WINDOW_SIZE,
DOMAIN,
FILTER_NAME_LOWPASS,
FILTER_NAME_OUTLIER,
FILTER_NAME_RANGE,
FILTER_NAME_THROTTLE,
FILTER_NAME_TIME_SMA,
FILTER_NAME_TIME_THROTTLE,
TIME_SMA_LAST,
)
FILTERS = [
FILTER_NAME_LOWPASS,
FILTER_NAME_OUTLIER,
FILTER_NAME_RANGE,
FILTER_NAME_THROTTLE,
FILTER_NAME_TIME_SMA,
FILTER_NAME_TIME_THROTTLE,
]
async def get_next_step(user_input: dict[str, Any]) -> str:
"""Return next step for options."""
return cast(str, user_input[CONF_FILTER_NAME])
async def validate_options(
handler: SchemaCommonFlowHandler, user_input: dict[str, Any]
) -> dict[str, Any]:
"""Validate options selected."""
if CONF_FILTER_WINDOW_SIZE in user_input and isinstance(
user_input[CONF_FILTER_WINDOW_SIZE], float
):
user_input[CONF_FILTER_WINDOW_SIZE] = int(user_input[CONF_FILTER_WINDOW_SIZE])
if CONF_FILTER_TIME_CONSTANT in user_input:
user_input[CONF_FILTER_TIME_CONSTANT] = int(
user_input[CONF_FILTER_TIME_CONSTANT]
)
if CONF_FILTER_PRECISION in user_input:
user_input[CONF_FILTER_PRECISION] = int(user_input[CONF_FILTER_PRECISION])
handler.parent_handler._async_abort_entries_match({**handler.options, **user_input}) # noqa: SLF001
return user_input
DATA_SCHEMA_SETUP = vol.Schema(
{
vol.Required(CONF_NAME, default=DEFAULT_NAME): TextSelector(),
vol.Required(CONF_ENTITY_ID): EntitySelector(
EntitySelectorConfig(domain=[SENSOR_DOMAIN])
),
vol.Required(CONF_FILTER_NAME): SelectSelector(
SelectSelectorConfig(
options=FILTERS,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_FILTER_NAME,
)
),
}
)
BASE_OPTIONS_SCHEMA = {
vol.Optional(CONF_FILTER_PRECISION, default=DEFAULT_PRECISION): NumberSelector(
NumberSelectorConfig(min=0, step=1, mode=NumberSelectorMode.BOX)
)
}
OUTLIER_SCHEMA = vol.Schema(
{
vol.Optional(
CONF_FILTER_WINDOW_SIZE, default=DEFAULT_WINDOW_SIZE
): NumberSelector(
NumberSelectorConfig(min=0, step=1, mode=NumberSelectorMode.BOX)
),
vol.Optional(CONF_FILTER_RADIUS, default=DEFAULT_FILTER_RADIUS): NumberSelector(
NumberSelectorConfig(min=0, step="any", mode=NumberSelectorMode.BOX)
),
}
).extend(BASE_OPTIONS_SCHEMA)
LOWPASS_SCHEMA = vol.Schema(
{
vol.Optional(
CONF_FILTER_WINDOW_SIZE, default=DEFAULT_WINDOW_SIZE
): NumberSelector(
NumberSelectorConfig(min=0, step=1, mode=NumberSelectorMode.BOX)
),
vol.Optional(
CONF_FILTER_TIME_CONSTANT, default=DEFAULT_FILTER_TIME_CONSTANT
): NumberSelector(
NumberSelectorConfig(min=0, step=1, mode=NumberSelectorMode.BOX)
),
}
).extend(BASE_OPTIONS_SCHEMA)
RANGE_SCHEMA = vol.Schema(
{
vol.Optional(CONF_FILTER_LOWER_BOUND): NumberSelector(
NumberSelectorConfig(min=0, step="any", mode=NumberSelectorMode.BOX)
),
vol.Optional(CONF_FILTER_UPPER_BOUND): NumberSelector(
NumberSelectorConfig(min=0, step="any", mode=NumberSelectorMode.BOX)
),
}
).extend(BASE_OPTIONS_SCHEMA)
TIME_SMA_SCHEMA = vol.Schema(
{
vol.Optional(CONF_TIME_SMA_TYPE, default=TIME_SMA_LAST): SelectSelector(
SelectSelectorConfig(
options=[TIME_SMA_LAST],
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_TIME_SMA_TYPE,
)
),
vol.Required(CONF_FILTER_WINDOW_SIZE): DurationSelector(
DurationSelectorConfig(enable_day=False, allow_negative=False)
),
}
).extend(BASE_OPTIONS_SCHEMA)
THROTTLE_SCHEMA = vol.Schema(
{
vol.Optional(
CONF_FILTER_WINDOW_SIZE, default=DEFAULT_WINDOW_SIZE
): NumberSelector(
NumberSelectorConfig(min=0, step=1, mode=NumberSelectorMode.BOX)
),
}
).extend(BASE_OPTIONS_SCHEMA)
TIME_THROTTLE_SCHEMA = vol.Schema(
{
vol.Required(CONF_FILTER_WINDOW_SIZE): DurationSelector(
DurationSelectorConfig(enable_day=False, allow_negative=False)
),
}
).extend(BASE_OPTIONS_SCHEMA)
CONFIG_FLOW = {
"user": SchemaFlowFormStep(
schema=DATA_SCHEMA_SETUP,
next_step=get_next_step,
),
"lowpass": SchemaFlowFormStep(
schema=LOWPASS_SCHEMA, validate_user_input=validate_options
),
"outlier": SchemaFlowFormStep(
schema=OUTLIER_SCHEMA, validate_user_input=validate_options
),
"range": SchemaFlowFormStep(
schema=RANGE_SCHEMA, validate_user_input=validate_options
),
"time_simple_moving_average": SchemaFlowFormStep(
schema=TIME_SMA_SCHEMA, validate_user_input=validate_options
),
"throttle": SchemaFlowFormStep(
schema=THROTTLE_SCHEMA, validate_user_input=validate_options
),
"time_throttle": SchemaFlowFormStep(
schema=TIME_THROTTLE_SCHEMA, validate_user_input=validate_options
),
}
OPTIONS_FLOW = {
"init": SchemaFlowFormStep(
schema=None,
next_step=get_next_step,
),
"lowpass": SchemaFlowFormStep(
schema=LOWPASS_SCHEMA, validate_user_input=validate_options
),
"outlier": SchemaFlowFormStep(
schema=OUTLIER_SCHEMA, validate_user_input=validate_options
),
"range": SchemaFlowFormStep(
schema=RANGE_SCHEMA, validate_user_input=validate_options
),
"time_simple_moving_average": SchemaFlowFormStep(
schema=TIME_SMA_SCHEMA, validate_user_input=validate_options
),
"throttle": SchemaFlowFormStep(
schema=THROTTLE_SCHEMA, validate_user_input=validate_options
),
"time_throttle": SchemaFlowFormStep(
schema=TIME_THROTTLE_SCHEMA, validate_user_input=validate_options
),
}
class FilterConfigFlowHandler(SchemaConfigFlowHandler, domain=DOMAIN):
"""Handle a config flow for Filter."""
config_flow = CONFIG_FLOW
options_flow = OPTIONS_FLOW
def async_config_entry_title(self, options: Mapping[str, Any]) -> str:
"""Return config entry title."""
return cast(str, options[CONF_NAME])

36
homeassistant/components/filter/const.py Normal file
View File

@ -0,0 +1,36 @@
"""The filter component constants."""
from homeassistant.const import Platform
DOMAIN = "filter"
PLATFORMS = [Platform.SENSOR]
CONF_INDEX = "index"
FILTER_NAME_RANGE = "range"
FILTER_NAME_LOWPASS = "lowpass"
FILTER_NAME_OUTLIER = "outlier"
FILTER_NAME_THROTTLE = "throttle"
FILTER_NAME_TIME_THROTTLE = "time_throttle"
FILTER_NAME_TIME_SMA = "time_simple_moving_average"
CONF_FILTERS = "filters"
CONF_FILTER_NAME = "filter"
CONF_FILTER_WINDOW_SIZE = "window_size"
CONF_FILTER_PRECISION = "precision"
CONF_FILTER_RADIUS = "radius"
CONF_FILTER_TIME_CONSTANT = "time_constant"
CONF_FILTER_LOWER_BOUND = "lower_bound"
CONF_FILTER_UPPER_BOUND = "upper_bound"
CONF_TIME_SMA_TYPE = "type"
TIME_SMA_LAST = "last"
WINDOW_SIZE_UNIT_NUMBER_EVENTS = 1
WINDOW_SIZE_UNIT_TIME = 2
DEFAULT_NAME = "Filtered sensor"
DEFAULT_WINDOW_SIZE = 1
DEFAULT_PRECISION = 2
DEFAULT_FILTER_RADIUS = 2.0
DEFAULT_FILTER_TIME_CONSTANT = 10

1
homeassistant/components/filter/manifest.json
View File

@ -2,6 +2,7 @@
"domain": "filter",
"name": "Filter",
"codeowners": ["@dgomes"],
"config_flow": true,
"dependencies": ["recorder"],
"documentation": "https://www.home-assistant.io/integrations/filter",
"integration_type": "helper",

81
homeassistant/components/filter/sensor.py
View File

@ -24,6 +24,7 @@ from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
@ -51,39 +52,37 @@ from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType, StateTyp
from homeassistant.util.decorator import Registry
import homeassistant.util.dt as dt_util
from . import DOMAIN, PLATFORMS
from .const import (
CONF_FILTER_LOWER_BOUND,
CONF_FILTER_NAME,
CONF_FILTER_PRECISION,
CONF_FILTER_RADIUS,
CONF_FILTER_TIME_CONSTANT,
CONF_FILTER_UPPER_BOUND,
CONF_FILTER_WINDOW_SIZE,
CONF_FILTERS,
CONF_TIME_SMA_TYPE,
DEFAULT_FILTER_RADIUS,
DEFAULT_FILTER_TIME_CONSTANT,
DEFAULT_PRECISION,
DEFAULT_WINDOW_SIZE,
DOMAIN,
FILTER_NAME_LOWPASS,
FILTER_NAME_OUTLIER,
FILTER_NAME_RANGE,
FILTER_NAME_THROTTLE,
FILTER_NAME_TIME_SMA,
FILTER_NAME_TIME_THROTTLE,
PLATFORMS,
TIME_SMA_LAST,
WINDOW_SIZE_UNIT_NUMBER_EVENTS,
WINDOW_SIZE_UNIT_TIME,
)
_LOGGER = logging.getLogger(__name__)
FILTER_NAME_RANGE = "range"
FILTER_NAME_LOWPASS = "lowpass"
FILTER_NAME_OUTLIER = "outlier"
FILTER_NAME_THROTTLE = "throttle"
FILTER_NAME_TIME_THROTTLE = "time_throttle"
FILTER_NAME_TIME_SMA = "time_simple_moving_average"
FILTERS: Registry[str, type[Filter]] = Registry()
CONF_FILTERS = "filters"
CONF_FILTER_NAME = "filter"
CONF_FILTER_WINDOW_SIZE = "window_size"
CONF_FILTER_PRECISION = "precision"
CONF_FILTER_RADIUS = "radius"
CONF_FILTER_TIME_CONSTANT = "time_constant"
CONF_FILTER_LOWER_BOUND = "lower_bound"
CONF_FILTER_UPPER_BOUND = "upper_bound"
CONF_TIME_SMA_TYPE = "type"
TIME_SMA_LAST = "last"
WINDOW_SIZE_UNIT_NUMBER_EVENTS = 1
WINDOW_SIZE_UNIT_TIME = 2
DEFAULT_WINDOW_SIZE = 1
DEFAULT_PRECISION = 2
DEFAULT_FILTER_RADIUS = 2.0
DEFAULT_FILTER_TIME_CONSTANT = 10
NAME_TEMPLATE = "{} filter"
ICON = "mdi:chart-line-variant"
FILTER_SCHEMA = vol.Schema({vol.Optional(CONF_FILTER_PRECISION): vol.Coerce(int)})
@ -199,6 +198,32 @@ async def async_setup_platform(
async_add_entities([SensorFilter(name, unique_id, entity_id, filters)])
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Filter sensor entry."""
name: str = entry.options[CONF_NAME]
entity_id: str = entry.options[CONF_ENTITY_ID]
filter_config = {
k: v for k, v in entry.options.items() if k not in (CONF_NAME, CONF_ENTITY_ID)
}
if CONF_FILTER_WINDOW_SIZE in filter_config and isinstance(
filter_config[CONF_FILTER_WINDOW_SIZE], dict
):
filter_config[CONF_FILTER_WINDOW_SIZE] = timedelta(
**filter_config[CONF_FILTER_WINDOW_SIZE]
)
filters = [
FILTERS[filter_config.pop(CONF_FILTER_NAME)](entity=entity_id, **filter_config)
]
async_add_entities([SensorFilter(name, entry.entry_id, entity_id, filters)])
class SensorFilter(SensorEntity):
"""Representation of a Filter Sensor."""

192
homeassistant/components/filter/strings.json
View File

@ -1,5 +1,197 @@
{
"title": "Filter",
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
},
"step": {
"user": {
"description": "Add a filter sensor. UI configuration is limited to a single filter, use YAML for filter chain.",
"data": {
"name": "[%key:common::config_flow::data::name%]",
"entity_id": "Entity",
"filter": "Filter"
},
"data_description": {
"name": "Name for the created entity.",
"entity_id": "Entity to filter from.",
"filter": "Select filter to configure."
}
},
"outlier": {
"description": "Read the documentation for further details on how to configure the filter sensor using these options.",
"data": {
"window_size": "Window size",
"precision": "Precision",
"radius": "Radius"
},
"data_description": {
"window_size": "Size of the window of previous states.",
"precision": "Defines the number of decimal places of the calculated sensor value.",
"radius": "Band radius from median of previous states."
}
},
"lowpass": {
"description": "[%key:component::filter::config::step::outlier::description%]",
"data": {
"window_size": "[%key:component::filter::config::step::outlier::data::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data::precision%]",
"time_constant": "Time constant"
},
"data_description": {
"window_size": "[%key:component::filter::config::step::outlier::data_description::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data_description::precision%]",
"time_constant": "Loosely relates to the amount of time it takes for a state to influence the output."
}
},
"range": {
"description": "[%key:component::filter::config::step::outlier::description%]",
"data": {
"precision": "[%key:component::filter::config::step::outlier::data::precision%]",
"lower_bound": "Lower bound",
"upper_bound": "Upper bound"
},
"data_description": {
"precision": "[%key:component::filter::config::step::outlier::data_description::precision%]",
"lower_bound": "Lower bound for filter range.",
"upper_bound": "Upper bound for filter range."
}
},
"time_simple_moving_average": {
"description": "[%key:component::filter::config::step::outlier::description%]",
"data": {
"window_size": "[%key:component::filter::config::step::outlier::data::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data::precision%]",
"type": "Type"
},
"data_description": {
"window_size": "[%key:component::filter::config::step::outlier::data_description::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data_description::precision%]",
"type": "Defines the type of Simple Moving Average."
}
},
"throttle": {
"description": "[%key:component::filter::config::step::outlier::description%]",
"data": {
"window_size": "[%key:component::filter::config::step::outlier::data::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data::precision%]"
},
"data_description": {
"window_size": "[%key:component::filter::config::step::outlier::data_description::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data_description::precision%]"
}
},
"time_throttle": {
"description": "[%key:component::filter::config::step::outlier::description%]",
"data": {
"window_size": "[%key:component::filter::config::step::outlier::data::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data::precision%]"
},
"data_description": {
"window_size": "[%key:component::filter::config::step::outlier::data_description::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data_description::precision%]"
}
}
}
},
"options": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
},
"step": {
"outlier": {
"description": "[%key:component::filter::config::step::outlier::description%]",
"data": {
"window_size": "[%key:component::filter::config::step::outlier::data::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data::precision%]",
"radius": "[%key:component::filter::config::step::outlier::data::radius%]"
},
"data_description": {
"window_size": "[%key:component::filter::config::step::outlier::data_description::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data_description::precision%]",
"radius": "[%key:component::filter::config::step::outlier::data_description::radius%]"
}
},
"lowpass": {
"description": "[%key:component::filter::config::step::outlier::description%]",
"data": {
"window_size": "[%key:component::filter::config::step::outlier::data::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data::precision%]",
"time_constant": "[%key:component::filter::config::step::lowpass::data::time_constant%]"
},
"data_description": {
"window_size": "[%key:component::filter::config::step::outlier::data_description::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data_description::precision%]",
"time_constant": "[%key:component::filter::config::step::lowpass::data_description::time_constant%]"
}
},
"range": {
"description": "[%key:component::filter::config::step::outlier::description%]",
"data": {
"precision": "[%key:component::filter::config::step::outlier::data::precision%]",
"lower_bound": "[%key:component::filter::config::step::range::data::lower_bound%]",
"upper_bound": "[%key:component::filter::config::step::range::data::upper_bound%]"
},
"data_description": {
"precision": "[%key:component::filter::config::step::outlier::data_description::precision%]",
"lower_bound": "[%key:component::filter::config::step::range::data_description::lower_bound%]",
"upper_bound": "[%key:component::filter::config::step::range::data_description::upper_bound%]"
}
},
"time_simple_moving_average": {
"description": "[%key:component::filter::config::step::outlier::description%]",
"data": {
"window_size": "[%key:component::filter::config::step::outlier::data::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data::precision%]",
"type": "[%key:component::filter::config::step::time_simple_moving_average::data::type%]"
},
"data_description": {
"window_size": "[%key:component::filter::config::step::outlier::data_description::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data_description::precision%]",
"type": "[%key:component::filter::config::step::time_simple_moving_average::data_description::type%]"
}
},
"throttle": {
"description": "[%key:component::filter::config::step::outlier::description%]",
"data": {
"window_size": "[%key:component::filter::config::step::outlier::data::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data::precision%]"
},
"data_description": {
"window_size": "[%key:component::filter::config::step::outlier::data_description::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data_description::precision%]"
}
},
"time_throttle": {
"description": "[%key:component::filter::config::step::outlier::description%]",
"data": {
"window_size": "[%key:component::filter::config::step::outlier::data::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data::precision%]"
},
"data_description": {
"window_size": "[%key:component::filter::config::step::outlier::data_description::window_size%]",
"precision": "[%key:component::filter::config::step::outlier::data_description::precision%]"
}
}
}
},
"selector": {
"filter": {
"options": {
"range": "Range",
"lowpass": "Lowpass",
"outlier": "Outlier",
"throttle": "Throttle",
"time_throttle": "Time throttle",
"time_simple_moving_average": "Moving Average (Time based)"
}
},
"type": {
"options": {
"last": "Last"
}
}
},
"services": {
"reload": {
"name": "[%key:common::action::reload%]",

1
homeassistant/generated/config_flows.py generated
View File

@ -6,6 +6,7 @@ To update, run python3 -m script.hassfest
FLOWS = {
"helper": [
"derivative",
"filter",
"generic_hygrostat",
"generic_thermostat",
"group",

2
homeassistant/generated/integrations.json
View File

@ -7436,7 +7436,7 @@
},
"filter": {
"integration_type": "helper",
"config_flow": false,
"config_flow": true,
"iot_class": "local_push"
},
"generic_hygrostat": {

93
tests/components/filter/conftest.py Normal file
View File

@ -0,0 +1,93 @@
"""Fixtures for the Filter integration."""
from __future__ import annotations
from collections.abc import Generator
from datetime import timedelta
from typing import Any
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant.components.filter.const import (
CONF_FILTER_NAME,
CONF_FILTER_PRECISION,
CONF_FILTER_RADIUS,
CONF_FILTER_WINDOW_SIZE,
DEFAULT_FILTER_RADIUS,
DEFAULT_NAME,
DEFAULT_PRECISION,
DEFAULT_WINDOW_SIZE,
DOMAIN,
FILTER_NAME_OUTLIER,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_ENTITY_ID, CONF_NAME
from homeassistant.core import HomeAssistant, State
import homeassistant.util.dt as dt_util
from tests.common import MockConfigEntry
@pytest.fixture(name="values")
def values_fixture() -> list[State]:
"""Fixture for a list of test States."""
values = []
raw_values = [20, 19, 18, 21, 22, 0]
timestamp = dt_util.utcnow()
for val in raw_values:
values.append(State("sensor.test_monitored", str(val), last_updated=timestamp))
timestamp += timedelta(minutes=1)
return values
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Automatically patch setup_entry."""
with patch(
"homeassistant.components.filter.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture(name="get_config")
async def get_config_to_integration_load() -> dict[str, Any]:
"""Return configuration.
To override the config, tests can be marked with:
@pytest.mark.parametrize("get_config", [{...}])
"""
return {
CONF_NAME: DEFAULT_NAME,
CONF_ENTITY_ID: "sensor.test_monitored",
CONF_FILTER_NAME: FILTER_NAME_OUTLIER,
CONF_FILTER_WINDOW_SIZE: DEFAULT_WINDOW_SIZE,
CONF_FILTER_RADIUS: DEFAULT_FILTER_RADIUS,
CONF_FILTER_PRECISION: DEFAULT_PRECISION,
}
@pytest.fixture(name="loaded_entry")
async def load_integration(
hass: HomeAssistant, get_config: dict[str, Any], values: list[State]
) -> MockConfigEntry:
"""Set up the Filter integration in Home Assistant."""
config_entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
options=get_config,
entry_id="1",
)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
for value in values:
hass.states.async_set(get_config["entity_id"], value.state)
await hass.async_block_till_done()
await hass.async_block_till_done()
return config_entry

227
tests/components/filter/test_config_flow.py Normal file
View File

@ -0,0 +1,227 @@
"""Test the Filter config flow."""
from __future__ import annotations
from typing import Any
from unittest.mock import AsyncMock
import pytest
from homeassistant import config_entries
from homeassistant.components.filter.const import (
CONF_FILTER_LOWER_BOUND,
CONF_FILTER_NAME,
CONF_FILTER_PRECISION,
CONF_FILTER_RADIUS,
CONF_FILTER_TIME_CONSTANT,
CONF_FILTER_UPPER_BOUND,
CONF_FILTER_WINDOW_SIZE,
CONF_TIME_SMA_TYPE,
DEFAULT_FILTER_RADIUS,
DEFAULT_NAME,
DEFAULT_PRECISION,
DEFAULT_WINDOW_SIZE,
DOMAIN,
FILTER_NAME_LOWPASS,
FILTER_NAME_OUTLIER,
FILTER_NAME_RANGE,
FILTER_NAME_THROTTLE,
FILTER_NAME_TIME_SMA,
FILTER_NAME_TIME_THROTTLE,
TIME_SMA_LAST,
)
from homeassistant.components.recorder import Recorder
from homeassistant.const import CONF_ENTITY_ID, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
("entry_config", "options", "result_options"),
[
(
{CONF_FILTER_NAME: FILTER_NAME_OUTLIER},
{
CONF_FILTER_WINDOW_SIZE: 1.0,
CONF_FILTER_RADIUS: 2.0,
},
{
CONF_FILTER_NAME: FILTER_NAME_OUTLIER,
CONF_FILTER_WINDOW_SIZE: 1,
CONF_FILTER_RADIUS: 2.0,
},
),
(
{CONF_FILTER_NAME: FILTER_NAME_LOWPASS},
{
CONF_FILTER_WINDOW_SIZE: 1.0,
CONF_FILTER_TIME_CONSTANT: 10.0,
},
{
CONF_FILTER_NAME: FILTER_NAME_LOWPASS,
CONF_FILTER_WINDOW_SIZE: 1,
CONF_FILTER_TIME_CONSTANT: 10,
},
),
(
{CONF_FILTER_NAME: FILTER_NAME_RANGE},
{
CONF_FILTER_LOWER_BOUND: 1.0,
CONF_FILTER_UPPER_BOUND: 10.0,
},
{
CONF_FILTER_NAME: FILTER_NAME_RANGE,
CONF_FILTER_LOWER_BOUND: 1.0,
CONF_FILTER_UPPER_BOUND: 10.0,
},
),
(
{CONF_FILTER_NAME: FILTER_NAME_TIME_SMA},
{
CONF_TIME_SMA_TYPE: TIME_SMA_LAST,
CONF_FILTER_WINDOW_SIZE: {"hours": 40, "minutes": 5, "seconds": 5},
},
{
CONF_FILTER_NAME: FILTER_NAME_TIME_SMA,
CONF_TIME_SMA_TYPE: TIME_SMA_LAST,
CONF_FILTER_WINDOW_SIZE: {"hours": 40, "minutes": 5, "seconds": 5},
},
),
(
{CONF_FILTER_NAME: FILTER_NAME_THROTTLE},
{
CONF_FILTER_WINDOW_SIZE: 1.0,
},
{
CONF_FILTER_NAME: FILTER_NAME_THROTTLE,
CONF_FILTER_WINDOW_SIZE: 1,
},
),
(
{CONF_FILTER_NAME: FILTER_NAME_TIME_THROTTLE},
{
CONF_FILTER_WINDOW_SIZE: {"hours": 40, "minutes": 5, "seconds": 5},
},
{
CONF_FILTER_NAME: FILTER_NAME_TIME_THROTTLE,
CONF_FILTER_WINDOW_SIZE: {"hours": 40, "minutes": 5, "seconds": 5},
},
),
],
)
async def test_form(
recorder_mock: Recorder,
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
entry_config: dict[str, Any],
options: dict[str, Any],
result_options: dict[str, Any],
) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: DEFAULT_NAME,
CONF_ENTITY_ID: "sensor.test_monitored",
**entry_config,
},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_FILTER_PRECISION: DEFAULT_PRECISION, **options},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["version"] == 1
assert result["options"] == {
CONF_NAME: DEFAULT_NAME,
CONF_ENTITY_ID: "sensor.test_monitored",
CONF_FILTER_PRECISION: DEFAULT_PRECISION,
**result_options,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_options_flow(
recorder_mock: Recorder, hass: HomeAssistant, loaded_entry: MockConfigEntry
) -> None:
"""Test options flow."""
result = await hass.config_entries.options.async_init(loaded_entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "outlier"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_FILTER_WINDOW_SIZE: 2.0,
CONF_FILTER_RADIUS: 3.0,
CONF_FILTER_PRECISION: DEFAULT_PRECISION,
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_NAME: DEFAULT_NAME,
CONF_ENTITY_ID: "sensor.test_monitored",
CONF_FILTER_NAME: FILTER_NAME_OUTLIER,
CONF_FILTER_WINDOW_SIZE: 2,
CONF_FILTER_RADIUS: 3.0,
CONF_FILTER_PRECISION: DEFAULT_PRECISION,
}
await hass.async_block_till_done()
# Check the entity was updated, no new entity was created
assert len(hass.states.async_all()) == 2
state = hass.states.get("sensor.filtered_sensor")
assert state is not None
async def test_entry_already_exist(
recorder_mock: Recorder, hass: HomeAssistant, loaded_entry: MockConfigEntry
) -> None:
"""Test abort when entry already exist."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: DEFAULT_NAME,
CONF_ENTITY_ID: "sensor.test_monitored",
CONF_FILTER_NAME: FILTER_NAME_OUTLIER,
},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_FILTER_WINDOW_SIZE: DEFAULT_WINDOW_SIZE,
CONF_FILTER_RADIUS: DEFAULT_FILTER_RADIUS,
CONF_FILTER_PRECISION: DEFAULT_PRECISION,
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"

20
tests/components/filter/test_init.py Normal file
View File

@ -0,0 +1,20 @@
"""Test Filter component setup process."""
from __future__ import annotations
from homeassistant.components.recorder import Recorder
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_unload_entry(
recorder_mock: Recorder, hass: HomeAssistant, loaded_entry: MockConfigEntry
) -> None:
"""Test unload an entry."""
assert loaded_entry.state is ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(loaded_entry.entry_id)
await hass.async_block_till_done()
assert loaded_entry.state is ConfigEntryState.NOT_LOADED

51
tests/components/filter/test_sensor.py
View File

@ -6,8 +6,18 @@ from unittest.mock import patch
import pytest
from homeassistant import config as hass_config
from homeassistant.components.filter.sensor import (
from homeassistant.components.filter.const import (
CONF_FILTER_NAME,
CONF_FILTER_PRECISION,
CONF_FILTER_WINDOW_SIZE,
CONF_TIME_SMA_TYPE,
DEFAULT_NAME,
DEFAULT_PRECISION,
DOMAIN,
FILTER_NAME_TIME_SMA,
TIME_SMA_LAST,
)
from homeassistant.components.filter.sensor import (
LowPassFilter,
OutlierFilter,
RangeFilter,
@ -24,6 +34,8 @@ from homeassistant.components.sensor import (
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_UNIT_OF_MEASUREMENT,
CONF_ENTITY_ID,
CONF_NAME,
SERVICE_RELOAD,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
@ -34,7 +46,7 @@ from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from tests.common import assert_setup_component, get_fixture_path
from tests.common import MockConfigEntry, assert_setup_component, get_fixture_path
@pytest.fixture(autouse=True, name="stub_blueprint_populate")
@ -97,6 +109,41 @@ async def test_chain(
assert state.state == "18.05"
async def test_from_config_entry(
recorder_mock: Recorder,
hass: HomeAssistant,
loaded_entry: MockConfigEntry,
) -> None:
"""Test if filter works loaded from config entry."""
state = hass.states.get("sensor.filtered_sensor")
assert state.state == "22.0"
@pytest.mark.parametrize(
"get_config",
[
{
CONF_NAME: DEFAULT_NAME,
CONF_ENTITY_ID: "sensor.test_monitored",
CONF_FILTER_NAME: FILTER_NAME_TIME_SMA,
CONF_TIME_SMA_TYPE: TIME_SMA_LAST,
CONF_FILTER_WINDOW_SIZE: {"hours": 40, "minutes": 5, "seconds": 5},
CONF_FILTER_PRECISION: DEFAULT_PRECISION,
}
],
)
async def test_from_config_entry_duration(
recorder_mock: Recorder,
hass: HomeAssistant,
loaded_entry: MockConfigEntry,
) -> None:
"""Test if filter works loaded from config entry with duration."""
state = hass.states.get("sensor.filtered_sensor")
assert state.state == "20.0"
@pytest.mark.parametrize("missing", [True, False])
async def test_chain_history(
recorder_mock: Recorder,