Xiaomi gw support (#8555)
* xiaomi support * xiaomi support * style * style * style * style * style * coveragerc * Update xiaomi.py * Update xiaomi.py * Update xiaomi.py * refactorization * refactorization * config validation * style * package * refactorization * refactorization * refactorization * HA integration
This commit is contained in:
Daniel Høyer Iversen
Pascal Vizeli
parent
966809c1a1
commit
90639d33ab
8 changed files with 874 additions and 1 deletions
|
|
@ -193,6 +193,9 @@ omit =
|
|||
homeassistant/components/wink.py
|
||||
homeassistant/components/*/wink.py
|
||||
|
||||
homeassistant/components/xiaomi.py
|
||||
homeassistant/components/*/xiaomi.py
|
||||
|
||||
homeassistant/components/zabbix.py
|
||||
homeassistant/components/*/zabbix.py
|
||||
|
||||
|
|
@ -274,7 +277,6 @@ omit =
|
|||
homeassistant/components/device_tracker/tplink.py
|
||||
homeassistant/components/device_tracker/trackr.py
|
||||
homeassistant/components/device_tracker/ubus.py
|
||||
homeassistant/components/device_tracker/xiaomi.py
|
||||
homeassistant/components/downloader.py
|
||||
homeassistant/components/emoncms_history.py
|
||||
homeassistant/components/emulated_hue/upnp.py
|
||||
|
|
|
|||
316
homeassistant/components/binary_sensor/xiaomi.py
Normal file
316
homeassistant/components/binary_sensor/xiaomi.py
Normal file
|
|
@ -0,0 +1,316 @@
|
|||
"""Support for Xiaomi binary sensors."""
|
||||
import logging
|
||||
|
||||
from homeassistant.components.binary_sensor import BinarySensorDevice
|
||||
from homeassistant.components.xiaomi import (PY_XIAOMI_GATEWAY, XiaomiDevice)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
NO_CLOSE = 'no_close'
|
||||
ATTR_OPEN_SINCE = 'Open since'
|
||||
|
||||
MOTION = 'motion'
|
||||
NO_MOTION = 'no_motion'
|
||||
ATTR_NO_MOTION_SINCE = 'No motion since'
|
||||
|
||||
DENSITY = 'density'
|
||||
ATTR_DENSITY = 'Density'
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Perform the setup for Xiaomi devices."""
|
||||
devices = []
|
||||
for (_, gateway) in hass.data[PY_XIAOMI_GATEWAY].gateways.items():
|
||||
for device in gateway.devices['binary_sensor']:
|
||||
model = device['model']
|
||||
if model == 'motion':
|
||||
devices.append(XiaomiMotionSensor(device, hass, gateway))
|
||||
elif model == 'magnet':
|
||||
devices.append(XiaomiDoorSensor(device, gateway))
|
||||
elif model == 'smoke':
|
||||
devices.append(XiaomiSmokeSensor(device, gateway))
|
||||
elif model == 'natgas':
|
||||
devices.append(XiaomiNatgasSensor(device, gateway))
|
||||
elif model == 'switch':
|
||||
devices.append(XiaomiButton(device, 'Switch', 'status',
|
||||
hass, gateway))
|
||||
elif model == 'sensor_switch.aq2':
|
||||
devices.append(XiaomiButton(device, 'Switch', 'status',
|
||||
hass, gateway))
|
||||
elif model == '86sw1':
|
||||
devices.append(XiaomiButton(device, 'Wall Switch', 'channel_0',
|
||||
hass, gateway))
|
||||
elif model == '86sw2':
|
||||
devices.append(XiaomiButton(device, 'Wall Switch (Left)',
|
||||
'channel_0', hass, gateway))
|
||||
devices.append(XiaomiButton(device, 'Wall Switch (Right)',
|
||||
'channel_1', hass, gateway))
|
||||
devices.append(XiaomiButton(device, 'Wall Switch (Both)',
|
||||
'dual_channel', hass, gateway))
|
||||
elif model == 'cube':
|
||||
devices.append(XiaomiCube(device, hass, gateway))
|
||||
add_devices(devices)
|
||||
|
||||
|
||||
class XiaomiBinarySensor(XiaomiDevice, BinarySensorDevice):
|
||||
"""Representation of a base XiaomiBinarySensor."""
|
||||
|
||||
def __init__(self, device, name, xiaomi_hub):
|
||||
"""Initialize the XiaomiSmokeSensor."""
|
||||
self._data_key = None
|
||||
self._device_class = None
|
||||
self._should_poll = False
|
||||
self._density = 0
|
||||
XiaomiDevice.__init__(self, device, name, xiaomi_hub)
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
"""Return True if entity has to be polled for state."""
|
||||
return self._should_poll
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return the state of the sensor."""
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def device_class(self):
|
||||
"""Return the class of binary sensor."""
|
||||
return self._device_class
|
||||
|
||||
def update(self):
|
||||
"""Update the sensor state."""
|
||||
_LOGGER.debug('Updating xiaomi sensor by polling')
|
||||
self._get_from_hub(self._sid)
|
||||
|
||||
|
||||
class XiaomiNatgasSensor(XiaomiBinarySensor):
|
||||
"""Representation of a XiaomiNatgasSensor."""
|
||||
|
||||
def __init__(self, device, xiaomi_hub):
|
||||
"""Initialize the XiaomiSmokeSensor."""
|
||||
self._data_key = 'alarm'
|
||||
self._density = None
|
||||
self._device_class = 'gas'
|
||||
XiaomiBinarySensor.__init__(self, device, 'Natgas Sensor', xiaomi_hub)
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
attrs = {ATTR_DENSITY: self._density}
|
||||
attrs.update(super().device_state_attributes)
|
||||
return attrs
|
||||
|
||||
def parse_data(self, data):
|
||||
"""Parse data sent by gateway."""
|
||||
if DENSITY in data:
|
||||
self._density = int(data.get(DENSITY))
|
||||
|
||||
value = data.get(self._data_key)
|
||||
if value is None:
|
||||
return False
|
||||
|
||||
if value == '1':
|
||||
if self._state:
|
||||
return False
|
||||
self._state = True
|
||||
return True
|
||||
elif value == '0':
|
||||
if self._state:
|
||||
self._state = False
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class XiaomiMotionSensor(XiaomiBinarySensor):
|
||||
"""Representation of a XiaomiMotionSensor."""
|
||||
|
||||
def __init__(self, device, hass, xiaomi_hub):
|
||||
"""Initialize the XiaomiMotionSensor."""
|
||||
self._hass = hass
|
||||
self._data_key = 'status'
|
||||
self._no_motion_since = 0
|
||||
self._device_class = 'motion'
|
||||
XiaomiBinarySensor.__init__(self, device, 'Motion Sensor', xiaomi_hub)
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
attrs = {ATTR_NO_MOTION_SINCE: self._no_motion_since}
|
||||
attrs.update(super().device_state_attributes)
|
||||
return attrs
|
||||
|
||||
def parse_data(self, data):
|
||||
"""Parse data sent by gateway."""
|
||||
self._should_poll = False
|
||||
if NO_MOTION in data: # handle push from the hub
|
||||
self._no_motion_since = data[NO_MOTION]
|
||||
self._state = False
|
||||
return True
|
||||
|
||||
value = data.get(self._data_key)
|
||||
if value is None:
|
||||
return False
|
||||
|
||||
if value == MOTION:
|
||||
self._should_poll = True
|
||||
if self.entity_id is not None:
|
||||
self._hass.bus.fire('motion', {
|
||||
'entity_id': self.entity_id
|
||||
})
|
||||
|
||||
self._no_motion_since = 0
|
||||
if self._state:
|
||||
return False
|
||||
self._state = True
|
||||
return True
|
||||
elif value == NO_MOTION:
|
||||
if not self._state:
|
||||
return False
|
||||
self._state = False
|
||||
return True
|
||||
|
||||
|
||||
class XiaomiDoorSensor(XiaomiBinarySensor):
|
||||
"""Representation of a XiaomiDoorSensor."""
|
||||
|
||||
def __init__(self, device, xiaomi_hub):
|
||||
"""Initialize the XiaomiDoorSensor."""
|
||||
self._data_key = 'status'
|
||||
self._open_since = 0
|
||||
self._device_class = 'opening'
|
||||
XiaomiBinarySensor.__init__(self, device, 'Door Window Sensor',
|
||||
xiaomi_hub)
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
attrs = {ATTR_OPEN_SINCE: self._open_since}
|
||||
attrs.update(super().device_state_attributes)
|
||||
return attrs
|
||||
|
||||
def parse_data(self, data):
|
||||
"""Parse data sent by gateway."""
|
||||
self._should_poll = False
|
||||
if NO_CLOSE in data: # handle push from the hub
|
||||
self._open_since = data[NO_CLOSE]
|
||||
return True
|
||||
|
||||
value = data.get(self._data_key)
|
||||
if value is None:
|
||||
return False
|
||||
|
||||
if value == 'open':
|
||||
self._should_poll = True
|
||||
if self._state:
|
||||
return False
|
||||
self._state = True
|
||||
return True
|
||||
elif value == 'close':
|
||||
self._open_since = 0
|
||||
if self._state:
|
||||
self._state = False
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class XiaomiSmokeSensor(XiaomiBinarySensor):
|
||||
"""Representation of a XiaomiSmokeSensor."""
|
||||
|
||||
def __init__(self, device, xiaomi_hub):
|
||||
"""Initialize the XiaomiSmokeSensor."""
|
||||
self._data_key = 'alarm'
|
||||
self._density = 0
|
||||
self._device_class = 'smoke'
|
||||
XiaomiBinarySensor.__init__(self, device, 'Smoke Sensor', xiaomi_hub)
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
attrs = {ATTR_DENSITY: self._density}
|
||||
attrs.update(super().device_state_attributes)
|
||||
return attrs
|
||||
|
||||
def parse_data(self, data):
|
||||
"""Parse data sent by gateway."""
|
||||
if DENSITY in data:
|
||||
self._density = int(data.get(DENSITY))
|
||||
value = data.get(self._data_key)
|
||||
if value is None:
|
||||
return False
|
||||
|
||||
if value == '1':
|
||||
if self._state:
|
||||
return False
|
||||
self._state = True
|
||||
return True
|
||||
elif value == '0':
|
||||
if self._state:
|
||||
self._state = False
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class XiaomiButton(XiaomiBinarySensor):
|
||||
"""Representation of a Xiaomi Button."""
|
||||
|
||||
def __init__(self, device, name, data_key, hass, xiaomi_hub):
|
||||
"""Initialize the XiaomiButton."""
|
||||
self._hass = hass
|
||||
self._data_key = data_key
|
||||
XiaomiBinarySensor.__init__(self, device, name, xiaomi_hub)
|
||||
|
||||
def parse_data(self, data):
|
||||
"""Parse data sent by gateway."""
|
||||
value = data.get(self._data_key)
|
||||
if value is None:
|
||||
return False
|
||||
|
||||
if value == 'long_click_press':
|
||||
self._state = True
|
||||
click_type = 'long_click_press'
|
||||
elif value == 'long_click_release':
|
||||
self._state = False
|
||||
click_type = 'hold'
|
||||
elif value == 'click':
|
||||
click_type = 'single'
|
||||
elif value == 'double_click':
|
||||
click_type = 'double'
|
||||
elif value == 'both_click':
|
||||
click_type = 'both'
|
||||
else:
|
||||
return False
|
||||
|
||||
self._hass.bus.fire('click', {
|
||||
'entity_id': self.entity_id,
|
||||
'click_type': click_type
|
||||
})
|
||||
if value in ['long_click_press', 'long_click_release']:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class XiaomiCube(XiaomiBinarySensor):
|
||||
"""Representation of a Xiaomi Cube."""
|
||||
|
||||
def __init__(self, device, hass, xiaomi_hub):
|
||||
"""Initialize the Xiaomi Cube."""
|
||||
self._hass = hass
|
||||
self._state = False
|
||||
XiaomiBinarySensor.__init__(self, device, 'Cube', xiaomi_hub)
|
||||
|
||||
def parse_data(self, data):
|
||||
"""Parse data sent by gateway."""
|
||||
if 'status' in data:
|
||||
self._hass.bus.fire('cube_action', {
|
||||
'entity_id': self.entity_id,
|
||||
'action_type': data['status']
|
||||
})
|
||||
|
||||
if 'rotate' in data:
|
||||
self._hass.bus.fire('cube_action', {
|
||||
'entity_id': self.entity_id,
|
||||
'action_type': 'rotate',
|
||||
'action_value': float(data['rotate'].replace(",", "."))
|
||||
})
|
||||
return False
|
||||
66
homeassistant/components/cover/xiaomi.py
Normal file
66
homeassistant/components/cover/xiaomi.py
Normal file
|
|
@ -0,0 +1,66 @@
|
|||
"""Support for Xiaomi curtain."""
|
||||
import logging
|
||||
|
||||
from homeassistant.components.cover import CoverDevice
|
||||
from homeassistant.components.xiaomi import (PY_XIAOMI_GATEWAY, XiaomiDevice)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ATTR_CURTAIN_LEVEL = 'curtain_level'
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Perform the setup for Xiaomi devices."""
|
||||
devices = []
|
||||
for (_, gateway) in hass.data[PY_XIAOMI_GATEWAY].gateways.items():
|
||||
for device in gateway.devices['cover']:
|
||||
model = device['model']
|
||||
if model == 'curtain':
|
||||
devices.append(XiaomiGenericCover(device, "Curtain",
|
||||
{'status': 'status',
|
||||
'pos': 'curtain_level'},
|
||||
gateway))
|
||||
add_devices(devices)
|
||||
|
||||
|
||||
class XiaomiGenericCover(XiaomiDevice, CoverDevice):
|
||||
"""Representation of a XiaomiPlug."""
|
||||
|
||||
def __init__(self, device, name, data_key, xiaomi_hub):
|
||||
"""Initialize the XiaomiPlug."""
|
||||
self._data_key = data_key
|
||||
self._pos = 0
|
||||
XiaomiDevice.__init__(self, device, name, xiaomi_hub)
|
||||
|
||||
@property
|
||||
def current_cover_position(self):
|
||||
"""Return the current position of the cover."""
|
||||
return self._pos
|
||||
|
||||
@property
|
||||
def is_closed(self):
|
||||
"""Return if the cover is closed."""
|
||||
return self.current_cover_position < 0
|
||||
|
||||
def close_cover(self, **kwargs):
|
||||
"""Close the cover."""
|
||||
self._write_to_hub(self._sid, self._data_key['status'], 'close')
|
||||
|
||||
def open_cover(self, **kwargs):
|
||||
"""Open the cover."""
|
||||
self._write_to_hub(self._sid, self._data_key['status'], 'open')
|
||||
|
||||
def stop_cover(self, **kwargs):
|
||||
"""Stop the cover."""
|
||||
self._write_to_hub(self._sid, self._data_key['status'], 'stop')
|
||||
|
||||
def set_cover_position(self, position, **kwargs):
|
||||
"""Move the cover to a specific position."""
|
||||
self._write_to_hub(self._sid, self._data_key['pos'], str(position))
|
||||
|
||||
def parse_data(self, data):
|
||||
"""Parse data sent by gateway."""
|
||||
if ATTR_CURTAIN_LEVEL in data:
|
||||
self._pos = int(data[ATTR_CURTAIN_LEVEL])
|
||||
return True
|
||||
return False
|
||||
103
homeassistant/components/light/xiaomi.py
Executable file
103
homeassistant/components/light/xiaomi.py
Executable file
|
|
@ -0,0 +1,103 @@
|
|||
"""Support for Xiaomi Gateway Light."""
|
||||
import logging
|
||||
import struct
|
||||
import binascii
|
||||
from homeassistant.components.xiaomi import (PY_XIAOMI_GATEWAY, XiaomiDevice)
|
||||
from homeassistant.components.light import (ATTR_BRIGHTNESS, ATTR_RGB_COLOR,
|
||||
SUPPORT_BRIGHTNESS,
|
||||
SUPPORT_RGB_COLOR, Light)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Perform the setup for Xiaomi devices."""
|
||||
devices = []
|
||||
for (_, gateway) in hass.data[PY_XIAOMI_GATEWAY].gateways.items():
|
||||
for device in gateway.devices['light']:
|
||||
model = device['model']
|
||||
if model == 'gateway':
|
||||
devices.append(XiaomiGatewayLight(device, 'Gateway Light',
|
||||
gateway))
|
||||
add_devices(devices)
|
||||
|
||||
|
||||
class XiaomiGatewayLight(XiaomiDevice, Light):
|
||||
"""Representation of a XiaomiGatewayLight."""
|
||||
|
||||
def __init__(self, device, name, xiaomi_hub):
|
||||
"""Initialize the XiaomiGatewayLight."""
|
||||
self._data_key = 'rgb'
|
||||
self._rgb = (255, 255, 255)
|
||||
self._brightness = 180
|
||||
|
||||
XiaomiDevice.__init__(self, device, name, xiaomi_hub)
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Return true if it is on."""
|
||||
return self._state
|
||||
|
||||
def parse_data(self, data):
|
||||
"""Parse data sent by gateway."""
|
||||
value = data.get(self._data_key)
|
||||
if value is None:
|
||||
return False
|
||||
|
||||
if value == 0:
|
||||
if self._state:
|
||||
self._state = False
|
||||
return True
|
||||
|
||||
rgbhexstr = "%x" % value
|
||||
if len(rgbhexstr) == 7:
|
||||
rgbhexstr = '0' + rgbhexstr
|
||||
elif len(rgbhexstr) != 8:
|
||||
_LOGGER.error('Light RGB data error.'
|
||||
' Must be 8 characters. Received: %s', rgbhexstr)
|
||||
return False
|
||||
|
||||
rgbhex = bytes.fromhex(rgbhexstr)
|
||||
rgba = struct.unpack('BBBB', rgbhex)
|
||||
brightness = rgba[0]
|
||||
rgb = rgba[1:]
|
||||
|
||||
self._brightness = int(255 * brightness / 100)
|
||||
self._rgb = rgb
|
||||
self._state = True
|
||||
return True
|
||||
|
||||
@property
|
||||
def brightness(self):
|
||||
"""Return the brightness of this light between 0..255."""
|
||||
return self._brightness
|
||||
|
||||
@property
|
||||
def rgb_color(self):
|
||||
"""Return the RBG color value."""
|
||||
return self._rgb
|
||||
|
||||
@property
|
||||
def supported_features(self):
|
||||
"""Return the supported features."""
|
||||
return SUPPORT_BRIGHTNESS | SUPPORT_RGB_COLOR
|
||||
|
||||
def turn_on(self, **kwargs):
|
||||
"""Turn the light on."""
|
||||
if ATTR_RGB_COLOR in kwargs:
|
||||
self._rgb = kwargs[ATTR_RGB_COLOR]
|
||||
|
||||
if ATTR_BRIGHTNESS in kwargs:
|
||||
self._brightness = int(100 * kwargs[ATTR_BRIGHTNESS] / 255)
|
||||
|
||||
rgba = (self._brightness,) + self._rgb
|
||||
rgbhex = binascii.hexlify(struct.pack('BBBB', *rgba)).decode("ASCII")
|
||||
rgbhex = int(rgbhex, 16)
|
||||
|
||||
if self._write_to_hub(self._sid, **{self._data_key: rgbhex}):
|
||||
self._state = True
|
||||
|
||||
def turn_off(self, **kwargs):
|
||||
"""Turn the light off."""
|
||||
if self._write_to_hub(self._sid, **{self._data_key: 0}):
|
||||
self._state = False
|
||||
77
homeassistant/components/sensor/xiaomi.py
Normal file
77
homeassistant/components/sensor/xiaomi.py
Normal file
|
|
@ -0,0 +1,77 @@
|
|||
"""Support for Xiaomi sensors."""
|
||||
import logging
|
||||
|
||||
from homeassistant.components.xiaomi import (PY_XIAOMI_GATEWAY, XiaomiDevice)
|
||||
from homeassistant.const import TEMP_CELSIUS
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Perform the setup for Xiaomi devices."""
|
||||
devices = []
|
||||
for (_, gateway) in hass.data[PY_XIAOMI_GATEWAY].gateways.items():
|
||||
for device in gateway.devices['sensor']:
|
||||
if device['model'] == 'sensor_ht':
|
||||
devices.append(XiaomiSensor(device, 'Temperature',
|
||||
'temperature', gateway))
|
||||
devices.append(XiaomiSensor(device, 'Humidity',
|
||||
'humidity', gateway))
|
||||
if device['model'] == 'weather.v1':
|
||||
devices.append(XiaomiSensor(device, 'Temperature',
|
||||
'temperature', gateway))
|
||||
devices.append(XiaomiSensor(device, 'Humidity',
|
||||
'humidity', gateway))
|
||||
devices.append(XiaomiSensor(device, 'Pressure',
|
||||
'pressure', gateway))
|
||||
elif device['model'] == 'gateway':
|
||||
devices.append(XiaomiSensor(device, 'Illumination',
|
||||
'illumination', gateway))
|
||||
add_devices(devices)
|
||||
|
||||
|
||||
class XiaomiSensor(XiaomiDevice):
|
||||
"""Representation of a XiaomiSensor."""
|
||||
|
||||
def __init__(self, device, name, data_key, xiaomi_hub):
|
||||
"""Initialize the XiaomiSensor."""
|
||||
self._data_key = data_key
|
||||
XiaomiDevice.__init__(self, device, name, xiaomi_hub)
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self):
|
||||
"""Return the unit of measurement of this entity, if any."""
|
||||
if self._data_key == 'temperature':
|
||||
return TEMP_CELSIUS
|
||||
elif self._data_key == 'humidity':
|
||||
return '%'
|
||||
elif self._data_key == 'illumination':
|
||||
return 'lm'
|
||||
elif self._data_key == 'pressure':
|
||||
return 'hPa'
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return the state of the sensor."""
|
||||
return self._state
|
||||
|
||||
def parse_data(self, data):
|
||||
"""Parse data sent by gateway."""
|
||||
value = data.get(self._data_key)
|
||||
if value is None:
|
||||
return False
|
||||
value = float(value)
|
||||
if self._data_key == 'temperature' and value == 10000:
|
||||
return False
|
||||
elif self._data_key == 'humidity' and value == 0:
|
||||
return False
|
||||
elif self._data_key == 'illumination' and value == 0:
|
||||
return False
|
||||
elif self._data_key == 'pressure' and value == 0:
|
||||
return False
|
||||
if self._data_key in ['temperature', 'humidity']:
|
||||
value /= 100
|
||||
elif self._data_key in ['illumination']:
|
||||
value = max(value - 300, 0)
|
||||
self._state = round(value, 2)
|
||||
return True
|
||||
111
homeassistant/components/switch/xiaomi.py
Normal file
111
homeassistant/components/switch/xiaomi.py
Normal file
|
|
@ -0,0 +1,111 @@
|
|||
"""Support for Xiaomi binary sensors."""
|
||||
import logging
|
||||
|
||||
from homeassistant.components.switch import SwitchDevice
|
||||
from homeassistant.components.xiaomi import (PY_XIAOMI_GATEWAY, XiaomiDevice)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ATTR_LOAD_POWER = 'Load power' # Load power in watts (W)
|
||||
ATTR_POWER_CONSUMED = 'Power consumed'
|
||||
ATTR_IN_USE = 'In use'
|
||||
LOAD_POWER = 'load_power'
|
||||
POWER_CONSUMED = 'power_consumed'
|
||||
IN_USE = 'inuse'
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Perform the setup for Xiaomi devices."""
|
||||
devices = []
|
||||
for (_, gateway) in hass.data[PY_XIAOMI_GATEWAY].gateways.items():
|
||||
for device in gateway.devices['switch']:
|
||||
model = device['model']
|
||||
if model == 'plug':
|
||||
devices.append(XiaomiGenericSwitch(device, "Plug", 'status',
|
||||
True, gateway))
|
||||
elif model == 'ctrl_neutral1':
|
||||
devices.append(XiaomiGenericSwitch(device, 'Wall Switch',
|
||||
'channel_0',
|
||||
False, gateway))
|
||||
elif model == 'ctrl_neutral2':
|
||||
devices.append(XiaomiGenericSwitch(device, 'Wall Switch Left',
|
||||
'channel_0',
|
||||
False, gateway))
|
||||
devices.append(XiaomiGenericSwitch(device, 'Wall Switch Right',
|
||||
'channel_1',
|
||||
False, gateway))
|
||||
elif model == '86plug':
|
||||
devices.append(XiaomiGenericSwitch(device, 'Wall Plug',
|
||||
'status', True, gateway))
|
||||
add_devices(devices)
|
||||
|
||||
|
||||
class XiaomiGenericSwitch(XiaomiDevice, SwitchDevice):
|
||||
"""Representation of a XiaomiPlug."""
|
||||
|
||||
def __init__(self, device, name, data_key, supports_power_consumption,
|
||||
xiaomi_hub):
|
||||
"""Initialize the XiaomiPlug."""
|
||||
self._data_key = data_key
|
||||
self._in_use = None
|
||||
self._load_power = None
|
||||
self._power_consumed = None
|
||||
self._supports_power_consumption = supports_power_consumption
|
||||
XiaomiDevice.__init__(self, device, name, xiaomi_hub)
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
"""Return the icon to use in the frontend, if any."""
|
||||
if self._data_key == 'status':
|
||||
return 'mdi:power-plug'
|
||||
return 'mdi:power-socket'
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Return true if it is on."""
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
if self._supports_power_consumption:
|
||||
attrs = {ATTR_IN_USE: self._in_use,
|
||||
ATTR_LOAD_POWER: self._load_power,
|
||||
ATTR_POWER_CONSUMED: self._power_consumed}
|
||||
else:
|
||||
attrs = {}
|
||||
attrs.update(super().device_state_attributes)
|
||||
return attrs
|
||||
|
||||
def turn_on(self, **kwargs):
|
||||
"""Turn the switch on."""
|
||||
if self._write_to_hub(self._sid, **{self._data_key: 'on'}):
|
||||
self._state = True
|
||||
self.schedule_update_ha_state()
|
||||
|
||||
def turn_off(self):
|
||||
"""Turn the switch off."""
|
||||
if self._write_to_hub(self._sid, **{self._data_key: 'off'}):
|
||||
self._state = False
|
||||
self.schedule_update_ha_state()
|
||||
|
||||
def parse_data(self, data):
|
||||
"""Parse data sent by gateway."""
|
||||
if IN_USE in data:
|
||||
self._in_use = int(data[IN_USE])
|
||||
if not self._in_use:
|
||||
self._load_power = 0
|
||||
if POWER_CONSUMED in data:
|
||||
self._power_consumed = round(float(data[POWER_CONSUMED]), 2)
|
||||
if LOAD_POWER in data:
|
||||
self._load_power = round(float(data[LOAD_POWER]), 2)
|
||||
|
||||
value = data.get(self._data_key)
|
||||
if value is None:
|
||||
return False
|
||||
|
||||
state = value == 'on'
|
||||
if self._state == state:
|
||||
return False
|
||||
self._state = state
|
||||
return True
|
||||
195
homeassistant/components/xiaomi.py
Normal file
195
homeassistant/components/xiaomi.py
Normal file
|
|
@ -0,0 +1,195 @@
|
|||
"""Support for Xiaomi Gateways."""
|
||||
import logging
|
||||
import voluptuous as vol
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers import discovery
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.const import (ATTR_BATTERY_LEVEL, EVENT_HOMEASSISTANT_STOP,
|
||||
CONF_MAC)
|
||||
|
||||
|
||||
REQUIREMENTS = ['https://github.com/Danielhiversen/PyXiaomiGateway/archive/'
|
||||
'877faec36e1bfa4177cae2a0d4f49570af083e1d.zip#'
|
||||
'PyXiaomiGateway==0.1.0']
|
||||
|
||||
ATTR_GW_SID = 'gw_sid'
|
||||
ATTR_RINGTONE_ID = 'ringtone_id'
|
||||
ATTR_RINGTONE_VOL = 'ringtone_vol'
|
||||
CONF_DISCOVERY_RETRY = 'discovery_retry'
|
||||
CONF_GATEWAYS = 'gateways'
|
||||
CONF_INTERFACE = 'interface'
|
||||
DOMAIN = 'xiaomi'
|
||||
PY_XIAOMI_GATEWAY = "xiaomi_gw"
|
||||
|
||||
|
||||
def _validate_conf(config):
|
||||
"""Validate a list of devices definitions."""
|
||||
res_config = []
|
||||
for gw_conf in config:
|
||||
res_gw_conf = {'sid': gw_conf.get(CONF_MAC)}
|
||||
if res_gw_conf['sid'] is not None:
|
||||
res_gw_conf['sid'] = res_gw_conf['sid'].replace(":", "").lower()
|
||||
if len(res_gw_conf['sid']) != 12:
|
||||
raise vol.Invalid('Invalid mac address', gw_conf.get(CONF_MAC))
|
||||
key = gw_conf.get('key')
|
||||
if key is None:
|
||||
_LOGGER.warning(
|
||||
'Gateway Key is not provided.'
|
||||
' Controlling gateway device will not be possible.')
|
||||
elif len(key) != 16:
|
||||
raise vol.Invalid('Invalid key %s.'
|
||||
' Key must be 16 characters', key)
|
||||
res_gw_conf['key'] = key
|
||||
res_config.append(res_gw_conf)
|
||||
return res_config
|
||||
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema({
|
||||
DOMAIN: vol.Schema({
|
||||
vol.Optional(CONF_GATEWAYS, default=[{CONF_MAC: None, "key": None}]):
|
||||
vol.All(cv.ensure_list, _validate_conf),
|
||||
vol.Optional(CONF_INTERFACE, default='any'): cv.string,
|
||||
vol.Optional(CONF_DISCOVERY_RETRY, default=3): cv.positive_int
|
||||
})
|
||||
}, extra=vol.ALLOW_EXTRA)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def setup(hass, config):
|
||||
"""Set up the Xiaomi component."""
|
||||
gateways = config[DOMAIN][CONF_GATEWAYS]
|
||||
interface = config[DOMAIN][CONF_INTERFACE]
|
||||
discovery_retry = config[DOMAIN][CONF_DISCOVERY_RETRY]
|
||||
|
||||
from PyXiaomiGateway import PyXiaomiGateway
|
||||
hass.data[PY_XIAOMI_GATEWAY] = PyXiaomiGateway(hass.add_job, gateways,
|
||||
interface)
|
||||
|
||||
_LOGGER.debug("Expecting %s gateways", len(gateways))
|
||||
for _ in range(discovery_retry):
|
||||
_LOGGER.info('Discovering Xiaomi Gateways (Try %s)', _ + 1)
|
||||
hass.data[PY_XIAOMI_GATEWAY].discover_gateways()
|
||||
if len(hass.data[PY_XIAOMI_GATEWAY].gateways) >= len(gateways):
|
||||
break
|
||||
|
||||
if not hass.data[PY_XIAOMI_GATEWAY].gateways:
|
||||
_LOGGER.error("No gateway discovered")
|
||||
return False
|
||||
hass.data[PY_XIAOMI_GATEWAY].listen()
|
||||
_LOGGER.debug("Listening for broadcast")
|
||||
|
||||
for component in ['binary_sensor', 'sensor', 'switch', 'light', 'cover']:
|
||||
discovery.load_platform(hass, component, DOMAIN, {}, config)
|
||||
|
||||
def stop_xiaomi(event):
|
||||
"""Stop Xiaomi Socket."""
|
||||
_LOGGER.info("Shutting down Xiaomi Hub.")
|
||||
hass.data[PY_XIAOMI_GATEWAY].stop_listen()
|
||||
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, stop_xiaomi)
|
||||
|
||||
def play_ringtone_service(call):
|
||||
"""Service to play ringtone through Gateway."""
|
||||
if call.data.get(ATTR_RINGTONE_ID) is None \
|
||||
or call.data.get(ATTR_GW_SID) is None:
|
||||
_LOGGER.error("Mandatory parameters is not specified.")
|
||||
return
|
||||
|
||||
ring_id = int(call.data.get(ATTR_RINGTONE_ID))
|
||||
if ring_id in [9, 14-19]:
|
||||
_LOGGER.error('Specified mid: %s is not defined in gateway.',
|
||||
ring_id)
|
||||
return
|
||||
|
||||
ring_vol = call.data.get(ATTR_RINGTONE_VOL)
|
||||
if ring_vol is None:
|
||||
ringtone = {'mid': ring_id}
|
||||
else:
|
||||
ringtone = {'mid': ring_id, 'vol': int(ring_vol)}
|
||||
|
||||
gw_sid = call.data.get(ATTR_GW_SID)
|
||||
|
||||
for (_, gateway) in hass.data[PY_XIAOMI_GATEWAY].gateways.items():
|
||||
if gateway.sid == gw_sid:
|
||||
gateway.write_to_hub(gateway.sid, **ringtone)
|
||||
break
|
||||
else:
|
||||
_LOGGER.error('Unknown gateway sid: %s was specified.', gw_sid)
|
||||
|
||||
def stop_ringtone_service(call):
|
||||
"""Service to stop playing ringtone on Gateway."""
|
||||
gw_sid = call.data.get(ATTR_GW_SID)
|
||||
if gw_sid is None:
|
||||
_LOGGER.error("Mandatory parameter (%s) is not specified.",
|
||||
ATTR_GW_SID)
|
||||
return
|
||||
|
||||
for (_, gateway) in hass.data[PY_XIAOMI_GATEWAY].gateways.items():
|
||||
if gateway.sid == gw_sid:
|
||||
ringtone = {'mid': 10000}
|
||||
gateway.write_to_hub(gateway.sid, **ringtone)
|
||||
break
|
||||
else:
|
||||
_LOGGER.error('Unknown gateway sid: %s was specified.', gw_sid)
|
||||
|
||||
hass.services.async_register(DOMAIN, 'play_ringtone',
|
||||
play_ringtone_service,
|
||||
description=None, schema=None)
|
||||
hass.services.async_register(DOMAIN, 'stop_ringtone',
|
||||
stop_ringtone_service,
|
||||
description=None, schema=None)
|
||||
return True
|
||||
|
||||
|
||||
class XiaomiDevice(Entity):
|
||||
"""Representation a base Xiaomi device."""
|
||||
|
||||
def __init__(self, device, name, xiaomi_hub):
|
||||
"""Initialize the xiaomi device."""
|
||||
self._state = None
|
||||
self._sid = device['sid']
|
||||
self._name = '{}_{}'.format(name, self._sid)
|
||||
self._write_to_hub = xiaomi_hub.write_to_hub
|
||||
self._get_from_hub = xiaomi_hub.get_from_hub
|
||||
xiaomi_hub.callbacks[self._sid].append(self.push_data)
|
||||
self._device_state_attributes = {}
|
||||
self.parse_data(device['data'])
|
||||
self.parse_voltage(device['data'])
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the device."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
"""Poll update device status."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
return self._device_state_attributes
|
||||
|
||||
def push_data(self, data):
|
||||
"""Push from Hub."""
|
||||
_LOGGER.debug("PUSH >> %s: %s", self, data)
|
||||
if self.parse_data(data) or self.parse_voltage(data):
|
||||
self.schedule_update_ha_state()
|
||||
|
||||
def parse_voltage(self, data):
|
||||
"""Parse battery level data sent by gateway."""
|
||||
if 'voltage' not in data:
|
||||
return False
|
||||
max_volt = 3300
|
||||
min_volt = 2800
|
||||
voltage = data['voltage']
|
||||
voltage = min(voltage, max_volt)
|
||||
voltage = max(voltage, min_volt)
|
||||
percent = ((voltage - min_volt) / (max_volt - min_volt)) * 100
|
||||
self._device_state_attributes[ATTR_BATTERY_LEVEL] = round(percent, 1)
|
||||
return True
|
||||
|
||||
def parse_data(self, data):
|
||||
"""Parse data sent by gateway."""
|
||||
raise NotImplementedError()
|
||||
|
|
@ -276,6 +276,9 @@ holidays==0.8.1
|
|||
# homeassistant.components.camera.onvif
|
||||
http://github.com/tgaugry/suds-passworddigest-py3/archive/86fc50e39b4d2b8997481967d6a7fe1c57118999.zip#suds-passworddigest-py3==0.1.2a
|
||||
|
||||
# homeassistant.components.xiaomi
|
||||
https://github.com/Danielhiversen/PyXiaomiGateway/archive/877faec36e1bfa4177cae2a0d4f49570af083e1d.zip#PyXiaomiGateway==0.1.0
|
||||
|
||||
# homeassistant.components.sensor.dht
|
||||
# https://github.com/adafruit/Adafruit_Python_DHT/archive/da8cddf7fb629c1ef4f046ca44f42523c9cf2d11.zip#Adafruit_DHT==1.3.0
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue