Normalize docstrings, address flake8 & pylint recommendations

Signed-off-by: Martin Weinelt <hexa@darmstadt.ccc.de>
This commit is contained in:
Martin Weinelt 2016-11-22 02:31:25 +01:00
parent 33b2800fe8
commit cb5899e3cc
No known key found for this signature in database
GPG Key ID: BD4AA0528F63F17E

View File

@ -1,16 +1,18 @@
# pyHS100 """
# Python Library supporting TP-Link Smart Plugs/Switches (HS100/HS110/Hs200) pyHS100
# Python library supporting TP-Link Smart Plugs/Switches (HS100/HS110/Hs200).
# The communication protocol was reverse engineered by Lubomir Stroetmann and
# Tobias Esser in 'Reverse Engineering the TP-Link HS110' The communication protocol was reverse engineered by Lubomir Stroetmann and
# https://www.softscheck.com/en/reverse-engineering-tp-link-hs110/ Tobias Esser in 'Reverse Engineering the TP-Link HS110':
# https://www.softscheck.com/en/reverse-engineering-tp-link-hs110/
# This library reuses codes and concepts of the TP-Link WiFi SmartPlug Client
# at https://github.com/softScheck/tplink-smartplug, developed by Lubomir This library reuses codes and concepts of the TP-Link WiFi SmartPlug Client
# Stroetmann which is licensed under the Apache License, Version 2.0. at https://github.com/softScheck/tplink-smartplug, developed by Lubomir
# Stroetmann which is licensed under the Apache License, Version 2.0.
# You may obtain a copy of the license at
# http://www.apache.org/licenses/LICENSE-2.0 You may obtain a copy of the license at
http://www.apache.org/licenses/LICENSE-2.0
"""
import datetime import datetime
import json import json
@ -20,6 +22,11 @@ import sys
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
# switch states
SWITCH_STATE_ON = 'on'
SWITCH_STATE_OFF = 'off'
SWITCH_STATE_UNKNOWN = 'unknown'
# possible device features # possible device features
FEATURE_ENERGY_METER = 'ENE' FEATURE_ENERGY_METER = 'ENE'
FEATURE_TIMER = 'TIM' FEATURE_TIMER = 'TIM'
@ -27,7 +34,7 @@ FEATURE_TIMER = 'TIM'
ALL_FEATURES = (FEATURE_ENERGY_METER, FEATURE_TIMER) ALL_FEATURES = (FEATURE_ENERGY_METER, FEATURE_TIMER)
class SmartPlug(object): class SmartPlug:
"""Representation of a TP-Link Smart Switch. """Representation of a TP-Link Smart Switch.
Usage example when used as library: Usage example when used as library:
@ -39,186 +46,225 @@ class SmartPlug(object):
p.state = "ON" p.state = "ON"
# query and print current state of plug # query and print current state of plug
print(p.state) print(p.state)
Note: Note:
The library references the same structure as defined for the D-Link Switch The library references the same structure as defined for the D-Link Switch
""" """
def __init__(self, ip): def __init__(self, ip_address):
"""Create a new SmartPlug instance identified by the IP.""" """
self.ip = ip Create a new SmartPlug instance, identified through its IP address.
:param ip_address: ip address on which the device listens
"""
socket.inet_pton(socket.AF_INET, ip_address)
self.ip_address = ip_address
self.alias, self.model, self.features = self.identify() self.alias, self.model, self.features = self.identify()
@property @property
def state(self): def state(self):
"""Get the device state (i.e. ON or OFF).""" """
Retrieve the switch state
:returns: one of
SWITCH_STATE_ON
SWITCH_STATE_OFF
SWITCH_STATE_UNKNOWN
"""
response = self.get_sysinfo() response = self.get_sysinfo()
relay_state = response['relay_state'] relay_state = response['relay_state']
if relay_state is None: if relay_state == 0:
return 'unknown' return SWITCH_STATE_OFF
elif relay_state == 0:
return "OFF"
elif relay_state == 1: elif relay_state == 1:
return "ON" return SWITCH_STATE_ON
else: else:
_LOGGER.warning("Unknown state %s returned" % str(relay_state)) _LOGGER.warning("Unknown state %s returned.", relay_state)
return 'unknown' return SWITCH_STATE_UNKNOWN
@state.setter @state.setter
def state(self, value): def state(self, value):
"""Set device state.
:type value: str
:param value: Future state (either ON or OFF)
""" """
if value.upper() == 'ON': Set the new switch state
:param value: one of
SWITCH_STATE_ON
SWITCH_STATE_OFF
:return: True if new state was successfully set
False if an error occured
"""
if value.lower() == SWITCH_STATE_ON:
self.turn_on() self.turn_on()
elif value.lower() == SWITCH_STATE_OFF:
elif value.upper() == 'OFF':
self.turn_off() self.turn_off()
else: else:
raise TypeError("State %s is not valid." % str(value))
raise ValueError("State %s is not valid.", value)
def get_sysinfo(self): def get_sysinfo(self):
"""Interrogate the switch""" """
return TPLinkSmartHomeProtocol.query( Retrieve system information.
host=self.ip, request='{"system":{"get_sysinfo":{}}}'
:return: dict sysinfo
"""
response = TPLinkSmartHomeProtocol.query(
host=self.ip_address,
request={'system': {'get_sysinfo': {}}}
)['system']['get_sysinfo'] )['system']['get_sysinfo']
def turn_on(self): if response['err_code'] != 0:
"""Turns the switch on return False
Return values: return response
True on success
False on failure def turn_on(self):
"""
Turn the switch on.
:return: True on success
:raises ProtocolError when device responds with err_code != 0
""" """
response = TPLinkSmartHomeProtocol.query( response = TPLinkSmartHomeProtocol.query(
host=self.ip, request='{"system":{"set_relay_state":{"state":1}}}') host=self.ip_address,
request={'system': {'set_relay_state': {'state': 1}}}
if response["system"]["set_relay_state"]["err_code"] == 0: )['system']['set_relay_state']
return True
if response['err_code'] != 0:
return False return False
return True
def turn_off(self): def turn_off(self):
"""Turns the switch off """
Turn the switch off.
Return values: :return: True on success
True on success False on error
False on failure
""" """
response = TPLinkSmartHomeProtocol.query( response = TPLinkSmartHomeProtocol.query(
host=self.ip, request='{"system":{"set_relay_state":{"state":0}}}') host=self.ip_address,
request={'system': {'set_relay_state': {'state': 0}}}
if response["system"]["set_relay_state"]["err_code"] == 0: )['system']['set_relay_state']
return True
if response['err_code'] != 0:
return False return False
return True
@property @property
def has_emeter(self): def has_emeter(self):
"""
Checks feature list for energey meter support.
:return: True if energey meter is available
False if energymeter is missing
"""
return FEATURE_ENERGY_METER in self.features return FEATURE_ENERGY_METER in self.features
def get_emeter_realtime(self): def get_emeter_realtime(self):
"""Gets the current energy readings from the switch """
Retrive current energy readings from device.
Return values: :returns: dict with current readings
False if command is not successful or the switch doesn't support energy metering False if device has no energy meter or error occured
Dict with the current readings
""" """
if not self.has_emeter: if not self.has_emeter:
return False return False
response = TPLinkSmartHomeProtocol.query( response = TPLinkSmartHomeProtocol.query(
host=self.ip, request='{"emeter":{"get_realtime":{}}}') host=self.ip_address, request={'emeter': {'get_realtime': {}}}
)['emeter']['get_realtime']
if response["emeter"]["get_realtime"]["err_code"] != 0: if response['err_code'] != 0:
return False return False
response["emeter"]["get_realtime"].pop('err_code', None) del response['err_code']
return response["emeter"]["get_realtime"]
def get_emeter_daily(self, year=datetime.datetime.now().year, month=datetime.datetime.now().month): return response
"""Gets daily statistics for a given month.
Arguments: def get_emeter_daily(self, year=None, month=None):
year (optional): The year for which to retrieve statistics, defaults to current year """
month (optional): The mont for which to retrieve statistics, defaults to current month Retrieve daily statistics for a given month
Return values: :param year: year for which to retrieve statistics (default: this year)
False if command is not successful or the switch doesn't support energy metering :param month: month for which to retrieve statistcs (default: this
Dict where the keys represent the days, and the values are the aggregated statistics month)
:return: dict: mapping of day of month to value
False if device has no energy meter or error occured
""" """
if not self.has_emeter: if not self.has_emeter:
return False return False
response = TPLinkSmartHomeProtocol.query( if year is None:
host=self.ip, request='{"emeter":{"get_daystat":{"month":' + str(month) + ',"year":' + str(year) + '}}}') year = datetime.datetime.now().year
if month is None:
month = datetime.datetime.now().month
if response["emeter"]["get_daystat"]["err_code"] != 0: response = TPLinkSmartHomeProtocol.query(
host=self.ip_address,
request={'emeter': {'get_daystat': {'month': str(month),
'year': str(year)}}}
)['emeter']['get_daystat']
if response['err_code'] != 0:
return False return False
data = dict() return {entry['day']: entry['energy']
for entry in response['day_list']}
for i, j in enumerate(response["emeter"]["get_daystat"]["day_list"]):
if j["energy"] > 0:
data[j["day"]] = j["energy"]
return data
def get_emeter_monthly(self, year=datetime.datetime.now().year): def get_emeter_monthly(self, year=datetime.datetime.now().year):
"""Gets monthly statistics for a given year. """
Retrieve monthly statistics for a given year.
Arguments: :param year: year for which to retrieve statistics (default: this year)
year (optional): The year for which to retrieve statistics, defaults to current year :return: dict: mapping of month to value
False if device has no energy meter or error occured
Return values:
False if command is not successful or the switch doesn't support energy metering
Dict - the keys represent the months, the values are the aggregated statistics
""" """
if not self.has_emeter: if not self.has_emeter:
return False return False
response = TPLinkSmartHomeProtocol.query( response = TPLinkSmartHomeProtocol.query(
host=self.ip, request='{"emeter":{"get_monthstat":{"year":' + str(year) + '}}}') host=self.ip_address,
request={'emeter': {'get_monthstat': {'year': str(year)}}}
)['emeter']['get_monthstat']
if response["emeter"]["get_monthstat"]["err_code"] != 0: if response['err_code'] != 0:
return False return False
data = dict() return {entry['month']: entry['energy']
for entry in response['month_list']}
for i, j in enumerate(response["emeter"]["get_monthstat"]["month_list"]):
if j["energy"] > 0:
data[j["month"]] = j["energy"]
return data
def erase_emeter_stats(self): def erase_emeter_stats(self):
"""Erases all statistics. """
Erase energy meter statistics
Return values: :return: True if statistics were deleted
True: Success False if device has no energy meter or error occured
False: Failure or not supported by switch
""" """
if not self.has_emeter: if not self.has_emeter:
return False return False
response = TPLinkSmartHomeProtocol.query( response = TPLinkSmartHomeProtocol.query(
host=self.ip, request='{"emeter":{"erase_emeter_stat":null}}') host=self.ip_address,
request={'emeter': {'erase_emeter_stat': None}}
)['emeter']['erase_emeter_stat']
if response["emeter"]["erase_emeter_stat"]["err_code"] != 0: return response['err_code'] == 0
return False
else:
return True
def current_consumption(self): def current_consumption(self):
"""Get the current power consumption in Watt.""" """
Get the current power consumption in Watt.
:return: the current power consumption in Watt.
False if device has no energy meter of error occured.
"""
if not self.has_emeter: if not self.has_emeter:
return False return False
response = self.get_emeter_realtime() response = self.get_emeter_realtime()
return response["power"] return response['power']
def identify(self): def identify(self):
""" """
@ -229,12 +275,13 @@ class SmartPlug(object):
sys_info = self.get_sysinfo() sys_info = self.get_sysinfo()
alias = sys_info['alias'] alias = sys_info['alias']
model = sys_info["model"] model = sys_info['model']
features = sys_info['feature'].split(':') features = sys_info['feature'].split(':')
for feature in features: for feature in features:
if feature not in ALL_FEATURES: if feature not in ALL_FEATURES:
_LOGGER.warn('Unknown feature %s on device %s.', feature, model) _LOGGER.warning("Unknown feature %s on device %s.",
feature, model)
return alias, model, features return alias, model, features
@ -252,7 +299,7 @@ class TPLinkSmartHomeProtocol:
which are licensed under the Apache License, Version 2.0 which are licensed under the Apache License, Version 2.0
http://www.apache.org/licenses/LICENSE-2.0 http://www.apache.org/licenses/LICENSE-2.0
""" """
IV = 171 initialization_vector = 171
@staticmethod @staticmethod
def query(host, request, port=9999): def query(host, request, port=9999):
@ -287,7 +334,7 @@ class TPLinkSmartHomeProtocol:
:param request: plaintext request data :param request: plaintext request data
:return: ciphertext request :return: ciphertext request
""" """
key = TPLinkSmartHomeProtocol.IV key = TPLinkSmartHomeProtocol.initialization_vector
buffer = ['\0\0\0\0'] buffer = ['\0\0\0\0']
for char in request: for char in request:
@ -309,7 +356,7 @@ class TPLinkSmartHomeProtocol:
:param ciphertext: encrypted response data :param ciphertext: encrypted response data
:return: plaintext response :return: plaintext response
""" """
key = TPLinkSmartHomeProtocol.IV key = TPLinkSmartHomeProtocol.initialization_vector
buffer = [] buffer = []
if sys.version_info.major > 2: if sys.version_info.major > 2: