1
0
mirror of https://github.com/home-assistant/core.git synced 2026-02-15 07:36:16 +00:00

Add Google Air Quality integration (#145237)

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
Co-authored-by: Norbert Rittel <norbert@rittel.de>
This commit is contained in:
Thomas55555
2025-11-25 21:18:44 +01:00
committed by GitHub
parent d2fd200469
commit 1069233851
23 changed files with 2336 additions and 0 deletions

2
CODEOWNERS generated
View File

@@ -595,6 +595,8 @@ build.json @home-assistant/supervisor
/tests/components/goodwe/ @mletenay @starkillerOG
/homeassistant/components/google/ @allenporter
/tests/components/google/ @allenporter
/homeassistant/components/google_air_quality/ @Thomas55555
/tests/components/google_air_quality/ @Thomas55555
/homeassistant/components/google_assistant/ @home-assistant/cloud
/tests/components/google_assistant/ @home-assistant/cloud
/homeassistant/components/google_assistant_sdk/ @tronikos

View File

@@ -2,6 +2,7 @@
"domain": "google",
"name": "Google",
"integrations": [
"google_air_quality",
"google_assistant",
"google_assistant_sdk",
"google_cloud",

View File

@@ -0,0 +1,64 @@
"""The Google Air Quality integration."""
import asyncio
from google_air_quality_api.api import GoogleAirQualityApi
from google_air_quality_api.auth import Auth
from homeassistant.const import CONF_API_KEY, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import CONF_REFERRER
from .coordinator import (
GoogleAirQualityConfigEntry,
GoogleAirQualityRuntimeData,
GoogleAirQualityUpdateCoordinator,
)
PLATFORMS: list[Platform] = [
Platform.SENSOR,
]
async def async_setup_entry(
hass: HomeAssistant, entry: GoogleAirQualityConfigEntry
) -> bool:
"""Set up Google Air Quality from a config entry."""
session = async_get_clientsession(hass)
api_key = entry.data[CONF_API_KEY]
referrer = entry.data.get(CONF_REFERRER)
auth = Auth(session, api_key, referrer=referrer)
client = GoogleAirQualityApi(auth)
coordinators: dict[str, GoogleAirQualityUpdateCoordinator] = {}
for subentry_id in entry.subentries:
coordinators[subentry_id] = GoogleAirQualityUpdateCoordinator(
hass, entry, subentry_id, client
)
await asyncio.gather(
*(
coordinator.async_config_entry_first_refresh()
for coordinator in coordinators.values()
)
)
entry.runtime_data = GoogleAirQualityRuntimeData(
api=client,
subentries_runtime_data=coordinators,
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(async_update_options))
return True
async def async_unload_entry(
hass: HomeAssistant, entry: GoogleAirQualityConfigEntry
) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def async_update_options(
hass: HomeAssistant, entry: GoogleAirQualityConfigEntry
) -> None:
"""Update options."""
await hass.config_entries.async_reload(entry.entry_id)

View File

@@ -0,0 +1,198 @@
"""Config flow for the Google Air Quality integration."""
from __future__ import annotations
import logging
from typing import Any
from google_air_quality_api.api import GoogleAirQualityApi
from google_air_quality_api.auth import Auth
from google_air_quality_api.exceptions import GoogleAirQualityApiError
import voluptuous as vol
from homeassistant.config_entries import (
ConfigEntry,
ConfigEntryState,
ConfigFlow,
ConfigFlowResult,
ConfigSubentryFlow,
SubentryFlowResult,
)
from homeassistant.const import (
CONF_API_KEY,
CONF_LATITUDE,
CONF_LOCATION,
CONF_LONGITUDE,
CONF_NAME,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import section
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.selector import LocationSelector, LocationSelectorConfig
from .const import CONF_REFERRER, DOMAIN, SECTION_API_KEY_OPTIONS
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_API_KEY): str,
vol.Optional(SECTION_API_KEY_OPTIONS): section(
vol.Schema({vol.Optional(CONF_REFERRER): str}), {"collapsed": True}
),
}
)
async def _validate_input(
user_input: dict[str, Any],
api: GoogleAirQualityApi,
errors: dict[str, str],
description_placeholders: dict[str, str],
) -> bool:
try:
await api.async_air_quality(
lat=user_input[CONF_LOCATION][CONF_LATITUDE],
long=user_input[CONF_LOCATION][CONF_LONGITUDE],
)
except GoogleAirQualityApiError as err:
errors["base"] = "cannot_connect"
description_placeholders["error_message"] = str(err)
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return True
return False
def _get_location_schema(hass: HomeAssistant) -> vol.Schema:
"""Return the schema for a location with default values from the hass config."""
return vol.Schema(
{
vol.Required(CONF_NAME, default=hass.config.location_name): str,
vol.Required(
CONF_LOCATION,
default={
CONF_LATITUDE: hass.config.latitude,
CONF_LONGITUDE: hass.config.longitude,
},
): LocationSelector(LocationSelectorConfig(radius=False)),
}
)
def _is_location_already_configured(
hass: HomeAssistant, new_data: dict[str, float], epsilon: float = 1e-4
) -> bool:
"""Check if the location is already configured."""
for entry in hass.config_entries.async_entries(DOMAIN):
for subentry in entry.subentries.values():
# A more accurate way is to use the haversine formula, but for simplicity
# we use a simple distance check. The epsilon value is small anyway.
# This is mostly to capture cases where the user has slightly moved the location pin.
if (
abs(subentry.data[CONF_LATITUDE] - new_data[CONF_LATITUDE]) <= epsilon
and abs(subentry.data[CONF_LONGITUDE] - new_data[CONF_LONGITUDE])
<= epsilon
):
return True
return False
class GoogleAirQualityConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Google AirQuality."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
description_placeholders: dict[str, str] = {
"api_key_url": "https://developers.google.com/maps/documentation/air-quality/get-api-key",
"restricting_api_keys_url": "https://developers.google.com/maps/api-security-best-practices#restricting-api-keys",
}
if user_input is not None:
api_key = user_input[CONF_API_KEY]
referrer = user_input.get(SECTION_API_KEY_OPTIONS, {}).get(CONF_REFERRER)
self._async_abort_entries_match({CONF_API_KEY: api_key})
if _is_location_already_configured(self.hass, user_input[CONF_LOCATION]):
return self.async_abort(reason="already_configured")
session = async_get_clientsession(self.hass)
referrer = user_input.get(SECTION_API_KEY_OPTIONS, {}).get(CONF_REFERRER)
auth = Auth(session, user_input[CONF_API_KEY], referrer=referrer)
api = GoogleAirQualityApi(auth)
if await _validate_input(user_input, api, errors, description_placeholders):
return self.async_create_entry(
title="Google Air Quality",
data={
CONF_API_KEY: api_key,
CONF_REFERRER: referrer,
},
subentries=[
{
"subentry_type": "location",
"data": user_input[CONF_LOCATION],
"title": user_input[CONF_NAME],
"unique_id": None,
},
],
)
else:
user_input = {}
schema = STEP_USER_DATA_SCHEMA.schema.copy()
schema.update(_get_location_schema(self.hass).schema)
return self.async_show_form(
step_id="user",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(schema), user_input
),
errors=errors,
description_placeholders=description_placeholders,
)
@classmethod
@callback
def async_get_supported_subentry_types(
cls, config_entry: ConfigEntry
) -> dict[str, type[ConfigSubentryFlow]]:
"""Return subentries supported by this integration."""
return {"location": LocationSubentryFlowHandler}
class LocationSubentryFlowHandler(ConfigSubentryFlow):
"""Handle a subentry flow for location."""
async def async_step_location(
self,
user_input: dict[str, Any] | None = None,
) -> SubentryFlowResult:
"""Handle the location step."""
if self._get_entry().state != ConfigEntryState.LOADED:
return self.async_abort(reason="entry_not_loaded")
errors: dict[str, str] = {}
description_placeholders: dict[str, str] = {}
if user_input is not None:
if _is_location_already_configured(self.hass, user_input[CONF_LOCATION]):
return self.async_abort(reason="already_configured")
api: GoogleAirQualityApi = self._get_entry().runtime_data.api
if await _validate_input(user_input, api, errors, description_placeholders):
return self.async_create_entry(
title=user_input[CONF_NAME],
data=user_input[CONF_LOCATION],
)
else:
user_input = {}
return self.async_show_form(
step_id="location",
data_schema=self.add_suggested_values_to_schema(
_get_location_schema(self.hass), user_input
),
errors=errors,
description_placeholders=description_placeholders,
)
async_step_user = async_step_location

