diff --git a/README.md b/README.md index d655e7da..2dd476b5 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,61 @@ # pyHS100 Python Library to control TPLink Switch (HS100 / HS110) + +# Usage + +For all available API functions run ```help(SmartPlug)``` + +```python +from pyHS100 import SmartPlug +from pprint import pformat as pf + +plug = SmartPlug("192.168.250.186") +print("Alias, type and supported features: %s" % (plug.identify(),)) +print("Hardware: %s" % pf(plug.hw_info)) +print("Full sysinfo: %s" % pf(plug.get_sysinfo())) # this prints lots of information about the device +``` + +## Time information +```python +print("Current time: %s" % plug.time) +print("Timezone: %s" % plug.timezone) +``` + +## Getting and setting the name +```python +print("Alias: %s" % plug.alias) +plug.alias = "My New Smartplug" +``` + +## State & switching +```python +print("Current state: %s" % plug.state) +plug.turn_off() +plug.turn_on() +``` +or +```python +plug.state = "ON" +plug.state = "OFF" +``` + +## Getting emeter status (on HS110) +```python +print("Current consumption: %s" % plug.get_emeter_realtime()) +print("Per day: %s" % plug.get_emeter_daily(year=2016, month=12)) +print("Per month: %s" % plug.get_emeter_monthly(year=2016)) +``` + +## Switching the led +```python +print("Current LED state: %s" % plug.led) +plug.led = 0 # turn off led +print("New LED state: %s" % plug.led) + +``` + +# Example +There is also a simple tool for testing connectivity in examples, to use: +```python +python -m examples.cli +``` diff --git a/examples/cli.py b/examples/cli.py new file mode 100644 index 00000000..aadab695 --- /dev/null +++ b/examples/cli.py @@ -0,0 +1,23 @@ +import sys +import logging + +from pyHS100 import SmartPlug + +logging.basicConfig(level=logging.DEBUG) + +if len(sys.argv) < 2: + print("%s " % sys.argv[0]) + sys.exit(1) + +hs = SmartPlug(sys.argv[1]) + +logging.info("Identify: %s", hs.identify) +logging.info("Sysinfo: %s", hs.get_sysinfo()) +has_emeter = hs.has_emeter +if has_emeter: + logging.info("== Emeter ==") + logging.info("- Current: %s", hs.get_emeter_realtime()) + logging.info("== Monthly ==") + logging.info(hs.get_emeter_monthly()) + logging.info("== Daily ==") + logging.info(hs.get_emeter_daily(month=11, year=2016)) diff --git a/pyHS100/__init__.py b/pyHS100/__init__.py index 8b137891..edac8b99 100644 --- a/pyHS100/__init__.py +++ b/pyHS100/__init__.py @@ -1 +1 @@ - +from pyHS100.pyHS100 import SmartPlug diff --git a/pyHS100/pyHS100.py b/pyHS100/pyHS100.py index 0dd59863..657764d2 100644 --- a/pyHS100/pyHS100.py +++ b/pyHS100/pyHS100.py @@ -23,6 +23,13 @@ import sys _LOGGER = logging.getLogger(__name__) +class SmartPlugException(Exception): + """ + SmartPlugException gets raised for errors reported by the plug. + """ + pass + + class SmartPlug: """Representation of a TP-Link Smart Switch. @@ -31,11 +38,14 @@ class SmartPlug: # print the devices alias print(p.alias) # change state of plug - p.state = "OFF" 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 """ @@ -54,12 +64,49 @@ class SmartPlug: """ Create a new SmartPlug instance, identified through its IP address. - :param ip_address: ip address on which the device listens + :param str ip_address: ip address on which the device listens + :raises SmartPlugException: when unable to communicate with the device """ socket.inet_pton(socket.AF_INET, ip_address) self.ip_address = ip_address - self.alias, self.model, self.features = self.identify() + self.initialize() + + def initialize(self): + """ + (Re-)Initializes the state. + + This should be called when the state of the plug is changed anyway. + + :raises: SmartPlugException: on error + """ + self.sys_info = self.get_sysinfo() + + self._alias, self.model, self.features = self.identify() + + def _query_helper(self, target, cmd, arg={}): + """ + Helper returning unwrapped result object and doing error handling. + + :param target: Target system {system, time, emeter, ..} + :param cmd: Command to execute + :param arg: JSON object passed as parameter to the command, defaults to {} + :return: Unwrapped result for the call. + :rtype: dict + :raises SmartPlugException: if command was not executed correctly + """ + response = TPLinkSmartHomeProtocol.query( + host=self.ip_address, + request={target: {cmd: arg}} + ) + + result = response[target][cmd] + if result["err_code"] != 0: + raise SmartPlugException("Error on {}.{}: {}".format(target, cmd, result)) + + del result["err_code"] + + return result @property def state(self): @@ -70,9 +117,9 @@ class SmartPlug: SWITCH_STATE_ON SWITCH_STATE_OFF SWITCH_STATE_UNKNOWN + :rtype: str """ - response = self.get_sysinfo() - relay_state = response['relay_state'] + relay_state = self.sys_info['relay_state'] if relay_state == 0: return SmartPlug.SWITCH_STATE_OFF @@ -90,8 +137,9 @@ class SmartPlug: :param value: one of SWITCH_STATE_ON SWITCH_STATE_OFF - :return: True if new state was successfully set - False if an error occured + :raises ValueError: on invalid state + :raises SmartPlugException: on error + """ if value.upper() == SmartPlug.SWITCH_STATE_ON: self.turn_on() @@ -100,55 +148,56 @@ class SmartPlug: else: raise ValueError("State %s is not valid.", value) + self.initialize() + def get_sysinfo(self): """ Retrieve system information. - :return: dict sysinfo + :return: sysinfo + :rtype dict + :raises SmartPlugException: on error """ - response = TPLinkSmartHomeProtocol.query( - host=self.ip_address, - request={'system': {'get_sysinfo': {}}} - )['system']['get_sysinfo'] + return self._query_helper("system", "get_sysinfo") - if response['err_code'] != 0: - return False + @property + def is_on(self): + """ + Returns whether device is on. - return response + :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. - :return: True on success - :raises ProtocolError when device responds with err_code != 0 + :raises SmartPlugException: on error """ - response = TPLinkSmartHomeProtocol.query( - host=self.ip_address, - request={'system': {'set_relay_state': {'state': 1}}} - )['system']['set_relay_state'] + self._query_helper("system", "set_relay_state", {"state": 1}) - if response['err_code'] != 0: - return False - - return True + self.initialize() def turn_off(self): """ Turn the switch off. - :return: True on success - False on error + :raises SmartPlugException: on error """ - response = TPLinkSmartHomeProtocol.query( - host=self.ip_address, - request={'system': {'set_relay_state': {'state': 0}}} - )['system']['set_relay_state'] + self._query_helper("system", "set_relay_state", {"state": 0}) - if response['err_code'] != 0: - return False - - return True + self.initialize() @property def has_emeter(self): @@ -164,22 +213,15 @@ class SmartPlug: """ Retrive current energy readings from device. - :returns: dict with current readings + :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 - response = TPLinkSmartHomeProtocol.query( - host=self.ip_address, request={'emeter': {'get_realtime': {}}} - )['emeter']['get_realtime'] - - if response['err_code'] != 0: - return False - - del response['err_code'] - - return response + return self._query_helper("emeter", "get_realtime") def get_emeter_daily(self, year=None, month=None): """ @@ -188,8 +230,10 @@ class SmartPlug: :param year: year for which to retrieve statistics (default: this year) :param month: month for which to retrieve statistcs (default: this month) - :return: dict: mapping of day of month to value + :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 @@ -199,14 +243,8 @@ class SmartPlug: if month is None: month = datetime.datetime.now().month - 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 + response = self._query_helper("emeter", "get_daystat", + {'month': month, 'year': year}) return {entry['day']: entry['energy'] for entry in response['day_list']} @@ -217,18 +255,15 @@ class SmartPlug: :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 or error occured + False if device has no energy meter + :rtype: dict + :raises SmartPlugException: on error """ if not self.has_emeter: return False - response = TPLinkSmartHomeProtocol.query( - host=self.ip_address, - request={'emeter': {'get_monthstat': {'year': str(year)}}} - )['emeter']['get_monthstat'] - - if response['err_code'] != 0: - return False + response = self._query_helper("emeter", "get_monthstat", + {'year': year}) return {entry['month']: entry['energy'] for entry in response['month_list']} @@ -238,24 +273,27 @@ class SmartPlug: Erase energy meter statistics :return: True if statistics were deleted - False if device has no energy meter or error occured + False if device has no energy meter. + :rtype: bool + :raises SmartPlugException: on error """ if not self.has_emeter: return False - response = TPLinkSmartHomeProtocol.query( - host=self.ip_address, - request={'emeter': {'erase_emeter_stat': None}} - )['emeter']['erase_emeter_stat'] + self._query_helper("emeter", "erase_emeter_stat", None) - return response['err_code'] == 0 + self.initialize() + + # 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 of error occured. + False if device has no energy meter. + :raises SmartPlugException: on error """ if not self.has_emeter: return False @@ -268,13 +306,12 @@ class SmartPlug: """ Query device information to identify model and featureset - :return: str model, list of supported features + :return: (alias, model, list of supported features) + :rtype: tuple """ - sys_info = self.get_sysinfo() - - alias = sys_info['alias'] - model = sys_info['model'] - features = sys_info['feature'].split(':') + alias = self.sys_info['alias'] + model = self.sys_info['model'] + features = self.sys_info['feature'].split(':') for feature in features: if feature not in SmartPlug.ALL_FEATURES: @@ -283,6 +320,199 @@ class SmartPlug: return alias, model, features + @property + def alias(self): + """ + Get current device alias (name) + + :return: Device name aka alias. + :rtype: str + """ + return self._alias + + @alias.setter + def alias(self, alias): + """ + Sets the device name aka alias. + + :param alias: New alias (name) + :raises SmartPlugException: on error + """ + self._query_helper("system", "set_dev_alias", {"alias": alias}) + + self.initialize() + + @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)}) + + self.initialize() + + @property + def icon(self): + """ + Returns device icon + + Note: not working on HS110, but is always empty. + + :return: icon and its hash + :rtype: dict + :raises SmartPlugException: on error + """ + return self._query_helper("system", "get_dev_icon") + + @icon.setter + def icon(self, icon): + """ + Content for hash and icon are unknown. + + :param str icon: Icon path(?) + :raises NotImplementedError: when not implemented + :raises SmartPlugError: on error + """ + raise NotImplementedError("Values for this call are unknown at this point.") + # here just for the sake of completeness + # self._query_helper("system", "set_dev_icon", {"icon": "", "hash": ""}) + # self.initialize() + + @property + def time(self): + """ + Returns current time from the device. + + :return: datetime for device's time + :rtype: datetime.datetime + :raises SmartPlugException: on error + """ + res = self._query_helper("time", "get_time") + return datetime.datetime(res["year"], res["month"], res["mday"], + res["hour"], res["min"], res["sec"]) + + @time.setter + def time(self, ts): + """ + Sets time based on datetime object. + Note: this calls set_timezone() for setting. + + :param datetime.datetime ts: New date and time + :return: result + :type: dict + :raises NotImplemented: when not implemented. + :raises SmartPlugException: on error + """ + raise NotImplementedError("Fails with err_code == 0 with HS110.") + """ here just for the sake of completeness / if someone figures out why it doesn't work. + ts_obj = { + "index": self.timezone["index"], + "hour": ts.hour, + "min": ts.minute, + "sec": ts.second, + "year": ts.year, + "month": ts.month, + "mday": ts.day, + } + + + response = self._query_helper("time", "set_timezone", ts_obj) + self.initialize() + + return response + """ + + @property + def timezone(self): + """ + Returns timezone information + + :return: Timezone information + :rtype: dict + :raises SmartPlugException: on error + """ + return self._query_helper("time", "get_timezone") + + @property + def hw_info(self): + """ + Returns information about hardware + + :return: Information about hardware + :rtype: dict + """ + keys = ["sw_ver", "hw_ver", "mac", "hwId", "fwId", "oemId", "dev_name"] + return {key: self.sys_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): + """ + Location of the device, as read from sysinfo + + :return: latitude and longitude + :rtype: dict + """ + + return {"latitude": self.sys_info["latitude"], + "longitude": self.sys_info["longitude"]} + + @property + def rssi(self): + """ + Returns WiFi signal strenth (rssi) + + :return: rssi + :rtype: int + """ + return self.sys_info["rssi"] + + @property + def mac(self): + """ + Returns mac address + + :return: mac address in hexadecimal with colons, e.g. 01:23:45:67:89:ab + :rtype: str + """ + return self.sys_info["mac"] + + @mac.setter + def mac(self, mac): + """ + Sets new mac address + + :param str mac: mac in hexadecimal with colons, e.g. 01:23:45:67:89:ab + :raises SmartPlugException: on error + """ + self._query_helper("system", "set_mac_addr", {"mac": mac}) + + self.initialize() + + class TPLinkSmartHomeProtocol: """ @@ -305,8 +535,8 @@ class TPLinkSmartHomeProtocol: Request information from a TP-Link SmartHome Device and return the response. - :param host: ip address of the device - :param port: port on the device (default: 9999) + :param str host: ip address of the device + :param int port: port on the device (default: 9999) :param request: command to send to the device (can be either dict or json string) :return: @@ -316,12 +546,23 @@ class TPLinkSmartHomeProtocol: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((host, port)) + + _LOGGER.debug("> (%i) %s", len(request), request) sock.send(TPLinkSmartHomeProtocol.encrypt(request)) - buffer = sock.recv(4096)[4:] + + buffer = bytes() + while True: + chunk = sock.recv(4096) + buffer += chunk + if not chunk: + break + sock.shutdown(socket.SHUT_RDWR) sock.close() - response = TPLinkSmartHomeProtocol.decrypt(buffer) + response = TPLinkSmartHomeProtocol.decrypt(buffer[4:]) + _LOGGER.debug("< (%i) %s", len(response), response) + return json.loads(response) @staticmethod