self-hosted/home-assistant-core
self-hosted/home-assistant-core
1
0
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-10-08 14:48:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add sensor platform to Airly integration (#27717)

Browse source 
* Add sesnor.py file

* Move AirlyData to __init__

* Cleaning

* Update .coveragerc file

* Sort consts

* Sort imports

* Remove icons from sensors with device_class
This commit is contained in:
Maciej Bieniek 2019-10-16 12:06:52 +02:00 committed by Pascal Vizeli
parent cc93dd4928
commit ec78821161
5 changed files with 283 additions and 86 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.coveragerc
View file

@ -29,6 +29,7 @@ omit =
homeassistant/components/aftership/sensor.py
homeassistant/components/airly/__init__.py
homeassistant/components/airly/air_quality.py
homeassistant/components/airly/sensor.py
homeassistant/components/airly/const.py
homeassistant/components/airvisual/sensor.py
homeassistant/components/aladdin_connect/cover.py

93
homeassistant/components/airly/__init__.py
View file

@ -1,5 +1,31 @@
"""The Airly component."""
import asyncio
import logging
from datetime import timedelta
import async_timeout
from aiohttp.client_exceptions import ClientConnectorError
from airly import Airly
from airly.exceptions import AirlyError
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import Config, HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.util import Throttle
from .const import (
ATTR_API_ADVICE,
ATTR_API_CAQI,
ATTR_API_CAQI_DESCRIPTION,
ATTR_API_CAQI_LEVEL,
DATA_CLIENT,
DOMAIN,
NO_AIRLY_SENSORS,
)
_LOGGER = logging.getLogger(__name__)
DEFAULT_SCAN_INTERVAL = timedelta(minutes=10)
async def async_setup(hass: HomeAssistant, config: Config) -> bool:
@ -9,13 +35,80 @@ async def async_setup(hass: HomeAssistant, config: Config) -> bool:
async def async_setup_entry(hass, config_entry):
"""Set up Airly as config entry."""
api_key = config_entry.data[CONF_API_KEY]
latitude = config_entry.data[CONF_LATITUDE]
longitude = config_entry.data[CONF_LONGITUDE]
websession = async_get_clientsession(hass)
airly = AirlyData(websession, api_key, latitude, longitude)
await airly.async_update()
hass.data[DOMAIN] = {}
hass.data[DOMAIN][DATA_CLIENT] = {}
hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id] = airly
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, "air_quality")
)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, "sensor")
)
return True
async def async_unload_entry(hass, config_entry):
"""Unload a config entry."""
hass.data[DOMAIN][DATA_CLIENT].pop(config_entry.entry_id)
await hass.config_entries.async_forward_entry_unload(config_entry, "air_quality")
await hass.config_entries.async_forward_entry_unload(config_entry, "sensor")
return True
class AirlyData:
"""Define an object to hold Airly data."""
def __init__(self, session, api_key, latitude, longitude):
"""Initialize."""
self.latitude = latitude
self.longitude = longitude
self.airly = Airly(api_key, session)
self.data = {}
@Throttle(DEFAULT_SCAN_INTERVAL)
async def async_update(self):
"""Update Airly data."""
try:
with async_timeout.timeout(10):
measurements = self.airly.create_measurements_session_point(
self.latitude, self.longitude
)
await measurements.update()
values = measurements.current["values"]
index = measurements.current["indexes"][0]
standards = measurements.current["standards"]
if index["description"] == NO_AIRLY_SENSORS:
_LOGGER.error("Can't retrieve data: no Airly sensors in this area")
return
for value in values:
self.data[value["name"]] = value["value"]
for standard in standards:
self.data[f"{standard['pollutant']}_LIMIT"] = standard["limit"]
self.data[f"{standard['pollutant']}_PERCENT"] = standard["percent"]
self.data[ATTR_API_CAQI] = index["value"]
self.data[ATTR_API_CAQI_LEVEL] = index["level"].lower().replace("_", " ")
self.data[ATTR_API_CAQI_DESCRIPTION] = index["description"]
self.data[ATTR_API_ADVICE] = index["advice"]
_LOGGER.debug("Data retrieved from Airly")
except (
ValueError,
AirlyError,
asyncio.TimeoutError,
ClientConnectorError,
) as error:
_LOGGER.error(error)
self.data = {}

106
homeassistant/components/airly/air_quality.py
View file