View File

@@ -0,0 +1,7 @@
"""Constants for the Google Air Quality integration."""
from typing import Final
DOMAIN = "google_air_quality"
SECTION_API_KEY_OPTIONS: Final = "api_key_options"
CONF_REFERRER: Final = "referrer"

View File

@@ -0,0 +1,68 @@
"""Coordinator for fetching data from Google Air Quality API."""
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import Final
from google_air_quality_api.api import GoogleAirQualityApi
from google_air_quality_api.exceptions import GoogleAirQualityApiError
from google_air_quality_api.model import AirQualityData
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
UPDATE_INTERVAL: Final = timedelta(hours=1)
type GoogleAirQualityConfigEntry = ConfigEntry[GoogleAirQualityRuntimeData]
class GoogleAirQualityUpdateCoordinator(DataUpdateCoordinator[AirQualityData]):
"""Coordinator for fetching Google AirQuality data."""
config_entry: GoogleAirQualityConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: GoogleAirQualityConfigEntry,
subentry_id: str,
client: GoogleAirQualityApi,
) -> None:
"""Initialize DataUpdateCoordinator."""
super().__init__(
hass,
_LOGGER,
config_entry=config_entry,
name=f"{DOMAIN}_{subentry_id}",
update_interval=UPDATE_INTERVAL,
)
self.client = client
subentry = config_entry.subentries[subentry_id]
self.lat = subentry.data[CONF_LATITUDE]
self.long = subentry.data[CONF_LONGITUDE]
async def _async_update_data(self) -> AirQualityData:
"""Fetch air quality data for this coordinate."""
try:
return await self.client.async_air_quality(self.lat, self.long)
except GoogleAirQualityApiError as ex:
_LOGGER.debug("Cannot fetch air quality data: %s", str(ex))
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="unable_to_fetch",
) from ex
@dataclass
class GoogleAirQualityRuntimeData:
"""Runtime data for the Google Air Quality integration."""
api: GoogleAirQualityApi
subentries_runtime_data: dict[str, GoogleAirQualityUpdateCoordinator]

View File

@@ -0,0 +1,15 @@
{
"entity": {
"sensor": {
"nitrogen_dioxide": {
"default": "mdi:molecule"
},
"ozone": {
"default": "mdi:molecule"
},
"sulphur_dioxide": {
"default": "mdi:molecule"
}
}
}
}

View File

