Add Ambee integration (#51645)

.coveragerc

@@ -45,6 +45,8 @@ omit =
     homeassistant/components/alarmdecoder/sensor.py
     homeassistant/components/alpha_vantage/sensor.py
     homeassistant/components/amazon_polly/*
+    homeassistant/components/ambee/__init__.py
+    homeassistant/components/ambee/sensor.py
     homeassistant/components/ambiclimate/climate.py
     homeassistant/components/ambient_station/*
     homeassistant/components/amcrest/*

.strict-typing

@@ -12,6 +12,7 @@ homeassistant.components.airly.*
 homeassistant.components.aladdin_connect.*
 homeassistant.components.alarm_control_panel.*
 homeassistant.components.amazon_polly.*
+homeassistant.components.ambee.*
 homeassistant.components.ampio.*
 homeassistant.components.automation.*
 homeassistant.components.binary_sensor.*

CODEOWNERS

@@ -33,6 +33,7 @@ homeassistant/components/alarmdecoder/* @ajschmidt8
 homeassistant/components/alexa/* @home-assistant/cloud @ochlocracy
 homeassistant/components/almond/* @gcampax @balloob
 homeassistant/components/alpha_vantage/* @fabaff
+homeassistant/components/ambee/* @frenck
 homeassistant/components/ambiclimate/* @danielhiversen
 homeassistant/components/ambient_station/* @bachya
 homeassistant/components/analytics/* @home-assistant/core @ludeeus

homeassistant/components/ambee/__init__.py

@@ -0,0 +1,46 @@
+"""Support for Ambee."""
+from __future__ import annotations
+
+from ambee import Ambee
+
+from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import DOMAIN, LOGGER, SCAN_INTERVAL
+
+PLATFORMS = (SENSOR_DOMAIN,)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Ambee from a config entry."""
+    client = Ambee(
+        api_key=entry.data[CONF_API_KEY],
+        latitude=entry.data[CONF_LATITUDE],
+        longitude=entry.data[CONF_LONGITUDE],
+    )
+
+    coordinator = DataUpdateCoordinator(
+        hass,
+        LOGGER,
+        name=DOMAIN,
+        update_interval=SCAN_INTERVAL,
+        update_method=client.air_quality,
+    )
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data.setdefault(DOMAIN, {})
+    hass.data[DOMAIN][entry.entry_id] = coordinator
+
+    hass.config_entries.async_setup_platforms(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload Ambee config entry."""
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+    if unload_ok:
+        del hass.data[DOMAIN][entry.entry_id]
+    return unload_ok

homeassistant/components/ambee/config_flow.py

@@ -0,0 +1,70 @@
+"""Config flow to configure the Ambee integration."""
+from __future__ import annotations
+
+from typing import Any
+
+from ambee import Ambee, AmbeeAuthenticationError, AmbeeError
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow
+from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+import homeassistant.helpers.config_validation as cv
+
+from .const import DOMAIN
+
+
+class AmbeeFlowHandler(ConfigFlow, domain=DOMAIN):
+    """Config flow for Ambee."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle a flow initialized by the user."""
+        errors = {}
+
+        if user_input is not None:
+            session = async_get_clientsession(self.hass)
+            try:
+                client = Ambee(
+                    api_key=user_input[CONF_API_KEY],
+                    latitude=user_input[CONF_LATITUDE],
+                    longitude=user_input[CONF_LONGITUDE],
+                    session=session,
+                )
+                await client.air_quality()
+            except AmbeeAuthenticationError:
+                errors["base"] = "invalid_api_key"
+            except AmbeeError:
+                errors["base"] = "cannot_connect"
+            else:
+                return self.async_create_entry(
+                    title=user_input[CONF_NAME],
+                    data={
+                        CONF_API_KEY: user_input[CONF_API_KEY],
+                        CONF_LATITUDE: user_input[CONF_LATITUDE],
+                        CONF_LONGITUDE: user_input[CONF_LONGITUDE],
+                    },
+                )
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_API_KEY): str,
+                    vol.Optional(
+                        CONF_NAME, default=self.hass.config.location_name
+                    ): str,
+                    vol.Optional(
+                        CONF_LATITUDE, default=self.hass.config.latitude
+                    ): cv.latitude,
+                    vol.Optional(
+                        CONF_LONGITUDE, default=self.hass.config.longitude
+                    ): cv.longitude,
+                }
+            ),
+            errors=errors,
+        )

homeassistant/components/ambee/const.py

@@ -0,0 +1,69 @@
+"""Constants for the Ambee integration."""
+from __future__ import annotations
+
+from datetime import timedelta
+import logging
+from typing import Any, Final
+
+from homeassistant.components.sensor import ATTR_STATE_CLASS, STATE_CLASS_MEASUREMENT
+from homeassistant.const import (
+    ATTR_DEVICE_CLASS,
+    ATTR_NAME,
+    ATTR_SERVICE,
+    ATTR_UNIT_OF_MEASUREMENT,
+    CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
+    CONCENTRATION_PARTS_PER_BILLION,
+    CONCENTRATION_PARTS_PER_MILLION,
+    DEVICE_CLASS_CO,
+)
+
+DOMAIN: Final = "ambee"
+LOGGER = logging.getLogger(__package__)
+SCAN_INTERVAL = timedelta(minutes=180)
+
+SERVICE_AIR_QUALITY: Final = ("air_quality", "Air Quality")
+
+SENSORS: dict[str, dict[str, Any]] = {
+    "particulate_matter_2_5": {
+        ATTR_SERVICE: SERVICE_AIR_QUALITY,
+        ATTR_NAME: "Particulate Matter < 2.5 μm",
+        ATTR_UNIT_OF_MEASUREMENT: CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
+        ATTR_STATE_CLASS: STATE_CLASS_MEASUREMENT,
+    },
+    "particulate_matter_10": {
+        ATTR_SERVICE: SERVICE_AIR_QUALITY,
+        ATTR_NAME: "Particulate Matter < 10 μm",
+        ATTR_UNIT_OF_MEASUREMENT: CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
+        ATTR_STATE_CLASS: STATE_CLASS_MEASUREMENT,
+    },
+    "sulphur_dioxide": {
+        ATTR_SERVICE: SERVICE_AIR_QUALITY,
+        ATTR_NAME: "Sulphur Dioxide (SO2)",
+        ATTR_UNIT_OF_MEASUREMENT: CONCENTRATION_PARTS_PER_BILLION,
+        ATTR_STATE_CLASS: STATE_CLASS_MEASUREMENT,
+    },
+    "nitrogen_dioxide": {
+        ATTR_SERVICE: SERVICE_AIR_QUALITY,
+        ATTR_NAME: "Nitrogen Dioxide (NO2)",
+        ATTR_UNIT_OF_MEASUREMENT: CONCENTRATION_PARTS_PER_BILLION,
+        ATTR_STATE_CLASS: STATE_CLASS_MEASUREMENT,
+    },
+    "ozone": {
+        ATTR_SERVICE: SERVICE_AIR_QUALITY,
+        ATTR_NAME: "Ozone",
+        ATTR_UNIT_OF_MEASUREMENT: CONCENTRATION_PARTS_PER_BILLION,
+        ATTR_STATE_CLASS: STATE_CLASS_MEASUREMENT,
+    },
+    "carbon_monoxide": {
+        ATTR_SERVICE: SERVICE_AIR_QUALITY,
+        ATTR_NAME: "Carbon Monoxide (CO)",
+        ATTR_DEVICE_CLASS: DEVICE_CLASS_CO,
+        ATTR_UNIT_OF_MEASUREMENT: CONCENTRATION_PARTS_PER_MILLION,
+        ATTR_STATE_CLASS: STATE_CLASS_MEASUREMENT,
+    },
+    "air_quality_index": {
+        ATTR_SERVICE: SERVICE_AIR_QUALITY,
+        ATTR_NAME: "Air Quality Index (AQI)",
+        ATTR_STATE_CLASS: STATE_CLASS_MEASUREMENT,
+    },
+}

homeassistant/components/ambee/manifest.json

@@ -0,0 +1,9 @@
+{
+  "domain": "ambee",
+  "name": "Ambee",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/ambee",
+  "requirements": ["ambee==0.2.1"],
+  "codeowners": ["@frenck"],
+  "iot_class": "cloud_polling"
+}

homeassistant/components/ambee/sensor.py

@@ -0,0 +1,69 @@
+"""Support for Ambee sensors."""
+from __future__ import annotations
+
+from homeassistant.components.sensor import ATTR_STATE_CLASS, SensorEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import (
+    ATTR_DEVICE_CLASS,
+    ATTR_IDENTIFIERS,
+    ATTR_MANUFACTURER,
+    ATTR_NAME,
+    ATTR_SERVICE,
+    ATTR_UNIT_OF_MEASUREMENT,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import StateType
+from homeassistant.helpers.update_coordinator import (
+    CoordinatorEntity,
+    DataUpdateCoordinator,
+)
+
+from .const import DOMAIN, SENSORS
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up Ambee sensor based on a config entry."""
+    coordinator: DataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
+    async_add_entities(
+        AmbeeSensor(coordinator=coordinator, entry_id=entry.entry_id, key=sensor)
+        for sensor in SENSORS
+    )
+
+
+class AmbeeSensor(CoordinatorEntity, SensorEntity):
+    """Defines an Ambee sensor."""
+
+    def __init__(
+        self, *, coordinator: DataUpdateCoordinator, entry_id: str, key: str
+    ) -> None:
+        """Initialize Ambee sensor."""
+        super().__init__(coordinator=coordinator)
+        self._key = key
+        self._entry_id = entry_id
+        self._service_key, self._service_name = SENSORS[key][ATTR_SERVICE]
+
+        self._attr_device_class = SENSORS[key].get(ATTR_DEVICE_CLASS)
+        self._attr_name = SENSORS[key][ATTR_NAME]
+        self._attr_state_class = SENSORS[key].get(ATTR_STATE_CLASS)
+        self._attr_unique_id = f"{entry_id}_{key}"
+        self._attr_unit_of_measurement = SENSORS[key].get(ATTR_UNIT_OF_MEASUREMENT)
+
+    @property
+    def state(self) -> StateType:
+        """Return the state of the sensor."""
+        return getattr(self.coordinator.data, self._key)  # type: ignore[no-any-return]
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        """Return device information about this Ambee Service."""
+        return {
+            ATTR_IDENTIFIERS: {(DOMAIN, f"{self._entry_id}_{self._service_key}")},
+            ATTR_NAME: self._service_name,
+            ATTR_MANUFACTURER: "Ambee",
+        }

homeassistant/components/ambee/strings.json

@@ -0,0 +1,19 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "description": "Set up Ambee to integrate with Home Assistant.",
+        "data": {
+          "api_key": "[%key:common::config_flow::data::api_key%]",
+          "latitude": "[%key:common::config_flow::data::latitude%]",
+          "longitude": "[%key:common::config_flow::data::longitude%]",
+          "name": "[%key:common::config_flow::data::name%]"
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "invalid_api_key": "[%key:common::config_flow::error::invalid_api_key%]"
+    }
+  }
+}

homeassistant/components/ambee/translations/en.json

@@ -0,0 +1,19 @@
+{
+    "config": {
+        "error": {
+            "cannot_connect": "Failed to connect",
+            "invalid_api_key": "Invalid API key"
+        },
+        "step": {
+            "user": {
+                "data": {
+                    "api_key": "API Key",
+                    "latitude": "Latitude",
+                    "longitude": "Longitude",
+                    "name": "Name"
+                },
+                "description": "Set up Ambee to integrate with Home Assistant."
+            }
+        }
+    }
+}

homeassistant/generated/config_flows.py

@@ -18,6 +18,7 @@ FLOWS = [
     "airvisual",
     "alarmdecoder",
     "almond",
+    "ambee",
     "ambiclimate",
     "ambient_station",
     "apple_tv",

mypy.ini

@@ -143,6 +143,17 @@ no_implicit_optional = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.ambee.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+no_implicit_optional = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.ampio.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true

requirements_all.txt

@@ -253,6 +253,9 @@ aladdin_connect==0.3
 # homeassistant.components.alpha_vantage
 alpha_vantage==2.3.1
 
+# homeassistant.components.ambee
+ambee==0.2.1
+
 # homeassistant.components.ambiclimate
 ambiclimate==0.2.1
 

requirements_test_all.txt

@@ -169,6 +169,9 @@ aioymaps==1.1.0
 # homeassistant.components.airly
 airly==1.1.0
 
+# homeassistant.components.ambee
+ambee==0.2.1
+
 # homeassistant.components.ambiclimate
 ambiclimate==0.2.1
 

tests/components/ambee/__init__.py

@@ -0,0 +1 @@
+"""Tests for the Ambee integration."""

tests/components/ambee/test_config_flow.py

@@ -0,0 +1,129 @@
+"""Tests for the Ambee config flow."""
+
+from unittest.mock import patch
+
+from ambee import AmbeeAuthenticationError, AmbeeError
+
+from homeassistant.components.ambee.const import DOMAIN
+from homeassistant.config_entries import SOURCE_USER
+from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import RESULT_TYPE_CREATE_ENTRY, RESULT_TYPE_FORM
+
+
+async def test_full_user_flow(hass: HomeAssistant) -> None:
+    """Test the full user configuration flow."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    assert result.get("type") == RESULT_TYPE_FORM
+    assert result.get("step_id") == SOURCE_USER
+    assert "flow_id" in result
+
+    with patch(
+        "homeassistant.components.ambee.config_flow.Ambee.air_quality"
+    ) as mock_ambee, patch(
+        "homeassistant.components.ambee.async_setup_entry", return_value=True
+    ) as mock_setup_entry:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            user_input={
+                CONF_NAME: "Name",
+                CONF_API_KEY: "example",
+                CONF_LATITUDE: 52.42,
+                CONF_LONGITUDE: 4.44,
+            },
+        )
+
+    assert result2.get("type") == RESULT_TYPE_CREATE_ENTRY
+    assert result2.get("title") == "Name"
+    assert result2.get("data") == {
+        CONF_API_KEY: "example",
+        CONF_LATITUDE: 52.42,
+        CONF_LONGITUDE: 4.44,
+    }
+
+    assert len(mock_setup_entry.mock_calls) == 1
+    assert len(mock_ambee.mock_calls) == 1
+
+
+async def test_full_flow_with_authentication_error(hass: HomeAssistant) -> None:
+    """Test the full user configuration flow with an authentication error.
+
+    This tests tests a full config flow, with a case the user enters an invalid
+    API token, but recover by entering the correct one.
+    """
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    assert result.get("type") == RESULT_TYPE_FORM
+    assert result.get("step_id") == SOURCE_USER
+    assert "flow_id" in result
+
+    with patch(
+        "homeassistant.components.ambee.config_flow.Ambee.air_quality",
+        side_effect=AmbeeAuthenticationError,
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            user_input={
+                CONF_NAME: "Name",
+                CONF_API_KEY: "invalid",
+                CONF_LATITUDE: 52.42,
+                CONF_LONGITUDE: 4.44,
+            },
+        )
+
+    assert result2.get("type") == RESULT_TYPE_FORM
+    assert result2.get("step_id") == SOURCE_USER
+    assert result2.get("errors") == {"base": "invalid_api_key"}
+    assert "flow_id" in result2
+
+    with patch(
+        "homeassistant.components.ambee.config_flow.Ambee.air_quality"
+    ) as mock_ambee, patch(
+        "homeassistant.components.ambee.async_setup_entry", return_value=True
+    ) as mock_setup_entry:
+        result3 = await hass.config_entries.flow.async_configure(
+            result2["flow_id"],
+            user_input={
+                CONF_NAME: "Name",
+                CONF_API_KEY: "example",
+                CONF_LATITUDE: 52.42,
+                CONF_LONGITUDE: 4.44,
+            },
+        )
+
+    assert result3.get("type") == RESULT_TYPE_CREATE_ENTRY
+    assert result3.get("title") == "Name"
+    assert result3.get("data") == {
+        CONF_API_KEY: "example",
+        CONF_LATITUDE: 52.42,
+        CONF_LONGITUDE: 4.44,
+    }
+
+    assert len(mock_setup_entry.mock_calls) == 1
+    assert len(mock_ambee.mock_calls) == 1
+
+
+async def test_api_error(hass: HomeAssistant) -> None:
+    """Test API error."""
+    with patch(
+        "homeassistant.components.ambee.Ambee.air_quality",
+        side_effect=AmbeeError,
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": SOURCE_USER},
+            data={
+                CONF_NAME: "Name",
+                CONF_API_KEY: "example",
+                CONF_LATITUDE: 52.42,
+                CONF_LONGITUDE: 4.44,
+            },
+        )
+
+    assert result.get("type") == RESULT_TYPE_FORM
+    assert result.get("errors") == {"base": "cannot_connect"}