Read all data from the device, disable double-encoding, implement more APIs, refactor querying, update README (#11)

* Read from socket until no data available, disable double string encoding

HS110 sends sometimes datagrams in chunks especially for get_daystat,
this patch makes it to read until there is no more data to be read.

As json.dumps() does JSON encoding already, there's no need to str()
the year or month either.

* Add cli.py, a simple script to query devices for debugging purposes.

* allow easier importing with from pyHS100 import SmartPlug

* move cli.py to examples, add short usage into README.md

* Implement more available APIs, refactor querying code.

This commit adds access to new properties, both read & write,  while keeping the old one (mostly) intact.
Querying is refactored to be done inside _query_helper() method,
which unwraps results automatically and rises SmartPlugException() in case of errors.
Errors are to be handled by clients.

New features:
* Setting device alias (plug.alias = "name")
* led read & write
* icon read (doesn't seem to return anything without cloud support at least), write API is not known, throws an exception currently
* time read (returns datetime), time write implemented, but not working even when no error is returned from the device
* timezone read
* mac read & write, writing is untested for now.

Properties for easier access:
* hw_info: return hw-specific elements from sysinfo
* on_since: pretty-printed from sysinfo
* location: latitude and longitued from sysinfo
* rssi: rssi from sysinfo

* Update README.md with examples of available features.

* Handle comments from mweinelt

* Refactor state handling, use booleans instead of strings

* Fix issues raised during the review.

Following issues are addressed by this commit:
* All API is more or less commented (including return types, exceptions, ..)
* Converted state to use
* Added properties is_on, is_off for those who don't want to check against strings.
* Handled most issues reported by pylint.
* Adjusted _query_helper() to strip off err_code from the result object.
* Fixed broken format() syntax for string formattings.

* Fix ci woes plus one typo.

* Do initialization after changing device properties, fix nits.
This commit is contained in:
Teemu R 2016-12-12 10:13:45 +01:00 committed by GadgetReactor
parent 1ffdc7bc67
commit 05a6bbb145
4 changed files with 404 additions and 81 deletions

View File

@ -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 <ip>
```

23
examples/cli.py Normal file
View File

@ -0,0 +1,23 @@
import sys
import logging
from pyHS100 import SmartPlug
logging.basicConfig(level=logging.DEBUG)
if len(sys.argv) < 2:
print("%s <ip>" % 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))

View File

@ -1 +1 @@
from pyHS100.pyHS100 import SmartPlug

View File

@ -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