Add airgradient integration (home-assistant#114113)

.strict-typing

@@ -48,6 +48,7 @@ homeassistant.components.adax.*
 homeassistant.components.adguard.*
 homeassistant.components.aftership.*
 homeassistant.components.air_quality.*
+homeassistant.components.airgradient.*
 homeassistant.components.airly.*
 homeassistant.components.airnow.*
 homeassistant.components.airq.*

CODEOWNERS

@@ -56,6 +56,8 @@ build.json @home-assistant/supervisor
 /tests/components/agent_dvr/ @ispysoftware
 /homeassistant/components/air_quality/ @home-assistant/core
 /tests/components/air_quality/ @home-assistant/core
+/homeassistant/components/airgradient/ @airgradienthq @joostlek
+/tests/components/airgradient/ @airgradienthq @joostlek
 /homeassistant/components/airly/ @bieniu
 /tests/components/airly/ @bieniu
 /homeassistant/components/airnow/ @asymworks

homeassistant/components/airgradient/__init__.py

@@ -0,0 +1,34 @@
+"""The Airgradient integration."""
+
+from __future__ import annotations
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, Platform
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN
+from .coordinator import AirGradientDataUpdateCoordinator
+
+PLATFORMS: list[Platform] = [Platform.SENSOR]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Airgradient from a config entry."""
+
+    coordinator = AirGradientDataUpdateCoordinator(hass, entry.data[CONF_HOST])
+
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/airgradient/config_flow.py

@@ -0,0 +1,83 @@
+"""Config flow for Airgradient."""
+
+from typing import Any
+
+from airgradient import AirGradientClient, AirGradientError
+import voluptuous as vol
+
+from homeassistant.components import zeroconf
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_HOST, CONF_MODEL
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import DOMAIN
+
+
+class AirGradientConfigFlow(ConfigFlow, domain=DOMAIN):
+    """AirGradient config flow."""
+
+    def __init__(self) -> None:
+        """Initialize the config flow."""
+        self.data: dict[str, Any] = {}
+
+    async def async_step_zeroconf(
+        self, discovery_info: zeroconf.ZeroconfServiceInfo
+    ) -> ConfigFlowResult:
+        """Handle zeroconf discovery."""
+        self.data[CONF_HOST] = host = discovery_info.host
+        self.data[CONF_MODEL] = discovery_info.properties["model"]
+
+        await self.async_set_unique_id(discovery_info.properties["serialno"])
+        self._abort_if_unique_id_configured(updates={CONF_HOST: host})
+
+        session = async_get_clientsession(self.hass)
+        air_gradient = AirGradientClient(host, session=session)
+        await air_gradient.get_current_measures()
+
+        self.context["title_placeholders"] = {
+            "model": self.data[CONF_MODEL],
+        }
+        return await self.async_step_discovery_confirm()
+
+    async def async_step_discovery_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Confirm discovery."""
+        if user_input is not None:
+            return self.async_create_entry(
+                title=self.data[CONF_MODEL],
+                data={CONF_HOST: self.data[CONF_HOST]},
+            )
+
+        self._set_confirm_only()
+        return self.async_show_form(
+            step_id="discovery_confirm",
+            description_placeholders={
+                "model": self.data[CONF_MODEL],
+            },
+        )
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle a flow initialized by the user."""
+        errors: dict[str, str] = {}
+        if user_input:
+            session = async_get_clientsession(self.hass)
+            air_gradient = AirGradientClient(user_input[CONF_HOST], session=session)
+            try:
+                current_measures = await air_gradient.get_current_measures()
+            except AirGradientError:
+                errors["base"] = "cannot_connect"
+            else:
+                await self.async_set_unique_id(current_measures.serial_number)
+                self._abort_if_unique_id_configured()
+                return self.async_create_entry(
+                    title=current_measures.model,
+                    data={CONF_HOST: user_input[CONF_HOST]},
+                )
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema({vol.Required(CONF_HOST): str}),
+            errors=errors,
+        )

homeassistant/components/airgradient/const.py

@@ -0,0 +1,7 @@
+"""Constants for the Airgradient integration."""
+
+import logging
+
+DOMAIN = "airgradient"
+
+LOGGER = logging.getLogger(__package__)

homeassistant/components/airgradient/coordinator.py

@@ -0,0 +1,32 @@
+"""Define an object to manage fetching AirGradient data."""
+
+from datetime import timedelta
+
+from airgradient import AirGradientClient, AirGradientError, Measures
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import LOGGER
+
+
+class AirGradientDataUpdateCoordinator(DataUpdateCoordinator[Measures]):
+    """Class to manage fetching AirGradient data."""
+
+    def __init__(self, hass: HomeAssistant, host: str) -> None:
+        """Initialize coordinator."""
+        super().__init__(
+            hass,
+            logger=LOGGER,
+            name=f"AirGradient {host}",
+            update_interval=timedelta(minutes=1),
+        )
+        session = async_get_clientsession(hass)
+        self.client = AirGradientClient(host, session=session)
+
+    async def _async_update_data(self) -> Measures:
+        try:
+            return await self.client.get_current_measures()
+        except AirGradientError as error:
+            raise UpdateFailed(error) from error

homeassistant/components/airgradient/icons.json

@@ -0,0 +1,15 @@
+{
+  "entity": {
+    "sensor": {
+      "total_volatile_organic_component_index": {
+        "default": "mdi:molecule"
+      },
+      "nitrogen_index": {
+        "default": "mdi:molecule"
+      },
+      "pm003_count": {
+        "default": "mdi:blur"
+      }
+    }
+  }
+}

homeassistant/components/airgradient/manifest.json

@@ -0,0 +1,11 @@
+{
+  "domain": "airgradient",
+  "name": "Airgradient",
+  "codeowners": ["@airgradienthq", "@joostlek"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/airgradient",
+  "integration_type": "device",
+  "iot_class": "local_polling",
+  "requirements": ["airgradient==0.4.0"],
+  "zeroconf": ["_airgradient._tcp.local."]
+}

homeassistant/components/airgradient/sensor.py

@@ -0,0 +1,192 @@
+"""Support for AirGradient sensors."""
+
+from collections.abc import Callable
+from dataclasses import dataclass
+
+from airgradient.models import Measures
+
+from homeassistant.components.sensor import (
+    SensorDeviceClass,
+    SensorEntity,
+    SensorEntityDescription,
+    SensorStateClass,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import (
+    CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
+    CONCENTRATION_PARTS_PER_MILLION,
+    PERCENTAGE,
+    SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
+    EntityCategory,
+    UnitOfTemperature,
+)
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import StateType
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from . import AirGradientDataUpdateCoordinator
+from .const import DOMAIN
+
+
+@dataclass(frozen=True, kw_only=True)
+class AirGradientSensorEntityDescription(SensorEntityDescription):
+    """Describes AirGradient sensor entity."""
+
+    value_fn: Callable[[Measures], StateType]
+
+
+SENSOR_TYPES: tuple[AirGradientSensorEntityDescription, ...] = (
+    AirGradientSensorEntityDescription(
+        key="pm01",
+        device_class=SensorDeviceClass.PM1,
+        native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
+        state_class=SensorStateClass.MEASUREMENT,
+        value_fn=lambda status: status.pm01,
+    ),
+    AirGradientSensorEntityDescription(
+        key="pm02",
+        device_class=SensorDeviceClass.PM25,
+        native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
+        state_class=SensorStateClass.MEASUREMENT,
+        value_fn=lambda status: status.pm02,
+    ),
+    AirGradientSensorEntityDescription(
+        key="pm10",
+        device_class=SensorDeviceClass.PM10,
+        native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
+        state_class=SensorStateClass.MEASUREMENT,
+        value_fn=lambda status: status.pm10,
+    ),
+    AirGradientSensorEntityDescription(
+        key="temperature",
+        device_class=SensorDeviceClass.TEMPERATURE,
+        native_unit_of_measurement=UnitOfTemperature.CELSIUS,
+        state_class=SensorStateClass.MEASUREMENT,
+        value_fn=lambda status: status.ambient_temperature,
+    ),
+    AirGradientSensorEntityDescription(
+        key="humidity",
+        device_class=SensorDeviceClass.HUMIDITY,
+        native_unit_of_measurement=PERCENTAGE,
+        state_class=SensorStateClass.MEASUREMENT,
+        value_fn=lambda status: status.relative_humidity,
+    ),
+    AirGradientSensorEntityDescription(
+        key="signal_strength",
+        device_class=SensorDeviceClass.SIGNAL_STRENGTH,
+        native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        state_class=SensorStateClass.MEASUREMENT,
+        entity_registry_enabled_default=False,
+        value_fn=lambda status: status.signal_strength,
+    ),
+    AirGradientSensorEntityDescription(
+        key="tvoc",
+        translation_key="total_volatile_organic_component_index",
+        state_class=SensorStateClass.MEASUREMENT,
+        value_fn=lambda status: status.total_volatile_organic_component_index,
+    ),
+    AirGradientSensorEntityDescription(
+        key="nitrogen_index",
+        translation_key="nitrogen_index",
+        state_class=SensorStateClass.MEASUREMENT,
+        value_fn=lambda status: status.nitrogen_index,
+    ),
+    AirGradientSensorEntityDescription(
+        key="co2",
+        device_class=SensorDeviceClass.CO2,
+        native_unit_of_measurement=CONCENTRATION_PARTS_PER_MILLION,
+        state_class=SensorStateClass.MEASUREMENT,
+        value_fn=lambda status: status.rco2,
+    ),
+    AirGradientSensorEntityDescription(
+        key="pm003",
+        translation_key="pm003_count",
+        state_class=SensorStateClass.MEASUREMENT,
+        value_fn=lambda status: status.pm003_count,
+    ),
+    AirGradientSensorEntityDescription(
+        key="nox_raw",
+        translation_key="raw_nitrogen",
+        native_unit_of_measurement="ticks",
+        state_class=SensorStateClass.MEASUREMENT,
+        entity_registry_enabled_default=False,
+        value_fn=lambda status: status.raw_nitrogen,
+    ),
+    AirGradientSensorEntityDescription(
+        key="tvoc_raw",
+        translation_key="raw_total_volatile_organic_component",
+        native_unit_of_measurement="ticks",
+        state_class=SensorStateClass.MEASUREMENT,
+        entity_registry_enabled_default=False,
+        value_fn=lambda status: status.raw_total_volatile_organic_component,
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Set up AirGradient sensor entities based on a config entry."""
+
+    coordinator: AirGradientDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
+    listener: Callable[[], None] | None = None
+    not_setup: set[AirGradientSensorEntityDescription] = set(SENSOR_TYPES)
+
+    @callback
+    def add_entities() -> None:
+        """Add new entities based on the latest data."""
+        nonlocal not_setup, listener
+        sensor_descriptions = not_setup
+        not_setup = set()
+        sensors = []
+        for description in sensor_descriptions:
+            if description.value_fn(coordinator.data) is None:
+                not_setup.add(description)
+            else:
+                sensors.append(AirGradientSensor(coordinator, description))
+
+        if sensors:
+            async_add_entities(sensors)
+        if not_setup:
+            if not listener:
+                listener = coordinator.async_add_listener(add_entities)
+        elif listener:
+            listener()
+
+    add_entities()
+
+
+class AirGradientSensor(
+    CoordinatorEntity[AirGradientDataUpdateCoordinator], SensorEntity
+):
+    """Defines an AirGradient sensor."""
+
+    _attr_has_entity_name = True
+
+    entity_description: AirGradientSensorEntityDescription
+
+    def __init__(
+        self,
+        coordinator: AirGradientDataUpdateCoordinator,
+        description: AirGradientSensorEntityDescription,
+    ) -> None:
+        """Initialize airgradient sensor."""
+        super().__init__(coordinator)
+
+        self.entity_description = description
+        self._attr_unique_id = f"{coordinator.data.serial_number}-{description.key}"
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, coordinator.data.serial_number)},
+            model=coordinator.data.model,
+            manufacturer="AirGradient",
+            serial_number=coordinator.data.serial_number,
+            sw_version=coordinator.data.firmware_version,
+        )
+
+    @property
+    def native_value(self) -> StateType:
+        """Return the state of the sensor."""
+        return self.entity_description.value_fn(self.coordinator.data)