Consolidate the netgear_lte configuration (#22105)

* Consolidate the netgear_lte configuration

* Simplfications from review

* Extract sensor_types

* Simplify defaults
This commit is contained in:
Anders Melchiorsen 2019-03-22 14:43:39 +01:00 committed by Martin Hjelmare
parent b125514655
commit 2b6e197deb
4 changed files with 96 additions and 49 deletions

View File

@ -8,12 +8,17 @@ import attr
import voluptuous as vol
from homeassistant.const import (
CONF_HOST, CONF_PASSWORD, EVENT_HOMEASSISTANT_STOP)
CONF_HOST, CONF_MONITORED_CONDITIONS, CONF_NAME, CONF_PASSWORD,
CONF_RECIPIENT, EVENT_HOMEASSISTANT_STOP)
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv
from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.helpers import config_validation as cv, discovery
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from homeassistant.util import Throttle
from . import sensor_types
REQUIREMENTS = ['eternalegypt==0.0.5']
_LOGGER = logging.getLogger(__name__)
@ -23,10 +28,26 @@ MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=10)
DOMAIN = 'netgear_lte'
DATA_KEY = 'netgear_lte'
NOTIFY_SCHEMA = vol.Schema({
vol.Optional(CONF_NAME, default=DOMAIN): cv.string,
vol.Optional(CONF_RECIPIENT, default=[]):
vol.All(cv.ensure_list, [cv.string]),
})
SENSOR_SCHEMA = vol.Schema({
vol.Optional(CONF_MONITORED_CONDITIONS, default=sensor_types.DEFAULT):
vol.All(cv.ensure_list, [vol.In(sensor_types.ALL)]),
})
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.All(cv.ensure_list, [vol.Schema({
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Optional(NOTIFY_DOMAIN, default={}):
vol.All(cv.ensure_list, [NOTIFY_SCHEMA]),
vol.Optional(SENSOR_DOMAIN, default={}):
SENSOR_SCHEMA,
})])
}, extra=vol.ALLOW_EXTRA)
@ -71,13 +92,8 @@ class LTEData:
modem_data = attr.ib(init=False, factory=dict)
def get_modem_data(self, config):
"""Get the requested or the only modem_data value."""
if CONF_HOST in config:
return self.modem_data.get(config[CONF_HOST])
if len(self.modem_data) == 1:
return next(iter(self.modem_data.values()))
return None
"""Get modem_data for the host in config."""
return self.modem_data.get(config[CONF_HOST])
async def async_setup(hass, config):
@ -87,9 +103,32 @@ async def async_setup(hass, config):
hass, cookie_jar=aiohttp.CookieJar(unsafe=True))
hass.data[DATA_KEY] = LTEData(websession)
tasks = [_setup_lte(hass, conf) for conf in config.get(DOMAIN, [])]
if tasks:
await asyncio.wait(tasks)
netgear_lte_config = config[DOMAIN]
# Set up each modem
tasks = [_setup_lte(hass, lte_conf) for lte_conf in netgear_lte_config]
await asyncio.wait(tasks)
# Load platforms for each modem
for lte_conf in netgear_lte_config:
# Notify
for notify_conf in lte_conf[NOTIFY_DOMAIN]:
discovery_info = {
CONF_HOST: lte_conf[CONF_HOST],
CONF_NAME: notify_conf.get(CONF_NAME),
NOTIFY_DOMAIN: notify_conf,
}
hass.async_create_task(discovery.async_load_platform(
hass, NOTIFY_DOMAIN, DOMAIN, discovery_info, config))
# Sensor
sensor_conf = lte_conf.get(SENSOR_DOMAIN)
discovery_info = {
CONF_HOST: lte_conf[CONF_HOST],
SENSOR_DOMAIN: sensor_conf,
}
hass.async_create_task(discovery.async_load_platform(
hass, SENSOR_DOMAIN, DOMAIN, discovery_info, config))
return True

