Add Indoor Air Quality integration

homeassistant/components/indoor_air_quality/__init__.py

@@ -0,0 +1,116 @@
+"""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 IndoorAirQualityController
+from .helpers import entity_ids_from_sources
+
+_LOGGER: Final = logging.getLogger(__name__)
+
+PLATFORMS: Final = [Platform.SENSOR]
+
+
+_ENTRY_DATA_SCHEMA: Final = vol.Schema(
+    {
+        vol.Required(CONF_SOURCES): vol.All(
+            vol.Schema({str: vol.Any(str, [str])}), vol.Length(min=1)
+        ),
+        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",
+]

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 (5 per source, normalized to 0-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]