267 lines
9.0 KiB
Python
267 lines
9.0 KiB
Python
"""Support for AirVisual air quality sensors."""
|
|
from logging import getLogger
|
|
|
|
from homeassistant.const import (
|
|
ATTR_LATITUDE,
|
|
ATTR_LONGITUDE,
|
|
ATTR_STATE,
|
|
CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
|
|
CONCENTRATION_PARTS_PER_BILLION,
|
|
CONCENTRATION_PARTS_PER_MILLION,
|
|
CONF_LATITUDE,
|
|
CONF_LONGITUDE,
|
|
CONF_SHOW_ON_MAP,
|
|
CONF_STATE,
|
|
DEVICE_CLASS_BATTERY,
|
|
DEVICE_CLASS_HUMIDITY,
|
|
DEVICE_CLASS_TEMPERATURE,
|
|
TEMP_CELSIUS,
|
|
UNIT_PERCENTAGE,
|
|
)
|
|
from homeassistant.core import callback
|
|
|
|
from . import AirVisualEntity
|
|
from .const import (
|
|
CONF_CITY,
|
|
CONF_COUNTRY,
|
|
CONF_INTEGRATION_TYPE,
|
|
DATA_COORDINATOR,
|
|
DOMAIN,
|
|
INTEGRATION_TYPE_GEOGRAPHY,
|
|
)
|
|
|
|
_LOGGER = getLogger(__name__)
|
|
|
|
ATTR_CITY = "city"
|
|
ATTR_COUNTRY = "country"
|
|
ATTR_POLLUTANT_SYMBOL = "pollutant_symbol"
|
|
ATTR_POLLUTANT_UNIT = "pollutant_unit"
|
|
ATTR_REGION = "region"
|
|
|
|
MASS_PARTS_PER_MILLION = "ppm"
|
|
MASS_PARTS_PER_BILLION = "ppb"
|
|
VOLUME_MICROGRAMS_PER_CUBIC_METER = "µg/m3"
|
|
|
|
SENSOR_KIND_LEVEL = "air_pollution_level"
|
|
SENSOR_KIND_AQI = "air_quality_index"
|
|
SENSOR_KIND_POLLUTANT = "main_pollutant"
|
|
SENSOR_KIND_BATTERY_LEVEL = "battery_level"
|
|
SENSOR_KIND_HUMIDITY = "humidity"
|
|
SENSOR_KIND_TEMPERATURE = "temperature"
|
|
|
|
GEOGRAPHY_SENSORS = [
|
|
(SENSOR_KIND_LEVEL, "Air Pollution Level", "mdi:gauge", None),
|
|
(SENSOR_KIND_AQI, "Air Quality Index", "mdi:chart-line", "AQI"),
|
|
(SENSOR_KIND_POLLUTANT, "Main Pollutant", "mdi:chemical-weapon", None),
|
|
]
|
|
GEOGRAPHY_SENSOR_LOCALES = {"cn": "Chinese", "us": "U.S."}
|
|
|
|
NODE_PRO_SENSORS = [
|
|
(SENSOR_KIND_BATTERY_LEVEL, "Battery", DEVICE_CLASS_BATTERY, UNIT_PERCENTAGE),
|
|
(SENSOR_KIND_HUMIDITY, "Humidity", DEVICE_CLASS_HUMIDITY, UNIT_PERCENTAGE),
|
|
(SENSOR_KIND_TEMPERATURE, "Temperature", DEVICE_CLASS_TEMPERATURE, TEMP_CELSIUS),
|
|
]
|
|
|
|
POLLUTANT_LEVEL_MAPPING = [
|
|
{"label": "Good", "icon": "mdi:emoticon-excited", "minimum": 0, "maximum": 50},
|
|
{"label": "Moderate", "icon": "mdi:emoticon-happy", "minimum": 51, "maximum": 100},
|
|
{
|
|
"label": "Unhealthy for sensitive groups",
|
|
"icon": "mdi:emoticon-neutral",
|
|
"minimum": 101,
|
|
"maximum": 150,
|
|
},
|
|
{"label": "Unhealthy", "icon": "mdi:emoticon-sad", "minimum": 151, "maximum": 200},
|
|
{
|
|
"label": "Very Unhealthy",
|
|
"icon": "mdi:emoticon-dead",
|
|
"minimum": 201,
|
|
"maximum": 300,
|
|
},
|
|
{"label": "Hazardous", "icon": "mdi:biohazard", "minimum": 301, "maximum": 10000},
|
|
]
|
|
|
|
POLLUTANT_MAPPING = {
|
|
"co": {"label": "Carbon Monoxide", "unit": CONCENTRATION_PARTS_PER_MILLION},
|
|
"n2": {"label": "Nitrogen Dioxide", "unit": CONCENTRATION_PARTS_PER_BILLION},
|
|
"o3": {"label": "Ozone", "unit": CONCENTRATION_PARTS_PER_BILLION},
|
|
"p1": {"label": "PM10", "unit": CONCENTRATION_MICROGRAMS_PER_CUBIC_METER},
|
|
"p2": {"label": "PM2.5", "unit": CONCENTRATION_MICROGRAMS_PER_CUBIC_METER},
|
|
"s2": {"label": "Sulfur Dioxide", "unit": CONCENTRATION_PARTS_PER_BILLION},
|
|
}
|
|
|
|
|
|
async def async_setup_entry(hass, config_entry, async_add_entities):
|
|
"""Set up AirVisual sensors based on a config entry."""
|
|
coordinator = hass.data[DOMAIN][DATA_COORDINATOR][config_entry.entry_id]
|
|
|
|
if config_entry.data[CONF_INTEGRATION_TYPE] == INTEGRATION_TYPE_GEOGRAPHY:
|
|
sensors = [
|
|
AirVisualGeographySensor(
|
|
coordinator, config_entry, kind, name, icon, unit, locale,
|
|
)
|
|
for locale in GEOGRAPHY_SENSOR_LOCALES
|
|
for kind, name, icon, unit in GEOGRAPHY_SENSORS
|
|
]
|
|
else:
|
|
sensors = [
|
|
AirVisualNodeProSensor(coordinator, kind, name, device_class, unit)
|
|
for kind, name, device_class, unit in NODE_PRO_SENSORS
|
|
]
|
|
|
|
async_add_entities(sensors, True)
|
|
|
|
|
|
class AirVisualGeographySensor(AirVisualEntity):
|
|
"""Define an AirVisual sensor related to geography data via the Cloud API."""
|
|
|
|
def __init__(self, coordinator, config_entry, kind, name, icon, unit, locale):
|
|
"""Initialize."""
|
|
super().__init__(coordinator)
|
|
|
|
self._attrs.update(
|
|
{
|
|
ATTR_CITY: config_entry.data.get(CONF_CITY),
|
|
ATTR_STATE: config_entry.data.get(CONF_STATE),
|
|
ATTR_COUNTRY: config_entry.data.get(CONF_COUNTRY),
|
|
}
|
|
)
|
|
self._config_entry = config_entry
|
|
self._icon = icon
|
|
self._kind = kind
|
|
self._locale = locale
|
|
self._name = name
|
|
self._state = None
|
|
self._unit = unit
|
|
|
|
@property
|
|
def available(self):
|
|
"""Return True if entity is available."""
|
|
try:
|
|
return self.coordinator.last_update_success and bool(
|
|
self.coordinator.data["current"]["pollution"]
|
|
)
|
|
except KeyError:
|
|
return False
|
|
|
|
@property
|
|
def name(self):
|
|
"""Return the name."""
|
|
return f"{GEOGRAPHY_SENSOR_LOCALES[self._locale]} {self._name}"
|
|
|
|
@property
|
|
def state(self):
|
|
"""Return the state."""
|
|
return self._state
|
|
|
|
@property
|
|
def unique_id(self):
|
|
"""Return a unique, Home Assistant friendly identifier for this entity."""
|
|
return f"{self._config_entry.unique_id}_{self._locale}_{self._kind}"
|
|
|
|
@callback
|
|
def update_from_latest_data(self):
|
|
"""Update the entity from the latest data."""
|
|
try:
|
|
data = self.coordinator.data["current"]["pollution"]
|
|
except KeyError:
|
|
return
|
|
|
|
if self._kind == SENSOR_KIND_LEVEL:
|
|
aqi = data[f"aqi{self._locale}"]
|
|
[level] = [
|
|
i
|
|
for i in POLLUTANT_LEVEL_MAPPING
|
|
if i["minimum"] <= aqi <= i["maximum"]
|
|
]
|
|
self._state = level["label"]
|
|
self._icon = level["icon"]
|
|
elif self._kind == SENSOR_KIND_AQI:
|
|
self._state = data[f"aqi{self._locale}"]
|
|
elif self._kind == SENSOR_KIND_POLLUTANT:
|
|
symbol = data[f"main{self._locale}"]
|
|
self._state = POLLUTANT_MAPPING[symbol]["label"]
|
|
self._attrs.update(
|
|
{
|
|
ATTR_POLLUTANT_SYMBOL: symbol,
|
|
ATTR_POLLUTANT_UNIT: POLLUTANT_MAPPING[symbol]["unit"],
|
|
}
|
|
)
|
|
|
|
if CONF_LATITUDE in self._config_entry.data:
|
|
if self._config_entry.options[CONF_SHOW_ON_MAP]:
|
|
self._attrs[ATTR_LATITUDE] = self._config_entry.data[CONF_LATITUDE]
|
|
self._attrs[ATTR_LONGITUDE] = self._config_entry.data[CONF_LONGITUDE]
|
|
self._attrs.pop("lati", None)
|
|
self._attrs.pop("long", None)
|
|
else:
|
|
self._attrs["lati"] = self._config_entry.data[CONF_LATITUDE]
|
|
self._attrs["long"] = self._config_entry.data[CONF_LONGITUDE]
|
|
self._attrs.pop(ATTR_LATITUDE, None)
|
|
self._attrs.pop(ATTR_LONGITUDE, None)
|
|
|
|
|
|
class AirVisualNodeProSensor(AirVisualEntity):
|
|
"""Define an AirVisual sensor related to a Node/Pro unit."""
|
|
|
|
def __init__(self, coordinator, kind, name, device_class, unit):
|
|
"""Initialize."""
|
|
super().__init__(coordinator)
|
|
|
|
self._device_class = device_class
|
|
self._kind = kind
|
|
self._name = name
|
|
self._state = None
|
|
self._unit = unit
|
|
|
|
@property
|
|
def device_class(self):
|
|
"""Return the device class."""
|
|
return self._device_class
|
|
|
|
@property
|
|
def device_info(self):
|
|
"""Return device registry information for this entity."""
|
|
return {
|
|
"identifiers": {
|
|
(DOMAIN, self.coordinator.data["current"]["serial_number"])
|
|
},
|
|
"name": self.coordinator.data["current"]["settings"]["node_name"],
|
|
"manufacturer": "AirVisual",
|
|
"model": f'{self.coordinator.data["current"]["status"]["model"]}',
|
|
"sw_version": (
|
|
f'Version {self.coordinator.data["current"]["status"]["system_version"]}'
|
|
f'{self.coordinator.data["current"]["status"]["app_version"]}'
|
|
),
|
|
}
|
|
|
|
@property
|
|
def name(self):
|
|
"""Return the name."""
|
|
node_name = self.coordinator.data["current"]["settings"]["node_name"]
|
|
return f"{node_name} Node/Pro: {self._name}"
|
|
|
|
@property
|
|
def state(self):
|
|
"""Return the state."""
|
|
return self._state
|
|
|
|
@property
|
|
def unique_id(self):
|
|
"""Return a unique, Home Assistant friendly identifier for this entity."""
|
|
return f"{self.coordinator.data['current']['serial_number']}_{self._kind}"
|
|
|
|
@callback
|
|
def update_from_latest_data(self):
|
|
"""Update the entity from the latest data."""
|
|
if self._kind == SENSOR_KIND_BATTERY_LEVEL:
|
|
self._state = self.coordinator.data["current"]["status"]["battery"]
|
|
elif self._kind == SENSOR_KIND_HUMIDITY:
|
|
self._state = self.coordinator.data["current"]["measurements"].get(
|
|
"humidity"
|
|
)
|
|
elif self._kind == SENSOR_KIND_TEMPERATURE:
|
|
self._state = self.coordinator.data["current"]["measurements"].get(
|
|
"temperature_C"
|
|
)
|