jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-17 10:17:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge netatmo_public sensor into the netatmo integration (#23531)

Browse Source 
* Merge netatmo public into netatmo integration

* Remove netatmo_public platform

* Remove dev log messages

* Improve error handling

* Check config for unsupported conditions

* Fix linter

* Reduce nested blocks
This commit is contained in:
cgtobi 2019-05-08 08:26:52 +02:00 committed by Fabian Affolter
parent 3e788aa1d6
commit f0f6787bf9
4 changed files with 199 additions and 254 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

262
homeassistant/components/netatmo/sensor.py
View File

@ -1,4 +1,5 @@
"""Support for the NetAtmo Weather Service."""
"""Support for the Netatmo Weather Service."""
from datetime import timedelta
import logging
from time import time
import threading
@ -7,10 +8,12 @@ import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import (
CONF_NAME, CONF_MODE, CONF_MONITORED_CONDITIONS,
TEMP_CELSIUS, DEVICE_CLASS_HUMIDITY, DEVICE_CLASS_TEMPERATURE,
DEVICE_CLASS_BATTERY)
from homeassistant.helpers.entity import Entity
import homeassistant.helpers.config_validation as cv
from homeassistant.util import Throttle
from .const import DATA_NETATMO_AUTH
@ -18,17 +21,35 @@ _LOGGER = logging.getLogger(__name__)
CONF_MODULES = 'modules'
CONF_STATION = 'station'
CONF_AREAS = 'areas'
CONF_LAT_NE = 'lat_ne'
CONF_LON_NE = 'lon_ne'
CONF_LAT_SW = 'lat_sw'
CONF_LON_SW = 'lon_sw'
# This is the NetAtmo data upload interval in seconds
DEFAULT_MODE = 'avg'
MODE_TYPES = {'max', 'avg'}
DEFAULT_NAME_PUBLIC = 'Netatmo Public Data'
# This is the Netatmo data upload interval in seconds
NETATMO_UPDATE_INTERVAL = 600
# NetAtmo Public Data is uploaded to server every 10 minutes
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=600)
SUPPORTED_PUBLIC_SENSOR_TYPES = [
'temperature', 'pressure', 'humidity', 'rain', 'windstrength',
'guststrength'
]
SENSOR_TYPES = {
'temperature': ['Temperature', TEMP_CELSIUS, None,
'temperature': ['Temperature', TEMP_CELSIUS, 'mdi:thermometer',
DEVICE_CLASS_TEMPERATURE],
'co2': ['CO2', 'ppm', 'mdi:cloud', None],
'pressure': ['Pressure', 'mbar', 'mdi:gauge', None],
'noise': ['Noise', 'dB', 'mdi:volume-high', None],
'humidity': ['Humidity', '%', None, DEVICE_CLASS_HUMIDITY],
'humidity': ['Humidity', '%', 'mdi:water-percent', DEVICE_CLASS_HUMIDITY],
'rain': ['Rain', 'mm', 'mdi:weather-rainy', None],
'sum_rain_1': ['sum_rain_1', 'mm', 'mdi:weather-rainy', None],
'sum_rain_24': ['sum_rain_24', 'mm', 'mdi:weather-rainy', None],
@ -39,7 +60,7 @@ SENSOR_TYPES = {
'max_temp': ['Max Temp.', TEMP_CELSIUS, 'mdi:thermometer', None],
'windangle': ['Angle', '', 'mdi:compass', None],
'windangle_value': ['Angle Value', 'º', 'mdi:compass', None],
'windstrength': ['Strength', 'km/h', 'mdi:weather-windy', None],
'windstrength': ['Wind Strength', 'km/h', 'mdi:weather-windy', None],
'gustangle': ['Gust Angle', '', 'mdi:compass', None],
'gustangle_value': ['Gust Angle Value', 'º', 'mdi:compass', None],
'guststrength': ['Gust Strength', 'km/h', 'mdi:weather-windy', None],
@ -57,6 +78,18 @@ MODULE_SCHEMA = vol.Schema({
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_STATION): cv.string,
vol.Optional(CONF_MODULES): MODULE_SCHEMA,
vol.Optional(CONF_AREAS): vol.All(cv.ensure_list, [
{
vol.Required(CONF_LAT_NE): cv.latitude,
vol.Required(CONF_LAT_SW): cv.latitude,
vol.Required(CONF_LON_NE): cv.longitude,
vol.Required(CONF_LON_SW): cv.longitude,
vol.Required(CONF_MONITORED_CONDITIONS): [vol.In(
SUPPORTED_PUBLIC_SENSOR_TYPES)],
vol.Optional(CONF_MODE, default=DEFAULT_MODE): vol.In(MODE_TYPES),
vol.Optional(CONF_NAME, default=DEFAULT_NAME_PUBLIC): cv.string
}
]),
})
MODULE_TYPE_OUTDOOR = 'NAModule1'
@ -68,31 +101,41 @@ MODULE_TYPE_INDOOR = 'NAModule4'
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the available Netatmo weather sensors."""
dev = []
not_handled = {}
auth = hass.data[DATA_NETATMO_AUTH]
if CONF_MODULES in config:
manual_config(auth, config, dev)
if config.get(CONF_AREAS) is not None:
for area in config[CONF_AREAS]:
data = NetatmoPublicData(
auth,
lat_ne=area[CONF_LAT_NE],
lon_ne=area[CONF_LON_NE],
lat_sw=area[CONF_LAT_SW],
lon_sw=area[CONF_LON_SW]
)
for sensor_type in area[CONF_MONITORED_CONDITIONS]:
dev.append(NetatmoPublicSensor(
area[CONF_NAME],
data,
sensor_type,
area[CONF_MODE]
))
else:
auto_config(auth, config, dev)
for data_class in all_product_classes():
data = NetatmoData(auth, data_class, config.get(CONF_STATION))
module_items = []
# Test if manually configured
if CONF_MODULES in config:
module_items = config[CONF_MODULES].items()
else:
# otherwise add all modules and conditions
for module_name in data.get_module_names():
monitored_conditions = \
data.station_data.monitoredConditions(module_name)
module_items.append(
(module_name, monitored_conditions))
if dev:
add_entities(dev, True)
def manual_config(auth, config, dev):
"""Handle manual configuration."""
import pyatmo
all_classes = all_product_classes()
not_handled = {}
for data_class in all_classes:
data = NetAtmoData(auth, data_class,
config.get(CONF_STATION))
try:
# Iterate each module
for module_name, monitored_conditions in \
config[CONF_MODULES].items():
for module_name, monitored_conditions in module_items:
# Test if module exists
if module_name not in data.get_module_names():
not_handled[module_name] = \
@ -100,33 +143,15 @@ def manual_config(auth, config, dev):
if module_name in not_handled else 1
else:
# Only create sensors for monitored properties
for variable in monitored_conditions:
dev.append(NetAtmoSensor(data, module_name, variable))
except pyatmo.NoDevice:
continue
for condition in monitored_conditions:
dev.append(NetatmoSensor(
data, module_name, condition))
for module_name, count in not_handled.items():
if count == len(all_classes):
for module_name, _ in not_handled.items():
_LOGGER.error('Module name: "%s" not found', module_name)
def auto_config(auth, config, dev):
"""Handle auto configuration."""
import pyatmo
for data_class in all_product_classes():
data = NetAtmoData(auth, data_class, config.get(CONF_STATION))
try:
for module_name in data.get_module_names():
for variable in \
data.station_data.monitoredConditions(module_name):
if variable in SENSOR_TYPES.keys():
dev.append(NetAtmoSensor(data, module_name, variable))
else:
_LOGGER.warning("Ignoring unknown var %s for mod %s",
variable, module_name)
except pyatmo.NoDevice:
continue
if dev:
add_entities(dev, True)
def all_product_classes():
@ -136,7 +161,7 @@ def all_product_classes():
return [pyatmo.WeatherStationData, pyatmo.HomeCoachData]
class NetAtmoSensor(Entity):
class NetatmoSensor(Entity):
"""Implementation of a Netatmo sensor."""
def __init__(self, netatmo_data, module_name, sensor_type):
@ -187,7 +212,7 @@ class NetAtmoSensor(Entity):
return self._unique_id
def update(self):
"""Get the latest data from NetAtmo API and updates the states."""
"""Get the latest data from Netatmo API and updates the states."""
self.netatmo_data.update()
if self.netatmo_data.data is None:
if self._state is None:
@ -362,14 +387,121 @@ class NetAtmoSensor(Entity):
return
class NetAtmoData:
"""Get the latest data from NetAtmo."""
class NetatmoPublicSensor(Entity):
"""Represent a single sensor in a Netatmo."""
def __init__(self, area_name, data, sensor_type, mode):
"""Initialize the sensor."""
self.netatmo_data = data
self.type = sensor_type
self._mode = mode
self._name = '{} {}'.format(area_name,
SENSOR_TYPES[self.type][0])
self._area_name = area_name
self._state = None
self._device_class = SENSOR_TYPES[self.type][3]
self._icon = SENSOR_TYPES[self.type][2]
self._unit_of_measurement = SENSOR_TYPES[self.type][1]
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def icon(self):
"""Icon to use in the frontend."""
return self._icon
@property
def device_class(self):
"""Return the device class of the sensor."""
return self._device_class
@property
def state(self):
"""Return the state of the device."""
return self._state
@property
def unit_of_measurement(self):
"""Return the unit of measurement of this entity."""
return self._unit_of_measurement
def update(self):
"""Get the latest data from Netatmo API and updates the states."""
self.netatmo_data.update()
if self.netatmo_data.data is None:
_LOGGER.warning("No data found for %s", self._name)
self._state = None
return
data = None
if self.type == 'temperature':
data = self.netatmo_data.data.getLatestTemperatures()
elif self.type == 'pressure':
data = self.netatmo_data.data.getLatestPressures()
elif self.type == 'humidity':
data = self.netatmo_data.data.getLatestHumidities()
elif self.type == 'rain':
data = self.netatmo_data.data.getLatestRain()
elif self.type == 'windstrength':
data = self.netatmo_data.data.getLatestWindStrengths()
elif self.type == 'guststrength':
data = self.netatmo_data.data.getLatestGustStrengths()
if not data:
_LOGGER.warning("No station provides %s data in the area %s",
self.type, self._area_name)
self._state = None
return
if self._mode == 'avg':
self._state = round(sum(data.values()) / len(data), 1)
elif self._mode == 'max':
self._state = max(data.values())
class NetatmoPublicData:
"""Get the latest data from Netatmo."""
def __init__(self, auth, lat_ne, lon_ne, lat_sw, lon_sw):
"""Initialize the data object."""
self.auth = auth
self.data = None
self.lat_ne = lat_ne
self.lon_ne = lon_ne
self.lat_sw = lat_sw
self.lon_sw = lon_sw
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
"""Request an update from the Netatmo API."""
import pyatmo
data = pyatmo.PublicData(self.auth,
LAT_NE=self.lat_ne,
LON_NE=self.lon_ne,
LAT_SW=self.lat_sw,
LON_SW=self.lon_sw,
filtering=True)
if data.CountStationInArea() == 0:
_LOGGER.warning('No Stations available in this area.')
return
self.data = data
class NetatmoData:
"""Get the latest data from Netatmo."""
def __init__(self, auth, data_class, station):
"""Initialize the data object."""
self.auth = auth
self.data_class = data_class
self.data = None
self.data = {}
self.station_data = None
self.station = station
self._next_update = time()
@ -378,8 +510,6 @@ class NetAtmoData:
def get_module_names(self):
"""Return all module available on the API as a list."""
self.update()
if not self.data:
return []
return self.data.keys()
def _detect_platform_type(self):
@ -387,12 +517,16 @@ class NetAtmoData:
The return can be a WeatherStationData or a HomeCoachData.
"""
from pyatmo import NoDevice
try:
station_data = self.data_class(self.auth)
_LOGGER.debug("%s detected!", str(self.data_class.__name__))
return station_data
except TypeError:
return
except NoDevice:
_LOGGER.error("No Weather or HomeCoach devices found for %s", str(
self.station
))
raise
def update(self):
"""Call the Netatmo API to update the data.
@ -405,11 +539,13 @@ class NetAtmoData:
not self._update_in_progress.acquire(False):
return
from pyatmo import NoDevice
try:
self.station_data = self._detect_platform_type()
if not self.station_data:
raise Exception("No Weather nor HomeCoach devices found")
except NoDevice:
return
try:
if self.station is not None:
self.data = self.station_data.lastData(
station=self.station, exclude=3600)
@ -432,11 +568,11 @@ class NetAtmoData:
newinterval = NETATMO_UPDATE_INTERVAL
else:
if newinterval < NETATMO_UPDATE_INTERVAL / 2:
# Never hammer the NetAtmo API more than
# Never hammer the Netatmo API more than
# twice per update interval
newinterval = NETATMO_UPDATE_INTERVAL / 2
_LOGGER.info(
"NetAtmo refresh interval reset to %d seconds",
"Netatmo refresh interval reset to %d seconds",
newinterval)
else:
# Last update time not found, fall back to default value

1
homeassistant/components/netatmo_public/__init__.py
View File

@ -1 +0,0 @@
"""The netatmo_public component."""

10
homeassistant/components/netatmo_public/manifest.json
View File

@ -1,10 +0,0 @@
{
"domain": "netatmo_public",
"name": "Netatmo public",
"documentation": "https://www.home-assistant.io/components/netatmo_public",
"requirements": [],
"dependencies": [
"netatmo"
],
"codeowners": []
}

180
homeassistant/components/netatmo_public/sensor.py
View File

@ -1,180 +0,0 @@
"""Support for Sensors using public Netatmo data."""
from datetime import timedelta
import logging
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import (
CONF_NAME, CONF_MODE, CONF_MONITORED_CONDITIONS, TEMP_CELSIUS,
DEVICE_CLASS_TEMPERATURE, DEVICE_CLASS_HUMIDITY)
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.util import Throttle
from homeassistant.components.netatmo.const import DATA_NETATMO_AUTH
_LOGGER = logging.getLogger(__name__)
CONF_AREAS = 'areas'
CONF_LAT_NE = 'lat_ne'
CONF_LON_NE = 'lon_ne'
CONF_LAT_SW = 'lat_sw'
CONF_LON_SW = 'lon_sw'
DEFAULT_NAME = 'Netatmo Public Data'
DEFAULT_MODE = 'avg'
MODE_TYPES = {'max', 'avg'}
SENSOR_TYPES = {
'temperature': ['Temperature', TEMP_CELSIUS, 'mdi:thermometer',
DEVICE_CLASS_TEMPERATURE],
'pressure': ['Pressure', 'mbar', 'mdi:gauge', None],
'humidity': ['Humidity', '%', 'mdi:water-percent', DEVICE_CLASS_HUMIDITY],
'rain': ['Rain', 'mm', 'mdi:weather-rainy', None],
'windstrength': ['Wind Strength', 'km/h', 'mdi:weather-windy', None],
'guststrength': ['Gust Strength', 'km/h', 'mdi:weather-windy', None],
}
# NetAtmo Data is uploaded to server every 10 minutes
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=600)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_AREAS): vol.All(cv.ensure_list, [
{
vol.Required(CONF_LAT_NE): cv.latitude,
vol.Required(CONF_LAT_SW): cv.latitude,
vol.Required(CONF_LON_NE): cv.longitude,
vol.Required(CONF_LON_SW): cv.longitude,
vol.Required(CONF_MONITORED_CONDITIONS): [vol.In(SENSOR_TYPES)],
vol.Optional(CONF_MODE, default=DEFAULT_MODE): vol.In(MODE_TYPES),
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string
}
]),
})
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the access to Netatmo binary sensor."""
auth = hass.data[DATA_NETATMO_AUTH]
sensors = []
areas = config.get(CONF_AREAS)
for area_conf in areas:
data = NetatmoPublicData(auth,
lat_ne=area_conf.get(CONF_LAT_NE),
lon_ne=area_conf.get(CONF_LON_NE),
lat_sw=area_conf.get(CONF_LAT_SW),
lon_sw=area_conf.get(CONF_LON_SW))
for sensor_type in area_conf.get(CONF_MONITORED_CONDITIONS):
sensors.append(NetatmoPublicSensor(area_conf.get(CONF_NAME),
data, sensor_type,
area_conf.get(CONF_MODE)))
add_entities(sensors, True)
class NetatmoPublicSensor(Entity):
"""Represent a single sensor in a Netatmo."""
def __init__(self, area_name, data, sensor_type, mode):
"""Initialize the sensor."""
self.netatmo_data = data
self.type = sensor_type
self._mode = mode
self._name = '{} {}'.format(area_name,
SENSOR_TYPES[self.type][0])
self._area_name = area_name
self._state = None
self._device_class = SENSOR_TYPES[self.type][3]
self._icon = SENSOR_TYPES[self.type][2]
self._unit_of_measurement = SENSOR_TYPES[self.type][1]
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def icon(self):
"""Icon to use in the frontend."""
return self._icon
@property
def device_class(self):
"""Return the device class of the sensor."""
return self._device_class
@property
def state(self):
"""Return the state of the device."""
return self._state
@property
def unit_of_measurement(self):
"""Return the unit of measurement of this entity."""
return self._unit_of_measurement
def update(self):
"""Get the latest data from NetAtmo API and updates the states."""
self.netatmo_data.update()
if self.netatmo_data.data is None:
_LOGGER.warning("No data found for %s", self._name)
self._state = None
return
data = None
if self.type == 'temperature':
data = self.netatmo_data.data.getLatestTemperatures()
elif self.type == 'pressure':
data = self.netatmo_data.data.getLatestPressures()
elif self.type == 'humidity':
data = self.netatmo_data.data.getLatestHumidities()
elif self.type == 'rain':
data = self.netatmo_data.data.getLatestRain()
elif self.type == 'windstrength':
data = self.netatmo_data.data.getLatestWindStrengths()
elif self.type == 'guststrength':
data = self.netatmo_data.data.getLatestGustStrengths()
if not data:
_LOGGER.warning("No station provides %s data in the area %s",
self.type, self._area_name)
self._state = None
return
if self._mode == 'avg':
self._state = round(sum(data.values()) / len(data), 1)
elif self._mode == 'max':
self._state = max(data.values())
class NetatmoPublicData:
"""Get the latest data from NetAtmo."""
def __init__(self, auth, lat_ne, lon_ne, lat_sw, lon_sw):
"""Initialize the data object."""
self.auth = auth
self.data = None
self.lat_ne = lat_ne
self.lon_ne = lon_ne
self.lat_sw = lat_sw
self.lon_sw = lon_sw
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
"""Request an update from the Netatmo API."""
import pyatmo
data = pyatmo.PublicData(self.auth,
LAT_NE=self.lat_ne,
LON_NE=self.lon_ne,
LAT_SW=self.lat_sw,
LON_SW=self.lon_sw,
filtering=True)
if data.CountStationInArea() == 0:
_LOGGER.warning('No Stations available in this area.')
return
self.data = data