Broadlink sensor and switch (#4834)

* Broadlink sensor and switch

* broadlink logging

* Use async

* style

* style
This commit is contained in:
Daniel Høyer Iversen 2016-12-16 06:42:00 +01:00 committed by Paulus Schoutsen
parent c125c4af4f
commit 394d53e748
4 changed files with 294 additions and 0 deletions

View file

@ -252,6 +252,7 @@ omit =
homeassistant/components/sensor/bbox.py
homeassistant/components/sensor/bitcoin.py
homeassistant/components/sensor/bom.py
homeassistant/components/sensor/broadlink.py
homeassistant/components/sensor/coinmarketcap.py
homeassistant/components/sensor/cpuspeed.py
homeassistant/components/sensor/cups.py
@ -323,6 +324,7 @@ omit =
homeassistant/components/switch/acer_projector.py
homeassistant/components/switch/anel_pwrctrl.py
homeassistant/components/switch/arest.py
homeassistant/components/switch/broadlink.py
homeassistant/components/switch/digitalloggers.py
homeassistant/components/switch/dlink.py
homeassistant/components/switch/edimax.py

View file

@ -0,0 +1,130 @@
"""
Support for the Broadlink RM2 Pro (only temperature) and A1 devices.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/sensor.broadlink/
"""
from datetime import timedelta
import binascii
import logging
import socket
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import (CONF_HOST, CONF_MAC,
CONF_MONITORED_CONDITIONS,
CONF_NAME, TEMP_CELSIUS, CONF_TIMEOUT)
from homeassistant.helpers.entity import Entity
from homeassistant.util import Throttle
import homeassistant.helpers.config_validation as cv
REQUIREMENTS = ['broadlink==0.2']
_LOGGER = logging.getLogger(__name__)
CONF_UPDATE_INTERVAL = 'update_interval'
DEVICE_DEFAULT_NAME = 'Broadlink sensor'
DEFAULT_TIMEOUT = 10
SENSOR_TYPES = {
'temperature': ['Temperature', TEMP_CELSIUS],
'air_quality': ['Air Quality', ' '],
'humidity': ['Humidity', '%'],
'light': ['Light', ' '],
'noise': ['Noise', ' ']
}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_NAME, default=DEVICE_DEFAULT_NAME): vol.Coerce(str),
vol.Optional(CONF_MONITORED_CONDITIONS, default=[]):
vol.All(cv.ensure_list, [vol.In(SENSOR_TYPES)]),
vol.Optional(CONF_UPDATE_INTERVAL, default=timedelta(seconds=300)): (
vol.All(cv.time_period, cv.positive_timedelta)),
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_MAC): cv.string,
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int
})
# pylint: disable=unused-argument
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Setup the Broadlink device sensors."""
mac = config.get(CONF_MAC).encode().replace(b':', b'')
mac_addr = binascii.unhexlify(mac)
broadlink_data = BroadlinkData(
config.get(CONF_UPDATE_INTERVAL),
config.get(CONF_HOST),
mac_addr, config.get(CONF_TIMEOUT))
dev = []
for variable in config[CONF_MONITORED_CONDITIONS]:
dev.append(BroadlinkSensor(
config.get(CONF_NAME),
broadlink_data,
variable))
add_devices(dev)
class BroadlinkSensor(Entity):
"""Representation of a Broadlink device sensor."""
def __init__(self, name, broadlink_data, sensor_type):
"""Initialize the sensor."""
self._name = "%s %s" % (name, SENSOR_TYPES[sensor_type][0])
self._state = None
self._type = sensor_type
self._broadlink_data = broadlink_data
self._unit_of_measurement = SENSOR_TYPES[sensor_type][1]
self.update()
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def state(self):
"""Return the state of the sensor."""
return self._state
@property
def unit_of_measurement(self):
"""Return the unit this state is expressed in."""
return self._unit_of_measurement
def update(self):
"""Get the latest data from the sensor."""
self._broadlink_data.update()
if self._broadlink_data.data is None:
return
self._state = self._broadlink_data.data[self._type]
class BroadlinkData(object):
"""Representation of a Broadlink data object."""
def __init__(self, interval, ip_addr, mac_addr, timeout):
"""Initialize the data object."""
import broadlink
self.data = None
self._device = broadlink.a1((ip_addr, 80), mac_addr)
self._device.timeout = timeout
self.update = Throttle(interval)(self._update)
try:
self._device.auth()
except socket.timeout:
_LOGGER.error("Failed to connect to device.")
def _update(self, retry=2):
try:
self.data = self._device.check_sensors_raw()
except socket.timeout as error:
if retry < 1:
_LOGGER.error(error)
return
try:
self._device.auth()
except socket.timeout:
pass
return self._update(max(0, retry-1))

View file

@ -0,0 +1,158 @@
"""
Support for Broadlink RM devices.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/switch.broadlink/
"""
from datetime import timedelta
from base64 import b64encode, b64decode
import asyncio
import binascii
import logging
import socket
import voluptuous as vol
import homeassistant.loader as loader
from homeassistant.util.dt import utcnow
from homeassistant.components.switch import (SwitchDevice, PLATFORM_SCHEMA)
from homeassistant.const import (CONF_FRIENDLY_NAME, CONF_SWITCHES,
CONF_COMMAND_OFF, CONF_COMMAND_ON,
CONF_TIMEOUT, CONF_HOST, CONF_MAC)
import homeassistant.helpers.config_validation as cv
REQUIREMENTS = ['broadlink==0.2']
_LOGGER = logging.getLogger(__name__)
DOMAIN = "broadlink"
DEFAULT_NAME = 'Broadlink switch'
DEFAULT_TIMEOUT = 10
SERVICE_LEARN = "learn_command"
SWITCH_SCHEMA = vol.Schema({
vol.Optional(CONF_COMMAND_OFF, default=None): cv.string,
vol.Optional(CONF_COMMAND_ON, default=None): cv.string,
vol.Optional(CONF_FRIENDLY_NAME, default=DEFAULT_NAME): cv.string,
})
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_SWITCHES): vol.Schema({cv.slug: SWITCH_SCHEMA}),
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_MAC): cv.string,
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Setup Broadlink switches."""
import broadlink
devices = config.get(CONF_SWITCHES, {})
switches = []
ip_addr = config.get(CONF_HOST)
mac_addr = binascii.unhexlify(
config.get(CONF_MAC).encode().replace(b':', b''))
broadlink_device = broadlink.rm((ip_addr, 80), mac_addr)
broadlink_device.timeout = config.get(CONF_TIMEOUT)
try:
broadlink_device.auth()
except socket.timeout:
_LOGGER.error("Failed to connect to device.")
persistent_notification = loader.get_component('persistent_notification')
@asyncio.coroutine
def _learn_command(call):
try:
yield from hass.loop.run_in_executor(None, broadlink_device.auth)
except socket.timeout:
_LOGGER.error("Failed to connect to device.")
return
yield from hass.loop.run_in_executor(None,
broadlink_device.enter_learning)
_LOGGER.info("Press the key you want HASS to learn")
start_time = utcnow()
while (utcnow() - start_time) < timedelta(seconds=20):
packet = yield from hass.loop.run_in_executor(None,
broadlink_device.
check_data)
if packet:
log_msg = 'Recieved packet is: {}'.\
format(b64encode(packet).decode('utf8'))
_LOGGER.info(log_msg)
persistent_notification.async_create(hass, log_msg,
title='Broadlink switch')
return
yield from asyncio.sleep(1, loop=hass.loop)
_LOGGER.error('Did not received any signal.')
persistent_notification.async_create(hass,
"Did not received any signal",
title='Broadlink switch')
hass.services.register(DOMAIN, SERVICE_LEARN, _learn_command)
for object_id, device_config in devices.items():
switches.append(
BroadlinkRM2Switch(
device_config.get(CONF_FRIENDLY_NAME, object_id),
device_config.get(CONF_COMMAND_ON),
device_config.get(CONF_COMMAND_OFF),
broadlink_device
)
)
add_devices(switches)
class BroadlinkRM2Switch(SwitchDevice):
"""Representation of an Broadlink switch."""
def __init__(self, friendly_name, command_on, command_off, device):
"""Initialize the switch."""
self._name = friendly_name
self._state = False
self._command_on = b64decode(command_on) if command_on else None
self._command_off = b64decode(command_off) if command_off else None
self._device = device
@property
def name(self):
"""Return the name of the switch."""
return self._name
@property
def assumed_state(self):
"""Return true if unable to access real state of entity."""
return True
@property
def is_on(self):
"""Return true if device is on."""
return self._state
def turn_on(self, **kwargs):
"""Turn the device on."""
if self._sendpacket(self._command_on):
self._state = True
def turn_off(self, **kwargs):
"""Turn the device off."""
if self._sendpacket(self._command_off):
self._state = False
def _sendpacket(self, packet, retry=2):
"""Send packet to device."""
if packet is None:
_LOGGER.debug("Empty packet.")
return True
try:
self._device.send_data(packet)
except socket.timeout as error:
if retry < 1:
_LOGGER.error(error)
return False
try:
self._device.auth()
except socket.timeout:
pass
return self._sendpacket(packet, max(0, retry-1))
return True

View file

@ -65,6 +65,10 @@ blockchain==1.3.3
# homeassistant.components.notify.aws_sqs
boto3==1.3.1
# homeassistant.components.sensor.broadlink
# homeassistant.components.switch.broadlink
broadlink==0.2
# homeassistant.components.sensor.coinmarketcap
coinmarketcap==2.0.1