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