feat: Add Amazon Indoor Air Quality Monitor (#1803)

custom_components/alexa_media/__init__.py

@@ -403,6 +403,10 @@ async def async_update_data() -> Optional[AlexaEntityData]:
             if temp and temp.enabled:
                 entities_to_monitor.add(temp.alexa_entity_id)
 
+            temp = sensor.get("Air_Quality")
+            if temp and temp.enabled:
+                entities_to_monitor.add(temp.alexa_entity_id)
+
         for light in hass.data[DATA_ALEXAMEDIA]["accounts"][email]["entities"]["light"]:
             if light.enabled:
                 entities_to_monitor.add(light.alexa_entity_id)

custom_components/alexa_media/alexa_entity.py

@@ -24,9 +24,9 @@ def has_capability(
     """Determine if an appliance from the Alexa network details offers a particular interface with enough support that is worth adding to Home Assistant.
 
     Args:
-        appliance(Dict[Text, Any]): An appliance from a call to AlexaAPI.get_network_details
-        interface_name(Text): One of the interfaces documented by the Alexa Smart Home Skills API
-        property_name(Text): The property that matches the interface name.
+        appliance(dict[str, Any]): An appliance from a call to AlexaAPI.get_network_details
+        interface_name(str): One of the interfaces documented by the Alexa Smart Home Skills API
+        property_name(str): The property that matches the interface name.
 
     """
     for cap in appliance["capabilities"]:
@@ -90,6 +90,17 @@ def is_temperature_sensor(appliance: dict[str, Any]) -> bool:
     )
 
 
+# Checks if air quality sensor
+def is_air_quality_sensor(appliance: dict[str, Any]) -> bool:
+    """Is the given appliance the Air Quality Sensor."""
+    return (
+        appliance["friendlyDescription"] == "Amazon Indoor Air Quality Monitor"
+        and "AIR_QUALITY_MONITOR" in appliance.get("applianceTypes", [])
+        and has_capability(appliance, "Alexa.TemperatureSensor", "temperature")
+        and has_capability(appliance, "Alexa.RangeController", "rangeValue")
+    )
+
+
 def is_light(appliance: dict[str, Any]) -> bool:
     """Is the given appliance a light controlled locally by an Echo."""
     return (
@@ -150,6 +161,12 @@ class AlexaTemperatureEntity(AlexaEntity):
     device_serial: str
 
 
+class AlexaAirQualityEntity(AlexaEntity):
+    """Class for AlexaAirQualityEntity."""
+
+    device_serial: str
+
+
 class AlexaBinaryEntity(AlexaEntity):
     """Class for AlexaBinaryEntity."""
 
@@ -162,16 +179,20 @@ class AlexaEntities(TypedDict):
     light: list[AlexaLightEntity]
     guard: list[AlexaEntity]
     temperature: list[AlexaTemperatureEntity]
+    air_quality: list[AlexaAirQualityEntity]
     binary_sensor: list[AlexaBinaryEntity]
 
 
 def parse_alexa_entities(network_details: Optional[dict[str, Any]]) -> AlexaEntities:
+    # pylint: disable=too-many-locals
     """Turn the network details into a list of useful entities with the important details extracted."""
     lights = []
     guards = []
     temperature_sensors = []
+    air_quality_sensors = []
     contact_sensors = []
     location_details = network_details["locationDetails"]["locationDetails"]
+    # pylint: disable=too-many-nested-blocks
     for location in location_details.values():
         amazon_bridge_details = location["amazonBridgeDetails"]["amazonBridgeDetails"]
         for bridge in amazon_bridge_details.values():
@@ -191,6 +212,44 @@ def parse_alexa_entities(network_details: Optional[dict[str, Any]]) -> AlexaEnti
                         serial if serial else appliance["entityId"]
                     )
                     temperature_sensors.append(processed_appliance)
+                # Code for Amazon Smart Air Quality Monitor
+                elif is_air_quality_sensor(appliance):
+                    serial = get_device_serial(appliance)
+                    processed_appliance["device_serial"] = (
+                        serial if serial else appliance["entityId"]
+                    )
+                    # create array of air quality sensors. We must store the instance id against
+                    # the assetId so we know which sensors are which.
+                    sensors = []
+                    if (
+                        appliance["friendlyDescription"]
+                        == "Amazon Indoor Air Quality Monitor"
+                    ):
+                        for cap in appliance["capabilities"]:
+                            instance = cap.get("instance")
+                            if instance:
+                                friendlyName = cap["resources"].get("friendlyNames")
+                                for entry in friendlyName:
+                                    assetId = entry["value"].get("assetId")
+                                    if assetId and assetId.startswith(
+                                        "Alexa.AirQuality"
+                                    ):
+                                        unit = cap["configuration"]["unitOfMeasure"]
+                                        sensor = {
+                                            "sensorType": assetId,
+                                            "instance": instance,
+                                            "unit": unit,
+                                        }
+                                        sensors.append(sensor)
+                                        _LOGGER.debug(
+                                            "AIAQM sensor detected %s", sensor
+                                        )
+                    processed_appliance["sensors"] = sensors
+
+                    # Add as both temperature and air quality sensor
+                    temperature_sensors.append(processed_appliance)
+                    air_quality_sensors.append(processed_appliance)
+
                 elif is_light(appliance):
                     processed_appliance["brightness"] = has_capability(
                         appliance, "Alexa.BrightnessController", "brightness"
@@ -216,6 +275,7 @@ def parse_alexa_entities(network_details: Optional[dict[str, Any]]) -> AlexaEnti
         "light": lights,
         "guard": guards,
         "temperature": temperature_sensors,
+        "air_quality": air_quality_sensors,
         "binary_sensor": contact_sensors,
     }
 
@@ -261,6 +321,20 @@ def parse_temperature_from_coordinator(
     return value.get("value") if value and "value" in value else None
 
 
+def parse_air_quality_from_coordinator(
+    coordinator: DataUpdateCoordinator, entity_id: str, instance_id: str
+) -> Optional[str]:
+    """Get the air quality of an entity from the coordinator data."""
+    value = parse_value_from_coordinator(
+        coordinator,
+        entity_id,
+        "Alexa.RangeController",
+        "rangeValue",
+        instance=instance_id,
+    )
+    return value
+
+
 def parse_brightness_from_coordinator(
     coordinator: DataUpdateCoordinator, entity_id: str, since: datetime
 ) -> Optional[int]:
@@ -330,13 +404,15 @@ def parse_value_from_coordinator(
     namespace: str,
     name: str,
     since: Optional[datetime] = None,
+    instance: str = None,
 ) -> Any:
     """Parse out values from coordinator for Alexa Entities."""
     if coordinator.data and entity_id in coordinator.data:
         for cap_state in coordinator.data[entity_id]:
             if (
                 cap_state.get("namespace") == namespace
                 and cap_state.get("name") == name
+                and (cap_state.get("instance") == instance or instance is None)
             ):
                 if is_cap_state_still_acceptable(cap_state, since):
                     return cap_state.get("value")

custom_components/alexa_media/const.py

@@ -8,6 +8,12 @@
 """
 from datetime import timedelta
 
+from homeassistant.const import (
+    CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
+    CONCENTRATION_PARTS_PER_MILLION,
+    PERCENTAGE,
+)
+
 __version__ = "4.4.0"
 PROJECT_URL = "https://github.com/custom-components/alexa_media_player/"
 ISSUE_URL = f"{PROJECT_URL}issues"
@@ -112,3 +118,16 @@
 AUTH_CALLBACK_NAME = "auth:alexamedia:callback"
 AUTH_PROXY_PATH = "/auth/alexamedia/proxy"
 AUTH_PROXY_NAME = "auth:alexamedia:proxy"
+
+ALEXA_UNIT_CONVERSION = {
+    "Alexa.Unit.Percent": PERCENTAGE,
+    "Alexa.Unit.PartsPerMillion": CONCENTRATION_PARTS_PER_MILLION,
+    "Alexa.Unit.Density.MicroGramsPerCubicMeter": CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
+}
+
+ALEXA_ICON_CONVERSION = {
+    "Alexa.AirQuality.CarbonMonoxide": "mdi:molecule-co",
+    "Alexa.AirQuality.Humidity": "mdi:water-percent",
+    "Alexa.AirQuality.IndoorAirQuality": "mdi:numeric",
+}
+ALEXA_ICON_DEFAULT = "mdi:molecule"

custom_components/alexa_media/sensor.py

@@ -36,8 +36,14 @@
     hide_email,
     hide_serial,
 )
-from .alexa_entity import parse_temperature_from_coordinator
+from .alexa_entity import (
+    parse_air_quality_from_coordinator,
+    parse_temperature_from_coordinator,
+)
 from .const import (
+    ALEXA_ICON_CONVERSION,
+    ALEXA_ICON_DEFAULT,
+    ALEXA_UNIT_CONVERSION,
     CONF_EXTENDED_ENTITY_DISCOVERY,
     RECURRING_DAY,
     RECURRING_PATTERN,
@@ -127,9 +133,19 @@ async def async_setup_platform(hass, config, add_devices_callback, discovery_inf
             account_dict, temperature_entities
         )
 
+    # AIAQM Sensors
+    air_quality_sensors = []
+    air_quality_entities = account_dict.get("devices", {}).get("air_quality", [])
+    if air_quality_entities and account_dict["options"].get(
+        CONF_EXTENDED_ENTITY_DISCOVERY
+    ):
+        air_quality_sensors = await create_air_quality_sensors(
+            account_dict, air_quality_entities
+        )
+
     return await add_devices(
         hide_email(account),
-        devices + temperature_sensors,
+        devices + temperature_sensors + air_quality_sensors,
         add_devices_callback,
         include_filter,
         exclude_filter,
@@ -174,6 +190,42 @@ async def create_temperature_sensors(account_dict, temperature_entities):
     return devices
 
 
+async def create_air_quality_sensors(account_dict, air_quality_entities):
+    devices = []
+    coordinator = account_dict["coordinator"]
+
+    for temp in air_quality_entities:
+        _LOGGER.debug(
+            "Creating entity %s for a air quality sensor with name %s",
+            temp["id"],
+            temp["name"],
+        )
+        # Each AIAQM has 5 different sensors.
+        for subsensor in temp["sensors"]:
+            sensor_type = subsensor["sensorType"]
+            instance = subsensor["instance"]
+            unit = subsensor["unit"]
+            serial = temp["device_serial"]
+            device_info = lookup_device_info(account_dict, serial)
+            sensor = AirQualitySensor(
+                coordinator,
+                temp["id"],
+                temp["name"],
+                device_info,
+                sensor_type,
+                instance,
+                unit,
+            )
+            _LOGGER.debug("Create air quality sensors %s", sensor)
+            account_dict["entities"]["sensor"].setdefault(serial, {})
+            account_dict["entities"]["sensor"][serial].setdefault(sensor_type, {})
+            account_dict["entities"]["sensor"][serial][sensor_type][
+                "Air_Quality"
+            ] = sensor
+            devices.append(sensor)
+    return devices
+
+
 def lookup_device_info(account_dict, device_serial):
     """Get the device to use for a given Echo based on a given device serial id.
 
@@ -240,6 +292,75 @@ def unique_id(self):
         return self.alexa_entity_id + "_temperature"
 
 
+class AirQualitySensor(SensorEntity, CoordinatorEntity):
+    """A air quality sensor reported by an Amazon indoor air quality monitor."""
+
+    def __init__(
+        self,
+        coordinator,
+        entity_id,
+        name,
+        media_player_device_id,
+        sensor_name,
+        instance,
+        unit,
+    ):
+        super().__init__(coordinator)
+        self.alexa_entity_id = entity_id
+        self._sensor_name = sensor_name
+        # tidy up name
+        self._sensor_name = self._sensor_name.replace("Alexa.AirQuality.", "")
+        self._sensor_name = "".join(
+            " " + char if char.isupper() else char.strip() for char in self._sensor_name
+        ).strip()
+        self._name = name + " " + self._sensor_name
+        self._media_player_device_id = media_player_device_id
+
+        self._instance = instance
+
+        self._unit = ALEXA_UNIT_CONVERSION.get(unit)
+        self._icon = ALEXA_ICON_CONVERSION.get(sensor_name, ALEXA_ICON_DEFAULT)
+
+    @property
+    def name(self):
+        return self._name
+
+    @property
+    def device_info(self):
+        """Return the device_info of the device."""
+        if self._media_player_device_id:
+            return {
+                "identifiers": {self._media_player_device_id},
+                "via_device": self._media_player_device_id,
+            }
+        return None
+
+    @property
+    def native_unit_of_measurement(self):
+        return self._unit
+
+    @property
+    def native_value(self):
+        return parse_air_quality_from_coordinator(
+            self.coordinator, self.alexa_entity_id, self._instance
+        )
+
+    @property
+    def device_class(self):
+        return self._sensor_name
+
+    @property
+    def unique_id(self):
+        # This includes "_temperature" because the Alexa entityId is for a physical device
+        # A single physical device could have multiple HA entities
+        return self.alexa_entity_id + " " + self._sensor_name
+
+    @property
+    def icon(self):
+        """Return the icon of the sensor."""
+        return self._icon
+
+
 class AlexaMediaNotificationSensor(SensorEntity):
     """Representation of Alexa Media sensors."""