Add netgear_lte component (#14687)

* Add netgear_lte component

* Improvements after review

* Allow multiple notify targets

* Require default notify target
This commit is contained in:
Anders Melchiorsen
2018-06-08 07:46:34 +02:00
committed by Martin Hjelmare
parent e3fba79126
commit b657cff6ba
5 changed files with 222 additions and 0 deletions

View File

@ -201,6 +201,9 @@ omit =
homeassistant/components/netatmo.py
homeassistant/components/*/netatmo.py
homeassistant/components/netgear_lte.py
homeassistant/components/*/netgear_lte.py
homeassistant/components/octoprint.py
homeassistant/components/*/octoprint.py

View File

@ -0,0 +1,86 @@
"""
Support for Netgear LTE modems.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/netgear_lte/
"""
import asyncio
from datetime import timedelta
import voluptuous as vol
import attr
from homeassistant.const import CONF_HOST, CONF_PASSWORD
import homeassistant.helpers.config_validation as cv
from homeassistant.util import Throttle
REQUIREMENTS = ['eternalegypt==0.0.1']
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=10)
DOMAIN = 'netgear_lte'
DATA_KEY = 'netgear_lte'
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.All(cv.ensure_list, [vol.Schema({
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
})])
}, extra=vol.ALLOW_EXTRA)
@attr.s
class LTEData:
"""Class for LTE state."""
eternalegypt = attr.ib()
unread_count = attr.ib(init=False)
usage = attr.ib(init=False)
@Throttle(MIN_TIME_BETWEEN_UPDATES)
async def async_update(self):
"""Call the API to update the data."""
information = await self.eternalegypt.information()
self.unread_count = sum(1 for x in information.sms if x.unread)
self.usage = information.usage
@attr.s
class LTEHostData:
"""Container for LTE states."""
hostdata = attr.ib(init=False, factory=dict)
def get(self, config):
"""Get the requested or the only hostdata value."""
if CONF_HOST in config:
return self.hostdata.get(config[CONF_HOST])
elif len(self.hostdata) == 1:
return next(iter(self.hostdata.values()))
return None
async def async_setup(hass, config):
"""Set up Netgear LTE component."""
if DATA_KEY not in hass.data:
hass.data[DATA_KEY] = LTEHostData()
tasks = [_setup_lte(hass, conf) for conf in config.get(DOMAIN, [])]
if tasks:
await asyncio.wait(tasks)
return True
async def _setup_lte(hass, lte_config):
"""Set up a Netgear LTE modem."""
import eternalegypt
host = lte_config[CONF_HOST]
password = lte_config[CONF_PASSWORD]
eternalegypt = eternalegypt.LB2120(host, password)
lte_data = LTEData(eternalegypt)
await lte_data.async_update()
hass.data[DATA_KEY].hostdata[host] = lte_data

View File

@ -0,0 +1,45 @@
"""Netgear LTE platform for notify component.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/notify.netgear_lte/
"""
import voluptuous as vol
import attr
from homeassistant.components.notify import (
BaseNotificationService, ATTR_TARGET, PLATFORM_SCHEMA)
from homeassistant.const import CONF_HOST
import homeassistant.helpers.config_validation as cv
from ..netgear_lte import DATA_KEY
DEPENDENCIES = ['netgear_lte']
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."""
lte_data = hass.data[DATA_KEY].get(config)
phone = config.get(ATTR_TARGET)
return NetgearNotifyService(lte_data, phone)
@attr.s
class NetgearNotifyService(BaseNotificationService):
"""Implementation of a notification service."""
lte_data = attr.ib()
phone = attr.ib()
async def async_send_message(self, message="", **kwargs):
"""Send a message to a user."""
targets = kwargs.get(ATTR_TARGET, self.phone)
if targets and message:
for target in targets:
await self.lte_data.eternalegypt.sms(target, message)

View File

@ -0,0 +1,85 @@
"""Netgear LTE sensors.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/sensor.netgear_lte/
"""
import voluptuous as vol
import attr
from homeassistant.const import CONF_HOST, CONF_SENSORS
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.helpers.entity import Entity
import homeassistant.helpers.config_validation as cv
from ..netgear_lte import DATA_KEY
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])])
})
async def async_setup_platform(
hass, config, async_add_devices, discovery_info):
"""Set up Netgear LTE sensor devices."""
lte_data = hass.data[DATA_KEY].get(config)
sensors = []
for sensortype in config[CONF_SENSORS]:
if sensortype == SENSOR_SMS:
sensors.append(SMSSensor(lte_data))
elif sensortype == SENSOR_USAGE:
sensors.append(UsageSensor(lte_data))
async_add_devices(sensors, True)
@attr.s
class LTESensor(Entity):
"""Data usage sensor entity."""
lte_data = attr.ib()
async def async_update(self):
"""Update state."""
await self.lte_data.async_update()
class SMSSensor(LTESensor):
"""Unread SMS sensor entity."""
@property
def name(self):
"""Return the name of the sensor."""
return "Netgear LTE SMS"
@property
def state(self):
"""Return the state of the sensor."""
return self.lte_data.unread_count
class UsageSensor(LTESensor):
"""Data usage sensor entity."""
@property
def unit_of_measurement(self):
"""Return the unit of measurement."""
return "MiB"
@property
def name(self):
"""Return the name of the sensor."""
return "Netgear LTE usage"
@property
def state(self):
"""Return the state of the sensor."""
return round(self.lte_data.usage / 1024**2, 1)

View File

@ -291,6 +291,9 @@ enocean==0.40
# homeassistant.components.sensor.season
ephem==3.7.6.0
# homeassistant.components.netgear_lte
eternalegypt==0.0.1
# homeassistant.components.keyboard_remote
# evdev==0.6.1