diff --git a/CODEOWNERS b/CODEOWNERS index d4557e226b2177..d984aa95c4f6d5 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -831,6 +831,8 @@ CLAUDE.md @home-assistant/core /tests/components/incomfort/ @jbouwh /homeassistant/components/indevolt/ @xirt /tests/components/indevolt/ @xirt +/homeassistant/components/indoor_air_quality/ @liudger +/tests/components/indoor_air_quality/ @liudger /homeassistant/components/inels/ @epdevlab /tests/components/inels/ @epdevlab /homeassistant/components/influxdb/ @mdegat01 @Robbie1221 diff --git a/homeassistant/components/indoor_air_quality/__init__.py b/homeassistant/components/indoor_air_quality/__init__.py new file mode 100644 index 00000000000000..543001dd13227a --- /dev/null +++ b/homeassistant/components/indoor_air_quality/__init__.py @@ -0,0 +1,125 @@ +"""The Indoor Air Quality integration. + +Calculates an Indoor Air Quality (IAQ) index from a configurable set of +sensor sources, using a configurable rating standard. +""" + +import logging +from typing import Any, Final, cast + +import voluptuous as vol + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import CONF_DEVICE_ID, Platform +from homeassistant.core import Event, HomeAssistant, callback +from homeassistant.exceptions import ConfigEntryError +from homeassistant.helpers.event import ( + EventStateChangedData, + async_track_state_change_event, +) + +from .const import CONF_SOURCES, CONF_STANDARD, DEFAULT_STANDARD, STANDARDS +from .coordinator import SOURCE_SPECS, IndoorAirQualityController +from .helpers import entity_ids_from_sources + +_LOGGER: Final = logging.getLogger(__name__) + +PLATFORMS: Final = [Platform.SENSOR] + + +def _has_known_source(sources: dict[str, Any]) -> dict[str, Any]: + """Ensure at least one source key is recognised by the controller.""" + if not any(key in SOURCE_SPECS for key in sources): + raise vol.Invalid(f"At least one of {sorted(SOURCE_SPECS)} must be configured") + return sources + + +_ENTRY_DATA_SCHEMA: Final = vol.Schema( + { + vol.Required(CONF_SOURCES): vol.All( + vol.Schema({str: vol.Any(str, [str])}), + vol.Length(min=1), + _has_known_source, + ), + vol.Optional(CONF_STANDARD, default=DEFAULT_STANDARD): vol.In(STANDARDS), + vol.Optional(CONF_DEVICE_ID): vol.Any(str, None), + }, + extra=vol.ALLOW_EXTRA, +) + + +type IndoorAirQualityConfigEntry = ConfigEntry[IndoorAirQualityController] + + +def _validate_entry_data(entry: IndoorAirQualityConfigEntry) -> dict[str, Any]: + """Validate ``entry.data``; raise :class:`ConfigEntryError` on failure.""" + try: + validated = _ENTRY_DATA_SCHEMA(dict(entry.data)) + except vol.Invalid as err: + raise ConfigEntryError( + f"Invalid Indoor Air Quality config entry data: {err}" + ) from err + return cast(dict[str, Any], validated) + + +async def async_setup_entry( + hass: HomeAssistant, entry: IndoorAirQualityConfigEntry +) -> bool: + """Set up Indoor Air Quality from a config entry.""" + data = _validate_entry_data(entry) + sources: dict[str, str | list[str]] = data[CONF_SOURCES] + device_id: str | None = data.get(CONF_DEVICE_ID) + standard: str = data[CONF_STANDARD] + + _LOGGER.debug( + "Initialize controller %s (standard=%s) for sources: %s", + entry.entry_id, + standard, + ", ".join(f"{key}={value}" for key, value in sources.items()), + ) + + controller = IndoorAirQualityController( + hass, entry.entry_id, entry.title, sources, device_id, standard=standard + ) + entry.runtime_data = controller + + @callback + def _handle_state_change(event: Event[EventStateChangedData]) -> None: + """Recalculate the IAQ index and notify subscribed entities.""" + controller.update() + controller.async_update_listeners() + + entity_ids = entity_ids_from_sources(sources) + if entity_ids: + entry.async_on_unload( + async_track_state_change_event(hass, entity_ids, _handle_state_change) + ) + + # Compute initial state. + controller.update() + + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + entry.async_on_unload(entry.add_update_listener(_async_reload_entry)) + return True + + +async def async_unload_entry( + hass: HomeAssistant, entry: IndoorAirQualityConfigEntry +) -> bool: + """Unload a config entry.""" + return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) + + +async def _async_reload_entry( + hass: HomeAssistant, entry: IndoorAirQualityConfigEntry +) -> None: + """Reload config entry on options/data change.""" + await hass.config_entries.async_reload(entry.entry_id) + + +__all__ = [ + "IndoorAirQualityConfigEntry", + "IndoorAirQualityController", + "async_setup_entry", + "async_unload_entry", +] diff --git a/homeassistant/components/indoor_air_quality/bands.py b/homeassistant/components/indoor_air_quality/bands.py new file mode 100644 index 00000000000000..7a0dbb6bf4a2b4 --- /dev/null +++ b/homeassistant/components/indoor_air_quality/bands.py @@ -0,0 +1,148 @@ +"""Index band tables and scoring for the Indoor Air Quality integration. + +Adding a new rating standard is purely additive: define its per-source band +tuple, its level cut-offs and its top level, and register it in ``BANDS``, +``LEVEL_BANDS`` and ``LEVEL_TOP``. +""" + +from typing import Final, NamedTuple + +from .const import ( + CONF_CO, + CONF_CO2, + CONF_HCHO, + CONF_HUMIDITY, + CONF_NO2, + CONF_PM, + CONF_RADON, + CONF_TEMPERATURE, + CONF_TVOC, + CONF_VOC_INDEX, + LEVEL_EXCELLENT, + LEVEL_FAIR, + LEVEL_GOOD, + LEVEL_INADEQUATE, + LEVEL_POOR, + STANDARD_UK, +) + + +class ScoreBand(NamedTuple): + """Range -> score mapping used by the index calculator. + + A value matches a band when it satisfies both the optional lower and + upper bound. ``low_strict`` / ``high_strict`` switch the corresponding + bound between ``<=`` (default) and ``<``. + """ + + score: int + low: float | None = None + high: float | None = None + low_strict: bool = False + high_strict: bool = False + + +def score_from_bands(value: float, bands: tuple[ScoreBand, ...]) -> int: + """Return the score for the first matching band, defaulting to 1.""" + for band in bands: + if band.low is not None: + if band.low_strict: + if not band.low < value: + continue + elif not band.low <= value: + continue + if band.high is not None: + if band.high_strict: + if not value < band.high: + continue + elif not value <= band.high: + continue + return band.score + return 1 + + +_BANDS_UK: Final[dict[str, tuple[ScoreBand, ...]]] = { + CONF_TEMPERATURE: ( + ScoreBand(5, low=18, high=21), + ScoreBand(4, low=16, high=23, low_strict=True, high_strict=True), + ScoreBand(3, low=15, high=24, low_strict=True, high_strict=True), + ScoreBand(2, low=14, high=25, low_strict=True, high_strict=True), + ), + CONF_HUMIDITY: ( + ScoreBand(5, low=40, high=60), + ScoreBand(4, low=30, high=70), + ScoreBand(3, low=20, high=80), + ScoreBand(2, low=10, high=90), + ), + CONF_CO2: ( + ScoreBand(5, high=600, high_strict=True), + ScoreBand(4, high=800), + ScoreBand(3, high=1500), + ScoreBand(2, high=1800), + ), + CONF_TVOC: ( + ScoreBand(5, high=0.1, high_strict=True), + ScoreBand(4, high=0.3), + ScoreBand(3, high=0.5), + ScoreBand(2, high=1.0), + ), + CONF_VOC_INDEX: ( + ScoreBand(5, high=50), + ScoreBand(4, high=115), + ScoreBand(3, high=180), + ScoreBand(2, high=260), + ), + CONF_PM: ( + ScoreBand(5, high=23), + ScoreBand(4, high=41), + ScoreBand(3, high=53), + ScoreBand(2, high=64), + ), + CONF_NO2: ( + ScoreBand(5, high=200, high_strict=True), + ScoreBand(3, high=400), + ), + CONF_CO: ( + ScoreBand(5, low=0, high=0), + ScoreBand(3, high=7), + ), + CONF_HCHO: ( + ScoreBand(5, high=20, high_strict=True), + ScoreBand(4, high=50), + ScoreBand(3, high=100), + ScoreBand(2, high=200), + ), + CONF_RADON: ( + ScoreBand(5, low=0, high=0), + ScoreBand(3, high=20, high_strict=True), + ScoreBand(2, high=100), + ), +} + +# IAQ score (per-source 1-5, normalized to 13-65) -> human level. +_LEVEL_BANDS_UK: Final[tuple[tuple[int, str], ...]] = ( + (25, LEVEL_INADEQUATE), + (38, LEVEL_POOR), + (51, LEVEL_FAIR), + (60, LEVEL_GOOD), +) + +BANDS: Final[dict[str, dict[str, tuple[ScoreBand, ...]]]] = { + STANDARD_UK: _BANDS_UK, +} + +LEVEL_BANDS: Final[dict[str, tuple[tuple[int, str], ...]]] = { + STANDARD_UK: _LEVEL_BANDS_UK, +} + +LEVEL_TOP: Final[dict[str, str]] = { + STANDARD_UK: LEVEL_EXCELLENT, +} + + +def level_for_index(standard: str, iaq_index: int) -> str: + """Return the human readable level for an IAQ index value.""" + for upper, level in LEVEL_BANDS[standard]: + if iaq_index <= upper: + return level + return LEVEL_TOP[standard] diff --git a/homeassistant/components/indoor_air_quality/config_flow.py b/homeassistant/components/indoor_air_quality/config_flow.py new file mode 100644 index 00000000000000..e56d8815849b1e --- /dev/null +++ b/homeassistant/components/indoor_air_quality/config_flow.py @@ -0,0 +1,477 @@ +"""Config flow for the Indoor Air Quality integration.""" + +from __future__ import annotations + +import hashlib +import logging +from typing import Any + +import voluptuous as vol + +from homeassistant import config_entries +from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN, SensorDeviceClass +from homeassistant.const import CONF_DEVICE_ID, CONF_NAME +from homeassistant.core import HomeAssistant +from homeassistant.helpers import device_registry as dr, entity_registry as er, selector + +from .const import ( + CONF_CO, + CONF_CO2, + CONF_HCHO, + CONF_HUMIDITY, + CONF_NO2, + CONF_PM, + CONF_RADON, + CONF_SOURCES, + CONF_STANDARD, + CONF_TEMPERATURE, + CONF_TVOC, + CONF_VOC_INDEX, + DEFAULT_STANDARD, + DOMAIN, + STANDARDS, +) + +_LOGGER = logging.getLogger(__name__) + +CONF_SHOW_SOURCE_OPTIONS = "show_source_options" +DEFAULT_NAME = "Indoor Air Quality" + +PM_DEVICE_CLASSES = { + SensorDeviceClass.PM1, + SensorDeviceClass.PM10, + SensorDeviceClass.PM25, +} +PM_DEVICE_CLASS_VALUES = {str(device_class) for device_class in PM_DEVICE_CLASSES} + +SOURCE_DEVICE_CLASSES = { + SensorDeviceClass.TEMPERATURE: CONF_TEMPERATURE, + SensorDeviceClass.HUMIDITY: CONF_HUMIDITY, + SensorDeviceClass.CO2: CONF_CO2, + SensorDeviceClass.CO: CONF_CO, + SensorDeviceClass.NITROGEN_DIOXIDE: CONF_NO2, +} + +TVOC_DEVICE_CLASSES = { + SensorDeviceClass.VOLATILE_ORGANIC_COMPOUNDS, + SensorDeviceClass.VOLATILE_ORGANIC_COMPOUNDS_PARTS, +} +TVOC_DEVICE_CLASS_VALUES = {str(device_class) for device_class in TVOC_DEVICE_CLASSES} + +DEVICE_SELECTOR = selector.DeviceSelector( + selector.DeviceSelectorConfig( + entity=selector.EntityFilterSelectorConfig(domain=SENSOR_DOMAIN), + ) +) + +STANDARD_SELECTOR = selector.SelectSelector( + selector.SelectSelectorConfig( + options=list(STANDARDS), + translation_key=CONF_STANDARD, + mode=selector.SelectSelectorMode.DROPDOWN, + ) +) + +# Source configuration options +SOURCE_SELECTORS = { + CONF_TEMPERATURE: selector.EntitySelector( + selector.EntitySelectorConfig( + domain=SENSOR_DOMAIN, + device_class=SensorDeviceClass.TEMPERATURE, + ) + ), + CONF_HUMIDITY: selector.EntitySelector( + selector.EntitySelectorConfig( + domain=SENSOR_DOMAIN, + device_class=SensorDeviceClass.HUMIDITY, + ) + ), + CONF_CO2: selector.EntitySelector( + selector.EntitySelectorConfig( + domain=SENSOR_DOMAIN, + device_class=SensorDeviceClass.CO2, + ) + ), + CONF_TVOC: selector.EntitySelector( + selector.EntitySelectorConfig( + domain=SENSOR_DOMAIN, + device_class=list(TVOC_DEVICE_CLASSES), + ) + ), + CONF_VOC_INDEX: selector.EntitySelector( + selector.EntitySelectorConfig(domain=SENSOR_DOMAIN) + ), + CONF_PM: selector.EntitySelector( + selector.EntitySelectorConfig( + domain=SENSOR_DOMAIN, + device_class=list(PM_DEVICE_CLASSES), + multiple=True, + ) + ), + CONF_NO2: selector.EntitySelector( + selector.EntitySelectorConfig( + domain=SENSOR_DOMAIN, + device_class=SensorDeviceClass.NITROGEN_DIOXIDE, + ) + ), + CONF_CO: selector.EntitySelector( + selector.EntitySelectorConfig( + domain=SENSOR_DOMAIN, + device_class=SensorDeviceClass.CO, + ) + ), + CONF_HCHO: selector.EntitySelector( + selector.EntitySelectorConfig(domain=SENSOR_DOMAIN) + ), + CONF_RADON: selector.EntitySelector( + selector.EntitySelectorConfig(domain=SENSOR_DOMAIN) + ), +} + + +def _source_schema_fields( + defaults: dict[str, Any] | None = None, +) -> dict[vol.Optional, selector.EntitySelector]: + """Return flat source selector fields for config and options flows.""" + defaults = defaults or {} + + return { + vol.Optional(source, default=defaults[source]) + if source in defaults + else vol.Optional(source): source_selector + for source, source_selector in SOURCE_SELECTORS.items() + } + + +def _has_at_least_one_source(sources: dict[str, Any]) -> bool: + """Check if at least one source is configured.""" + return any(sources.values()) + + +def _validate_voc_sources(sources: dict[str, Any]) -> dict[str, str]: + """Validate that only one of TVOC or VOC_INDEX is provided.""" + errors = {} + + if sources.get(CONF_TVOC) and sources.get(CONF_VOC_INDEX): + errors["base"] = "only_one_voc_sensor" + + return errors + + +def _clean_sources(sources: dict[str, Any]) -> dict[str, Any]: + """Remove empty source selections.""" + return {key: value for key, value in sources.items() if value} + + +def _device_name(hass: HomeAssistant, device_id: str | None) -> str | None: + """Return the best available friendly name for a device.""" + if not device_id: + return None + + device_registry = dr.async_get(hass) + device_entry = device_registry.async_get(device_id) + if device_entry is None: + return None + + return device_entry.name_by_user or device_entry.name + + +def _sources_from_user_input(user_input: dict[str, Any]) -> dict[str, Any]: + """Return source selections submitted by the user.""" + return _clean_sources( + {source: user_input.get(source) for source in SOURCE_SELECTORS} + ) + + +def _entity_labels(entry: er.RegistryEntry) -> str: + """Return searchable labels for an entity registry entry.""" + return " ".join( + str(value).lower().replace("_", " ") + for value in ( + entry.entity_id, + entry.name, + entry.original_name, + entry.translation_key, + ) + if value + ) + + +def _entity_device_classes(entry: er.RegistryEntry) -> set[str]: + """Return all available device classes for an entity registry entry.""" + return { + str(device_class) + for device_class in (entry.device_class, entry.original_device_class) + if device_class + } + + +def _source_key_from_entry(labels: str, device_classes: set[str]) -> str | None: + """Return a source key for an entity registry entry.""" + for device_class, source_key in SOURCE_DEVICE_CLASSES.items(): + if str(device_class) in device_classes: + return source_key + + if "hcho" in labels or "formaldehyde" in labels: + return CONF_HCHO + if "radon" in labels: + return CONF_RADON + + return None + + +def _is_voc_index_entry(labels: str) -> bool: + """Return whether labels describe a VOC index sensor.""" + return "voc index" in labels or "vocindex" in labels + + +def _is_tvoc_entry(labels: str, device_classes: set[str]) -> bool: + """Return whether labels or device classes describe a tVOC sensor.""" + return bool(device_classes & TVOC_DEVICE_CLASS_VALUES) or bool( + "tvoc" in labels or "volatile organic" in labels + ) + + +def _sources_from_device(hass: HomeAssistant, device_id: str) -> dict[str, Any]: + """Build source configuration from sensors attached to a device.""" + entity_registry = er.async_get(hass) + sources: dict[str, Any] = {} + pm_sources: list[str] = [] + tvoc_source = None + voc_index_source = None + + for entry in er.async_entries_for_device(entity_registry, device_id): + if entry.domain != SENSOR_DOMAIN: + continue + + labels = _entity_labels(entry) + device_classes = _entity_device_classes(entry) + + if device_classes & PM_DEVICE_CLASS_VALUES: + pm_sources.append(entry.entity_id) + continue + + if labels and not voc_index_source and _is_voc_index_entry(labels): + voc_index_source = entry.entity_id + continue + + if _is_tvoc_entry(labels, device_classes): + tvoc_source = tvoc_source or entry.entity_id + continue + + if source_key := _source_key_from_entry(labels, device_classes): + sources.setdefault(source_key, entry.entity_id) + + if pm_sources: + sources[CONF_PM] = sorted(pm_sources) + + if voc_index_source: + sources[CONF_VOC_INDEX] = voc_index_source + elif tvoc_source: + sources[CONF_TVOC] = tvoc_source + + return sources + + +def _sources_from_input( + hass: HomeAssistant, user_input: dict[str, Any] +) -> dict[str, Any]: + """Build source configuration from a selected device and manual overrides.""" + sources = {} + if device_id := user_input.get(CONF_DEVICE_ID): + sources.update(_sources_from_device(hass, device_id)) + + sources.update(_sources_from_user_input(user_input)) + return sources + + +def _unique_id_from_input(user_input: dict[str, Any], name: str) -> str: + """Return the config entry unique ID for the selected input. + + For a device-based entry the device id is used directly. For a manual + setup the unique id is a stable hash of the user-provided name so that + the entry id stays stable even if the user later renames the entry. + """ + if device_id := user_input.get(CONF_DEVICE_ID): + return device_id + + return hashlib.sha256(name.encode("utf-8")).hexdigest()[:16] + + +class IndoorAirQualityConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow for Indoor Air Quality.""" + + VERSION = 1 + + def __init__(self) -> None: + """Initialize the config flow.""" + self._device_id: str | None = None + self._detected_sources: dict[str, Any] = {} + self._name = DEFAULT_NAME + self._standard: str = DEFAULT_STANDARD + + async def _async_create_config_entry( + self, + user_input: dict[str, Any], + sources: dict[str, Any], + ) -> config_entries.ConfigFlowResult: + """Validate input and create a config entry.""" + errors = {} + + if not _has_at_least_one_source(sources): + errors["base"] = ( + "no_matching_sources" + if user_input.get(CONF_DEVICE_ID) + else "no_sources" + ) + else: + errors.update(_validate_voc_sources(sources)) + + if errors: + return self.async_show_form( + step_id="user", + data_schema=self._user_schema(), + errors=errors, + ) + + name = ( + user_input.get(CONF_NAME) + or _device_name(self.hass, user_input.get(CONF_DEVICE_ID)) + or DEFAULT_NAME + ) + + data: dict[str, Any] = { + CONF_SOURCES: sources, + CONF_STANDARD: user_input.get(CONF_STANDARD, DEFAULT_STANDARD), + } + if device_id := user_input.get(CONF_DEVICE_ID): + data[CONF_DEVICE_ID] = device_id + + await self.async_set_unique_id(_unique_id_from_input(user_input, name)) + self._abort_if_unique_id_configured() + + return self.async_create_entry( + title=name, + data=data, + ) + + def _user_schema(self) -> vol.Schema: + """Return the first-step config schema.""" + return vol.Schema( + { + vol.Optional(CONF_DEVICE_ID): DEVICE_SELECTOR, + vol.Optional( + CONF_STANDARD, default=DEFAULT_STANDARD + ): STANDARD_SELECTOR, + vol.Optional( + CONF_SHOW_SOURCE_OPTIONS, + default=False, + ): selector.BooleanSelector(), + } + ) + + def _sources_schema(self) -> vol.Schema: + """Return the manual source selection schema.""" + return vol.Schema( + { + vol.Optional(CONF_NAME, default=self._name): str, + **_source_schema_fields(self._detected_sources), + } + ) + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> config_entries.ConfigFlowResult: + """Handle the initial step.""" + errors = {} + + if user_input is not None: + self._standard = user_input.get(CONF_STANDARD, DEFAULT_STANDARD) + device_id = user_input.get(CONF_DEVICE_ID) + sources = _sources_from_input(self.hass, user_input) + + if user_input.get(CONF_SHOW_SOURCE_OPTIONS): + self._device_id = device_id + self._detected_sources = sources + self._name = _device_name(self.hass, device_id) or DEFAULT_NAME + return await self.async_step_sources() + + if not device_id: + errors["base"] = "device_or_sources" + else: + return await self._async_create_config_entry(user_input, sources) + + return self.async_show_form( + step_id="user", + data_schema=self._user_schema(), + errors=errors, + ) + + async def async_step_sources( + self, user_input: dict[str, Any] | None = None + ) -> config_entries.ConfigFlowResult: + """Handle the optional manual source selection step.""" + errors = {} + + if user_input is not None: + sources = self._detected_sources | _sources_from_user_input(user_input) + errors.update(_validate_voc_sources(sources)) + + if not _has_at_least_one_source(sources): + errors["base"] = "no_sources" + + if not errors: + return await self._async_create_config_entry( + { + CONF_NAME: user_input.get(CONF_NAME) or self._name, + CONF_DEVICE_ID: self._device_id, + CONF_STANDARD: self._standard, + }, + sources, + ) + + return self.async_show_form( + step_id="sources", + data_schema=self._sources_schema(), + errors=errors, + ) + + async def async_step_reconfigure( + self, user_input: dict[str, Any] | None = None + ) -> config_entries.ConfigFlowResult: + """Allow the user to reconfigure standard and sources of an entry.""" + entry = self._get_reconfigure_entry() + current_sources = entry.data.get(CONF_SOURCES, {}) + current_standard = entry.data.get(CONF_STANDARD, DEFAULT_STANDARD) + + errors: dict[str, str] = {} + + if user_input is not None: + sources = _sources_from_user_input(user_input) + + if not _has_at_least_one_source(sources): + errors["base"] = "no_sources" + else: + errors.update(_validate_voc_sources(sources)) + + if not errors: + return self.async_update_reload_and_abort( + entry, + data_updates={ + CONF_SOURCES: sources, + CONF_STANDARD: user_input.get(CONF_STANDARD, current_standard), + }, + ) + + return self.async_show_form( + step_id="reconfigure", + data_schema=vol.Schema( + { + vol.Optional( + CONF_STANDARD, default=current_standard + ): STANDARD_SELECTOR, + **_source_schema_fields(current_sources), + } + ), + errors=errors, + ) diff --git a/homeassistant/components/indoor_air_quality/const.py b/homeassistant/components/indoor_air_quality/const.py new file mode 100644 index 00000000000000..11a24e2caebf80 --- /dev/null +++ b/homeassistant/components/indoor_air_quality/const.py @@ -0,0 +1,104 @@ +"""Constants for the Indoor Air Quality integration.""" + +from typing import Final + +from homeassistant.const import ( + CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, + CONCENTRATION_MILLIGRAMS_PER_CUBIC_METER, +) + +# Base component constants +NAME: Final = "Indoor Air Quality" +DOMAIN: Final = "indoor_air_quality" + +# Supported air quality rating standards. +# Additional standards (e.g. EPA AQI, EU CAQI, WHO 2021) can be added here +# without changing the config-flow schema. +STANDARD_UK: Final = "uk" +STANDARDS: Final = (STANDARD_UK,) +DEFAULT_STANDARD: Final = STANDARD_UK +CONF_STANDARD: Final = "standard" + +SENSOR_INDEX: Final = "iaq_index" +SENSOR_LEVEL: Final = "iaq_level" + +SENSOR_TYPES: Final = (SENSOR_INDEX, SENSOR_LEVEL) + +# Configuration and options +CONF_SOURCES: Final = "sources" +CONF_TEMPERATURE: Final = "temperature" +CONF_HUMIDITY: Final = "humidity" +CONF_CO2: Final = "co2" +CONF_TVOC: Final = "tvoc" +CONF_VOC_INDEX: Final = "voc_index" +CONF_PM: Final = "pm" +CONF_NO2: Final = "no2" +CONF_CO: Final = "co" +CONF_HCHO: Final = "hcho" # Formaldehyde +CONF_RADON: Final = "radon" + +# Attributes +ATTR_SOURCES_SET: Final = "sources_set" +ATTR_SOURCES_USED: Final = "sources_used" +ATTR_SOURCE_INDEX_TPL: Final = "{}_index" + + +# Lowercase translation-key style level identifiers used as the +# ``iaq_level`` sensor's state. Display labels live in ``strings.json``. +LEVEL_EXCELLENT: Final = "excellent" +LEVEL_GOOD: Final = "good" +LEVEL_FAIR: Final = "fair" +LEVEL_POOR: Final = "poor" +LEVEL_INADEQUATE: Final = "inadequate" + +LEVELS: Final = ( + LEVEL_EXCELLENT, + LEVEL_GOOD, + LEVEL_FAIR, + LEVEL_POOR, + LEVEL_INADEQUATE, +) + +UNIT_PPM: Final = { + "ppm": 1, # Target unit -- conversion rate will be ignored + "ppb": 0.001, +} +UNIT_PPB: Final = { + "ppb": 1, # Target unit -- conversion rate will be ignored + "ppm": 1000, +} +UNIT_UGM3: Final = { + # First entry is the target unit; remaining entries are aliases / sibling + # units with their multiplicative factor to the target. + CONCENTRATION_MICROGRAMS_PER_CUBIC_METER: 1, # "μg/m³" (Greek mu) + "µg/m³": 1, # micro sign alias + "µg/m3": 1, + "µg/m^3": 1, + "μg/m3": 1, # Greek mu alias variants + "μg/m^3": 1, + "ug/m³": 1, + "ug/m3": 1, + "ug/m^3": 1, + CONCENTRATION_MILLIGRAMS_PER_CUBIC_METER: 1000, + "mg/m3": 1000, + "mg/m^3": 1000, +} +UNIT_MGM3: Final = { + CONCENTRATION_MILLIGRAMS_PER_CUBIC_METER: 1, # Target unit + "mg/m3": 1, + "mg/m^3": 1, + CONCENTRATION_MICROGRAMS_PER_CUBIC_METER: 0.001, + "µg/m³": 0.001, + "µg/m3": 0.001, + "µg/m^3": 0.001, + "μg/m3": 0.001, + "μg/m^3": 0.001, + "ug/m³": 0.001, + "ug/m3": 0.001, + "ug/m^3": 0.001, +} + +# Molar masses (g/mol) used to convert between ppm/ppb and µg/m³ / mg/m³. +MOLAR_MASS_TVOC: Final = 78.9516 +MOLAR_MASS_HCHO: Final = 30.0260 +MOLAR_MASS_CO2: Final = 44.0100 diff --git a/homeassistant/components/indoor_air_quality/coordinator.py b/homeassistant/components/indoor_air_quality/coordinator.py new file mode 100644 index 00000000000000..2698494d583ea2 --- /dev/null +++ b/homeassistant/components/indoor_air_quality/coordinator.py @@ -0,0 +1,380 @@ +"""Indoor Air Quality controller.""" + +from collections.abc import Callable, Mapping +from dataclasses import dataclass +import logging +from typing import Any, Final + +from homeassistant.const import ( + CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, + CONCENTRATION_MILLIGRAMS_PER_CUBIC_METER, + PERCENTAGE, + UnitOfTemperature, +) +from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback +from homeassistant.util.unit_conversion import ( + BaseUnitConverter, + CarbonMonoxideConcentrationConverter, + MassVolumeConcentrationConverter, + NitrogenDioxideConcentrationConverter, + TemperatureConverter, +) + +from .bands import BANDS, level_for_index, score_from_bands +from .const import ( + ATTR_SOURCE_INDEX_TPL, + ATTR_SOURCES_SET, + ATTR_SOURCES_USED, + CONF_CO, + CONF_CO2, + CONF_HCHO, + CONF_HUMIDITY, + CONF_NO2, + CONF_PM, + CONF_RADON, + CONF_TEMPERATURE, + CONF_TVOC, + CONF_VOC_INDEX, + DEFAULT_STANDARD, + MOLAR_MASS_CO2, + MOLAR_MASS_HCHO, + MOLAR_MASS_TVOC, + UNIT_MGM3, + UNIT_PPM, + UNIT_UGM3, +) +from .helpers import convert_value, normalise_unit, resolve_state + +_LOGGER: Final = logging.getLogger(__name__) + + +@dataclass(frozen=True, slots=True) +class SourceSpec: + """Declarative configuration for a single IAQ source. + + Either ``converter`` (a Home Assistant core unit converter) or + ``target_units`` (a custom alias/factor mapping with optional + ``molar_mass``) is used to normalise the source value into + ``target_unit``. ``valid_units`` rejects sources that report a unit + outside the listed set without performing any conversion. + """ + + converter: type[BaseUnitConverter] | None = None + target_unit: str | None = None + target_units: Mapping[str, float] | None = None + molar_mass: float | None = None + is_list: bool = False + valid_units: frozenset[str | None] | None = None + + +# Per-source conversion declarations. Sources missing here are ignored. +SOURCE_SPECS: Final[dict[str, SourceSpec]] = { + CONF_TEMPERATURE: SourceSpec( + valid_units=frozenset({UnitOfTemperature.CELSIUS, UnitOfTemperature.FAHRENHEIT}) + ), + CONF_HUMIDITY: SourceSpec(valid_units=frozenset({PERCENTAGE})), + CONF_CO2: SourceSpec(target_units=UNIT_PPM, molar_mass=MOLAR_MASS_CO2), + CONF_TVOC: SourceSpec(target_units=UNIT_MGM3, molar_mass=MOLAR_MASS_TVOC), + CONF_VOC_INDEX: SourceSpec(), + CONF_PM: SourceSpec( + converter=MassVolumeConcentrationConverter, + target_unit=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, + is_list=True, + ), + CONF_NO2: SourceSpec( + converter=NitrogenDioxideConcentrationConverter, + target_unit=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, + ), + CONF_CO: SourceSpec( + converter=CarbonMonoxideConcentrationConverter, + target_unit=CONCENTRATION_MILLIGRAMS_PER_CUBIC_METER, + ), + CONF_HCHO: SourceSpec(target_units=UNIT_UGM3, molar_mass=MOLAR_MASS_HCHO), + CONF_RADON: SourceSpec(valid_units=frozenset({None, "Bq/m3", "Bq/m³"})), +} + + +class IndoorAirQualityController: + """Indoor Air Quality controller.""" + + def __init__( + self, + hass: HomeAssistant, + entry_id: str, + name: str, + sources: Mapping[str, str | list[str]], + device_id: str | None = None, + *, + standard: str = DEFAULT_STANDARD, + ) -> None: + """Initialize controller.""" + self.hass = hass + self._entry_id = entry_id + self._name = name + self._sources = sources + self._device_id = device_id + self._standard = standard + + self._iaq_index: int | None = None + self._iaq_sources = 0 + self._indexes: dict[str, int] = {} + self._listeners: list[CALLBACK_TYPE] = [] + + # --- public read-only properties --------------------------------------- + + @property + def entry_id(self) -> str: + """Return the config entry id this controller belongs to.""" + return self._entry_id + + @property + def unique_id(self) -> str: + """Return a unique ID.""" + return self._entry_id + + @property + def name(self) -> str: + """Get controller name.""" + return self._name + + @property + def device_id(self) -> str | None: + """Return the source device ID.""" + return self._device_id + + @property + def standard(self) -> str: + """Return the active rating standard.""" + return self._standard + + @property + def iaq_index(self) -> int | None: + """Get IAQ index.""" + return self._iaq_index + + @property + def iaq_level(self) -> str | None: + """Get IAQ level.""" + if self._iaq_index is None: + return None + return level_for_index(self._standard, self._iaq_index) + + @property + def extra_state_attributes(self) -> dict[str, Any]: + """Return diagnostic attributes for the IAQ sensors.""" + attrs: dict[str, Any] = { + ATTR_SOURCES_SET: len(self._sources), + ATTR_SOURCES_USED: self._iaq_sources, + } + for src, idx in self._indexes.items(): + attrs[ATTR_SOURCE_INDEX_TPL.format(src)] = idx + return attrs + + # --- listener API ------------------------------------------------------ + + @callback + def async_add_listener(self, update_callback: CALLBACK_TYPE) -> CALLBACK_TYPE: + """Subscribe to controller updates and return an unsubscribe callback.""" + self._listeners.append(update_callback) + + @callback + def remove_listener() -> None: + if update_callback in self._listeners: + self._listeners.remove(update_callback) + + return remove_listener + + @callback + def async_update_listeners(self) -> None: + """Notify all subscribed listeners of an update.""" + for listener in list(self._listeners): + listener() + + # --- index calculation ------------------------------------------------- + + def update(self) -> None: + """Recalculate the IAQ index from the configured sources.""" + _LOGGER.debug("[%s] State update", self._entry_id) + + bands_for_standard = BANDS.get(self._standard) + if bands_for_standard is None: + _LOGGER.warning( + "[%s] Unknown IAQ standard %r", self._entry_id, self._standard + ) + self._reset_state() + return + + total = 0 + used = 0 + indexes: dict[str, int] = {} + + for src in self._sources: + spec = SOURCE_SPECS.get(src) + if spec is None: + _LOGGER.debug("[%s] Unknown source %s", self._entry_id, src) + continue + + value = self._resolve_source(src, spec) + if value is None: + continue + + bands = bands_for_standard.get(src) + if bands is None: + _LOGGER.debug( + "[%s] No bands for source %s in standard %s", + self._entry_id, + src, + self._standard, + ) + continue + + idx = score_from_bands(value, bands) + _LOGGER.debug( + "[%s] %s_index=%s (value=%s)", self._entry_id, src, idx, value + ) + indexes[src] = idx + total += idx + used += 1 + + if used: + self._indexes = indexes + self._iaq_index = int((65 * total) / (5 * used)) + self._iaq_sources = used + _LOGGER.debug( + "[%s] Update IAQ index to %d (%d sources used)", + self._entry_id, + self._iaq_index, + self._iaq_sources, + ) + else: + self._reset_state() + _LOGGER.debug("[%s] No usable sources, IAQ index cleared", self._entry_id) + + def _reset_state(self) -> None: + """Clear all computed state so sensors report unavailable.""" + self._iaq_index = None + self._iaq_sources = 0 + self._indexes = {} + + # --- per-source resolution --------------------------------------------- + + def _resolve_source(self, source: str, spec: SourceSpec) -> float | None: + """Resolve a configured source into the controller's calculation unit.""" + # Specialised paths first; these don't fit the generic conversion model. + if source == CONF_TEMPERATURE: + return self._resolve_temperature() + if source == CONF_VOC_INDEX: + return self._resolve_dimensionless(source) + + if spec.is_list: + return self._resolve_list(source, spec) + + entity_id = self._single_entity_id(source) + if entity_id is None: + return None + resolved = resolve_state(self.hass, entity_id) + if resolved is None: + return None + value, unit = resolved + + if spec.valid_units is not None: + if unit not in spec.valid_units: + _LOGGER.debug( + "Entity %s has unsupported %s unit %r", entity_id, source, unit + ) + return None + return value + + return self._normalise(source, entity_id, value, unit, spec) + + def _normalise( + self, + source: str, + entity_id: str, + value: float, + unit: str | None, + spec: SourceSpec, + ) -> float | None: + """Normalise ``value`` into the source's target unit.""" + if spec.converter is not None and spec.target_unit is not None: + normalised_unit = normalise_unit(unit) + if normalised_unit not in spec.converter.VALID_UNITS: + _LOGGER.debug( + "Entity %s has unsupported %s unit %r", entity_id, source, unit + ) + return None + return spec.converter.convert(value, normalised_unit, spec.target_unit) + + if spec.target_units is None: + return value + + return convert_value( + value, + unit, + spec.target_units, + molar_mass=spec.molar_mass, + source_type=source, + ) + + def _resolve_dimensionless(self, source: str) -> float | None: + """Resolve a numeric, unit-agnostic source (e.g. VOC index).""" + entity_id = self._single_entity_id(source) + if entity_id is None: + return None + resolved = resolve_state(self.hass, entity_id) + if resolved is None: + return None + return resolved[0] + + def _resolve_temperature(self) -> float | None: + """Resolve temperature in °C.""" + entity_id = self._single_entity_id(CONF_TEMPERATURE) + if entity_id is None: + return None + resolved = resolve_state(self.hass, entity_id) + if resolved is None: + return None + value, unit = resolved + if unit not in (UnitOfTemperature.CELSIUS, UnitOfTemperature.FAHRENHEIT): + _LOGGER.debug( + "Entity %s has unsupported temperature unit %r", + entity_id, + unit, + ) + return None + if unit != UnitOfTemperature.CELSIUS: + value = TemperatureConverter.convert(value, unit, UnitOfTemperature.CELSIUS) + return value + + def _resolve_list(self, source: str, spec: SourceSpec) -> float | None: + """Resolve a list-of-entities source, summing converted values.""" + entity_ids = self._sources.get(source) + if not entity_ids: + return None + if not isinstance(entity_ids, list): + entity_ids = [entity_ids] + + values: list[float] = [] + for eid in entity_ids: + resolved = resolve_state(self.hass, eid) + if resolved is None: + continue + value, unit = resolved + converted = self._normalise(source, eid, value, unit, spec) + if converted is not None: + values.append(converted) + + if not values: + return None + return sum(values) + + def _single_entity_id(self, source: str) -> str | None: + """Return a configured single-entity source id, ignoring lists.""" + entity_id = self._sources.get(source) + if not entity_id or isinstance(entity_id, list): + return None + return entity_id + + +# Subscribers that can receive notifications from a controller. +ControllerListener = Callable[[], None] diff --git a/homeassistant/components/indoor_air_quality/helpers.py b/homeassistant/components/indoor_air_quality/helpers.py new file mode 100644 index 00000000000000..337e4848e7b493 --- /dev/null +++ b/homeassistant/components/indoor_air_quality/helpers.py @@ -0,0 +1,143 @@ +"""Pure helpers for the Indoor Air Quality integration.""" + +from collections.abc import Mapping +import logging +from typing import Final + +from homeassistant.const import ( + ATTR_UNIT_OF_MEASUREMENT, + CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, + CONCENTRATION_MILLIGRAMS_PER_CUBIC_METER, + STATE_UNAVAILABLE, + STATE_UNKNOWN, +) +from homeassistant.core import HomeAssistant + +_LOGGER: Final = logging.getLogger(__name__) + +_WARNED_MISSING_ENTITIES: set[str] = set() + + +_UGM3_UNITS: Final = frozenset( + { + CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, # canonical "μg/m³" (Greek mu) + "µg/m³", # micro sign (U+00B5) + "µg/m3", + "µg/m^3", + "μg/m3", # Greek mu variants without superscript + "μg/m^3", + "ug/m³", + "ug/m3", + "ug/m^3", + } +) +_MGM3_UNITS: Final = frozenset( + {CONCENTRATION_MILLIGRAMS_PER_CUBIC_METER, "mg/m3", "mg/m^3"} +) + + +def normalise_unit(unit: str | None) -> str | None: + """Normalise common ASCII / micro-sign concentration unit aliases. + + Sensors typically report µg/m³ using the micro-sign (U+00B5) while + Home Assistant's canonical constants use the Greek mu (U+03BC). + Variants like ``ug/m3`` or ``mg/m^3`` are also normalised so that the + core unit converters accept the value. + """ + if unit is None: + return None + if unit in _UGM3_UNITS: + return CONCENTRATION_MICROGRAMS_PER_CUBIC_METER + if unit in _MGM3_UNITS: + return CONCENTRATION_MILLIGRAMS_PER_CUBIC_METER + return unit + + +def has_state(state: str | None) -> bool: + """Return True if a state has a usable value.""" + return state is not None and state not in (STATE_UNKNOWN, STATE_UNAVAILABLE) + + +def resolve_state( + hass: HomeAssistant, entity_id: str +) -> tuple[float, str | None] | None: + """Resolve an entity's numeric state and unit, or ``None``.""" + entity = hass.states.get(entity_id) + if entity is None: + if entity_id not in _WARNED_MISSING_ENTITIES: + _WARNED_MISSING_ENTITIES.add(entity_id) + _LOGGER.warning("Entity %s not found", entity_id) + else: + _LOGGER.debug("Entity %s not found", entity_id) + return None + _WARNED_MISSING_ENTITIES.discard(entity_id) + if not has_state(entity.state): + _LOGGER.debug("State of entity %s is unknown", entity_id) + return None + try: + value = float(entity.state) + except TypeError, ValueError: + _LOGGER.debug("State of entity %s is not numeric: %r", entity_id, entity.state) + return None + return value, entity.attributes.get(ATTR_UNIT_OF_MEASUREMENT) + + +def convert_value( + value: float, + unit: str | None, + target_units: Mapping[str, float], + *, + molar_mass: float | None = None, + source_type: str = "", +) -> float | None: + """Convert ``value`` from ``unit`` to the target unit. + + The first key of ``target_units`` is the target unit; remaining keys list + aliases and like-family conversions with their multiplicative factor. + + Cross-family conversion is supported in one direction only: from ppm/ppb + to a mass concentration (mg/m³ or µg/m³) using the supplied + ``molar_mass`` and the standard 24.45 L/mol molar volume at 25 °C / + 1 atm. The reverse direction (mass concentration → ppm/ppb) is not + implemented because the only ppm-targeted source (CO₂) is reported in + ppm by virtually every sensor in practice; sources reporting in another + family are treated as non-convertible. + """ + if unit is not None and unit in target_units: + return value * target_units[unit] + + target_unit = next(iter(target_units)) + + if molar_mass is None or unit not in {"ppm", "ppb"}: + _LOGGER.debug( + "Unit %r is not convertible to %r for %s source", + unit, + target_unit, + source_type, + ) + return None + + mm = molar_mass + if "ppb" in (unit, target_unit): + mm /= 1000 + if target_unit in _UGM3_UNITS: + mm *= 1000 + return value * (mm / 24.45) + + +def entity_ids_from_sources(sources: Mapping[str, str | list[str]]) -> list[str]: + """Flatten configured sources to a de-duplicated list of entity ids. + + Order is preserved so that callers (e.g. tests, listener registration) + see a stable sequence even when the same entity is selected for + multiple source fields. + """ + entity_ids: dict[str, None] = {} + for value in sources.values(): + if isinstance(value, list): + for entity_id in value: + if entity_id: + entity_ids.setdefault(entity_id) + elif value: + entity_ids.setdefault(value) + return list(entity_ids) diff --git a/homeassistant/components/indoor_air_quality/icons.json b/homeassistant/components/indoor_air_quality/icons.json new file mode 100644 index 00000000000000..55bea1f60b2526 --- /dev/null +++ b/homeassistant/components/indoor_air_quality/icons.json @@ -0,0 +1,19 @@ +{ + "entity": { + "sensor": { + "iaq_index": { + "default": "mdi:air-filter" + }, + "iaq_level": { + "default": "mdi:air-filter", + "state": { + "excellent": "mdi:emoticon-excited", + "fair": "mdi:emoticon-neutral", + "good": "mdi:emoticon-happy", + "inadequate": "mdi:emoticon-dead", + "poor": "mdi:emoticon-sad" + } + } + } + } +} diff --git a/homeassistant/components/indoor_air_quality/manifest.json b/homeassistant/components/indoor_air_quality/manifest.json new file mode 100644 index 00000000000000..4234f3130129c8 --- /dev/null +++ b/homeassistant/components/indoor_air_quality/manifest.json @@ -0,0 +1,11 @@ +{ + "domain": "indoor_air_quality", + "name": "Indoor Air Quality", + "codeowners": ["@liudger"], + "config_flow": true, + "documentation": "https://www.home-assistant.io/integrations/indoor_air_quality", + "integration_type": "helper", + "iot_class": "calculated", + "quality_scale": "bronze", + "requirements": [] +} diff --git a/homeassistant/components/indoor_air_quality/quality_scale.yaml b/homeassistant/components/indoor_air_quality/quality_scale.yaml new file mode 100644 index 00000000000000..791d1be7c25ee6 --- /dev/null +++ b/homeassistant/components/indoor_air_quality/quality_scale.yaml @@ -0,0 +1,112 @@ +rules: + # Bronze + action-setup: + status: exempt + comment: | + This integration does not provide any service actions; it only exposes + sensor entities derived from configured source entities. + appropriate-polling: + status: exempt + comment: | + The integration does not poll. Updates are driven by state changes of + the configured source entities via async_track_state_change_event. + brands: done + common-modules: done + config-flow-test-coverage: done + config-flow: done + dependency-transparency: + status: exempt + comment: The integration has no external runtime dependencies. + docs-actions: + status: exempt + comment: This integration does not provide any service actions. + docs-high-level-description: done + docs-installation-instructions: done + docs-removal-instructions: done + entity-event-setup: done + entity-unique-id: done + has-entity-name: done + runtime-data: done + test-before-configure: + status: exempt + comment: | + This integration is a calculated helper. It does not connect to any + external service or device that would benefit from a connection test. + test-before-setup: + status: exempt + comment: | + Setup only wires up state-change listeners for already-configured + entities; there is nothing to test before setup. + unique-config-entry: done + # Silver + action-exceptions: + status: exempt + comment: This integration does not provide any service actions. + config-entry-unloading: done + docs-configuration-parameters: todo + docs-installation-parameters: todo + entity-unavailable: todo + integration-owner: done + log-when-unavailable: + status: exempt + comment: | + Source-entity availability is reflected in the calculated index values + directly; there is no separate connection state to log. + parallel-updates: + status: done + comment: | + The sensor platform is push-based: entities set ``_attr_should_poll = + False`` and are written to Home Assistant from controller listeners + that fire on source state-change events. ``PARALLEL_UPDATES = 0`` is + set on the platform to document that no concurrent-update cap is + needed. + reauthentication-flow: + status: exempt + comment: This integration does not authenticate against any service. + test-coverage: todo + + # Gold + devices: todo + diagnostics: todo + discovery-update-info: + status: exempt + comment: This integration is configured manually; nothing to discover. + discovery: + status: exempt + comment: This integration is configured manually; nothing to discover. + docs-data-update: todo + docs-examples: todo + docs-known-limitations: todo + docs-supported-devices: + status: exempt + comment: This is a helper integration; it does not target devices. + docs-supported-functions: todo + docs-troubleshooting: todo + docs-use-cases: todo + dynamic-devices: + status: exempt + comment: This integration does not represent dynamic devices. + entity-category: todo + entity-device-class: done + entity-disabled-by-default: todo + entity-translations: done + exception-translations: + status: exempt + comment: This integration does not raise user-facing exceptions. + icon-translations: done + reconfiguration-flow: done + repair-issues: + status: exempt + comment: There are no user-actionable repair issues to surface. + stale-devices: + status: exempt + comment: This integration does not represent dynamic devices. + + # Platinum + async-dependency: + status: exempt + comment: The integration has no external runtime dependencies. + inject-websession: + status: exempt + comment: The integration does not perform any HTTP requests. + strict-typing: todo diff --git a/homeassistant/components/indoor_air_quality/sensor.py b/homeassistant/components/indoor_air_quality/sensor.py new file mode 100644 index 00000000000000..991517df1abc44 --- /dev/null +++ b/homeassistant/components/indoor_air_quality/sensor.py @@ -0,0 +1,115 @@ +"""Sensor platform for the Indoor Air Quality integration.""" + +from __future__ import annotations + +import logging +from typing import Final + +from homeassistant.components.sensor import ( + SensorDeviceClass, + SensorEntity, + SensorStateClass, +) +from homeassistant.const import CONF_DEVICE_ID +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers import device_registry as dr +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback + +from . import IndoorAirQualityConfigEntry, IndoorAirQualityController +from .const import DOMAIN, LEVELS, NAME, SENSOR_INDEX, SENSOR_LEVEL, SENSOR_TYPES + +_LOGGER: Final = logging.getLogger(__name__) + +PARALLEL_UPDATES = 0 + + +def _device_info_from_device_id( + hass: HomeAssistant, device_id: str | None +) -> DeviceInfo | None: + """Return DeviceInfo linking to a parent device, when configured.""" + if not device_id: + return None + device_registry = dr.async_get(hass) + device = device_registry.async_get(device_id) + if device is None: + return None + if device.identifiers: + return DeviceInfo(identifiers=device.identifiers) + if device.connections: + return DeviceInfo(connections=device.connections) + return None + + +async def async_setup_entry( + hass: HomeAssistant, + entry: IndoorAirQualityConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up Indoor Air Quality sensors from a config entry.""" + controller = entry.runtime_data + device_id = entry.data.get(CONF_DEVICE_ID) + parent_device = _device_info_from_device_id(hass, device_id) + + async_add_entities( + IndoorAirQualitySensor(controller, sensor_type, parent_device) + for sensor_type in SENSOR_TYPES + ) + + +class IndoorAirQualitySensor(SensorEntity): + """Indoor Air Quality sensor.""" + + _attr_has_entity_name = True + _attr_should_poll = False + + def __init__( + self, + controller: IndoorAirQualityController, + sensor_type: str, + parent_device: DeviceInfo | None, + ) -> None: + """Initialize the sensor.""" + self._controller = controller + self._sensor_type = sensor_type + self._attr_translation_key = sensor_type + self._attr_unique_id = f"{controller.unique_id}_{sensor_type}" + + if sensor_type == SENSOR_LEVEL: + self._attr_device_class = SensorDeviceClass.ENUM + self._attr_options = list(LEVELS) + elif sensor_type == SENSOR_INDEX: + self._attr_state_class = SensorStateClass.MEASUREMENT + + if parent_device is not None: + self._attr_device_info = parent_device + else: + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, controller.unique_id)}, + manufacturer=NAME, + name=controller.name, + entry_type=dr.DeviceEntryType.SERVICE, + ) + + @property + def native_value(self) -> int | str | None: + """Return the sensor's current value.""" + if self._sensor_type == SENSOR_INDEX: + return self._controller.iaq_index + if self._sensor_type == SENSOR_LEVEL: + return self._controller.iaq_level + return None + + @property + def extra_state_attributes(self) -> dict[str, object]: + """Return controller-wide diagnostic attributes.""" + return self._controller.extra_state_attributes + + async def async_added_to_hass(self) -> None: + """Subscribe to controller updates.""" + + @callback + def _async_handle_update() -> None: + self.async_write_ha_state() + + self.async_on_remove(self._controller.async_add_listener(_async_handle_update)) diff --git a/homeassistant/components/indoor_air_quality/strings.json b/homeassistant/components/indoor_air_quality/strings.json new file mode 100644 index 00000000000000..2037739edaf1e1 --- /dev/null +++ b/homeassistant/components/indoor_air_quality/strings.json @@ -0,0 +1,119 @@ +{ + "config": { + "abort": { + "already_configured": "[%key:common::config_flow::abort::already_configured_device%]", + "reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]" + }, + "error": { + "device_or_sources": "Select a device or enable additional sensors to choose sources manually.", + "no_matching_sources": "The selected device does not have any supported sensor sources. Select sources manually or choose another device.", + "no_sources": "At least one sensor source must be selected.", + "only_one_voc_sensor": "Select either a tVOC sensor or a VOC index sensor, not both." + }, + "step": { + "reconfigure": { + "data": { + "co": "[%key:component::indoor_air_quality::config::step::sources::data::co%]", + "co2": "[%key:component::indoor_air_quality::config::step::sources::data::co2%]", + "hcho": "[%key:component::indoor_air_quality::config::step::sources::data::hcho%]", + "humidity": "[%key:component::indoor_air_quality::config::step::sources::data::humidity%]", + "no2": "[%key:component::indoor_air_quality::config::step::sources::data::no2%]", + "pm": "[%key:component::indoor_air_quality::config::step::sources::data::pm%]", + "radon": "[%key:component::indoor_air_quality::config::step::sources::data::radon%]", + "standard": "[%key:component::indoor_air_quality::config::step::user::data::standard%]", + "temperature": "[%key:component::indoor_air_quality::config::step::sources::data::temperature%]", + "tvoc": "[%key:component::indoor_air_quality::config::step::sources::data::tvoc%]", + "voc_index": "[%key:component::indoor_air_quality::config::step::sources::data::voc_index%]" + }, + "data_description": { + "co": "[%key:component::indoor_air_quality::config::step::sources::data_description::co%]", + "co2": "[%key:component::indoor_air_quality::config::step::sources::data_description::co2%]", + "hcho": "[%key:component::indoor_air_quality::config::step::sources::data_description::hcho%]", + "humidity": "[%key:component::indoor_air_quality::config::step::sources::data_description::humidity%]", + "no2": "[%key:component::indoor_air_quality::config::step::sources::data_description::no2%]", + "pm": "[%key:component::indoor_air_quality::config::step::sources::data_description::pm%]", + "radon": "[%key:component::indoor_air_quality::config::step::sources::data_description::radon%]", + "standard": "[%key:component::indoor_air_quality::config::step::user::data_description::standard%]", + "temperature": "[%key:component::indoor_air_quality::config::step::sources::data_description::temperature%]", + "tvoc": "[%key:component::indoor_air_quality::config::step::sources::data_description::tvoc%]", + "voc_index": "[%key:component::indoor_air_quality::config::step::sources::data_description::voc_index%]" + }, + "description": "Update the rating standard or sensor sources for this Indoor Air Quality entry." + }, + "sources": { + "data": { + "co": "Carbon monoxide sensor", + "co2": "Carbon dioxide sensor", + "hcho": "Formaldehyde sensor", + "humidity": "Humidity sensor", + "name": "[%key:common::config_flow::data::name%]", + "no2": "Nitrogen dioxide sensor", + "pm": "Particulate matter sensors", + "radon": "Radon sensor", + "temperature": "Temperature sensor", + "tvoc": "tVOC sensor", + "voc_index": "VOC index sensor" + }, + "data_description": { + "co": "Optional CO source used for the air quality calculation.", + "co2": "Optional CO\u2082 source used for the air quality calculation.", + "hcho": "Optional formaldehyde source used for the air quality calculation.", + "humidity": "Optional humidity source used for the air quality calculation.", + "name": "Friendly name for this air quality sensor. Defaults to the device name.", + "no2": "Optional NO\u2082 source used for the air quality calculation.", + "pm": "Optional particulate matter sources used for the air quality calculation.", + "radon": "Optional radon source used for the air quality calculation.", + "temperature": "Optional temperature source used for the air quality calculation.", + "tvoc": "Optional total VOC source used for the air quality calculation.", + "voc_index": "Optional VOC index source used for the air quality calculation." + }, + "description": "Review the automatically selected sensors or add more sources." + }, + "user": { + "data": { + "device_id": "Device", + "show_source_options": "Choose additional sensors", + "standard": "Rating standard" + }, + "data_description": { + "device_id": "The integration will automatically use supported sensor entities from this device.", + "show_source_options": "Enable this to add or override source sensors before creating the IAQ sensor.", + "standard": "Air quality rating standard used to score the configured sources." + }, + "description": "Select the device that provides your air quality readings, or enable manual sensor selection." + } + } + }, + "entity": { + "sensor": { + "iaq_index": { + "name": "Index", + "state_attributes": { + "sources_set": { + "name": "Sources configured" + }, + "sources_used": { + "name": "Sources used" + } + } + }, + "iaq_level": { + "name": "Level", + "state": { + "excellent": "Excellent", + "fair": "Fair", + "good": "Good", + "inadequate": "Inadequate", + "poor": "Poor" + } + } + } + }, + "selector": { + "standard": { + "options": { + "uk": "United Kingdom" + } + } + } +} diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index d981856b0e4925..3d55de344f27eb 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -11,6 +11,7 @@ "generic_thermostat", "group", "history_stats", + "indoor_air_quality", "integration", "min_max", "mold_indicator", diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index 9c76b93ae4669d..c1367c538dcb7d 100644 --- a/homeassistant/generated/integrations.json +++ b/homeassistant/generated/integrations.json @@ -8278,6 +8278,12 @@ "config_flow": true, "iot_class": "local_polling" }, + "indoor_air_quality": { + "name": "Indoor Air Quality", + "integration_type": "helper", + "config_flow": true, + "iot_class": "calculated" + }, "input_boolean": { "integration_type": "helper", "config_flow": false diff --git a/tests/components/indoor_air_quality/__init__.py b/tests/components/indoor_air_quality/__init__.py new file mode 100644 index 00000000000000..cd16568181f6a2 --- /dev/null +++ b/tests/components/indoor_air_quality/__init__.py @@ -0,0 +1 @@ +"""Tests for the Indoor Air Quality integration.""" diff --git a/tests/components/indoor_air_quality/test_config_flow.py b/tests/components/indoor_air_quality/test_config_flow.py new file mode 100644 index 00000000000000..b07ce4aad3d2bc --- /dev/null +++ b/tests/components/indoor_air_quality/test_config_flow.py @@ -0,0 +1,429 @@ +"""Test the Indoor Air Quality config flow.""" + +from homeassistant import config_entries +from homeassistant.components.indoor_air_quality.config_flow import ( + CONF_SHOW_SOURCE_OPTIONS, + _device_name, +) +from homeassistant.components.indoor_air_quality.const import ( + CONF_HCHO, + CONF_HUMIDITY, + CONF_PM, + CONF_RADON, + CONF_SOURCES, + CONF_STANDARD, + CONF_TEMPERATURE, + CONF_TVOC, + CONF_VOC_INDEX, + DOMAIN, + STANDARD_UK, +) +from homeassistant.components.sensor import SensorDeviceClass +from homeassistant.const import CONF_DEVICE_ID, CONF_NAME +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResultType +from homeassistant.helpers import device_registry as dr, entity_registry as er + +from tests.common import MockConfigEntry + + +def _create_mock_device(hass: HomeAssistant, device_name: str = "VINDSTYRKA") -> str: + """Create a mock source device.""" + config_entry = MockConfigEntry(domain="zha") + config_entry.add_to_hass(hass) + + device_registry = dr.async_get(hass) + device_entry = device_registry.async_get_or_create( + config_entry_id=config_entry.entry_id, + identifiers={("zha", device_name.lower())}, + manufacturer="IKEA of Sweden", + name=device_name, + ) + + return device_entry.id + + +def _create_mock_sensor( + hass: HomeAssistant, + device_id: str, + suggested_object_id: str, + device_class: SensorDeviceClass | None = None, + original_name: str | None = None, +) -> None: + """Create a mock source sensor for a device.""" + entity_registry = er.async_get(hass) + entity_registry.async_get_or_create( + "sensor", + "zha", + suggested_object_id, + suggested_object_id=suggested_object_id, + device_id=device_id, + original_device_class=device_class, + original_name=original_name, + ) + + +async def test_form_user(hass: HomeAssistant) -> None: + """Test manual source setup.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {} + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_SHOW_SOURCE_OPTIONS: True}, + ) + assert result2["type"] is FlowResultType.FORM + assert result2["step_id"] == "sources" + + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + { + CONF_NAME: "Test Air Quality", + CONF_TEMPERATURE: "sensor.temperature", + CONF_HUMIDITY: "sensor.humidity", + }, + ) + await hass.async_block_till_done() + + assert result3["type"] is FlowResultType.CREATE_ENTRY + assert result3["title"] == "Test Air Quality" + assert result3["data"] == { + CONF_SOURCES: { + CONF_TEMPERATURE: "sensor.temperature", + CONF_HUMIDITY: "sensor.humidity", + }, + CONF_STANDARD: STANDARD_UK, + } + + +async def test_form_user_no_sources(hass: HomeAssistant) -> None: + """Test user form with no device and no sources errors out.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {}, + ) + assert result2["type"] is FlowResultType.FORM + assert result2["errors"] == {"base": "device_or_sources"} + + result3 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_SHOW_SOURCE_OPTIONS: True}, + ) + result4 = await hass.config_entries.flow.async_configure( + result3["flow_id"], + {CONF_NAME: "Test Air Quality"}, + ) + + assert result4["type"] is FlowResultType.FORM + assert result4["step_id"] == "sources" + assert result4["errors"] == {"base": "no_sources"} + + +async def test_form_user_both_voc_sensors(hass: HomeAssistant) -> None: + """Test user form rejects both VOC sources at once.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_SHOW_SOURCE_OPTIONS: True}, + ) + + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + { + CONF_NAME: "Test Air Quality", + CONF_TVOC: "sensor.tvoc", + CONF_VOC_INDEX: "sensor.voc_index", + }, + ) + + assert result3["type"] is FlowResultType.FORM + assert result3["step_id"] == "sources" + assert result3["errors"] == {"base": "only_one_voc_sensor"} + + +async def test_form_user_device_sources(hass: HomeAssistant) -> None: + """Test user form auto-detects sources from a selected device.""" + device_id = _create_mock_device(hass) + _create_mock_sensor( + hass, device_id, "kitchen_temperature", SensorDeviceClass.TEMPERATURE + ) + _create_mock_sensor(hass, device_id, "kitchen_humidity", SensorDeviceClass.HUMIDITY) + _create_mock_sensor(hass, device_id, "kitchen_pm25", SensorDeviceClass.PM25) + _create_mock_sensor(hass, device_id, "kitchen_voc_index", original_name="VOC index") + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_DEVICE_ID: device_id}, + ) + await hass.async_block_till_done() + + assert result2["type"] is FlowResultType.CREATE_ENTRY + assert result2["title"] == "VINDSTYRKA" + assert result2["data"] == { + CONF_DEVICE_ID: device_id, + CONF_SOURCES: { + CONF_TEMPERATURE: "sensor.kitchen_temperature", + CONF_HUMIDITY: "sensor.kitchen_humidity", + CONF_PM: ["sensor.kitchen_pm25"], + CONF_VOC_INDEX: "sensor.kitchen_voc_index", + }, + CONF_STANDARD: STANDARD_UK, + } + + +async def test_form_user_device_extra_sources(hass: HomeAssistant) -> None: + """Test user form merges manual sources with device-detected sources.""" + device_id = _create_mock_device(hass, "Kitchen Monitor") + _create_mock_sensor( + hass, device_id, "kitchen_temperature", SensorDeviceClass.TEMPERATURE + ) + _create_mock_sensor(hass, device_id, "kitchen_humidity", SensorDeviceClass.HUMIDITY) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_DEVICE_ID: device_id, + CONF_SHOW_SOURCE_OPTIONS: True, + }, + ) + assert result2["type"] is FlowResultType.FORM + assert result2["step_id"] == "sources" + + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + {CONF_HCHO: "sensor.kitchen_formaldehyde"}, + ) + await hass.async_block_till_done() + + assert result3["type"] is FlowResultType.CREATE_ENTRY + assert result3["title"] == "Kitchen Monitor" + assert result3["data"] == { + CONF_DEVICE_ID: device_id, + CONF_SOURCES: { + CONF_TEMPERATURE: "sensor.kitchen_temperature", + CONF_HUMIDITY: "sensor.kitchen_humidity", + CONF_HCHO: "sensor.kitchen_formaldehyde", + }, + CONF_STANDARD: STANDARD_UK, + } + + +async def test_form_user_device_no_matching_sources(hass: HomeAssistant) -> None: + """Test user form errors when a device has no supported sensors.""" + device_id = _create_mock_device(hass) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_DEVICE_ID: device_id}, + ) + + assert result2["type"] is FlowResultType.FORM + assert result2["errors"] == {"base": "no_matching_sources"} + + +async def test_form_user_already_configured(hass: HomeAssistant) -> None: + """Test user form aborts when an entry with the same name already exists.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + sources_step = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_SHOW_SOURCE_OPTIONS: True}, + ) + await hass.config_entries.flow.async_configure( + sources_step["flow_id"], + { + CONF_NAME: "Test Air Quality", + CONF_TEMPERATURE: "sensor.temperature", + }, + ) + await hass.async_block_till_done() + + result2 = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + {CONF_SHOW_SOURCE_OPTIONS: True}, + ) + result4 = await hass.config_entries.flow.async_configure( + result3["flow_id"], + { + CONF_NAME: "Test Air Quality", + CONF_TEMPERATURE: "sensor.temperature", + }, + ) + + assert result4["type"] is FlowResultType.ABORT + assert result4["reason"] == "already_configured" + + +async def test_reconfigure_flow(hass: HomeAssistant) -> None: + """Test the reconfigure flow updates and reloads the entry.""" + hass.states.async_set("sensor.test", "20", {"unit_of_measurement": "°C"}) + hass.states.async_set("sensor.new_temp", "21", {"unit_of_measurement": "°C"}) + config_entry = MockConfigEntry( + domain=DOMAIN, + data={ + CONF_SOURCES: {CONF_TEMPERATURE: "sensor.test"}, + CONF_STANDARD: STANDARD_UK, + }, + title="Test", + unique_id="test", + ) + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + result = await config_entry.start_reconfigure_flow(hass) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "reconfigure" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_TEMPERATURE: "sensor.new_temp"}, + ) + await hass.async_block_till_done() + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "reconfigure_successful" + assert config_entry.data[CONF_SOURCES] == {CONF_TEMPERATURE: "sensor.new_temp"} + + +async def test_reconfigure_no_sources(hass: HomeAssistant) -> None: + """Test reconfigure errors when the user clears every source.""" + hass.states.async_set( + "sensor.pm", "5", {"unit_of_measurement": "µg/m³", "device_class": "pm25"} + ) + config_entry = MockConfigEntry( + domain=DOMAIN, + data={ + CONF_SOURCES: {CONF_PM: ["sensor.pm"]}, + CONF_STANDARD: STANDARD_UK, + }, + title="Test", + unique_id="test", + ) + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + result = await config_entry.start_reconfigure_flow(hass) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_PM: []} + ) + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": "no_sources"} + + +async def test_form_user_device_with_tvoc(hass: HomeAssistant) -> None: + """Device-detection picks up a tVOC sensor when no VOC index is present.""" + device_id = _create_mock_device(hass, "Air Monitor") + _create_mock_sensor( + hass, + device_id, + "air_tvoc", + SensorDeviceClass.VOLATILE_ORGANIC_COMPOUNDS, + ) + _create_mock_sensor( + hass, + device_id, + "air_tvoc_alt", + SensorDeviceClass.VOLATILE_ORGANIC_COMPOUNDS_PARTS, + ) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_DEVICE_ID: device_id} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + # The first matching tVOC entity is picked; the second is ignored. + assert result["data"][CONF_SOURCES] == {CONF_TVOC: "sensor.air_tvoc"} + + +async def test_form_user_device_with_hcho_and_radon_labels( + hass: HomeAssistant, +) -> None: + """HCHO/radon detection falls back to label matching when no device class.""" + device_id = _create_mock_device(hass, "Multi Sensor") + _create_mock_sensor( + hass, device_id, "multi_formaldehyde", original_name="Formaldehyde" + ) + _create_mock_sensor(hass, device_id, "multi_radon", original_name="Radon") + _create_mock_sensor(hass, device_id, "multi_unrelated", original_name="Battery") + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_DEVICE_ID: device_id} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"][CONF_SOURCES] == { + CONF_HCHO: "sensor.multi_formaldehyde", + CONF_RADON: "sensor.multi_radon", + } + + +async def test_form_user_device_skips_non_sensor_entities( + hass: HomeAssistant, +) -> None: + """Non-sensor entities on a device are skipped during detection.""" + device_id = _create_mock_device(hass, "Mixed Device") + entity_registry = er.async_get(hass) + entity_registry.async_get_or_create( + "switch", + "zha", + "mixed_switch", + suggested_object_id="mixed_switch", + device_id=device_id, + ) + _create_mock_sensor( + hass, device_id, "mixed_temperature", SensorDeviceClass.TEMPERATURE + ) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_DEVICE_ID: device_id} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"][CONF_SOURCES] == { + CONF_TEMPERATURE: "sensor.mixed_temperature", + } + + +def test_device_name_returns_none_for_unknown_device(hass: HomeAssistant) -> None: + """_device_name returns None when the device is not in the registry.""" + assert _device_name(hass, None) is None + assert _device_name(hass, "does-not-exist") is None diff --git a/tests/components/indoor_air_quality/test_init.py b/tests/components/indoor_air_quality/test_init.py new file mode 100644 index 00000000000000..5a29a7ad0c8c38 --- /dev/null +++ b/tests/components/indoor_air_quality/test_init.py @@ -0,0 +1,538 @@ +"""Test the Indoor Air Quality controller and setup.""" + +from typing import Any + +import pytest + +from homeassistant import config_entries +from homeassistant.components.indoor_air_quality import IndoorAirQualityController +from homeassistant.components.indoor_air_quality.const import ( + ATTR_SOURCE_INDEX_TPL, + ATTR_SOURCES_SET, + ATTR_SOURCES_USED, + CONF_CO, + CONF_CO2, + CONF_HCHO, + CONF_HUMIDITY, + CONF_NO2, + CONF_PM, + CONF_RADON, + CONF_SOURCES, + CONF_STANDARD, + CONF_TEMPERATURE, + CONF_TVOC, + CONF_VOC_INDEX, + DOMAIN, + LEVEL_EXCELLENT, + LEVEL_FAIR, + LEVEL_GOOD, + LEVEL_INADEQUATE, + LEVEL_POOR, + MOLAR_MASS_CO2, + MOLAR_MASS_HCHO, + MOLAR_MASS_TVOC, + STANDARD_UK, + UNIT_MGM3, + UNIT_PPM, + UNIT_UGM3, +) +from homeassistant.components.indoor_air_quality.helpers import ( + convert_value, + resolve_state, +) +from homeassistant.components.sensor import ATTR_STATE_CLASS, SensorStateClass +from homeassistant.const import ( + ATTR_UNIT_OF_MEASUREMENT, + CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, + CONCENTRATION_MILLIGRAMS_PER_CUBIC_METER, + PERCENTAGE, + STATE_UNAVAILABLE, + STATE_UNKNOWN, + UnitOfTemperature, +) +from homeassistant.core import HomeAssistant + +from tests.common import MockConfigEntry + + +async def test_setup_and_unload(hass: HomeAssistant) -> None: + """Test a config entry is set up, exposes sensors, and unloads cleanly.""" + hass.states.async_set( + "sensor.temp", + 20, + {ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS}, + ) + + entry = MockConfigEntry( + domain=DOMAIN, + data={ + CONF_SOURCES: {CONF_TEMPERATURE: "sensor.temp"}, + CONF_STANDARD: STANDARD_UK, + }, + title="Living Room", + unique_id="living-room", + ) + entry.add_to_hass(hass) + + assert await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + state_index = hass.states.get("sensor.living_room_index") + state_level = hass.states.get("sensor.living_room_level") + assert state_index is not None + assert state_level is not None + assert state_index.state == "65" + assert state_level.state == LEVEL_EXCELLENT + + assert await hass.config_entries.async_unload(entry.entry_id) + await hass.async_block_till_done() + + +async def test_state_change_updates_sensor(hass: HomeAssistant) -> None: + """Test the calculated sensors update when a source state changes.""" + hass.states.async_set( + "sensor.temp", + 20, + {ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS}, + ) + + entry = MockConfigEntry( + domain=DOMAIN, + data={ + CONF_SOURCES: {CONF_TEMPERATURE: "sensor.temp"}, + CONF_STANDARD: STANDARD_UK, + }, + title="Bedroom", + unique_id="bedroom", + ) + entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + assert hass.states.get("sensor.bedroom_level").state == LEVEL_EXCELLENT + + hass.states.async_set( + "sensor.temp", + 14, + {ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS}, + ) + await hass.async_block_till_done() + + assert hass.states.get("sensor.bedroom_level").state == LEVEL_INADEQUATE + + +async def test_controller_initial_state(hass: HomeAssistant) -> None: + """Test the controller starts with no calculated value.""" + controller = IndoorAirQualityController( + hass, "test", "Test", {CONF_TEMPERATURE: "sensor.test"} + ) + + assert controller.unique_id == "test" + assert controller.name == "Test" + assert controller.standard == STANDARD_UK + assert controller.iaq_index is None + assert controller.iaq_level is None + assert controller.extra_state_attributes == { + ATTR_SOURCES_SET: 1, + ATTR_SOURCES_USED: 0, + } + + +async def test_controller_update_multi_source(hass: HomeAssistant) -> None: + """Test the controller combines multiple sources into a level.""" + hass.states.async_set( + "sensor.t", 17, {ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS} + ) + hass.states.async_set("sensor.h", 50, {ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE}) + hass.states.async_set("sensor.c", 800, {ATTR_UNIT_OF_MEASUREMENT: "ppm"}) + + controller = IndoorAirQualityController( + hass, + "test", + "Test", + { + CONF_TEMPERATURE: "sensor.t", + CONF_HUMIDITY: "sensor.h", + CONF_CO2: "sensor.c", + }, + ) + controller.update() + + assert controller.iaq_index == 56 + assert controller.iaq_level == LEVEL_GOOD + assert controller.extra_state_attributes == { + ATTR_SOURCES_SET: 3, + ATTR_SOURCES_USED: 3, + ATTR_SOURCE_INDEX_TPL.format(CONF_TEMPERATURE): 4, + ATTR_SOURCE_INDEX_TPL.format(CONF_HUMIDITY): 5, + ATTR_SOURCE_INDEX_TPL.format(CONF_CO2): 4, + } + + +@pytest.mark.parametrize( + ("celsius", "expected_index", "expected_level"), + [ + (18, 65, LEVEL_EXCELLENT), + (16, 39, LEVEL_FAIR), + (15, 26, LEVEL_POOR), + (14, 13, LEVEL_INADEQUATE), + ], +) +async def test_temperature_levels( + hass: HomeAssistant, + celsius: float, + expected_index: int, + expected_level: str, +) -> None: + """Test temperature-only calculations across each band.""" + hass.states.async_set( + "sensor.t", + celsius, + {ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS}, + ) + controller = IndoorAirQualityController( + hass, "test", "Test", {CONF_TEMPERATURE: "sensor.t"} + ) + controller.update() + assert controller.iaq_index == expected_index + assert controller.iaq_level == expected_level + + +async def test_temperature_fahrenheit_conversion(hass: HomeAssistant) -> None: + """Test temperature input in °F is converted to the matching band.""" + controller = IndoorAirQualityController( + hass, "test", "Test", {CONF_TEMPERATURE: "sensor.t"} + ) + for fahrenheit, expected_score in ((57, 1), (59, 2), (60, 3), (63, 4), (67, 5)): + hass.states.async_set( + "sensor.t", + fahrenheit, + {ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT}, + ) + controller.update() + assert ( + controller.extra_state_attributes[ + ATTR_SOURCE_INDEX_TPL.format("temperature") + ] + == expected_score + ) + + +async def test_humidity_levels(hass: HomeAssistant) -> None: + """Test humidity bands score correctly on both sides of the optimum.""" + controller = IndoorAirQualityController( + hass, "test", "Test", {CONF_HUMIDITY: "sensor.h"} + ) + for value, expected_score in ( + (5, 1), + (15, 2), + (25, 3), + (35, 4), + (50, 5), + (75, 3), + (85, 2), + (95, 1), + ): + hass.states.async_set("sensor.h", value, {ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE}) + controller.update() + assert ( + controller.extra_state_attributes[ATTR_SOURCE_INDEX_TPL.format("humidity")] + == expected_score + ) + + +async def test_co2_levels(hass: HomeAssistant) -> None: + """Test CO2 bands.""" + controller = IndoorAirQualityController( + hass, "test", "Test", {CONF_CO2: "sensor.c"} + ) + for value, expected_score in ( + (500, 5), + (600, 4), + (1500, 3), + (1800, 2), + (1801, 1), + ): + hass.states.async_set("sensor.c", value, {ATTR_UNIT_OF_MEASUREMENT: "ppm"}) + controller.update() + assert ( + controller.extra_state_attributes[ATTR_SOURCE_INDEX_TPL.format("co2")] + == expected_score + ) + + +async def test_pm_summed_levels(hass: HomeAssistant) -> None: + """Test PM source list is summed in µg/m³.""" + controller = IndoorAirQualityController( + hass, "test", "Test", {CONF_PM: ["sensor.pm1", "sensor.pm2"]} + ) + + hass.states.async_set("sensor.pm1", 10, {ATTR_UNIT_OF_MEASUREMENT: "µg/m³"}) + hass.states.async_set("sensor.pm2", 0.01, {ATTR_UNIT_OF_MEASUREMENT: "mg/m³"}) + controller.update() + # 10 + 0.01 mg/m³ * 1000 = 20 µg/m³ → score 5 + assert controller.extra_state_attributes[ATTR_SOURCE_INDEX_TPL.format("pm")] == 5 + + +async def test_co_levels(hass: HomeAssistant) -> None: + """Test CO bands and zero-special case.""" + controller = IndoorAirQualityController( + hass, "test", "Test", {CONF_CO: "sensor.co"} + ) + for value, expected_score in ((0, 5), (5, 3), (8, 1)): + hass.states.async_set("sensor.co", value, {ATTR_UNIT_OF_MEASUREMENT: "mg/m³"}) + controller.update() + assert ( + controller.extra_state_attributes[ATTR_SOURCE_INDEX_TPL.format("co")] + == expected_score + ) + + +async def test_radon_levels(hass: HomeAssistant) -> None: + """Test radon bands and zero-special case.""" + controller = IndoorAirQualityController( + hass, "test", "Test", {CONF_RADON: "sensor.r"} + ) + for value, expected_score in ((0, 5), (10, 3), (50, 2), (200, 1)): + hass.states.async_set("sensor.r", value, {ATTR_UNIT_OF_MEASUREMENT: "Bq/m³"}) + controller.update() + assert ( + controller.extra_state_attributes[ATTR_SOURCE_INDEX_TPL.format("radon")] + == expected_score + ) + + +async def test_voc_index_levels(hass: HomeAssistant) -> None: + """Test dimensionless VOC index bands.""" + controller = IndoorAirQualityController( + hass, "test", "Test", {CONF_VOC_INDEX: "sensor.v"} + ) + for value, expected_score in ((0, 5), (115, 4), (180, 3), (260, 2), (261, 1)): + hass.states.async_set("sensor.v", value) + controller.update() + assert ( + controller.extra_state_attributes[ATTR_SOURCE_INDEX_TPL.format("voc_index")] + == expected_score + ) + + +async def test_resolve_state(hass: HomeAssistant) -> None: + """Test resolve_state behaviour for valid, invalid and missing states.""" + assert resolve_state(hass, "sensor.missing") is None + + hass.states.async_set("sensor.s", STATE_UNKNOWN) + assert resolve_state(hass, "sensor.s") is None + + hass.states.async_set("sensor.s", STATE_UNAVAILABLE) + assert resolve_state(hass, "sensor.s") is None + + hass.states.async_set("sensor.s", "not a number") + assert resolve_state(hass, "sensor.s") is None + + hass.states.async_set("sensor.s", "12.5", {ATTR_UNIT_OF_MEASUREMENT: "ppm"}) + value, unit = resolve_state(hass, "sensor.s") + assert value == 12.5 + assert unit == "ppm" + + +def test_convert_value_same_family() -> None: + """Test direct unit conversions inside a single family.""" + # ppb -> ppm (target unit ppm has factor 1, ppb has factor 0.001) + assert convert_value(1500, "ppb", UNIT_PPM) == pytest.approx(1.5) + # mg/m³ -> µg/m³ + assert convert_value(0.01, "mg/m³", UNIT_UGM3) == pytest.approx(10) + + +@pytest.mark.parametrize( + "source_unit", + [ + CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, # canonical Greek mu + "µg/m³", # micro sign alias + "ug/m³", # ASCII alias + ], +) +def test_convert_value_ugm3_aliases(source_unit: str) -> None: + """Canonical and alias µg/m³ unit strings all hit the target unit.""" + assert convert_value(7, source_unit, UNIT_UGM3) == pytest.approx(7) + + +@pytest.mark.parametrize( + "source_unit", + [ + CONCENTRATION_MILLIGRAMS_PER_CUBIC_METER, # canonical + "mg/m³", + ], +) +def test_convert_value_mgm3_aliases(source_unit: str) -> None: + """Canonical and alias mg/m³ unit strings all hit the target unit.""" + assert convert_value(2, source_unit, UNIT_MGM3) == pytest.approx(2) + + +def test_convert_value_molar_mass() -> None: + """Test ppm/ppb to mass concentration conversions via molar mass.""" + # 1 ppm CO2 ≈ MOLAR_MASS_CO2 / 24.45 mg/m³ + expected = MOLAR_MASS_CO2 / 24.45 + assert convert_value( + 1, "ppm", UNIT_MGM3, molar_mass=MOLAR_MASS_CO2 + ) == pytest.approx(expected) + + # Same in µg/m³ → 1000× larger + assert convert_value( + 1, "ppm", UNIT_UGM3, molar_mass=MOLAR_MASS_CO2 + ) == pytest.approx(expected * 1000) + + +def test_convert_value_unknown_unit() -> None: + """Test convert returns None for unknown source units without molar mass.""" + assert convert_value(1, "weird", UNIT_PPM) is None + # ppm into a mass unit without molar mass cannot be resolved. + assert convert_value(1, "ppm", UNIT_MGM3) is None + + +@pytest.mark.parametrize( + "molar_mass", + [MOLAR_MASS_TVOC, MOLAR_MASS_HCHO], +) +def test_convert_value_molar_masses_present(molar_mass: float) -> None: + """Smoke-test all renamed molar-mass constants are usable.""" + assert convert_value(1, "ppm", UNIT_MGM3, molar_mass=molar_mass) == pytest.approx( + molar_mass / 24.45 + ) + + +async def test_voc_conflict_excluded(hass: HomeAssistant) -> None: + """Test that configuring both tVOC and VOC index lets controller skip neither.""" + # The flow rejects this combination, but the controller itself should still + # tolerate it: each resolver runs independently. + hass.states.async_set("sensor.tvoc", 0.05, {ATTR_UNIT_OF_MEASUREMENT: "mg/m³"}) + hass.states.async_set("sensor.voc", 50) + controller = IndoorAirQualityController( + hass, + "test", + "Test", + {CONF_TVOC: "sensor.tvoc", CONF_VOC_INDEX: "sensor.voc"}, + ) + controller.update() + attrs = controller.extra_state_attributes + assert attrs[ATTR_SOURCE_INDEX_TPL.format("tvoc")] == 5 + assert attrs[ATTR_SOURCE_INDEX_TPL.format("voc_index")] == 5 + + +async def test_hcho_and_no2_levels(hass: HomeAssistant) -> None: + """Test HCHO and NO2 bands using the µg/m³ targets.""" + controller = IndoorAirQualityController( + hass, + "test", + "Test", + {CONF_HCHO: "sensor.hcho", CONF_NO2: "sensor.no2"}, + ) + hass.states.async_set("sensor.hcho", 5, {ATTR_UNIT_OF_MEASUREMENT: "µg/m³"}) + hass.states.async_set("sensor.no2", 100, {ATTR_UNIT_OF_MEASUREMENT: "µg/m³"}) + controller.update() + attrs = controller.extra_state_attributes + assert attrs[ATTR_SOURCE_INDEX_TPL.format("hcho")] == 5 + assert attrs[ATTR_SOURCE_INDEX_TPL.format("no2")] == 5 + + +async def test_iaq_index_state_class(hass: HomeAssistant) -> None: + """The numeric IAQ index sensor exposes a measurement state class.""" + hass.states.async_set( + "sensor.temp", 20, {ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS} + ) + entry = MockConfigEntry( + domain=DOMAIN, + data={ + CONF_SOURCES: {CONF_TEMPERATURE: "sensor.temp"}, + CONF_STANDARD: STANDARD_UK, + }, + title="Bedroom", + unique_id="bedroom", + ) + entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + state = hass.states.get("sensor.bedroom_index") + assert state is not None + assert state.attributes[ATTR_STATE_CLASS] == SensorStateClass.MEASUREMENT + + +@pytest.mark.parametrize( + "data", + [ + {CONF_SOURCES: {}, CONF_STANDARD: "uk"}, + {CONF_SOURCES: {CONF_TEMPERATURE: "sensor.t"}, CONF_STANDARD: "not_a_standard"}, + {CONF_SOURCES: {"unknown_key": "sensor.x"}, CONF_STANDARD: "uk"}, + ], + ids=["empty_sources", "unknown_standard", "only_unknown_source_keys"], +) +async def test_invalid_entry_data_raises( + hass: HomeAssistant, data: dict[str, Any] +) -> None: + """Invalid entry data should raise ConfigEntryError.""" + entry = MockConfigEntry( + domain=DOMAIN, + data=data, + title="Bad", + unique_id="bad", + ) + entry.add_to_hass(hass) + assert not await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + assert entry.state is config_entries.ConfigEntryState.SETUP_ERROR + + +async def test_state_resets_when_all_sources_unavailable( + hass: HomeAssistant, +) -> None: + """Index, level and per-source attributes clear when no source resolves.""" + hass.states.async_set( + "sensor.t", 17, {ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS} + ) + hass.states.async_set("sensor.h", 50, {ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE}) + + controller = IndoorAirQualityController( + hass, + "test", + "Test", + {CONF_TEMPERATURE: "sensor.t", CONF_HUMIDITY: "sensor.h"}, + ) + controller.update() + assert controller.iaq_index is not None + assert controller.iaq_level is not None + assert ATTR_SOURCE_INDEX_TPL.format(CONF_TEMPERATURE) in ( + controller.extra_state_attributes + ) + + hass.states.async_remove("sensor.t") + hass.states.async_remove("sensor.h") + controller.update() + + assert controller.iaq_index is None + assert controller.iaq_level is None + assert controller.extra_state_attributes == { + ATTR_SOURCES_SET: 2, + ATTR_SOURCES_USED: 0, + } + + +async def test_state_resets_for_unknown_standard(hass: HomeAssistant) -> None: + """An unknown rating standard clears any previously computed state.""" + hass.states.async_set( + "sensor.t", 17, {ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS} + ) + controller = IndoorAirQualityController( + hass, "test", "Test", {CONF_TEMPERATURE: "sensor.t"} + ) + controller.update() + assert controller.iaq_index is not None + + controller._standard = "not_a_standard" + controller.update() + + assert controller.iaq_index is None + assert controller.iaq_level is None + assert controller.extra_state_attributes == { + ATTR_SOURCES_SET: 1, + ATTR_SOURCES_USED: 0, + }