From 0eabf264e457eb5e7b950fea0efc2bb37e93dbcd Mon Sep 17 00:00:00 2001 From: Georgi Kirichkov Date: Thu, 13 Oct 2016 18:31:34 +0300 Subject: [PATCH 1/8] Adds Energy Meter commands available on the TP-Link HS110 Also adds turn_on() and turn_off() commands to supplement the state --- pyHS100/pyHS100.py | 182 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 182 insertions(+) diff --git a/pyHS100/pyHS100.py b/pyHS100/pyHS100.py index 33c92619..8f15f63a 100644 --- a/pyHS100/pyHS100.py +++ b/pyHS100/pyHS100.py @@ -1,7 +1,13 @@ +# Parts of this code reuse code and concepts by Lubomir Stroetmann from softScheck GmbH +# licensed under the Apache License v 2.0. +# Copy of the Apache License can be found at http://www.apache.org/licenses/LICENSE-2.0 +# The code from Lubomir Stroetmann is located at http://github.com/softScheck/tplink-smartplug + import logging import socket import codecs import json +import datetime _LOGGER = logging.getLogger(__name__) @@ -25,6 +31,7 @@ class SmartPlug(object): self.ip = ip self.port = 9999 self._error_report = False + self.model = self._identify_model() @property def state(self): @@ -102,3 +109,178 @@ class SmartPlug(object): sys_info = info["system"]["get_sysinfo"] relay_state = sys_info["relay_state"] return relay_state + + def get_info(self): + """Interrogate the switch""" + return self._send_command('{"system":{"get_sysinfo":{}}}') + + def turn_on(self): + """Turns the switch on + + Return values: + True on success + False on failure + """ + response = self._send_command('{"system":{"set_relay_state":{"state":1}}}') + + if response["system"]["set_relay_state"]["err_code"] == 0: + return True + + return False + + def turn_off(self): + """Turns the switch off + + Return values: + True on success + False on failure + """ + response = self._send_command('{"system":{"set_relay_state":{"state":0}}}') + + if response["system"]["set_relay_state"]["err_code"] == 0: + return True + + return False + + def get_emeter_realtime(self): + """Gets the current energy readings from the switch + + Return values: + False if command is not successful or the switch doesn't support energy metering + Dict with the current readings + """ + if self.model == 100: + return False + + response = self._send_command('{"emeter":{"get_realtime":{}}}') + + if response["emeter"]["get_realtime"]["err_code"] != 0: + return False + + response["emeter"]["get_realtime"].pop('err_code', None) + return response["emeter"]["get_realtime"] + + def get_emeter_daily(self, year = datetime.datetime.now().year, month = datetime.datetime.now().month): + """Gets daily statistics for a given month. + + Arguments: + 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 + + Return values: + False if command is not successful or the switch doesn't support energy metering + Dict where the keys represent the days, and the values are the aggregated statistics + """ + if self.model == 100: + return False + + response = self._send_command('{"emeter":{"get_daystat":{"month":' + str(month) + ',"year":' + str(year) + '}}}') + + if response["emeter"]["get_daystat"]["err_code"] != 0: + return False + + data = dict() + + 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): + """Gets monthly statistics for a given year. + + Arguments: + year (optional): The year for which to retrieve statistics, defaults to current year + + 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 self.model == 100: + return False + + response = self._send_command('{"emeter":{"get_monthstat":{"year":' + str(year) + '}}}') + + if response["emeter"]["get_monthstat"]["err_code"] != 0: + return False + + data = dict() + + 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): + """Erases all statistics. + + Return values: + True: Success + False: Failure or not supported by switch + """ + + if self.model == 100: + return False + + response = self._send_command('{"emeter":{"erase_emeter_stat":null}}') + + if response["emeter"]["erase_emeter_stat"]["err_code"] != 0: + return False + else: + return True + + def current_consumption(self): + """Get the current power consumption in Watt.""" + + response = self.get_emeter_realtime() + + return response["power"] + + def _encrypt(self, string): + """Encrypts a command.""" + key = 171 + result = "\0\0\0\0" + for i in string: + a = key ^ ord(i) + key = a + result += chr(a) + return result + + def _decrypt(self, string): + """Decrypts a command.""" + key = 171 + result = "" + for i in string: + a = key ^ ord(i) + key = ord(i) + result += chr(a) + return result + + def _send_command(self, command): + """Sends a command to the switch. + + Accepts one argument - the command as a string + + Return values: + The decrypted JSON + """ + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.connect((self.ip, self.port)) + s.send(self._encrypt(command)) + response = self._decrypt(s.recv(4096)[4:]) + s.close() + + return json.loads(response) + + def _identify_model(self): + """Query sysinfo and determine model""" + sys_info = self.get_info() + + if sys_info["system"]["get_sysinfo"]["model"][:5] == 'HS100': + model = 100 + elif sys_info["system"]["get_sysinfo"]["model"][:5] == 'HS110': + model = 110 + + return model \ No newline at end of file From 77d524ecf2150dfef35e38229f3123a5b007d4e1 Mon Sep 17 00:00:00 2001 From: Georgi Kirichkov Date: Thu, 13 Oct 2016 18:33:50 +0300 Subject: [PATCH 2/8] Refactors state() to use turn_on() and turn_off() --- pyHS100/pyHS100.py | 22 ++-------------------- 1 file changed, 2 insertions(+), 20 deletions(-) diff --git a/pyHS100/pyHS100.py b/pyHS100/pyHS100.py index 8f15f63a..23b48afc 100644 --- a/pyHS100/pyHS100.py +++ b/pyHS100/pyHS100.py @@ -54,28 +54,10 @@ class SmartPlug(object): :param value: Future state (either ON or OFF) """ if value.upper() == 'ON': - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.connect((self.ip, self.port)) - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.connect((self.ip, self.port)) - on_str = ('0000002ad0f281f88bff9af7d5' - 'ef94b6c5a0d48bf99cf091e8b7' - 'c4b0d1a5c0e2d8a381f286e793' - 'f6d4eedfa2dfa2') - data = codecs.decode(on_str, 'hex_codec') - s.send(data) - s.close() + self.turn_on() elif value.upper() == 'OFF': - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.connect((self.ip, self.port)) - off_str = ('0000002ad0f281f88bff9af7d5' - 'ef94b6c5a0d48bf99cf091e8b7' - 'c4b0d1a5c0e2d8a381f286e793' - 'f6d4eedea3dea3') - data = codecs.decode(off_str, 'hex_codec') - s.send(data) - s.close() + self.turn_off() else: raise TypeError("State %s is not valid." % str(value)) From 093899c588b163f1a199c0357d9926228db176e6 Mon Sep 17 00:00:00 2001 From: Georgi Kirichkov Date: Sat, 15 Oct 2016 11:18:31 +0300 Subject: [PATCH 3/8] Makes the socket sending code compatible with both Python 2 and python 3 Adds a shutdown to the socket used to send commands --- pyHS100/pyHS100.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/pyHS100/pyHS100.py b/pyHS100/pyHS100.py index 23b48afc..a68d17fd 100644 --- a/pyHS100/pyHS100.py +++ b/pyHS100/pyHS100.py @@ -8,6 +8,7 @@ import socket import codecs import json import datetime +import sys _LOGGER = logging.getLogger(__name__) @@ -228,16 +229,24 @@ class SmartPlug(object): a = key ^ ord(i) key = a result += chr(a) + + if sys.version_info.major > 2: + return result.encode('latin-1') + return result def _decrypt(self, string): """Decrypts a command.""" + if sys.version_info.major > 2: + string = string.decode('latin-1') + key = 171 result = "" for i in string: a = key ^ ord(i) key = ord(i) result += chr(a) + return result def _send_command(self, command): @@ -252,6 +261,7 @@ class SmartPlug(object): s.connect((self.ip, self.port)) s.send(self._encrypt(command)) response = self._decrypt(s.recv(4096)[4:]) + s.shutdown(1) s.close() return json.loads(response) From 9cd61fdcc8038d865662fe3d19a5b72bb89aae79 Mon Sep 17 00:00:00 2001 From: Georgi Kirichkov Date: Mon, 17 Oct 2016 15:17:28 +0300 Subject: [PATCH 4/8] Adds additional comments, for better compliance with the Apache license --- pyHS100/pyHS100.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/pyHS100/pyHS100.py b/pyHS100/pyHS100.py index a68d17fd..a5051b53 100644 --- a/pyHS100/pyHS100.py +++ b/pyHS100/pyHS100.py @@ -223,6 +223,11 @@ class SmartPlug(object): def _encrypt(self, string): """Encrypts a command.""" + + """ + Taken from https://raw.githubusercontent.com/softScheck/tplink-smartplug/master/tplink-smartplug.py + Changes: the return value is encoded in latin-1 in Python 3 and later + """ key = 171 result = "\0\0\0\0" for i in string: @@ -237,6 +242,11 @@ class SmartPlug(object): def _decrypt(self, string): """Decrypts a command.""" + + """ + Taken from https://raw.githubusercontent.com/softScheck/tplink-smartplug/master/tplink-smartplug.py + Changes: the string parameter is decoded from latin-1 in Python 3 and later + """ if sys.version_info.major > 2: string = string.decode('latin-1') From 3744a075c330e0d9dc8adb065bc67a034aaa1967 Mon Sep 17 00:00:00 2001 From: Georgi Kirichkov Date: Mon, 17 Oct 2016 15:36:07 +0300 Subject: [PATCH 5/8] Bumps the module version to 0.2.0 --- setup.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/setup.py b/setup.py index 9eeaf459..5992aa64 100644 --- a/setup.py +++ b/setup.py @@ -1,7 +1,7 @@ from setuptools import setup setup(name='pyHS100', - version='0.1.2', + version='0.2.0', description='Interface for TPLink HS100 Smart Plugs.', url='https://github.com/GadgetReactor/pyHS100', author='Sean Seah (GadgetReactor)', From 605abfdebb592053b09dfdce633a0a801601d2d5 Mon Sep 17 00:00:00 2001 From: Georgi Kirichkov Date: Tue, 18 Oct 2016 10:59:30 +0300 Subject: [PATCH 6/8] Fixes indentation and removes extra whitespaces --- pyHS100/pyHS100.py | 268 ++++++++++++++++++++++----------------------- 1 file changed, 134 insertions(+), 134 deletions(-) diff --git a/pyHS100/pyHS100.py b/pyHS100/pyHS100.py index a5051b53..8664222c 100644 --- a/pyHS100/pyHS100.py +++ b/pyHS100/pyHS100.py @@ -1,4 +1,4 @@ -# Parts of this code reuse code and concepts by Lubomir Stroetmann from softScheck GmbH +# Parts of this code reuse code and concepts by Lubomir Stroetmann from softScheck GmbH # licensed under the Apache License v 2.0. # Copy of the Apache License can be found at http://www.apache.org/licenses/LICENSE-2.0 # The code from Lubomir Stroetmann is located at http://github.com/softScheck/tplink-smartplug @@ -55,10 +55,10 @@ class SmartPlug(object): :param value: Future state (either ON or OFF) """ if value.upper() == 'ON': - self.turn_on() + self.turn_on() elif value.upper() == 'OFF': - self.turn_off() + self.turn_off() else: raise TypeError("State %s is not valid." % str(value)) @@ -94,84 +94,84 @@ class SmartPlug(object): return relay_state def get_info(self): - """Interrogate the switch""" - return self._send_command('{"system":{"get_sysinfo":{}}}') + """Interrogate the switch""" + return self._send_command('{"system":{"get_sysinfo":{}}}') def turn_on(self): - """Turns the switch on + """Turns the switch on - Return values: - True on success - False on failure - """ - response = self._send_command('{"system":{"set_relay_state":{"state":1}}}') + Return values: + True on success + False on failure + """ + response = self._send_command('{"system":{"set_relay_state":{"state":1}}}') - if response["system"]["set_relay_state"]["err_code"] == 0: - return True + if response["system"]["set_relay_state"]["err_code"] == 0: + return True - return False + return False def turn_off(self): - """Turns the switch off + """Turns the switch off - Return values: - True on success - False on failure - """ - response = self._send_command('{"system":{"set_relay_state":{"state":0}}}') + Return values: + True on success + False on failure + """ + response = self._send_command('{"system":{"set_relay_state":{"state":0}}}') - if response["system"]["set_relay_state"]["err_code"] == 0: - return True + if response["system"]["set_relay_state"]["err_code"] == 0: + return True + + return False - return False - def get_emeter_realtime(self): - """Gets the current energy readings from the switch + """Gets the current energy readings from the switch - Return values: - False if command is not successful or the switch doesn't support energy metering - Dict with the current readings - """ - if self.model == 100: - return False + Return values: + False if command is not successful or the switch doesn't support energy metering + Dict with the current readings + """ + if self.model == 100: + return False - response = self._send_command('{"emeter":{"get_realtime":{}}}') + response = self._send_command('{"emeter":{"get_realtime":{}}}') - if response["emeter"]["get_realtime"]["err_code"] != 0: - return False + if response["emeter"]["get_realtime"]["err_code"] != 0: + return False - response["emeter"]["get_realtime"].pop('err_code', None) - return response["emeter"]["get_realtime"] + response["emeter"]["get_realtime"].pop('err_code', None) + return response["emeter"]["get_realtime"] - def get_emeter_daily(self, year = datetime.datetime.now().year, month = datetime.datetime.now().month): - """Gets daily statistics for a given month. - - Arguments: - 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 + def get_emeter_daily(self, year=datetime.datetime.now().year, month=datetime.datetime.now().month): + """Gets daily statistics for a given month. - Return values: - False if command is not successful or the switch doesn't support energy metering - Dict where the keys represent the days, and the values are the aggregated statistics - """ - if self.model == 100: - return False + Arguments: + 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 - response = self._send_command('{"emeter":{"get_daystat":{"month":' + str(month) + ',"year":' + str(year) + '}}}') + Return values: + False if command is not successful or the switch doesn't support energy metering + Dict where the keys represent the days, and the values are the aggregated statistics + """ + if self.model == 100: + return False - if response["emeter"]["get_daystat"]["err_code"] != 0: - return False + response = self._send_command('{"emeter":{"get_daystat":{"month":' + str(month) + ',"year":' + str(year) + '}}}') - data = dict() + if response["emeter"]["get_daystat"]["err_code"] != 0: + return False - for i, j in enumerate(response["emeter"]["get_daystat"]["day_list"]): - if j["energy"] > 0: - data[j["day"]] = j["energy"] + data = dict() - return data + for i, j in enumerate(response["emeter"]["get_daystat"]["day_list"]): + if j["energy"] > 0: + data[j["day"]] = j["energy"] - def get_emeter_monthly(self, year = datetime.datetime.now().year): - """Gets monthly statistics for a given year. + return data + + def get_emeter_monthly(self, year=datetime.datetime.now().year): + """Gets monthly statistics for a given year. Arguments: year (optional): The year for which to retrieve statistics, defaults to current year @@ -179,110 +179,110 @@ class SmartPlug(object): 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 self.model == 100: - return False + """ + if self.model == 100: + return False - response = self._send_command('{"emeter":{"get_monthstat":{"year":' + str(year) + '}}}') + response = self._send_command('{"emeter":{"get_monthstat":{"year":' + str(year) + '}}}') - if response["emeter"]["get_monthstat"]["err_code"] != 0: - return False + if response["emeter"]["get_monthstat"]["err_code"] != 0: + return False - data = dict() + data = dict() - for i, j in enumerate(response["emeter"]["get_monthstat"]["month_list"]): - if j["energy"] > 0: - data[j["month"]] = j["energy"] + for i, j in enumerate(response["emeter"]["get_monthstat"]["month_list"]): + if j["energy"] > 0: + data[j["month"]] = j["energy"] + + return data - return data - def erase_emeter_stats(self): - """Erases all statistics. + """Erases all statistics. - Return values: - True: Success - False: Failure or not supported by switch - """ + Return values: + True: Success + False: Failure or not supported by switch + """ - if self.model == 100: - return False + if self.model == 100: + return False - response = self._send_command('{"emeter":{"erase_emeter_stat":null}}') + response = self._send_command('{"emeter":{"erase_emeter_stat":null}}') - if response["emeter"]["erase_emeter_stat"]["err_code"] != 0: - return False - else: - return True + if response["emeter"]["erase_emeter_stat"]["err_code"] != 0: + return False + else: + return True def current_consumption(self): - """Get the current power consumption in Watt.""" + """Get the current power consumption in Watt.""" - response = self.get_emeter_realtime() + response = self.get_emeter_realtime() - return response["power"] + return response["power"] def _encrypt(self, string): - """Encrypts a command.""" + """Encrypts a command.""" - """ - Taken from https://raw.githubusercontent.com/softScheck/tplink-smartplug/master/tplink-smartplug.py - Changes: the return value is encoded in latin-1 in Python 3 and later - """ - key = 171 - result = "\0\0\0\0" - for i in string: - a = key ^ ord(i) - key = a - result += chr(a) + """ + Taken from https://raw.githubusercontent.com/softScheck/tplink-smartplug/master/tplink-smartplug.py + Changes: the return value is encoded in latin-1 in Python 3 and later + """ + key = 171 + result = "\0\0\0\0" + for i in string: + a = key ^ ord(i) + key = a + result += chr(a) - if sys.version_info.major > 2: - return result.encode('latin-1') + if sys.version_info.major > 2: + return result.encode('latin-1') - return result + return result def _decrypt(self, string): - """Decrypts a command.""" + """Decrypts a command.""" - """ - Taken from https://raw.githubusercontent.com/softScheck/tplink-smartplug/master/tplink-smartplug.py - Changes: the string parameter is decoded from latin-1 in Python 3 and later - """ - if sys.version_info.major > 2: - string = string.decode('latin-1') + """ + Taken from https://raw.githubusercontent.com/softScheck/tplink-smartplug/master/tplink-smartplug.py + Changes: the string parameter is decoded from latin-1 in Python 3 and later + """ + if sys.version_info.major > 2: + string = string.decode('latin-1') - key = 171 - result = "" - for i in string: - a = key ^ ord(i) - key = ord(i) - result += chr(a) + key = 171 + result = "" + for i in string: + a = key ^ ord(i) + key = ord(i) + result += chr(a) - return result + return result def _send_command(self, command): - """Sends a command to the switch. - - Accepts one argument - the command as a string - - Return values: - The decrypted JSON - """ - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.connect((self.ip, self.port)) - s.send(self._encrypt(command)) - response = self._decrypt(s.recv(4096)[4:]) - s.shutdown(1) - s.close() + """Sends a command to the switch. - return json.loads(response) + Accepts one argument - the command as a string + + Return values: + The decrypted JSON + """ + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.connect((self.ip, self.port)) + s.send(self._encrypt(command)) + response = self._decrypt(s.recv(4096)[4:]) + s.shutdown(1) + s.close() + + return json.loads(response) def _identify_model(self): - """Query sysinfo and determine model""" - sys_info = self.get_info() + """Query sysinfo and determine model""" + sys_info = self.get_info() - if sys_info["system"]["get_sysinfo"]["model"][:5] == 'HS100': - model = 100 - elif sys_info["system"]["get_sysinfo"]["model"][:5] == 'HS110': - model = 110 + if sys_info["system"]["get_sysinfo"]["model"][:5] == 'HS100': + model = 100 + elif sys_info["system"]["get_sysinfo"]["model"][:5] == 'HS110': + model = 110 - return model \ No newline at end of file + return model From dc3de3fa103e347d485ba9a803dc4bca89ddc548 Mon Sep 17 00:00:00 2001 From: Georgi Kirichkov Date: Wed, 19 Oct 2016 10:28:48 +0300 Subject: [PATCH 7/8] Adds model check to current_consumption() and removes whitespace --- pyHS100/pyHS100.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pyHS100/pyHS100.py b/pyHS100/pyHS100.py index 8664222c..5c06e77f 100644 --- a/pyHS100/pyHS100.py +++ b/pyHS100/pyHS100.py @@ -203,7 +203,6 @@ class SmartPlug(object): True: Success False: Failure or not supported by switch """ - if self.model == 100: return False @@ -216,6 +215,8 @@ class SmartPlug(object): def current_consumption(self): """Get the current power consumption in Watt.""" + if self.model == 100: + return False response = self.get_emeter_realtime() From 61714ac1100c5824ac2742ad618117ae3b6f3ece Mon Sep 17 00:00:00 2001 From: Georgi Kirichkov Date: Wed, 19 Oct 2016 10:53:21 +0300 Subject: [PATCH 8/8] Refactors state property to use get_info() and removes hs100_status() --- pyHS100/pyHS100.py | 42 +++++++----------------------------------- 1 file changed, 7 insertions(+), 35 deletions(-) diff --git a/pyHS100/pyHS100.py b/pyHS100/pyHS100.py index 5c06e77f..4143b076 100644 --- a/pyHS100/pyHS100.py +++ b/pyHS100/pyHS100.py @@ -37,15 +37,17 @@ class SmartPlug(object): @property def state(self): """Get the device state (i.e. ON or OFF).""" - response = self.hs100_status() - if response is None: + response = self.get_info() + relay_state = response["system"]["get_sysinfo"]["relay_state"] + + if relay_state is None: return 'unknown' - elif response == 0: + elif relay_state == 0: return "OFF" - elif response == 1: + elif relay_state == 1: return "ON" else: - _LOGGER.warning("Unknown state %s returned" % str(response)) + _LOGGER.warning("Unknown state %s returned" % str(relay_state)) return 'unknown' @state.setter @@ -63,36 +65,6 @@ class SmartPlug(object): else: raise TypeError("State %s is not valid." % str(value)) - def hs100_status(self): - """Query HS100 for relay status.""" - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.connect((self.ip, self.port)) - skip = 4 - code = 171 - response = "" - query_str = ('00000023d0f0d2a1d8abdfbad7' - 'f5cfb494b6d1b4c09fec95e68f' - 'e187e8caf09eeb87ebcbb696eb') - data = codecs.decode(query_str, 'hex_codec') - s.send(data) - reply = s.recv(4096) - s.shutdown(1) - s.close() - - for value in reply: - if skip > 0: - skip = skip - 1 - else: - change = (value ^ code) - response = response + chr(change) - code = value - - info = json.loads(response) - # info is reserved for future expansion. - sys_info = info["system"]["get_sysinfo"] - relay_state = sys_info["relay_state"] - return relay_state - def get_info(self): """Interrogate the switch""" return self._send_command('{"system":{"get_sysinfo":{}}}')