238 lines
7.9 KiB
Python
238 lines
7.9 KiB
Python
"""Support for Awair sensors."""
|
|
|
|
from typing import Callable, List, Optional
|
|
|
|
from python_awair.devices import AwairDevice
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.components.awair import AwairDataUpdateCoordinator, AwairResult
|
|
from homeassistant.components.sensor import PLATFORM_SCHEMA
|
|
from homeassistant.config_entries import SOURCE_IMPORT
|
|
from homeassistant.const import ATTR_ATTRIBUTION, ATTR_DEVICE_CLASS, CONF_ACCESS_TOKEN
|
|
from homeassistant.helpers import device_registry as dr
|
|
import homeassistant.helpers.config_validation as cv
|
|
from homeassistant.helpers.entity import Entity
|
|
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
|
|
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
|
|
|
from .const import (
|
|
API_DUST,
|
|
API_PM25,
|
|
API_SCORE,
|
|
API_TEMP,
|
|
API_VOC,
|
|
ATTR_ICON,
|
|
ATTR_LABEL,
|
|
ATTR_UNIQUE_ID,
|
|
ATTR_UNIT,
|
|
ATTRIBUTION,
|
|
DOMAIN,
|
|
DUST_ALIASES,
|
|
LOGGER,
|
|
SENSOR_TYPES,
|
|
)
|
|
|
|
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
|
{vol.Required(CONF_ACCESS_TOKEN): cv.string},
|
|
extra=vol.ALLOW_EXTRA,
|
|
)
|
|
|
|
|
|
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
|
|
"""Import Awair configuration from YAML."""
|
|
LOGGER.warning(
|
|
"Loading Awair via platform setup is deprecated. Please remove it from your configuration."
|
|
)
|
|
hass.async_create_task(
|
|
hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": SOURCE_IMPORT},
|
|
data=config,
|
|
)
|
|
)
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistantType,
|
|
config_entry: ConfigType,
|
|
async_add_entities: Callable[[List[Entity], bool], None],
|
|
):
|
|
"""Set up Awair sensor entity based on a config entry."""
|
|
coordinator = hass.data[DOMAIN][config_entry.entry_id]
|
|
sensors = []
|
|
|
|
data: List[AwairResult] = coordinator.data.values()
|
|
for result in data:
|
|
if result.air_data:
|
|
sensors.append(AwairSensor(API_SCORE, result.device, coordinator))
|
|
device_sensors = result.air_data.sensors.keys()
|
|
for sensor in device_sensors:
|
|
if sensor in SENSOR_TYPES:
|
|
sensors.append(AwairSensor(sensor, result.device, coordinator))
|
|
|
|
# The "DUST" sensor for Awair is a combo pm2.5/pm10 sensor only
|
|
# present on first-gen devices in lieu of separate pm2.5/pm10 sensors.
|
|
# We handle that by creating fake pm2.5/pm10 sensors that will always
|
|
# report identical values, and we let users decide how they want to use
|
|
# that data - because we can't really tell what kind of particles the
|
|
# "DUST" sensor actually detected. However, it's still useful data.
|
|
if API_DUST in device_sensors:
|
|
for alias_kind in DUST_ALIASES:
|
|
sensors.append(AwairSensor(alias_kind, result.device, coordinator))
|
|
|
|
async_add_entities(sensors)
|
|
|
|
|
|
class AwairSensor(CoordinatorEntity):
|
|
"""Defines an Awair sensor entity."""
|
|
|
|
def __init__(
|
|
self,
|
|
kind: str,
|
|
device: AwairDevice,
|
|
coordinator: AwairDataUpdateCoordinator,
|
|
) -> None:
|
|
"""Set up an individual AwairSensor."""
|
|
super().__init__(coordinator)
|
|
self._kind = kind
|
|
self._device = device
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
"""Return the name of the sensor."""
|
|
name = SENSOR_TYPES[self._kind][ATTR_LABEL]
|
|
if self._device.name:
|
|
name = f"{self._device.name} {name}"
|
|
|
|
return name
|
|
|
|
@property
|
|
def unique_id(self) -> str:
|
|
"""Return the uuid as the unique_id."""
|
|
unique_id_tag = SENSOR_TYPES[self._kind][ATTR_UNIQUE_ID]
|
|
|
|
# This integration used to create a sensor that was labelled as a "PM2.5"
|
|
# sensor for first-gen Awair devices, but its unique_id reflected the truth:
|
|
# under the hood, it was a "DUST" sensor. So we preserve that specific unique_id
|
|
# for users with first-gen devices that are upgrading.
|
|
if self._kind == API_PM25 and API_DUST in self._air_data.sensors:
|
|
unique_id_tag = "DUST"
|
|
|
|
return f"{self._device.uuid}_{unique_id_tag}"
|
|
|
|
@property
|
|
def available(self) -> bool:
|
|
"""Determine if the sensor is available based on API results."""
|
|
# If the last update was successful...
|
|
if self.coordinator.last_update_success and self._air_data:
|
|
# and the results included our sensor type...
|
|
if self._kind in self._air_data.sensors:
|
|
# then we are available.
|
|
return True
|
|
|
|
# or, we're a dust alias
|
|
if self._kind in DUST_ALIASES and API_DUST in self._air_data.sensors:
|
|
return True
|
|
|
|
# or we are API_SCORE
|
|
if self._kind == API_SCORE:
|
|
# then we are available.
|
|
return True
|
|
|
|
# Otherwise, we are not.
|
|
return False
|
|
|
|
@property
|
|
def state(self) -> float:
|
|
"""Return the state, rounding off to reasonable values."""
|
|
state: float
|
|
|
|
# Special-case for "SCORE", which we treat as the AQI
|
|
if self._kind == API_SCORE:
|
|
state = self._air_data.score
|
|
elif self._kind in DUST_ALIASES and API_DUST in self._air_data.sensors:
|
|
state = self._air_data.sensors.dust
|
|
else:
|
|
state = self._air_data.sensors[self._kind]
|
|
|
|
if self._kind == API_VOC or self._kind == API_SCORE:
|
|
return round(state)
|
|
|
|
if self._kind == API_TEMP:
|
|
return round(state, 1)
|
|
|
|
return round(state, 2)
|
|
|
|
@property
|
|
def icon(self) -> str:
|
|
"""Return the icon."""
|
|
return SENSOR_TYPES[self._kind][ATTR_ICON]
|
|
|
|
@property
|
|
def device_class(self) -> str:
|
|
"""Return the device_class."""
|
|
return SENSOR_TYPES[self._kind][ATTR_DEVICE_CLASS]
|
|
|
|
@property
|
|
def unit_of_measurement(self) -> str:
|
|
"""Return the unit the value is expressed in."""
|
|
return SENSOR_TYPES[self._kind][ATTR_UNIT]
|
|
|
|
@property
|
|
def device_state_attributes(self) -> dict:
|
|
"""Return the Awair Index alongside state attributes.
|
|
|
|
The Awair Index is a subjective score ranging from 0-4 (inclusive) that
|
|
is is used by the Awair app when displaying the relative "safety" of a
|
|
given measurement. Each value is mapped to a color indicating the safety:
|
|
|
|
0: green
|
|
1: yellow
|
|
2: light-orange
|
|
3: orange
|
|
4: red
|
|
|
|
The API indicates that both positive and negative values may be returned,
|
|
but the negative values are mapped to identical colors as the positive values.
|
|
Knowing that, we just return the absolute value of a given index so that
|
|
users don't have to handle positive/negative values that ultimately "mean"
|
|
the same thing.
|
|
|
|
https://docs.developer.getawair.com/?version=latest#awair-score-and-index
|
|
"""
|
|
attrs = {ATTR_ATTRIBUTION: ATTRIBUTION}
|
|
if self._kind in self._air_data.indices:
|
|
attrs["awair_index"] = abs(self._air_data.indices[self._kind])
|
|
elif self._kind in DUST_ALIASES and API_DUST in self._air_data.indices:
|
|
attrs["awair_index"] = abs(self._air_data.indices.dust)
|
|
|
|
return attrs
|
|
|
|
@property
|
|
def device_info(self) -> dict:
|
|
"""Device information."""
|
|
info = {
|
|
"identifiers": {(DOMAIN, self._device.uuid)},
|
|
"manufacturer": "Awair",
|
|
"model": self._device.model,
|
|
}
|
|
|
|
if self._device.name:
|
|
info["name"] = self._device.name
|
|
|
|
if self._device.mac_address:
|
|
info["connections"] = {
|
|
(dr.CONNECTION_NETWORK_MAC, self._device.mac_address)
|
|
}
|
|
|
|
return info
|
|
|
|
@property
|
|
def _air_data(self) -> Optional[AwairResult]:
|
|
"""Return the latest data for our device, or None."""
|
|
result: Optional[AwairResult] = self.coordinator.data.get(self._device.uuid)
|
|
if result:
|
|
return result.air_data
|
|
|
|
return None
|