mirror of
https://github.com/python-kasa/python-kasa.git
synced 2024-12-22 11:13:34 +00:00
Add support for TP-Link smartbulbs (#30)
* Add support for new-style protocol Newer devices (including my LB130) seem to include the request length in the previously empty message header, and ignore requests that lack it. They also don't send an empty packet as the final part of a response, which can lead to hangs. Add support for this, with luck not breaking existing devices in the process. * Fix tests We now include the request length in the encrypted packet header, so strip the header rather than assuming that it's just zeroes. * Create a SmartDevice parent class Add a generic SmartDevice class that SmartPlug can inherit from, in preparation for adding support for other device types. * Add support for TP-Link smartbulbs These bulbs use the same protocol as the smart plugs, but have additional commands for controlling bulb-specific features. In addition, the bulbs have their emeter under a different target and return responses that include the energy unit in the key names. * Add tests for bulbs Not entirely comprehensive, but has pretty much the same level of testing as plugs
This commit is contained in:
Matthew Garrett
GadgetReactor
parent
04185706f8
commit
2d6376b597
@ -1,3 +1,3 @@
|
||||
from __future__ import absolute_import
|
||||
from __future__ import unicode_literals
|
||||
from pyHS100.pyHS100 import SmartPlug, TPLinkSmartHomeProtocol, SmartPlugException
|
||||
from pyHS100.pyHS100 import SmartBulb, SmartPlug, TPLinkSmartHomeProtocol, SmartPlugException
|
||||
|
||||
@ -3,6 +3,7 @@ from __future__ import unicode_literals
|
||||
|
||||
import json
|
||||
import socket
|
||||
import struct
|
||||
import logging
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
@ -49,10 +50,16 @@ class TPLinkSmartHomeProtocol:
|
||||
sock.send(TPLinkSmartHomeProtocol.encrypt(request))
|
||||
|
||||
buffer = bytes()
|
||||
# Some devices send responses with a length header of 0 and
|
||||
# terminate with a zero size chunk. Others send the length and
|
||||
# will hang if we attempt to read more data.
|
||||
length = -1
|
||||
while True:
|
||||
chunk = sock.recv(4096)
|
||||
if length == -1:
|
||||
length = struct.unpack(">I", chunk[0:4])[0]
|
||||
buffer += chunk
|
||||
if not chunk:
|
||||
if (length > 0 and len(buffer) >= length + 4) or not chunk:
|
||||
break
|
||||
|
||||
finally:
|
||||
@ -115,7 +122,7 @@ class TPLinkSmartHomeProtocol:
|
||||
:return: ciphertext request
|
||||
"""
|
||||
key = TPLinkSmartHomeProtocol.INITIALIZATION_VECTOR
|
||||
buffer = bytearray(4) # 4 nullbytes
|
||||
buffer = bytearray(struct.pack(">I", len(request)))
|
||||
|
||||
for char in request:
|
||||
cipher = key ^ ord(char)
|
||||
|
||||
@ -38,39 +38,10 @@ class SmartPlugException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class SmartPlug(object):
|
||||
"""Representation of a TP-Link Smart Switch.
|
||||
|
||||
Usage example when used as library:
|
||||
p = SmartPlug("192.168.1.105")
|
||||
# print the devices alias
|
||||
print(p.alias)
|
||||
# change state of plug
|
||||
p.state = "ON"
|
||||
p.state = "OFF"
|
||||
# query and print current state of plug
|
||||
print(p.state)
|
||||
|
||||
Errors reported by the device are raised as SmartPlugExceptions,
|
||||
and should be handled by the user of the library.
|
||||
|
||||
Note:
|
||||
The library references the same structure as defined for the D-Link Switch
|
||||
"""
|
||||
# switch states
|
||||
SWITCH_STATE_ON = 'ON'
|
||||
SWITCH_STATE_OFF = 'OFF'
|
||||
SWITCH_STATE_UNKNOWN = 'UNKNOWN'
|
||||
|
||||
# possible device features
|
||||
FEATURE_ENERGY_METER = 'ENE'
|
||||
FEATURE_TIMER = 'TIM'
|
||||
|
||||
ALL_FEATURES = (FEATURE_ENERGY_METER, FEATURE_TIMER)
|
||||
|
||||
class SmartDevice(object):
|
||||
def __init__(self, ip_address, protocol=None):
|
||||
"""
|
||||
Create a new SmartPlug instance, identified through its IP address.
|
||||
Create a new SmartDevice instance, identified through its IP address.
|
||||
|
||||
:param str ip_address: ip address on which the device listens
|
||||
:raises SmartPlugException: when unable to communicate with the device
|
||||
@ -118,48 +89,6 @@ class SmartPlug(object):
|
||||
# TODO use volyptuous
|
||||
return self.get_sysinfo()
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""
|
||||
Retrieve the switch state
|
||||
|
||||
:returns: one of
|
||||
SWITCH_STATE_ON
|
||||
SWITCH_STATE_OFF
|
||||
SWITCH_STATE_UNKNOWN
|
||||
:rtype: str
|
||||
"""
|
||||
relay_state = self.sys_info['relay_state']
|
||||
|
||||
if relay_state == 0:
|
||||
return SmartPlug.SWITCH_STATE_OFF
|
||||
elif relay_state == 1:
|
||||
return SmartPlug.SWITCH_STATE_ON
|
||||
else:
|
||||
_LOGGER.warning("Unknown state %s returned.", relay_state)
|
||||
return SmartPlug.SWITCH_STATE_UNKNOWN
|
||||
|
||||
@state.setter
|
||||
def state(self, value):
|
||||
"""
|
||||
Set the new switch state
|
||||
|
||||
:param value: one of
|
||||
SWITCH_STATE_ON
|
||||
SWITCH_STATE_OFF
|
||||
:raises ValueError: on invalid state
|
||||
:raises SmartPlugException: on error
|
||||
|
||||
"""
|
||||
if not isinstance(value, basestring):
|
||||
raise ValueError("State must be str, not of %s.", type(value))
|
||||
elif value.upper() == SmartPlug.SWITCH_STATE_ON:
|
||||
self.turn_on()
|
||||
elif value.upper() == SmartPlug.SWITCH_STATE_OFF:
|
||||
self.turn_off()
|
||||
else:
|
||||
raise ValueError("State %s is not valid.", value)
|
||||
|
||||
def get_sysinfo(self):
|
||||
"""
|
||||
Retrieve system information.
|
||||
@ -170,142 +99,6 @@ class SmartPlug(object):
|
||||
"""
|
||||
return self._query_helper("system", "get_sysinfo")
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""
|
||||
Returns whether device is on.
|
||||
|
||||
:return: True if device is on, False otherwise
|
||||
"""
|
||||
return bool(self.sys_info['relay_state'])
|
||||
|
||||
@property
|
||||
def is_off(self):
|
||||
"""
|
||||
Returns whether device is off.
|
||||
|
||||
:return: True if device is off, False otherwise.
|
||||
:rtype: bool
|
||||
"""
|
||||
return not self.is_on
|
||||
|
||||
def turn_on(self):
|
||||
"""
|
||||
Turn the switch on.
|
||||
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
self._query_helper("system", "set_relay_state", {"state": 1})
|
||||
|
||||
def turn_off(self):
|
||||
"""
|
||||
Turn the switch off.
|
||||
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
self._query_helper("system", "set_relay_state", {"state": 0})
|
||||
|
||||
@property
|
||||
def has_emeter(self):
|
||||
"""
|
||||
Checks feature list for energey meter support.
|
||||
|
||||
:return: True if energey meter is available
|
||||
False if energymeter is missing
|
||||
"""
|
||||
return SmartPlug.FEATURE_ENERGY_METER in self.features
|
||||
|
||||
def get_emeter_realtime(self):
|
||||
"""
|
||||
Retrive current energy readings from device.
|
||||
|
||||
:returns: current readings or False
|
||||
:rtype: dict, False
|
||||
False if device has no energy meter or error occured
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
if not self.has_emeter:
|
||||
return False
|
||||
|
||||
return self._query_helper("emeter", "get_realtime")
|
||||
|
||||
def get_emeter_daily(self, year=None, month=None):
|
||||
"""
|
||||
Retrieve daily statistics for a given month
|
||||
|
||||
:param year: year for which to retrieve statistics (default: this year)
|
||||
:param month: month for which to retrieve statistcs (default: this
|
||||
month)
|
||||
:return: mapping of day of month to value
|
||||
False if device has no energy meter or error occured
|
||||
:rtype: dict
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
if not self.has_emeter:
|
||||
return False
|
||||
|
||||
if year is None:
|
||||
year = datetime.datetime.now().year
|
||||
if month is None:
|
||||
month = datetime.datetime.now().month
|
||||
|
||||
response = self._query_helper("emeter", "get_daystat",
|
||||
{'month': month, 'year': year})
|
||||
|
||||
return {entry['day']: entry['energy']
|
||||
for entry in response['day_list']}
|
||||
|
||||
def get_emeter_monthly(self, year=datetime.datetime.now().year):
|
||||
"""
|
||||
Retrieve monthly statistics for a given year.
|
||||
|
||||
:param year: year for which to retrieve statistics (default: this year)
|
||||
:return: dict: mapping of month to value
|
||||
False if device has no energy meter
|
||||
:rtype: dict
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
if not self.has_emeter:
|
||||
return False
|
||||
|
||||
response = self._query_helper("emeter", "get_monthstat",
|
||||
{'year': year})
|
||||
|
||||
return {entry['month']: entry['energy']
|
||||
for entry in response['month_list']}
|
||||
|
||||
def erase_emeter_stats(self):
|
||||
"""
|
||||
Erase energy meter statistics
|
||||
|
||||
:return: True if statistics were deleted
|
||||
False if device has no energy meter.
|
||||
:rtype: bool
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
if not self.has_emeter:
|
||||
return False
|
||||
|
||||
self._query_helper("emeter", "erase_emeter_stat", None)
|
||||
|
||||
# As query_helper raises exception in case of failure, we have succeeded when we are this far.
|
||||
return True
|
||||
|
||||
def current_consumption(self):
|
||||
"""
|
||||
Get the current power consumption in Watt.
|
||||
|
||||
:return: the current power consumption in Watt.
|
||||
False if device has no energy meter.
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
if not self.has_emeter:
|
||||
return False
|
||||
|
||||
response = self.get_emeter_realtime()
|
||||
|
||||
return response['power']
|
||||
|
||||
def identify(self):
|
||||
"""
|
||||
Query device information to identify model and featureset
|
||||
@ -331,23 +124,6 @@ class SmartPlug(object):
|
||||
"""
|
||||
return self.sys_info['model']
|
||||
|
||||
@property
|
||||
def features(self):
|
||||
"""
|
||||
Returns features of the devices
|
||||
|
||||
:return: list of features
|
||||
:rtype: list
|
||||
"""
|
||||
features = self.sys_info['feature'].split(':')
|
||||
|
||||
for feature in features:
|
||||
if feature not in SmartPlug.ALL_FEATURES:
|
||||
_LOGGER.warning("Unknown feature %s on device %s.",
|
||||
feature, self.model)
|
||||
|
||||
return features
|
||||
|
||||
@property
|
||||
def alias(self):
|
||||
"""
|
||||
@ -368,26 +144,6 @@ class SmartPlug(object):
|
||||
"""
|
||||
self._query_helper("system", "set_dev_alias", {"alias": alias})
|
||||
|
||||
@property
|
||||
def led(self):
|
||||
"""
|
||||
Returns the state of the led.
|
||||
|
||||
:return: True if led is on, False otherwise
|
||||
:rtype: bool
|
||||
"""
|
||||
return bool(1 - self.sys_info["led_off"])
|
||||
|
||||
@led.setter
|
||||
def led(self, state):
|
||||
"""
|
||||
Sets the state of the led (night mode)
|
||||
|
||||
:param bool state: True to set led on, False to set led off
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
self._query_helper("system", "set_led_off", {"off": int(not state)})
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
"""
|
||||
@ -482,17 +238,6 @@ class SmartPlug(object):
|
||||
info = self.sys_info
|
||||
return {key: info[key] for key in keys}
|
||||
|
||||
@property
|
||||
def on_since(self):
|
||||
"""
|
||||
Returns pretty-printed on-time
|
||||
|
||||
:return: datetime for on since
|
||||
:rtype: datetime
|
||||
"""
|
||||
return datetime.datetime.now() - \
|
||||
datetime.timedelta(seconds=self.sys_info["on_time"])
|
||||
|
||||
@property
|
||||
def location(self):
|
||||
"""
|
||||
@ -534,3 +279,510 @@ class SmartPlug(object):
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
self._query_helper("system", "set_mac_addr", {"mac": mac})
|
||||
|
||||
def get_emeter_realtime(self):
|
||||
"""
|
||||
Retrive current energy readings from device.
|
||||
|
||||
:returns: current readings or False
|
||||
:rtype: dict, False
|
||||
False if device has no energy meter or error occured
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
if not self.has_emeter:
|
||||
return False
|
||||
|
||||
return self._query_helper(self.emeter_type, "get_realtime")
|
||||
|
||||
def get_emeter_daily(self, year=None, month=None):
|
||||
"""
|
||||
Retrieve daily statistics for a given month
|
||||
|
||||
:param year: year for which to retrieve statistics (default: this year)
|
||||
:param month: month for which to retrieve statistcs (default: this
|
||||
month)
|
||||
:return: mapping of day of month to value
|
||||
False if device has no energy meter or error occured
|
||||
:rtype: dict
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
if not self.has_emeter:
|
||||
return False
|
||||
|
||||
if year is None:
|
||||
year = datetime.datetime.now().year
|
||||
if month is None:
|
||||
month = datetime.datetime.now().month
|
||||
|
||||
response = self._query_helper(self.emeter_type, "get_daystat",
|
||||
{'month': month, 'year': year})
|
||||
|
||||
if self.emeter_units:
|
||||
key = 'energy_wh'
|
||||
else:
|
||||
key = 'energy'
|
||||
|
||||
return {entry['day']: entry[key]
|
||||
for entry in response['day_list']}
|
||||
|
||||
def get_emeter_monthly(self, year=datetime.datetime.now().year):
|
||||
"""
|
||||
Retrieve monthly statistics for a given year.
|
||||
|
||||
:param year: year for which to retrieve statistics (default: this year)
|
||||
:return: dict: mapping of month to value
|
||||
False if device has no energy meter
|
||||
:rtype: dict
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
if not self.has_emeter:
|
||||
return False
|
||||
|
||||
response = self._query_helper(self.emeter_type, "get_monthstat",
|
||||
{'year': year})
|
||||
|
||||
if self.emeter_units:
|
||||
key = 'energy_wh'
|
||||
else:
|
||||
key = 'energy'
|
||||
|
||||
return {entry['month']: entry[key]
|
||||
for entry in response['month_list']}
|
||||
|
||||
def erase_emeter_stats(self):
|
||||
"""
|
||||
Erase energy meter statistics
|
||||
|
||||
:return: True if statistics were deleted
|
||||
False if device has no energy meter.
|
||||
:rtype: bool
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
if not self.has_emeter:
|
||||
return False
|
||||
|
||||
self._query_helper(self.emeter_type, "erase_emeter_stat", None)
|
||||
|
||||
# As query_helper raises exception in case of failure, we have
|
||||
# succeeded when we are this far.
|
||||
return True
|
||||
|
||||
def current_consumption(self):
|
||||
"""
|
||||
Get the current power consumption in Watt.
|
||||
|
||||
:return: the current power consumption in Watt.
|
||||
False if device has no energy meter.
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
if not self.has_emeter:
|
||||
return False
|
||||
|
||||
response = self.get_emeter_realtime()
|
||||
if self.emeter_units:
|
||||
return response['power_mw']
|
||||
else:
|
||||
return response['power']
|
||||
|
||||
|
||||
class SmartBulb(SmartDevice):
|
||||
"""Representation of a TP-Link Smart Bulb.
|
||||
|
||||
Usage example when used as library:
|
||||
p = SmartBulb("192.168.1.105")
|
||||
# print the devices alias
|
||||
print(p.alias)
|
||||
# change state of bulb
|
||||
p.state = "ON"
|
||||
p.state = "OFF"
|
||||
# query and print current state of plug
|
||||
print(p.state)
|
||||
# check whether the bulb supports color changes
|
||||
if p.is_color:
|
||||
# set the color to an HSV tuple
|
||||
p.hsv = (100, 0, 255)
|
||||
# get the current HSV value
|
||||
print(p.hsv)
|
||||
# check whether the bulb supports setting color temperature
|
||||
if p.is_variable_color_temp:
|
||||
# set the color temperature in Kelvin
|
||||
p.color_temp = 3000
|
||||
# get the current color temperature
|
||||
print(p.color_temp)
|
||||
# check whether the bulb is dimmable
|
||||
if p.is_dimmable:
|
||||
# set the bulb to 50% brightness
|
||||
p.brightness = 50
|
||||
# check the current brightness
|
||||
print(p.brightness)
|
||||
|
||||
Errors reported by the device are raised as SmartPlugExceptions,
|
||||
and should be handled by the user of the library.
|
||||
|
||||
"""
|
||||
# bulb states
|
||||
BULB_STATE_ON = 'ON'
|
||||
BULB_STATE_OFF = 'OFF'
|
||||
|
||||
def __init__(self, ip_address, protocol=None):
|
||||
SmartDevice.__init__(self, ip_address, protocol)
|
||||
self.emeter_type = "smartlife.iot.common.emeter"
|
||||
self.emeter_units = True
|
||||
|
||||
@property
|
||||
def is_color(self):
|
||||
"""
|
||||
Whether the bulb supports color changes
|
||||
|
||||
:return: True if the bulb supports color changes, False otherwise
|
||||
:rtype: bool
|
||||
"""
|
||||
return bool(self.sys_info['is_color'])
|
||||
|
||||
@property
|
||||
def is_dimmable(self):
|
||||
"""
|
||||
Whether the bulb supports brightness changes
|
||||
|
||||
:return: True if the bulb supports brightness changes, False otherwise
|
||||
:rtype: bool
|
||||
"""
|
||||
return bool(self.sys_info['is_dimmable'])
|
||||
|
||||
@property
|
||||
def is_variable_color_temp(self):
|
||||
"""
|
||||
Whether the bulb supports color temperature changes
|
||||
|
||||
:return: True if the bulb supports color temperature changes, False
|
||||
otherwise
|
||||
:rtype: bool
|
||||
"""
|
||||
return bool(self.sys_info['is_variable_color_temp'])
|
||||
|
||||
def get_light_state(self):
|
||||
return self._query_helper("smartlife.iot.smartbulb.lightingservice",
|
||||
"get_light_state")
|
||||
|
||||
def set_light_state(self, state):
|
||||
return self._query_helper("smartlife.iot.smartbulb.lightingservice",
|
||||
"transition_light_state", state)
|
||||
|
||||
@property
|
||||
def hsv(self):
|
||||
"""
|
||||
Returns the current HSV state of the bulb, if supported
|
||||
|
||||
:return: tuple containing current hue, saturation and value (0-255)
|
||||
:rtype: tuple
|
||||
"""
|
||||
|
||||
if not self.is_color:
|
||||
return None
|
||||
|
||||
light_state = self.get_light_state()
|
||||
if light_state['on_off'] == 0:
|
||||
hue = light_state['dft_on_state']['hue']
|
||||
saturation = light_state['dft_on_state']['saturation']
|
||||
value = int(light_state['dft_on_state']['brightness'] * 255 / 100)
|
||||
else:
|
||||
hue = light_state['hue']
|
||||
saturation = light_state['saturation']
|
||||
value = int(light_state['brightness'] * 255 / 100)
|
||||
|
||||
return(hue, saturation, value)
|
||||
|
||||
@hsv.setter
|
||||
def hsv(self, state):
|
||||
"""
|
||||
Sets new HSV, if supported
|
||||
|
||||
:param tuple state: hue, saturation and value (0-255 each)
|
||||
"""
|
||||
if not self.is_color:
|
||||
return None
|
||||
|
||||
light_state = {
|
||||
"hue": state[0],
|
||||
"saturation": state[1],
|
||||
"brightness": int(state[2] * 100 / 255),
|
||||
}
|
||||
return self.set_light_state(light_state)
|
||||
|
||||
@property
|
||||
def color_temp(self):
|
||||
"""
|
||||
Color temperature of the device, if supported
|
||||
|
||||
:return: Color temperature in Kelvin
|
||||
:rtype: int
|
||||
"""
|
||||
if not self.is_variable_color_temp:
|
||||
return None
|
||||
|
||||
light_state = self.get_light_state()
|
||||
if light_state['on_off'] == 0:
|
||||
return(light_state['dft_on_state']['color_temp'])
|
||||
else:
|
||||
return(light_state['color_temp'])
|
||||
|
||||
@color_temp.setter
|
||||
def color_temp(self, temp):
|
||||
"""
|
||||
Set the color temperature of the device, if supported
|
||||
|
||||
:param int temp: The new color temperature, in Kelvin
|
||||
"""
|
||||
if not self.is_variable_color_temp:
|
||||
return None
|
||||
|
||||
light_state = {
|
||||
"color_temp": temp,
|
||||
}
|
||||
return self.set_light_state(light_state)
|
||||
|
||||
@property
|
||||
def brightness(self):
|
||||
"""
|
||||
Current brightness of the device, if supported
|
||||
|
||||
:return: brightness in percent
|
||||
:rtype: int
|
||||
"""
|
||||
if not self.is_dimmable:
|
||||
return None
|
||||
|
||||
light_state = self.get_light_state()
|
||||
if light_state['on_off'] == 0:
|
||||
return(light_state['dft_on_state']['brightness'])
|
||||
else:
|
||||
return(light_state['brightness'])
|
||||
|
||||
@brightness.setter
|
||||
def brightness(self, brightness):
|
||||
"""
|
||||
Set the current brightness of the device, if supported
|
||||
|
||||
:param int brightness: brightness in percent
|
||||
"""
|
||||
if not self.is_dimmable:
|
||||
return None
|
||||
|
||||
light_state = {
|
||||
"brightness": brightness,
|
||||
}
|
||||
return self.set_light_state(light_state)
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""
|
||||
Retrieve the bulb state
|
||||
|
||||
:returns: one of
|
||||
BULB_STATE_ON
|
||||
BULB_STATE_OFF
|
||||
:rtype: str
|
||||
"""
|
||||
light_state = self.get_light_state()
|
||||
if light_state['on_off']:
|
||||
return self.BULB_STATE_ON
|
||||
return self.BULB_STATE_OFF
|
||||
|
||||
@state.setter
|
||||
def state(self, bulb_state):
|
||||
"""
|
||||
Set the new bulb state
|
||||
|
||||
:param bulb_state: one of
|
||||
BULB_STATE_ON
|
||||
BULB_STATE_OFF
|
||||
"""
|
||||
print(bulb_state)
|
||||
print(self.BULB_STATE_ON)
|
||||
print(self.BULB_STATE_OFF)
|
||||
if bulb_state == self.BULB_STATE_ON:
|
||||
bulb_state = 1
|
||||
elif bulb_state == self.BULB_STATE_OFF:
|
||||
bulb_state = 0
|
||||
else:
|
||||
raise ValueError
|
||||
|
||||
light_state = {
|
||||
"on_off": bulb_state,
|
||||
}
|
||||
return self.set_light_state(light_state)
|
||||
|
||||
@property
|
||||
def has_emeter(self):
|
||||
return True
|
||||
|
||||
|
||||
class SmartPlug(SmartDevice):
|
||||
"""Representation of a TP-Link Smart Switch.
|
||||
|
||||
Usage example when used as library:
|
||||
p = SmartPlug("192.168.1.105")
|
||||
# print the devices alias
|
||||
print(p.alias)
|
||||
# change state of plug
|
||||
p.state = "ON"
|
||||
p.state = "OFF"
|
||||
# query and print current state of plug
|
||||
print(p.state)
|
||||
|
||||
Errors reported by the device are raised as SmartPlugExceptions,
|
||||
and should be handled by the user of the library.
|
||||
|
||||
Note:
|
||||
The library references the same structure as defined for the D-Link Switch
|
||||
"""
|
||||
# switch states
|
||||
SWITCH_STATE_ON = 'ON'
|
||||
SWITCH_STATE_OFF = 'OFF'
|
||||
SWITCH_STATE_UNKNOWN = 'UNKNOWN'
|
||||
|
||||
# possible device features
|
||||
FEATURE_ENERGY_METER = 'ENE'
|
||||
FEATURE_TIMER = 'TIM'
|
||||
|
||||
ALL_FEATURES = (FEATURE_ENERGY_METER, FEATURE_TIMER)
|
||||
|
||||
def __init__(self, ip_address, protocol=None):
|
||||
SmartDevice.__init__(self, ip_address, protocol)
|
||||
self.emeter_type = "emeter"
|
||||
self.emeter_units = False
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""
|
||||
Retrieve the switch state
|
||||
|
||||
:returns: one of
|
||||
SWITCH_STATE_ON
|
||||
SWITCH_STATE_OFF
|
||||
SWITCH_STATE_UNKNOWN
|
||||
:rtype: str
|
||||
"""
|
||||
relay_state = self.sys_info['relay_state']
|
||||
|
||||
if relay_state == 0:
|
||||
return SmartPlug.SWITCH_STATE_OFF
|
||||
elif relay_state == 1:
|
||||
return SmartPlug.SWITCH_STATE_ON
|
||||
else:
|
||||
_LOGGER.warning("Unknown state %s returned.", relay_state)
|
||||
return SmartPlug.SWITCH_STATE_UNKNOWN
|
||||
|
||||
@state.setter
|
||||
def state(self, value):
|
||||
"""
|
||||
Set the new switch state
|
||||
|
||||
:param value: one of
|
||||
SWITCH_STATE_ON
|
||||
SWITCH_STATE_OFF
|
||||
:raises ValueError: on invalid state
|
||||
:raises SmartPlugException: on error
|
||||
|
||||
"""
|
||||
if not isinstance(value, basestring):
|
||||
raise ValueError("State must be str, not of %s.", type(value))
|
||||
elif value.upper() == SmartPlug.SWITCH_STATE_ON:
|
||||
self.turn_on()
|
||||
elif value.upper() == SmartPlug.SWITCH_STATE_OFF:
|
||||
self.turn_off()
|
||||
else:
|
||||
raise ValueError("State %s is not valid.", value)
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""
|
||||
Returns whether device is on.
|
||||
|
||||
:return: True if device is on, False otherwise
|
||||
"""
|
||||
return bool(self.sys_info['relay_state'])
|
||||
|
||||
@property
|
||||
def is_off(self):
|
||||
"""
|
||||
Returns whether device is off.
|
||||
|
||||
:return: True if device is off, False otherwise.
|
||||
:rtype: bool
|
||||
"""
|
||||
return not self.is_on
|
||||
|
||||
def turn_on(self):
|
||||
"""
|
||||
Turn the switch on.
|
||||
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
self._query_helper("system", "set_relay_state", {"state": 1})
|
||||
|
||||
def turn_off(self):
|
||||
"""
|
||||
Turn the switch off.
|
||||
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
self._query_helper("system", "set_relay_state", {"state": 0})
|
||||
|
||||
@property
|
||||
def has_emeter(self):
|
||||
"""
|
||||
Checks feature list for energey meter support.
|
||||
|
||||
:return: True if energey meter is available
|
||||
False if energymeter is missing
|
||||
"""
|
||||
return SmartPlug.FEATURE_ENERGY_METER in self.features
|
||||
|
||||
@property
|
||||
def features(self):
|
||||
"""
|
||||
Returns features of the devices
|
||||
|
||||
:return: list of features
|
||||
:rtype: list
|
||||
"""
|
||||
features = self.sys_info['feature'].split(':')
|
||||
|
||||
for feature in features:
|
||||
if feature not in SmartPlug.ALL_FEATURES:
|
||||
_LOGGER.warning("Unknown feature %s on device %s.",
|
||||
feature, self.model)
|
||||
|
||||
return features
|
||||
|
||||
@property
|
||||
def led(self):
|
||||
"""
|
||||
Returns the state of the led.
|
||||
|
||||
:return: True if led is on, False otherwise
|
||||
:rtype: bool
|
||||
"""
|
||||
return bool(1 - self.sys_info["led_off"])
|
||||
|
||||
@led.setter
|
||||
def led(self, state):
|
||||
"""
|
||||
Sets the state of the led (night mode)
|
||||
|
||||
:param bool state: True to set led on, False to set led off
|
||||
:raises SmartPlugException: on error
|
||||
"""
|
||||
self._query_helper("system", "set_led_off", {"off": int(not state)})
|
||||
|
||||
@property
|
||||
def on_since(self):
|
||||
"""
|
||||
Returns pretty-printed on-time
|
||||
|
||||
:return: datetime for on since
|
||||
:rtype: datetime
|
||||
"""
|
||||
return datetime.datetime.now() - \
|
||||
datetime.timedelta(seconds=self.sys_info["on_time"])
|
||||
|
||||
@ -25,6 +25,26 @@ emeter_support = {"get_realtime": get_realtime,
|
||||
"get_monthstat": get_monthstat,
|
||||
"get_daystat": get_daystat,}
|
||||
|
||||
def get_realtime_units(obj, x):
|
||||
return {"power_mw": 10800}
|
||||
|
||||
def get_monthstat_units(obj, x):
|
||||
if x["year"] < 2016:
|
||||
return {"month_list":[]}
|
||||
|
||||
return {"month_list": [{"year": 2016, "month": 11, "energy_wh": 32}, {"year": 2016, "month": 12, "energy_wh": 16}]}
|
||||
|
||||
def get_daystat_units(obj, x):
|
||||
if x["year"] < 2016:
|
||||
return {"day_list":[]}
|
||||
|
||||
return {"day_list": [{"year": 2016, "month": 11, "day": 24, "energy_wh": 20},
|
||||
{"year": 2016, "month": 11, "day": 25, "energy_wh": 32}]}
|
||||
|
||||
emeter_units_support = {"get_realtime": get_realtime_units,
|
||||
"get_monthstat": get_monthstat_units,
|
||||
"get_daystat": get_daystat_units,}
|
||||
|
||||
sysinfo_hs110 = {'system': {'get_sysinfo':
|
||||
{'active_mode': 'schedule',
|
||||
'alias': 'Mobile Plug',
|
||||
@ -76,6 +96,63 @@ sysinfo_hs200 = {'system': {'get_sysinfo': {'active_mode': 'schedule',
|
||||
'updating': 0}}
|
||||
}
|
||||
|
||||
sysinfo_lb130 = {'system': {'get_sysinfo':
|
||||
{'active_mode': 'none',
|
||||
'alias': 'Living Room Side Table',
|
||||
'ctrl_protocols': {'name': 'Linkie', 'version': '1.0'},
|
||||
'description': 'Smart Wi-Fi LED Bulb with Color Changing',
|
||||
'dev_state': 'normal',
|
||||
'deviceId': '80123C4640E9FC33A9019A0F3FD8BF5C17B7D9A8',
|
||||
'disco_ver': '1.0',
|
||||
'heapsize': 347000,
|
||||
'hwId': '111E35908497A05512E259BB76801E10',
|
||||
'hw_ver': '1.0',
|
||||
'is_color': 1,
|
||||
'is_dimmable': 1,
|
||||
'is_factory': False,
|
||||
'is_variable_color_temp': 1,
|
||||
'light_state': {'brightness': 100,
|
||||
'color_temp': 3700,
|
||||
'hue': 0,
|
||||
'mode': 'normal',
|
||||
'on_off': 1,
|
||||
'saturation': 0},
|
||||
'mic_mac': '50C7BF104865',
|
||||
'mic_type': 'IOT.SMARTBULB',
|
||||
'model': 'LB130(US)',
|
||||
'oemId': '05BF7B3BE1675C5A6867B7A7E4C9F6F7',
|
||||
'preferred_state': [{'brightness': 50,
|
||||
'color_temp': 2700,
|
||||
'hue': 0,
|
||||
'index': 0,
|
||||
'saturation': 0},
|
||||
{'brightness': 100,
|
||||
'color_temp': 0,
|
||||
'hue': 0,
|
||||
'index': 1,
|
||||
'saturation': 75},
|
||||
{'brightness': 100,
|
||||
'color_temp': 0,
|
||||
'hue': 120,
|
||||
'index': 2,
|
||||
'saturation': 75},
|
||||
{'brightness': 100,
|
||||
'color_temp': 0,
|
||||
'hue': 240,
|
||||
'index': 3,
|
||||
'saturation': 75}],
|
||||
'rssi': -55,
|
||||
'sw_ver': '1.1.2 Build 160927 Rel.111100'}},
|
||||
'smartlife.iot.smartbulb.lightingservice': {'get_light_state':
|
||||
{'on_off':1,
|
||||
'mode':'normal',
|
||||
'hue': 0,
|
||||
'saturation': 0,
|
||||
'color_temp': 3700,
|
||||
'brightness': 100,
|
||||
'err_code': 0}},
|
||||
'smartlife.iot.common.emeter': emeter_units_support,
|
||||
}
|
||||
|
||||
def error(cls, target, cmd="no-command", msg="default msg"):
|
||||
return {target: {cmd: {"err_code": -1323, "msg": msg}}}
|
||||
@ -90,13 +167,13 @@ def success(target, cmd, res):
|
||||
|
||||
|
||||
class FakeTransportProtocol(TPLinkSmartHomeProtocol):
|
||||
def __init__(self, invalid=False):
|
||||
def __init__(self, sysinfo, invalid=False):
|
||||
""" invalid is set only for testing
|
||||
to force query() to throw the exception for non-connected """
|
||||
proto = FakeTransportProtocol.baseproto
|
||||
for target in sysinfo_hs110:
|
||||
for cmd in sysinfo_hs110[target]:
|
||||
proto[target][cmd] = sysinfo_hs110[target][cmd]
|
||||
for target in sysinfo:
|
||||
for cmd in sysinfo[target]:
|
||||
proto[target][cmd] = sysinfo[target][cmd]
|
||||
self.proto = proto
|
||||
self.invalid = invalid
|
||||
|
||||
@ -116,6 +193,11 @@ class FakeTransportProtocol(TPLinkSmartHomeProtocol):
|
||||
_LOGGER.debug("Setting mac to %s", x)
|
||||
self.proto["system"]["get_sysinfo"][""]
|
||||
|
||||
def transition_light_state(self, x):
|
||||
_LOGGER.debug("Setting light state to %s", x)
|
||||
for key in x:
|
||||
self.proto["smartlife.iot.smartbulb.lightingservice"]["get_light_state"][key]=x[key]
|
||||
|
||||
baseproto = {
|
||||
"system": { "set_relay_state": set_relay_state,
|
||||
"set_dev_alias": set_alias,
|
||||
@ -129,6 +211,14 @@ class FakeTransportProtocol(TPLinkSmartHomeProtocol):
|
||||
"get_monthstat": None,
|
||||
"erase_emeter_state": None
|
||||
},
|
||||
"smartlife.iot.common.emeter": { "get_realtime": None,
|
||||
"get_daystat": None,
|
||||
"get_monthstat": None,
|
||||
"erase_emeter_state": None
|
||||
},
|
||||
"smartlife.iot.smartbulb.lightingservice": { "get_light_state": None,
|
||||
"transition_light_state": transition_light_state,
|
||||
},
|
||||
"time": { "get_time": { "year": 2017, "month": 1, "mday": 2, "hour": 3, "min": 4, "sec": 5 },
|
||||
"get_timezone": {'zone_str': "test", 'dst_offset': -1, 'index': 12, 'tz_str': "test2" },
|
||||
"set_timezone": None,
|
||||
|
||||
198
pyHS100/tests/test_bulb.py
Normal file
198
pyHS100/tests/test_bulb.py
Normal file
@ -0,0 +1,198 @@
|
||||
from __future__ import absolute_import
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from unittest import TestCase, skip, skipIf
|
||||
from voluptuous import Schema, Invalid, All, Range
|
||||
from functools import partial
|
||||
|
||||
from pyHS100 import SmartBulb, SmartPlugException
|
||||
from pyHS100.tests.fakes import FakeTransportProtocol, sysinfo_lb130
|
||||
|
||||
BULB_IP = '192.168.250.186'
|
||||
SKIP_STATE_TESTS = False
|
||||
|
||||
# python2 compatibility
|
||||
try:
|
||||
basestring
|
||||
except NameError:
|
||||
basestring = str
|
||||
|
||||
|
||||
def check_int_bool(x):
|
||||
if x != 0 and x != 1:
|
||||
raise Invalid(x)
|
||||
return x
|
||||
|
||||
|
||||
def check_mode(x):
|
||||
if x in ['schedule', 'none']:
|
||||
return x
|
||||
|
||||
raise Invalid("invalid mode {}".format(x))
|
||||
|
||||
|
||||
class TestSmartBulb(TestCase):
|
||||
# these schemas should go to the mainlib as
|
||||
# they can be useful when adding support for new features/devices
|
||||
# as well as to check that faked devices are operating properly.
|
||||
sysinfo_schema = Schema({
|
||||
'active_mode': check_mode,
|
||||
'alias': basestring,
|
||||
'ctrl_protocols': {
|
||||
'name': basestring,
|
||||
'version': basestring,
|
||||
},
|
||||
'description': basestring,
|
||||
'dev_state': basestring,
|
||||
'deviceId': basestring,
|
||||
'disco_ver': basestring,
|
||||
'heapsize': int,
|
||||
'hwId': basestring,
|
||||
'hw_ver': basestring,
|
||||
'is_color': check_int_bool,
|
||||
'is_dimmable': check_int_bool,
|
||||
'is_factory': bool,
|
||||
'is_variable_color_temp': check_int_bool,
|
||||
'light_state': {
|
||||
'brightness': All(int, Range(min=0, max=100)),
|
||||
'color_temp': int,
|
||||
'hue': All(int, Range(min=0, max=255)),
|
||||
'mode': basestring,
|
||||
'on_off': check_int_bool,
|
||||
'saturation': All(int, Range(min=0, max=255)),
|
||||
},
|
||||
'mic_mac': basestring,
|
||||
'mic_type': basestring,
|
||||
'model': basestring,
|
||||
'oemId': basestring,
|
||||
'preferred_state': [{
|
||||
'brightness': All(int, Range(min=0, max=100)),
|
||||
'color_temp': int,
|
||||
'hue': All(int, Range(min=0, max=255)),
|
||||
'index': int,
|
||||
'saturation': All(int, Range(min=0, max=255)),
|
||||
}],
|
||||
'rssi': All(int, Range(max=0)),
|
||||
'sw_ver': basestring,
|
||||
})
|
||||
|
||||
current_consumption_schema = Schema({
|
||||
'power_mw': int,
|
||||
})
|
||||
|
||||
tz_schema = Schema({
|
||||
'zone_str': basestring,
|
||||
'dst_offset': int,
|
||||
'index': All(int, Range(min=0)),
|
||||
'tz_str': basestring,
|
||||
})
|
||||
|
||||
def setUp(self):
|
||||
self.bulb = SmartBulb(BULB_IP,
|
||||
protocol=FakeTransportProtocol(sysinfo_lb130))
|
||||
|
||||
def tearDown(self):
|
||||
self.bulb = None
|
||||
|
||||
def test_initialize(self):
|
||||
self.assertIsNotNone(self.bulb.sys_info)
|
||||
self.sysinfo_schema(self.bulb.sys_info)
|
||||
|
||||
def test_initialize_invalid_connection(self):
|
||||
bulb = SmartBulb('127.0.0.1',
|
||||
protocol=FakeTransportProtocol(sysinfo_lb130,
|
||||
invalid=True))
|
||||
with self.assertRaises(SmartPlugException):
|
||||
bulb.sys_info['model']
|
||||
|
||||
def test_query_helper(self):
|
||||
with self.assertRaises(SmartPlugException):
|
||||
self.bulb._query_helper("test", "testcmd", {})
|
||||
# TODO check for unwrapping?
|
||||
|
||||
@skipIf(SKIP_STATE_TESTS, "SKIP_STATE_TESTS is True, skipping")
|
||||
def test_state(self):
|
||||
def set_invalid(x):
|
||||
self.bulb.state = x
|
||||
|
||||
set_invalid_int = partial(set_invalid, 1234)
|
||||
self.assertRaises(ValueError, set_invalid_int)
|
||||
|
||||
set_invalid_str = partial(set_invalid, "1234")
|
||||
self.assertRaises(ValueError, set_invalid_str)
|
||||
|
||||
set_invalid_bool = partial(set_invalid, True)
|
||||
self.assertRaises(ValueError, set_invalid_bool)
|
||||
|
||||
orig_state = self.bulb.state
|
||||
if orig_state == SmartBulb.BULB_STATE_OFF:
|
||||
self.bulb.state = SmartBulb.BULB_STATE_ON
|
||||
self.assertTrue(self.bulb.state == SmartBulb.BULB_STATE_ON)
|
||||
self.bulb.state = SmartBulb.BULB_STATE_OFF
|
||||
self.assertTrue(self.bulb.state == SmartBulb.BULB_STATE_OFF)
|
||||
elif orig_state == SmartBulb.BULB_STATE_ON:
|
||||
self.bulb.state = SmartBulb.BULB_STATE_OFF
|
||||
self.assertTrue(self.bulb.state == SmartBulb.BULB_STATE_OFF)
|
||||
self.bulb.state = SmartBulb.BULB_STATE_ON
|
||||
self.assertTrue(self.bulb.state == SmartBulb.BULB_STATE_ON)
|
||||
|
||||
def test_get_sysinfo(self):
|
||||
# initialize checks for this already, but just to be sure
|
||||
self.sysinfo_schema(self.bulb.get_sysinfo())
|
||||
|
||||
@skipIf(SKIP_STATE_TESTS, "SKIP_STATE_TESTS is True, skipping")
|
||||
def test_turns_and_isses(self):
|
||||
orig_state = self.bulb.state
|
||||
|
||||
if orig_state == SmartBulb.BULB_STATE_ON:
|
||||
self.bulb.state = SmartBulb.BULB_STATE_OFF
|
||||
self.assertTrue(self.bulb.state == SmartBulb.BULB_STATE_OFF)
|
||||
self.bulb.state = SmartBulb.BULB_STATE_ON
|
||||
self.assertTrue(self.bulb.state == SmartBulb.BULB_STATE_ON)
|
||||
else:
|
||||
self.bulb.state = SmartBulb.BULB_STATE_ON
|
||||
self.assertTrue(self.bulb.state == SmartBulb.BULB_STATE_ON)
|
||||
self.bulb.state = SmartBulb.BULB_STATE_OFF
|
||||
self.assertTrue(self.bulb.state == SmartBulb.BULB_STATE_OFF)
|
||||
|
||||
def test_get_emeter_realtime(self):
|
||||
self.current_consumption_schema((self.bulb.get_emeter_realtime()))
|
||||
|
||||
def test_get_emeter_daily(self):
|
||||
self.assertEqual(self.bulb.get_emeter_daily(year=1900, month=1), {})
|
||||
|
||||
k, v = self.bulb.get_emeter_daily().popitem()
|
||||
self.assertTrue(isinstance(k, int))
|
||||
self.assertTrue(isinstance(v, int))
|
||||
|
||||
def test_get_emeter_monthly(self):
|
||||
self.assertEqual(self.bulb.get_emeter_monthly(year=1900), {})
|
||||
|
||||
d = self.bulb.get_emeter_monthly()
|
||||
k, v = d.popitem()
|
||||
self.assertTrue(isinstance(k, int))
|
||||
self.assertTrue(isinstance(v, int))
|
||||
|
||||
@skip("not clearing your stats..")
|
||||
def test_erase_emeter_stats(self):
|
||||
self.fail()
|
||||
|
||||
def test_current_consumption(self):
|
||||
x = self.bulb.current_consumption()
|
||||
self.assertTrue(isinstance(x, int))
|
||||
self.assertTrue(x >= 0.0)
|
||||
|
||||
def test_alias(self):
|
||||
test_alias = "TEST1234"
|
||||
original = self.bulb.alias
|
||||
self.assertTrue(isinstance(original, basestring))
|
||||
self.bulb.alias = test_alias
|
||||
self.assertEqual(self.bulb.alias, test_alias)
|
||||
self.bulb.alias = original
|
||||
self.assertEqual(self.bulb.alias, original)
|
||||
|
||||
def test_icon(self):
|
||||
self.assertEqual(set(self.bulb.icon.keys()), {'icon', 'hash'})
|
||||
|
||||
def test_rssi(self):
|
||||
self.sysinfo_schema({'rssi': self.bulb.rssi}) # wrapping for vol
|
||||
@ -10,6 +10,6 @@ class TestTPLinkSmartHomeProtocol(TestCase):
|
||||
def test_encrypt(self):
|
||||
d = json.dumps({'foo': 1, 'bar': 2})
|
||||
encrypted = TPLinkSmartHomeProtocol.encrypt(d)
|
||||
# encrypt appends nullbytes for the protocol sends
|
||||
encrypted = encrypted.lstrip(b'\0')
|
||||
# encrypt adds a 4 byte header
|
||||
encrypted = encrypted[4:]
|
||||
self.assertEqual(d, TPLinkSmartHomeProtocol.decrypt(encrypted))
|
||||
|
||||
@ -8,7 +8,7 @@ import datetime
|
||||
import re
|
||||
|
||||
from pyHS100 import SmartPlug, SmartPlugException
|
||||
from pyHS100.tests.fakes import FakeTransportProtocol
|
||||
from pyHS100.tests.fakes import FakeTransportProtocol, sysinfo_hs110
|
||||
|
||||
PLUG_IP = '192.168.250.186'
|
||||
SKIP_STATE_TESTS = False
|
||||
@ -82,7 +82,8 @@ class TestSmartPlug(TestCase):
|
||||
})
|
||||
|
||||
def setUp(self):
|
||||
self.plug = SmartPlug(PLUG_IP, protocol=FakeTransportProtocol())
|
||||
self.plug = SmartPlug(PLUG_IP,
|
||||
protocol=FakeTransportProtocol(sysinfo_hs110))
|
||||
|
||||
def tearDown(self):
|
||||
self.plug = None
|
||||
@ -92,7 +93,9 @@ class TestSmartPlug(TestCase):
|
||||
self.sysinfo_schema(self.plug.sys_info)
|
||||
|
||||
def test_initialize_invalid_connection(self):
|
||||
plug = SmartPlug('127.0.0.1', protocol=FakeTransportProtocol(invalid=True))
|
||||
plug = SmartPlug('127.0.0.1',
|
||||
protocol=FakeTransportProtocol(sysinfo_hs110,
|
||||
invalid=True))
|
||||
with self.assertRaises(SmartPlugException):
|
||||
plug.sys_info['model']
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user