@@ -0,0 +1,12 @@
{
"domain": "google_air_quality",
"name": "Google Air Quality",
"codeowners": ["@Thomas55555"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/google_air_quality",
"integration_type": "service",
"iot_class": "cloud_polling",
"loggers": ["google_air_quality_api"],
"quality_scale": "bronze",
"requirements": ["google_air_quality_api==1.1.1"]
}

View File

@@ -0,0 +1,80 @@
rules:
# Bronze
config-flow: done
brands: done
dependency-transparency: done
common-modules: done
has-entity-name: done
action-setup:
status: exempt
comment: Integration does not have actions
appropriate-polling: done
test-before-configure: done
entity-event-setup:
status: exempt
comment: Integration does not subscribe to events.
unique-config-entry: done
entity-unique-id: done
docs-installation-instructions: done
docs-removal-instructions: done
test-before-setup: done
docs-high-level-description: done
config-flow-test-coverage: done
docs-actions:
status: exempt
comment: Integration does not have actions
runtime-data: done
# Silver
log-when-unavailable: done
config-entry-unloading: done
reauthentication-flow: todo
action-exceptions:
status: exempt
comment: Integration does not have actions
docs-installation-parameters: todo
integration-owner: done
parallel-updates: done
test-coverage: todo
docs-configuration-parameters: todo
entity-unavailable: done
# Gold
docs-examples: todo
discovery-update-info: todo
entity-device-class: done
entity-translations: done
docs-data-update: done
entity-disabled-by-default:
status: exempt
comment: Nothing to disable at the moment
discovery:
status: exempt
comment: There is no physical device to discover.
exception-translations: done
devices: done
docs-supported-devices:
status: todo
comment: There is no physical device which can be supported. We can add a link to the supported countries.
icon-translations: done
docs-known-limitations: done
stale-devices:
status: exempt
comment: There are no device which can get stale.
docs-supported-functions: done
repair-issues:
status: exempt
comment: There is nothing to repair right now.
reconfiguration-flow: todo
entity-category: done
dynamic-devices:
status: exempt
comment: There are no device which can be added.
docs-troubleshooting: done
diagnostics: todo
docs-use-cases: todo
# Platinum
async-dependency: done
strict-typing: todo
inject-websession: done

View File

@@ -0,0 +1,213 @@
"""Creates the sensor entities for Google Air Quality."""
from collections.abc import Callable
from dataclasses import dataclass
import logging
from google_air_quality_api.model import AirQualityData
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigSubentry
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import GoogleAirQualityConfigEntry
from .const import DOMAIN
from .coordinator import GoogleAirQualityUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
# Coordinator is used to centralize the data updates
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class AirQualitySensorEntityDescription(SensorEntityDescription):
"""Describes Air Quality sensor entity."""
exists_fn: Callable[[AirQualityData], bool] = lambda _: True
options_fn: Callable[[AirQualityData], list[str] | None] = lambda _: None
value_fn: Callable[[AirQualityData], StateType]
native_unit_of_measurement_fn: Callable[[AirQualityData], str | None] = (
lambda _: None
)
translation_placeholders_fn: Callable[[AirQualityData], dict[str, str]] | None = (
None
)
AIR_QUALITY_SENSOR_TYPES: tuple[AirQualitySensorEntityDescription, ...] = (
AirQualitySensorEntityDescription(
key="uaqi",
translation_key="uaqi",
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.AQI,
value_fn=lambda x: x.indexes[0].aqi,
),
AirQualitySensorEntityDescription(
key="uaqi_category",
translation_key="uaqi_category",
device_class=SensorDeviceClass.ENUM,
options_fn=lambda x: x.indexes[0].category_options,
value_fn=lambda x: x.indexes[0].category,
),
AirQualitySensorEntityDescription(
key="local_aqi",
translation_key="local_aqi",
exists_fn=lambda x: x.indexes[1].aqi is not None,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.AQI,
value_fn=lambda x: x.indexes[1].aqi,
translation_placeholders_fn=lambda data: {
"local_aqi": data.indexes[1].display_name
},
),
AirQualitySensorEntityDescription(
key="local_category",
translation_key="local_category",
device_class=SensorDeviceClass.ENUM,
value_fn=lambda x: x.indexes[1].category,
options_fn=lambda x: x.indexes[1].category_options,
translation_placeholders_fn=lambda data: {
"local_aqi": data.indexes[1].display_name
},
),
AirQualitySensorEntityDescription(
key="uaqi_dominant_pollutant",
translation_key="uaqi_dominant_pollutant",
device_class=SensorDeviceClass.ENUM,
value_fn=lambda x: x.indexes[0].dominant_pollutant,
options_fn=lambda x: x.indexes[0].pollutant_options,
),
AirQualitySensorEntityDescription(
key="local_dominant_pollutant",
translation_key="local_dominant_pollutant",
device_class=SensorDeviceClass.ENUM,
value_fn=lambda x: x.indexes[1].dominant_pollutant,
options_fn=lambda x: x.indexes[1].pollutant_options,
translation_placeholders_fn=lambda data: {
"local_aqi": data.indexes[1].display_name
},
),
AirQualitySensorEntityDescription(
key="co",
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.CO,
native_unit_of_measurement_fn=lambda x: x.pollutants.co.concentration.units,
value_fn=lambda x: x.pollutants.co.concentration.value,
),
AirQualitySensorEntityDescription(
key="no2",
translation_key="nitrogen_dioxide",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement_fn=lambda x: x.pollutants.no2.concentration.units,
value_fn=lambda x: x.pollutants.no2.concentration.value,
),
AirQualitySensorEntityDescription(
key="o3",
translation_key="ozone",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement_fn=lambda x: x.pollutants.o3.concentration.units,
value_fn=lambda x: x.pollutants.o3.concentration.value,
),
AirQualitySensorEntityDescription(
key="pm10",
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.PM10,
native_unit_of_measurement_fn=lambda x: x.pollutants.pm10.concentration.units,
value_fn=lambda x: x.pollutants.pm10.concentration.value,
),
AirQualitySensorEntityDescription(
key="pm25",
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.PM25,
native_unit_of_measurement_fn=lambda x: x.pollutants.pm25.concentration.units,
value_fn=lambda x: x.pollutants.pm25.concentration.value,
),
AirQualitySensorEntityDescription(
key="so2",
translation_key="sulphur_dioxide",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement_fn=lambda x: x.pollutants.so2.concentration.units,
value_fn=lambda x: x.pollutants.so2.concentration.value,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: GoogleAirQualityConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up sensor platform."""
coordinators = entry.runtime_data.subentries_runtime_data
for subentry_id, subentry in entry.subentries.items():
coordinator = coordinators[subentry_id]
_LOGGER.debug("subentry.data: %s", subentry.data)
async_add_entities(
(
AirQualitySensorEntity(coordinator, description, subentry_id, subentry)
for description in AIR_QUALITY_SENSOR_TYPES
if description.exists_fn(coordinator.data)
),
config_subentry_id=subentry_id,
)
class AirQualitySensorEntity(
CoordinatorEntity[GoogleAirQualityUpdateCoordinator], SensorEntity
):
"""Defining the Air Quality Sensors with AirQualitySensorEntityDescription."""
entity_description: AirQualitySensorEntityDescription
_attr_attribution = "Data provided by Google Air Quality"
_attr_has_entity_name = True
def __init__(
self,
coordinator: GoogleAirQualityUpdateCoordinator,
description: AirQualitySensorEntityDescription,
subentry_id: str,
subentry: ConfigSubentry,
) -> None:
"""Set up Air Quality Sensors."""
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = f"{description.key}_{subentry.data[CONF_LATITUDE]}_{subentry.data[CONF_LONGITUDE]}"
self._attr_device_info = DeviceInfo(
identifiers={
(DOMAIN, f"{self.coordinator.config_entry.entry_id}_{subentry_id}")
},
name=subentry.title,
entry_type=DeviceEntryType.SERVICE,
)
if description.translation_placeholders_fn:
self._attr_translation_placeholders = (
description.translation_placeholders_fn(coordinator.data)
)
@property
def native_value(self) -> StateType:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.coordinator.data)
@property
def options(self) -> list[str] | None:
"""Return the option of the sensor."""
return self.entity_description.options_fn(self.coordinator.data)
@property
def native_unit_of_measurement(self) -> str | None:
"""Return the native unit of measurement of the sensor."""
return self.entity_description.native_unit_of_measurement_fn(
self.coordinator.data
)

View File

@@ -0,0 +1,237 @@
{
"common": {
"unable_to_fetch": "Unable to access the Google API. See the debug logs for more details."
},
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"unable_to_fetch": "[%key:component::google_air_quality::common::unable_to_fetch%]",
"unknown": "[%key:common::config_flow::error::unknown%]",
"wrong_account": "Wrong account: Please authenticate with the right account."
},
"create_entry": {
"default": "[%key:common::config_flow::create_entry::authenticated%]"
},
"error": {
"cannot_connect": "Unable to connect to the Google Air Quality API:\n\n{error_message}",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"user": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]",
"location": "[%key:common::config_flow::data::location%]",
"name": "[%key:common::config_flow::data::name%]"
},
"data_description": {
"api_key": "A unique alphanumeric string that associates your Google billing account with Google Air Quality API",
"location": "Location coordinates",
"name": "Location name"
},
"description": "Get your API key from [here]({api_key_url}).",
"sections": {
"api_key_options": {
"data": {
"referrer": "HTTP referrer"
},
"data_description": {
"referrer": "Specify this only if the API key has a [website application restriction]({restricting_api_keys_url})."
},
"name": "Optional API key options"
}
}
}
}
},
"config_subentries": {
"location": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_location%]",
"unable_to_fetch": "[%key:component::google_air_quality::common::unable_to_fetch%]"
},
"entry_type": "Air quality location",
"error": {
"no_data_for_location": "Information is unavailable for this location. Please try a different location.",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"initiate_flow": {
"user": "Add location"
},
"step": {
"user": {
"data": {
"location": "[%key:common::config_flow::data::location%]"
},
"description": "Select the coordinates for which you want to create an entry.",
"title": "Air quality data location"
}
}
}
},
"entity": {
"sensor": {
"local_aqi": {
"name": "{local_aqi} AQI"
},
"local_category": {
"name": "{local_aqi} category",
"state": {
"1_blue": "1 - Blue",
"1_green": "1 - Green",
"1a_very_good_air_quality": "1A - Very good air quality",
"1b_good_air_quality": "1B - Good air quality",
"2_cyan": "2 - Cyan",
"2_light_green": "2 - Light green",
"2_orange": "4 - Orange",
"2_red": "5 - Red",
"2_yellow": "3 - Yellow",
"2a_acceptable_air_quality": "2A - Acceptable air quality",
"2b_acceptable_air_quality": "2B - Acceptable air quality",
"3_green": "3 - Green",
"3a_aggravated_air_quality": "3A - Aggravated air quality",
"3b_bad_air_quality": "3B - Bad air quality",
"4_yellow_watch": "4 - Yellow/Watch",
"5_orange_alert": "5 - Orange/Alert",
"6_red_alert": "6 - Red/Alert+",
"10_33": "10-33% of guideline",
"33_66": "33-66% of guideline",
"66_100": "66-100% of guideline",
"above_average_air_pollution": "Above average air pollution",
"acceptable": "Acceptable",
"acceptable_air_quality": "Acceptable air quality",
"acutely_unhealthy": "Acutely unhealthy air quality",
"alarm_level": "Alarm level",
"alert": "Alert threshold",
"alert_level": "Alert level",
"average_air_pollution": "Average air pollution",
"average_air_quality": "Average air quality",
"bad_air_quality": "Bad air quality",
"below_10": "Less than 10% of guideline",
"below_average_air_pollution": "Below average air pollution",
"caution": "Caution",
"clean": "Clean",
"considerable_air_pollution": "Considerable air pollution",
"degraded_air_quality": "Degraded air quality",
"desirable_air_quality": "Desirable air quality",
"emergency": "Emergency",
"emergency_level": "Emergency level",
"evident_air_pollution": "Evident air pollution",
"excellent": "Excellent",
"excellent_air_quality": "Excellent air quality",
"extremely_bad_air_quality": "Extremely bad air quality",
"extremely_poor_air_quality": "Extremely poor air quality",
"extremely_unfavorable_air_quality": "Extremely unfavorable air quality",
"extremely_unhealthy_air_quality": "Extremely Unhealthy air quality",
"fair_air_quality": "Fair air quality",
"fairly_good_air_quality": "Fairly good air quality",
"good": "Good",
"good_air_quality": "Good air quality",
"greater_100": "Greater than 100% of guideline",
"hazardous_air_quality": "Hazardous air quality",
"heavily_polluted": "Heavily polluted",
"heavy_air_pollution": "Heavy air pollution",
"high_air_pollution": "High air pollution",
"high_air_quality": "High air pollution",
"high_health_risk": "High health risk",
"horrible_air_quality": "Horrible air quality",
"light_air_pollution": "Light air pollution",
"low": "Low pollution",
"low_air_pollution": "Low air pollution",
"low_air_quality": "Low air quality",
"low_health_risk": "Low health risk",
"medium_air_pollution": "Medium air pollution",
"medium_air_quality": "Medium air quality",
"moderate_air_pollution": "Moderate air pollution",
"moderate_air_quality": "Moderate air quality",
"moderate_health_risk": "Moderate health risk",
"moderately_polluted": "Moderately polluted",
"n1_good_air_quality": "N1 - Good air quality",
"n2_moderate_air_quality": "N2 - Moderate air quality",
"n3_bad_air_quality": "N3 - Bad air quality",
"n4_very_bad_air_quality": "N4 - Very bad air quality",
"n5_poor_air_quality": "N5 - Poor air quality",
"normal": "Normal",
"polluted": "Polluted",
"poor_air_quality": "Poor air quality",
"precautionary_level": "Precautionary level",
"reasonably_good_air_quality": "Reasonably good air quality",
"regular_air_quality": "Regular air quality",
"satisfactory_air_quality": "Satisfactory air quality",
"serious_air_pollution": "Serious air pollution",
"seriously_polluted": "Seriously polluted",
"severe_air_pollution": "Severe air pollution",
"severe_air_quality": "Severe air quality",
"slightly_polluted": "Slightly polluted",
"sufficient_air_quality": "Sufficient air quality",
"unfavorable_air_quality": "Unfavorable air quality",
"unfavorable_sensitive": "Unfavorable air quality for sensitive groups",
"unhealthy_air_quality": "Unhealthy air quality",
"unhealthy_sensitive": "Unhealthy air quality for sensitive groups",
"unsatisfactory_air_quality": "Unsatisfactory air quality",
"very_bad_air_quality": "Very bad air quality",
"very_good_air_quality": "Very good air quality",
"very_high_air_pollution": "Very high air pollution",
"very_high_air_quality": "Very High air pollution",
"very_high_health_risk": "Very high health risk",
"very_low_air_pollution": "Very low air pollution",
"very_polluted": "Very polluted",
"very_poor_air_quality": "Very poor air quality",
"very_unfavorable_air_quality": "Very unfavorable air quality",
"very_unhealthy": "Very unhealthy air quality",
"very_unhealthy_air_quality": "Very unhealthy air quality",
"warning_air_pollution": "Warning level air pollution"
}
},
"local_dominant_pollutant": {
"name": "{local_aqi} dominant pollutant",
"state": {
"co": "[%key:component::sensor::entity_component::carbon_monoxide::name%]",
"no2": "[%key:component::sensor::entity_component::nitrogen_dioxide::name%]",
"o3": "[%key:component::sensor::entity_component::ozone::name%]",
"pm10": "[%key:component::sensor::entity_component::pm10::name%]",
"pm25": "[%key:component::sensor::entity_component::pm25::name%]",
"so2": "[%key:component::sensor::entity_component::sulphur_dioxide::name%]"
}
},
"nitrogen_dioxide": {
"name": "[%key:component::sensor::entity_component::nitrogen_dioxide::name%]"
},
"ozone": {
"name": "[%key:component::sensor::entity_component::ozone::name%]"
},
"sulphur_dioxide": {
"name": "[%key:component::sensor::entity_component::sulphur_dioxide::name%]"
},
"uaqi": {
"name": "Universal Air Quality Index"
},
"uaqi_category": {
"name": "UAQI category",
"state": {
"excellent_air_quality": "[%key:component::google_air_quality::entity::sensor::local_category::state::excellent_air_quality%]",
"good_air_quality": "[%key:component::google_air_quality::entity::sensor::local_category::state::good_air_quality%]",
"low_air_quality": "[%key:component::google_air_quality::entity::sensor::local_category::state::low_air_quality%]",
"moderate_air_quality": "[%key:component::google_air_quality::entity::sensor::local_category::state::moderate_air_quality%]",
"poor_air_quality": "[%key:component::google_air_quality::entity::sensor::local_category::state::poor_air_quality%]"
}
},
"uaqi_dominant_pollutant": {
"name": "UAQI dominant pollutant",
"state": {
"co": "[%key:component::sensor::entity_component::carbon_monoxide::name%]",
"no2": "[%key:component::sensor::entity_component::nitrogen_dioxide::name%]",
"o3": "[%key:component::sensor::entity_component::ozone::name%]",
"pm10": "[%key:component::sensor::entity_component::pm10::name%]",
"pm25": "[%key:component::sensor::entity_component::pm25::name%]",
"so2": "[%key:component::sensor::entity_component::sulphur_dioxide::name%]"
}
}
}
},
"exceptions": {
"unable_to_fetch": {
"message": "[%key:component::google_air_quality::common::unable_to_fetch%]"
}
}
}

View File

@@ -249,6 +249,7 @@ FLOWS = {
"gogogate2",
"goodwe",
"google",
"google_air_quality",
"google_assistant_sdk",
"google_cloud",
"google_drive",

View File

@@ -2392,6 +2392,12 @@
"google": {
"name": "Google",
"integrations": {
"google_air_quality": {
"integration_type": "service",
"config_flow": true,
"iot_class": "cloud_polling",
"name": "Google Air Quality"
},
"google_assistant_sdk": {
"integration_type": "service",
"config_flow": true,

3
requirements_all.txt generated
View File

@@ -1086,6 +1086,9 @@ google-nest-sdm==9.1.0
# homeassistant.components.google_photos
google-photos-library-api==0.12.1
# homeassistant.components.google_air_quality
google_air_quality_api==1.1.1
# homeassistant.components.slide
# homeassistant.components.slide_local
goslide-api==0.7.0

View File

@@ -962,6 +962,9 @@ google-nest-sdm==9.1.0
# homeassistant.components.google_photos
google-photos-library-api==0.12.1
# homeassistant.components.google_air_quality
google_air_quality_api==1.1.1
# homeassistant.components.slide
# homeassistant.components.slide_local
goslide-api==0.7.0

View File

@@ -0,0 +1 @@
"""Tests for the Google Air Quality integration."""

View File

@@ -0,0 +1,103 @@
"""Test fixtures for Google Air Quality."""
from collections.abc import AsyncGenerator, Generator
from typing import Any
from unittest.mock import AsyncMock, Mock, patch
from google_air_quality_api.model import AirQualityData
import pytest
from homeassistant.components.google_air_quality import CONF_REFERRER
from homeassistant.components.google_air_quality.const import DOMAIN
from homeassistant.config_entries import ConfigSubentryDataWithId
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, load_json_object_fixture
USER_IDENTIFIER = "user-identifier-1"
CONFIG_ENTRY_ID = "api-key-1234"
CONFIG_ENTRY_ID_2 = "user-identifier-2"
CLIENT_ID = "1234"
CLIENT_SECRET = "5678"
FAKE_ACCESS_TOKEN = "some-access-token"
FAKE_REFRESH_TOKEN = "some-refresh-token"
EXPIRES_IN = 3600
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.google_air_quality.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_subentries() -> list[ConfigSubentryDataWithId]:
"""Fixture for subentries."""
return [
ConfigSubentryDataWithId(
data={
CONF_LATITUDE: 10.1,
CONF_LONGITUDE: 20.1,
},
subentry_type="location",
title="Home",
subentry_id="home-subentry-id",
unique_id=None,
)
]
@pytest.fixture
def mock_config_entry(
hass: HomeAssistant, mock_subentries: list[ConfigSubentryDataWithId]
) -> MockConfigEntry:
"""Fixture for a config and a subentry."""
return MockConfigEntry(
domain=DOMAIN,
title=DOMAIN,
data={CONF_API_KEY: "test-api-key", CONF_REFERRER: None},
entry_id="123456789",
subentries_data=[*mock_subentries],
)
@pytest.fixture(name="mock_api")
def mock_client_api() -> Generator[Mock]:
"""Set up fake Google Air Quality API responses from fixtures."""
responses = load_json_object_fixture("air_quality_data.json", DOMAIN)
with (
patch(
"homeassistant.components.google_air_quality.GoogleAirQualityApi",
autospec=True,
) as mock_api,
patch(
"homeassistant.components.google_air_quality.config_flow.GoogleAirQualityApi",
new=mock_api,
),
):
api = mock_api.return_value
api.async_air_quality.return_value = AirQualityData.from_dict(responses)
yield api
@pytest.fixture(name="setup_integration")
async def mock_setup_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_api: Mock,
) -> AsyncGenerator[Any, Any]:
"""Fixture to set up the integration."""
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.google_air_quality.GoogleAirQualityApi",
return_value=mock_api,
):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
yield

View File

@@ -0,0 +1,86 @@
{
"dateTime": "2025-05-14T20:00:00Z",
"regionCode": "de",
"indexes": [
{
"code": "uaqi",
"displayName": "Universal AQI",
"aqi": 80,
"aqiDisplay": "80",
"color": {
"red": 0.34509805,
"green": 0.74509805,
"blue": 0.20784314
},
"category": "Excellent air quality",
"dominantPollutant": "o3"
},
{
"code": "deu_uba",
"displayName": "LQI (DE)",
"color": {
"red": 0.3137255,
"green": 0.8039216,
"blue": 0.6666667
},
"category": "Good air quality",
"dominantPollutant": "no2"
}
],
"pollutants": [
{
"code": "co",
"displayName": "CO",
"fullName": "Carbon monoxide",
"concentration": {
"value": 269.02,
"units": "PARTS_PER_BILLION"
}
},
{
"code": "no2",
"displayName": "NO2",
"fullName": "Nitrogen dioxide",
"concentration": {
"value": 14.18,
"units": "PARTS_PER_BILLION"
}
},
{
"code": "o3",
"displayName": "O3",
"fullName": "Ozone",
"concentration": {
"value": 24.94,
"units": "PARTS_PER_BILLION"
}
},
{
"code": "pm10",
"displayName": "PM10",
"fullName": "Inhalable particulate matter (\u003c10µm)",
"concentration": {
"value": 21.95,
"units": "MICROGRAMS_PER_CUBIC_METER"
}
},
{
"code": "pm25",
"displayName": "PM2.5",
"fullName": "Fine particulate matter (\u003c2.5µm)",
"concentration": {
"value": 10.6,
"units": "MICROGRAMS_PER_CUBIC_METER"
}
},
{
"code": "so2",
"displayName": "SO2",
"fullName": "Sulfur dioxide",
"concentration": {
"value": 1.2,
"units": "PARTS_PER_BILLION"
}
}
]
}

View File

@@ -0,0 +1,159 @@
{
"results": [
{
"place": "//places.googleapis.com/places/ChIJ41tlyVtsmkcRjM3xQYYTiB8",
"placeId": "ChIJ41tlyVtsmkcRjM3xQYYTiB8",
"location": {
"latitude": 48.0002263,
"longitude": 8.9992729999999987
},
"granularity": "GEOMETRIC_CENTER",
"viewport": {
"low": {
"latitude": 47.9988721697085,
"longitude": 8.9979237197084974
},
"high": {
"latitude": 48.0015701302915,
"longitude": 9.000621680291502
}
},
"bounds": {
"low": {
"latitude": 48.0001829,
"longitude": 8.9983465
},
"high": {
"latitude": 48.0002594,
"longitude": 9.0001989
}
},
"formattedAddress": "Straße Ohne Straßennamen, 88637 Buchheim, Deutschland",
"addressComponents": [
{
"longText": "Straße Ohne Straßennamen",
"shortText": "Straße Ohne Straßennamen",
"types": ["route"],
"languageCode": "de"
},
{
"longText": "Buchheim",
"shortText": "Buchheim",
"types": ["locality", "political"],
"languageCode": "de"
},
{
"longText": "Landkreis Tuttlingen",
"shortText": "TUT",
"types": ["administrative_area_level_3", "political"],
"languageCode": "de"
},
{
"longText": "Freiburg",
"shortText": "Freiburg",
"types": ["administrative_area_level_2", "political"],
"languageCode": "de"
},
{
"longText": "Baden-Württemberg",
"shortText": "BW",
"types": ["administrative_area_level_1", "political"],
"languageCode": "de"
},
{
"longText": "Deutschland",
"shortText": "DE",
"types": ["country", "political"],
"languageCode": "de"
},
{
"longText": "88637",
"shortText": "88637",
"types": ["postal_code"]
}
],
"types": ["route"]
},
{
"place": "//places.googleapis.com/places/GhIJAQAAAAAASEARAAAAAAAAIkA",
"placeId": "GhIJAQAAAAAASEARAAAAAAAAIkA",
"location": {
"latitude": 48.000000000000007,
"longitude": 9
},
"granularity": "GEOMETRIC_CENTER",
"viewport": {
"low": {
"latitude": 47.9987135197085,
"longitude": 8.9987135197084971
},
"high": {
"latitude": 48.0014114802915,
"longitude": 9.0014114802915017
}
},
"bounds": {
"low": {
"latitude": 48.000000000000007,
"longitude": 9
},
"high": {
"latitude": 48.000125,
"longitude": 9.0001249999999988
}
},
"formattedAddress": "2222+22, 88637 Buchheim, Deutschland",
"addressComponents": [
{
"longText": "2222+22",
"shortText": "2222+22",
"types": ["plus_code"]
},
{
"longText": "Buchheim",
"shortText": "Buchheim",
"types": ["locality", "political"],
"languageCode": "de"
},
{
"longText": "Landkreis Tuttlingen",
"shortText": "TUT",
"types": ["administrative_area_level_3", "political"],
"languageCode": "de"
},
{
"longText": "Freiburg",
"shortText": "Freiburg",
"types": ["administrative_area_level_2", "political"],
"languageCode": "de"
},
{
"longText": "Baden-Württemberg",
"shortText": "BW",
"types": ["administrative_area_level_1", "political"],
"languageCode": "de"
},
{
"longText": "Deutschland",
"shortText": "DE",
"types": ["country", "political"],
"languageCode": "de"
},
{
"longText": "88637",
"shortText": "88637",
"types": ["postal_code"]
}
],
"types": ["plus_code"],
"plusCode": {
"globalCode": "8FWF2222+22",
"compoundCode": "2222+22 Buchheim, Deutschland"
}
}
],
"plusCode": {
"globalCode": "8FWF2222+222",
"compoundCode": "2222+222 Buchheim, Deutschland"
}
}

View File

@@ -0,0 +1,635 @@
# serializer version: 1
# name: test_sensor_snapshot[sensor.home_carbon_monoxide-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.home_carbon_monoxide',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.CO: 'carbon_monoxide'>,
'original_icon': None,
'original_name': 'Carbon monoxide',
'platform': 'google_air_quality',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'co_10.1_20.1',
'unit_of_measurement': 'ppm',
})
# ---
# name: test_sensor_snapshot[sensor.home_carbon_monoxide-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by Google Air Quality',
'device_class': 'carbon_monoxide',
'friendly_name': 'Home Carbon monoxide',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': 'ppm',
}),
'context': <ANY>,
'entity_id': 'sensor.home_carbon_monoxide',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.26902',
})
# ---
# name: test_sensor_snapshot[sensor.home_lqi_de_category-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'very_good_air_quality',
'good_air_quality',
'moderate_air_quality',
'poor_air_quality',
'very_poor_air_quality',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.home_lqi_de_category',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'LQI (DE) category',
'platform': 'google_air_quality',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'local_category',
'unique_id': 'local_category_10.1_20.1',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_snapshot[sensor.home_lqi_de_category-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by Google Air Quality',
'device_class': 'enum',
'friendly_name': 'Home LQI (DE) category',
'options': list([
'very_good_air_quality',
'good_air_quality',
'moderate_air_quality',
'poor_air_quality',
'very_poor_air_quality',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.home_lqi_de_category',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'good_air_quality',
})
# ---
# name: test_sensor_snapshot[sensor.home_lqi_de_dominant_pollutant-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'no2',
'o3',
'pm25',
'pm10',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.home_lqi_de_dominant_pollutant',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'LQI (DE) dominant pollutant',
'platform': 'google_air_quality',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'local_dominant_pollutant',
'unique_id': 'local_dominant_pollutant_10.1_20.1',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_snapshot[sensor.home_lqi_de_dominant_pollutant-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by Google Air Quality',
'device_class': 'enum',
'friendly_name': 'Home LQI (DE) dominant pollutant',
'options': list([
'no2',
'o3',
'pm25',
'pm10',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.home_lqi_de_dominant_pollutant',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'no2',
})
# ---
# name: test_sensor_snapshot[sensor.home_nitrogen_dioxide-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.home_nitrogen_dioxide',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Nitrogen dioxide',
'platform': 'google_air_quality',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'nitrogen_dioxide',
'unique_id': 'no2_10.1_20.1',
'unit_of_measurement': 'ppb',
})
# ---
# name: test_sensor_snapshot[sensor.home_nitrogen_dioxide-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by Google Air Quality',
'friendly_name': 'Home Nitrogen dioxide',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': 'ppb',
}),
'context': <ANY>,
'entity_id': 'sensor.home_nitrogen_dioxide',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '14.18',
})
# ---
# name: test_sensor_snapshot[sensor.home_ozone-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.home_ozone',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Ozone',
'platform': 'google_air_quality',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'ozone',
'unique_id': 'o3_10.1_20.1',
'unit_of_measurement': 'ppb',
})
# ---
# name: test_sensor_snapshot[sensor.home_ozone-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by Google Air Quality',
'friendly_name': 'Home Ozone',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': 'ppb',
}),
'context': <ANY>,
'entity_id': 'sensor.home_ozone',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '24.94',
})
# ---
# name: test_sensor_snapshot[sensor.home_pm10-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.home_pm10',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.PM10: 'pm10'>,
'original_icon': None,
'original_name': 'PM10',
'platform': 'google_air_quality',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'pm10_10.1_20.1',
'unit_of_measurement': 'μg/m³',
})
# ---
# name: test_sensor_snapshot[sensor.home_pm10-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by Google Air Quality',
'device_class': 'pm10',
'friendly_name': 'Home PM10',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': 'μg/m³',
}),
'context': <ANY>,
'entity_id': 'sensor.home_pm10',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '21.95',
})
# ---
# name: test_sensor_snapshot[sensor.home_pm2_5-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.home_pm2_5',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.PM25: 'pm25'>,
'original_icon': None,
'original_name': 'PM2.5',
'platform': 'google_air_quality',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'pm25_10.1_20.1',
'unit_of_measurement': 'μg/m³',
})
# ---
# name: test_sensor_snapshot[sensor.home_pm2_5-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by Google Air Quality',
'device_class': 'pm25',
'friendly_name': 'Home PM2.5',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': 'μg/m³',
}),
'context': <ANY>,
'entity_id': 'sensor.home_pm2_5',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '10.6',
})
# ---
# name: test_sensor_snapshot[sensor.home_sulphur_dioxide-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.home_sulphur_dioxide',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sulphur dioxide',
'platform': 'google_air_quality',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sulphur_dioxide',
'unique_id': 'so2_10.1_20.1',
'unit_of_measurement': 'ppb',
})
# ---
# name: test_sensor_snapshot[sensor.home_sulphur_dioxide-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by Google Air Quality',
'friendly_name': 'Home Sulphur dioxide',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': 'ppb',
}),
'context': <ANY>,
'entity_id': 'sensor.home_sulphur_dioxide',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '1.2',
})
# ---
# name: test_sensor_snapshot[sensor.home_uaqi_category-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'excellent_air_quality',
'good_air_quality',
'moderate_air_quality',
'low_air_quality',
'poor_air_quality',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.home_uaqi_category',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'UAQI category',
'platform': 'google_air_quality',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'uaqi_category',
'unique_id': 'uaqi_category_10.1_20.1',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_snapshot[sensor.home_uaqi_category-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by Google Air Quality',
'device_class': 'enum',
'friendly_name': 'Home UAQI category',
'options': list([
'excellent_air_quality',
'good_air_quality',
'moderate_air_quality',
'low_air_quality',
'poor_air_quality',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.home_uaqi_category',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'excellent_air_quality',
})
# ---
# name: test_sensor_snapshot[sensor.home_uaqi_dominant_pollutant-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'co',
'no2',
'o3',
'pm10',
'pm25',
'so2',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.home_uaqi_dominant_pollutant',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'UAQI dominant pollutant',
'platform': 'google_air_quality',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'uaqi_dominant_pollutant',
'unique_id': 'uaqi_dominant_pollutant_10.1_20.1',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_snapshot[sensor.home_uaqi_dominant_pollutant-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by Google Air Quality',
'device_class': 'enum',
'friendly_name': 'Home UAQI dominant pollutant',
'options': list([
'co',
'no2',
'o3',
'pm10',
'pm25',
'so2',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.home_uaqi_dominant_pollutant',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'o3',
})
# ---
# name: test_sensor_snapshot[sensor.home_universal_air_quality_index-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.home_universal_air_quality_index',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.AQI: 'aqi'>,
'original_icon': None,
'original_name': 'Universal Air Quality Index',
'platform': 'google_air_quality',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'uaqi',
'unique_id': 'uaqi_10.1_20.1',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_snapshot[sensor.home_universal_air_quality_index-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by Google Air Quality',
'device_class': 'aqi',
'friendly_name': 'Home Universal Air Quality Index',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.home_universal_air_quality_index',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '80',
})
# ---

View File

@@ -0,0 +1,369 @@
"""Test the Google Air Quality config flow."""
from unittest.mock import AsyncMock
from google_air_quality_api.exceptions import GoogleAirQualityApiError
import pytest
from homeassistant.components.google_air_quality.const import (
CONF_REFERRER,
DOMAIN,
SECTION_API_KEY_OPTIONS,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import (
CONF_API_KEY,
CONF_LATITUDE,
CONF_LOCATION,
CONF_LONGITUDE,
CONF_NAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry, get_schema_suggested_value
def _assert_create_entry_result(
result: dict, expected_referrer: str | None = None
) -> None:
"""Assert that the result is a create entry result."""
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Google Air Quality"
assert result["data"] == {
CONF_API_KEY: "test-api-key",
CONF_REFERRER: expected_referrer,
}
assert len(result["subentries"]) == 1
subentry = result["subentries"][0]
assert subentry["subentry_type"] == "location"
assert subentry["title"] == "test-name"
assert subentry["data"] == {
CONF_LATITUDE: 10.1,
CONF_LONGITUDE: 20.1,
}
async def test_create_entry(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_api: AsyncMock,
) -> None:
"""Test creating a config entry."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: "test-name",
CONF_API_KEY: "test-api-key",
CONF_LOCATION: {
CONF_LATITUDE: 10.1,
CONF_LONGITUDE: 20.1,
},
},
)
mock_api.async_air_quality.assert_called_once_with(lat=10.1, long=20.1)
_assert_create_entry_result(result)
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_with_referrer(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_api: AsyncMock,
) -> None:
"""Test we get the form and optional referrer is specified."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: "test-name",
CONF_API_KEY: "test-api-key",
SECTION_API_KEY_OPTIONS: {
CONF_REFERRER: "test-referrer",
},
CONF_LOCATION: {
CONF_LATITUDE: 10.1,
CONF_LONGITUDE: 20.1,
},
},
)
mock_api.async_air_quality.assert_called_once_with(lat=10.1, long=20.1)
_assert_create_entry_result(result, expected_referrer="test-referrer")
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("api_exception", "expected_error"),
[
(GoogleAirQualityApiError(), "cannot_connect"),
(ValueError(), "unknown"),
],
)
async def test_form_exceptions(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_api: AsyncMock,
api_exception,
expected_error,
) -> None:
"""Test we handle exceptions."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_api.async_air_quality.side_effect = api_exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: "test-name",
CONF_API_KEY: "test-api-key",
CONF_LOCATION: {
CONF_LATITUDE: 10.1,
CONF_LONGITUDE: 20.1,
},
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": expected_error}
# On error, the form should have the previous user input
data_schema = result["data_schema"].schema
assert get_schema_suggested_value(data_schema, CONF_NAME) == "test-name"
assert get_schema_suggested_value(data_schema, CONF_API_KEY) == "test-api-key"
assert get_schema_suggested_value(data_schema, CONF_LOCATION) == {
CONF_LATITUDE: 10.1,
CONF_LONGITUDE: 20.1,
}
# Make sure the config flow tests finish with either an
# FlowResultType.CREATE_ENTRY or FlowResultType.ABORT so
# we can show the config flow is able to recover from an error.
mock_api.async_air_quality.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: "test-name",
CONF_API_KEY: "test-api-key",
CONF_LOCATION: {
CONF_LATITUDE: 10.1,
CONF_LONGITUDE: 20.1,
},
},
)
_assert_create_entry_result(result)
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_api_key_already_configured(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_api: AsyncMock,
) -> None:
"""Test user input for config_entry with API key that already exists."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: "test-name",
CONF_API_KEY: "test-api-key",
CONF_LOCATION: {
CONF_LATITUDE: 10.2,
CONF_LONGITUDE: 20.2,
},
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert mock_api.async_air_quality.call_count == 0
async def test_form_location_already_configured(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_api: AsyncMock,
) -> None:
"""Test user input for a location that already exists."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: "test-name",
CONF_API_KEY: "another-api-key",
CONF_LOCATION: {
CONF_LATITUDE: 10.1001,
CONF_LONGITUDE: 20.0999,
},
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert mock_api.async_air_quality.call_count == 0
async def test_form_not_already_configured(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_api: AsyncMock,
) -> None:
"""Test user input for config_entry different than the existing one."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: "new-test-name",
CONF_API_KEY: "new-test-api-key",
CONF_LOCATION: {
CONF_LATITUDE: 10.1002,
CONF_LONGITUDE: 20.0998,
},
},
)
mock_api.async_air_quality.assert_called_once_with(lat=10.1002, long=20.0998)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Google Air Quality"
assert result["data"] == {
CONF_API_KEY: "new-test-api-key",
CONF_REFERRER: None,
}
assert len(result["subentries"]) == 1
subentry = result["subentries"][0]
assert subentry["subentry_type"] == "location"
assert subentry["title"] == "new-test-name"
assert subentry["data"] == {
CONF_LATITUDE: 10.1002,
CONF_LONGITUDE: 20.0998,
}
assert len(mock_setup_entry.mock_calls) == 2
async def test_subentry_flow(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_api: AsyncMock,
) -> None:
"""Test creating a location subentry."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# After initial setup for 1 subentry, each API is called once
assert mock_api.async_air_quality.call_count == 1
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "location"),
context={"source": "user"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "location"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{
CONF_NAME: "Work",
CONF_LOCATION: {
CONF_LATITUDE: 30.1,
CONF_LONGITUDE: 40.1,
},
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Work"
assert result["data"] == {
CONF_LATITUDE: 30.1,
CONF_LONGITUDE: 40.1,
}
# Initial setup: 1 of each API call
# Subentry flow validation: 1 current conditions call
# Reload with 2 subentries: 2 of each API call
assert mock_api.async_air_quality.call_count == 1 + 1 + 2
entry = hass.config_entries.async_get_entry(mock_config_entry.entry_id)
assert len(entry.subentries) == 2
async def test_subentry_flow_location_already_configured(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_api: AsyncMock,
) -> None:
"""Test user input for a location that already exists."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "location"),
context={"source": "user"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "location"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{
CONF_NAME: "Work",
CONF_LOCATION: {
CONF_LATITUDE: 10.1,
CONF_LONGITUDE: 20.1,
},
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
entry = hass.config_entries.async_get_entry(mock_config_entry.entry_id)
assert len(entry.subentries) == 1
async def test_subentry_flow_entry_not_loaded(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test creating a location subentry when the parent entry is not loaded."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "location"),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "entry_not_loaded"

View File

@@ -0,0 +1,40 @@
"""Tests for Google Air Quality."""
from unittest.mock import AsyncMock
from google_air_quality_api.exceptions import GoogleAirQualityApiError
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_setup(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_api: AsyncMock,
) -> None:
"""Test successful setup and unload."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
async def test_config_not_ready(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_api: AsyncMock,
) -> None:
"""Test for setup failure if an API call fails."""
mock_config_entry.add_to_hass(hass)
mock_api.async_air_quality.side_effect = GoogleAirQualityApiError()
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY

View File

@@ -0,0 +1,33 @@
"""Test the Google Air Quality sensor."""
from unittest.mock import AsyncMock, patch
from syrupy.assertion import SnapshotAssertion
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
async def test_sensor_snapshot(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
mock_api: AsyncMock,
snapshot: SnapshotAssertion,
) -> None:
"""Snapshot test of the sensors."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.LOADED
with patch(
"homeassistant.components.google_air_quality.PLATFORMS",
[Platform.SENSOR],
):
await snapshot_platform(
hass, entity_registry, snapshot, mock_config_entry.entry_id
)