View File

@ -2,28 +2,23 @@
import logging
import attr
import voluptuous as vol
from homeassistant.components.notify import (
ATTR_TARGET, PLATFORM_SCHEMA, BaseNotificationService)
from homeassistant.const import CONF_HOST
import homeassistant.helpers.config_validation as cv
ATTR_TARGET, BaseNotificationService, DOMAIN)
from ..netgear_lte import DATA_KEY
from . import CONF_RECIPIENT, DATA_KEY
DEPENDENCIES = ['netgear_lte']
_LOGGER = logging.getLogger(__name__)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_HOST): cv.string,
vol.Required(ATTR_TARGET): vol.All(cv.ensure_list, [cv.string]),
})
async def async_get_service(hass, config, discovery_info=None):
"""Get the notification service."""
return NetgearNotifyService(hass, config)
if discovery_info is None:
return
return NetgearNotifyService(hass, discovery_info)
@attr.s
@ -35,17 +30,23 @@ class NetgearNotifyService(BaseNotificationService):
async def async_send_message(self, message="", **kwargs):
"""Send a message to a user."""
import eternalegypt
modem_data = self.hass.data[DATA_KEY].get_modem_data(self.config)
if not modem_data:
_LOGGER.error("No modem available")
_LOGGER.error("Modem not ready")
return
phone = self.config.get(ATTR_TARGET)
targets = kwargs.get(ATTR_TARGET, phone)
if targets and message:
for target in targets:
import eternalegypt
try:
await modem_data.modem.sms(target, message)
except eternalegypt.Error:
_LOGGER.error("Unable to send to %s", target)
targets = kwargs.get(ATTR_TARGET, self.config[DOMAIN][CONF_RECIPIENT])
if not targets:
_LOGGER.warning("No recipients")
return
if not message:
return
for target in targets:
try:
await modem_data.modem.sms(target, message)
except eternalegypt.Error:
_LOGGER.error("Unable to send to %s", target)

View File

@ -1,37 +1,36 @@
"""Support for Netgear LTE sensors."""
import attr
import voluptuous as vol
import logging
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import CONF_HOST, CONF_SENSORS
import attr
from homeassistant.components.sensor import DOMAIN
from homeassistant.exceptions import PlatformNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from ..netgear_lte import DATA_KEY
from . import CONF_MONITORED_CONDITIONS, DATA_KEY
from .sensor_types import SENSOR_SMS, SENSOR_USAGE
DEPENDENCIES = ['netgear_lte']
SENSOR_SMS = 'sms'
SENSOR_USAGE = 'usage'
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_HOST): cv.string,
vol.Required(CONF_SENSORS): vol.All(
cv.ensure_list, [vol.In([SENSOR_SMS, SENSOR_USAGE])]),
})
_LOGGER = logging.getLogger(__name__)
async def async_setup_platform(
hass, config, async_add_entities, discovery_info):
"""Set up Netgear LTE sensor devices."""
modem_data = hass.data[DATA_KEY].get_modem_data(config)
if discovery_info is None:
return
modem_data = hass.data[DATA_KEY].get_modem_data(discovery_info)
if not modem_data:
raise PlatformNotReady
sensor_conf = discovery_info[DOMAIN]
monitored_conditions = sensor_conf[CONF_MONITORED_CONDITIONS]
sensors = []
for sensor_type in config[CONF_SENSORS]:
for sensor_type in monitored_conditions:
if sensor_type == SENSOR_SMS:
sensors.append(SMSSensor(modem_data, sensor_type))
elif sensor_type == SENSOR_USAGE:

View File

@ -0,0 +1,8 @@
"""Define possible sensor types."""
SENSOR_SMS = 'sms'
SENSOR_USAGE = 'usage'
ALL = [SENSOR_SMS, SENSOR_USAGE]
DEFAULT = [SENSOR_USAGE]