@ -1,40 +1,29 @@
"""Support for the Airly service."""
import asyncio
import logging
from datetime import timedelta
import async_timeout
from aiohttp.client_exceptions import ClientConnectorError
from airly import Airly
from airly.exceptions import AirlyError
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME
"""Support for the Airly air_quality service."""
from homeassistant.components.air_quality import (
AirQualityEntity,
ATTR_AQI,
ATTR_PM_10,
ATTR_PM_2_5,
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.util import Throttle
from homeassistant.const import CONF_NAME
from .const import NO_AIRLY_SENSORS
_LOGGER = logging.getLogger(__name__)
from .const import (
ATTR_API_ADVICE,
ATTR_API_CAQI,
ATTR_API_CAQI_DESCRIPTION,
ATTR_API_CAQI_LEVEL,
ATTR_API_PM10,
ATTR_API_PM10_LIMIT,
ATTR_API_PM10_PERCENT,
ATTR_API_PM25,
ATTR_API_PM25_LIMIT,
ATTR_API_PM25_PERCENT,
DATA_CLIENT,
DOMAIN,
)
ATTRIBUTION = "Data provided by Airly"
ATTR_API_ADVICE = "ADVICE"
ATTR_API_CAQI = "CAQI"
ATTR_API_CAQI_DESCRIPTION = "DESCRIPTION"
ATTR_API_CAQI_LEVEL = "LEVEL"
ATTR_API_PM10 = "PM10"
ATTR_API_PM10_LIMIT = "PM10_LIMIT"
ATTR_API_PM10_PERCENT = "PM10_PERCENT"
ATTR_API_PM25 = "PM25"
ATTR_API_PM25_LIMIT = "PM25_LIMIT"
ATTR_API_PM25_PERCENT = "PM25_PERCENT"
LABEL_ADVICE = "advice"
LABEL_AQI_LEVEL = f"{ATTR_AQI}_level"
LABEL_PM_2_5_LIMIT = f"{ATTR_PM_2_5}_limit"
@ -42,19 +31,12 @@ LABEL_PM_2_5_PERCENT = f"{ATTR_PM_2_5}_percent_of_limit"
LABEL_PM_10_LIMIT = f"{ATTR_PM_10}_limit"
LABEL_PM_10_PERCENT = f"{ATTR_PM_10}_percent_of_limit"
DEFAULT_SCAN_INTERVAL = timedelta(minutes=10)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Add a Airly entities from a config_entry."""
api_key = config_entry.data[CONF_API_KEY]
"""Set up Airly air_quality entity based on a config entry."""
name = config_entry.data[CONF_NAME]
latitude = config_entry.data[CONF_LATITUDE]
longitude = config_entry.data[CONF_LONGITUDE]
websession = async_get_clientsession(hass)
data = AirlyData(websession, api_key, latitude, longitude)
data = hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id]
async_add_entities([AirlyAirQuality(data, name)], True)
@ -72,7 +54,7 @@ def round_state(func):
class AirlyAirQuality(AirQualityEntity):
"""Define an Airly air_quality."""
"""Define an Airly air quality."""
def __init__(self, airly, name):
"""Initialize."""
@ -145,7 +127,7 @@ class AirlyAirQuality(AirQualityEntity):
return self._attrs
async def async_update(self):
"""Get the data from Airly."""
"""Update the entity."""
await self.airly.async_update()
if self.airly.data:
@ -154,51 +136,3 @@ class AirlyAirQuality(AirQualityEntity):
self._pm_10 = self.data[ATTR_API_PM10]
self._pm_2_5 = self.data[ATTR_API_PM25]
self._aqi = self.data[ATTR_API_CAQI]
class AirlyData:
"""Define an object to hold sensor data."""
def __init__(self, session, api_key, latitude, longitude):
"""Initialize."""
self.latitude = latitude
self.longitude = longitude
self.airly = Airly(api_key, session)
self.data = {}
@Throttle(DEFAULT_SCAN_INTERVAL)
async def async_update(self):
"""Update Airly data."""
try:
with async_timeout.timeout(10):
measurements = self.airly.create_measurements_session_point(
self.latitude, self.longitude
)
await measurements.update()
values = measurements.current["values"]
index = measurements.current["indexes"][0]
standards = measurements.current["standards"]
if index["description"] == NO_AIRLY_SENSORS:
_LOGGER.error("Can't retrieve data: no Airly sensors in this area")
return
for value in values:
self.data[value["name"]] = value["value"]
for standard in standards:
self.data[f"{standard['pollutant']}_LIMIT"] = standard["limit"]
self.data[f"{standard['pollutant']}_PERCENT"] = standard["percent"]
self.data[ATTR_API_CAQI] = index["value"]
self.data[ATTR_API_CAQI_LEVEL] = index["level"].lower().replace("_", " ")
self.data[ATTR_API_CAQI_DESCRIPTION] = index["description"]
self.data[ATTR_API_ADVICE] = index["advice"]
_LOGGER.debug("Data retrieved from Airly")
except (
ValueError,
AirlyError,
asyncio.TimeoutError,
ClientConnectorError,
) as error:
_LOGGER.error(error)
self.data = {}

15
homeassistant/components/airly/const.py
View file

@ -1,4 +1,19 @@
"""Constants for Airly integration."""
ATTR_API_ADVICE = "ADVICE"
ATTR_API_CAQI = "CAQI"
ATTR_API_CAQI_DESCRIPTION = "DESCRIPTION"
ATTR_API_CAQI_LEVEL = "LEVEL"
ATTR_API_HUMIDITY = "HUMIDITY"
ATTR_API_PM1 = "PM1"
ATTR_API_PM10 = "PM10"
ATTR_API_PM10_LIMIT = "PM10_LIMIT"
ATTR_API_PM10_PERCENT = "PM10_PERCENT"
ATTR_API_PM25 = "PM25"
ATTR_API_PM25_LIMIT = "PM25_LIMIT"
ATTR_API_PM25_PERCENT = "PM25_PERCENT"
ATTR_API_PRESSURE = "PRESSURE"
ATTR_API_TEMPERATURE = "TEMPERATURE"
DATA_CLIENT = "client"
DEFAULT_NAME = "Airly"
DOMAIN = "airly"
NO_AIRLY_SENSORS = "There are no Airly sensors in this area yet."

154
homeassistant/components/airly/sensor.py Normal file
View file

@ -0,0 +1,154 @@
"""Support for the Airly sensor service."""
from homeassistant.const import (
ATTR_DEVICE_CLASS,
CONF_NAME,
DEVICE_CLASS_HUMIDITY,
DEVICE_CLASS_PRESSURE,
DEVICE_CLASS_TEMPERATURE,
PRESSURE_HPA,
TEMP_CELSIUS,
)
from homeassistant.helpers.entity import Entity
from .const import (
ATTR_API_HUMIDITY,
ATTR_API_PM1,
ATTR_API_PRESSURE,
ATTR_API_TEMPERATURE,
DATA_CLIENT,
DOMAIN,
)
ATTRIBUTION = "Data provided by Airly"
ATTR_ICON = "icon"
ATTR_LABEL = "label"
ATTR_UNIT = "unit"
HUMI_PERCENT = "%"
VOLUME_MICROGRAMS_PER_CUBIC_METER = "µg/m³"
SENSOR_TYPES = {
ATTR_API_PM1: {
ATTR_DEVICE_CLASS: None,
ATTR_ICON: "mdi:blur",
ATTR_LABEL: ATTR_API_PM1,
ATTR_UNIT: VOLUME_MICROGRAMS_PER_CUBIC_METER,
},
ATTR_API_HUMIDITY: {
ATTR_DEVICE_CLASS: DEVICE_CLASS_HUMIDITY,
ATTR_ICON: None,
ATTR_LABEL: ATTR_API_HUMIDITY.capitalize(),
ATTR_UNIT: HUMI_PERCENT,
},
ATTR_API_PRESSURE: {
ATTR_DEVICE_CLASS: DEVICE_CLASS_PRESSURE,
ATTR_ICON: None,
ATTR_LABEL: ATTR_API_PRESSURE.capitalize(),
ATTR_UNIT: PRESSURE_HPA,
},
ATTR_API_TEMPERATURE: {
ATTR_DEVICE_CLASS: DEVICE_CLASS_TEMPERATURE,
ATTR_ICON: None,
ATTR_LABEL: ATTR_API_TEMPERATURE.capitalize(),
ATTR_UNIT: TEMP_CELSIUS,
},
}
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up Airly sensor entities based on a config entry."""
name = config_entry.data[CONF_NAME]
data = hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id]
sensors = []
for sensor in SENSOR_TYPES:
sensors.append(AirlySensor(data, name, sensor))
async_add_entities(sensors, True)
def round_state(func):
"""Round state."""
def _decorator(self):
res = func(self)
if isinstance(res, float):
return round(res)
return res
return _decorator
class AirlySensor(Entity):
"""Define an Airly sensor."""
def __init__(self, airly, name, kind):
"""Initialize."""
self.airly = airly
self.data = airly.data
self._name = name
self.kind = kind
self._device_class = None
self._state = None
self._icon = None
self._unit_of_measurement = None
self._attrs = {}
@property
def name(self):
"""Return the name."""
return f"{self._name} {SENSOR_TYPES[self.kind][ATTR_LABEL]}"
@property
def state(self):
"""Return the state."""
self._state = self.data[self.kind]
if self.kind in [ATTR_API_PM1, ATTR_API_PRESSURE]:
self._state = round(self._state)
if self.kind in [ATTR_API_TEMPERATURE, ATTR_API_HUMIDITY]:
self._state = round(self._state, 1)
return self._state
@property
def device_state_attributes(self):
"""Return the state attributes."""
return self._attrs
@property
def icon(self):
"""Return the icon."""
self._icon = SENSOR_TYPES[self.kind][ATTR_ICON]
return self._icon
@property
def device_class(self):
"""Return the device_class."""
return SENSOR_TYPES[self.kind][ATTR_DEVICE_CLASS]
@property
def unique_id(self):
"""Return a unique_id for this entity."""
return f"{self.airly.latitude}-{self.airly.longitude}-{self.kind.lower()}"
@property
def unit_of_measurement(self):
"""Return the unit the value is expressed in."""
return SENSOR_TYPES[self.kind][ATTR_UNIT]
@property
def attribution(self):
"""Return the attribution."""
return ATTRIBUTION
@property
def available(self):
"""Return True if entity is available."""
return bool(self.airly.data)
async def async_update(self):
"""Update the sensor."""
await self.airly.async_update()
if self.airly.data:
self.data = self.airly.data