add typing hints to make it easier for 3rd party developers to use the library (#90)

* add typing hints to make it easier for 3rd party developers to use the library

* remove unused devicetype enum to support python3.3

* add python 3.3 to travis and tox, install typing module in setup.py
This commit is contained in:
Teemu R 2017-09-18 18:13:06 +02:00 committed by GitHub
parent 3ddd31f3c1
commit af90a36153
11 changed files with 163 additions and 135 deletions

View File

@ -1,6 +1,7 @@
sudo: false
language: python
python:
- "3.3"
- "3.4"
- "3.5"
- "3.6"

View File

@ -1,20 +1,24 @@
import socket
import logging
import json
from typing import Dict
from pyHS100 import TPLinkSmartHomeProtocol, SmartPlug, SmartBulb
from pyHS100 import TPLinkSmartHomeProtocol, SmartDevice, SmartPlug, SmartBulb
_LOGGER = logging.getLogger(__name__)
class Discover:
@staticmethod
def discover(protocol=None, port=9999, timeout=3):
def discover(protocol: TPLinkSmartHomeProtocol = None,
port: int = 9999,
timeout: int = 3) -> Dict[str, SmartDevice]:
"""
Sends discovery message to 255.255.255.255:9999 in order
to detect available supported devices in the local network,
and waits for given timeout for answers from devices.
:param protocol: Protocol implementation to use
:param timeout: How long to wait for responses, defaults to 5
:param port: port to send broadcast messages, defaults to 9999.
:rtype: dict

View File

@ -2,6 +2,7 @@ import json
import socket
import struct
import logging
from typing import Any, Dict, Union
_LOGGER = logging.getLogger(__name__)
@ -24,7 +25,9 @@ class TPLinkSmartHomeProtocol:
DEFAULT_TIMEOUT = 5
@staticmethod
def query(host, request, port=DEFAULT_PORT):
def query(host: str,
request: Union[str, Dict],
port: int = DEFAULT_PORT) -> Any:
"""
Request information from a TP-Link SmartHome Device and return the
response.
@ -76,7 +79,7 @@ class TPLinkSmartHomeProtocol:
return json.loads(response)
@staticmethod
def encrypt(request):
def encrypt(request: str) -> bytearray:
"""
Encrypt a request for a TP-Link Smart Home Device.
@ -94,7 +97,7 @@ class TPLinkSmartHomeProtocol:
return buffer
@staticmethod
def decrypt(ciphertext):
def decrypt(ciphertext: bytes) -> str:
"""
Decrypt a response of a TP-Link Smart Home Device.
@ -104,9 +107,9 @@ class TPLinkSmartHomeProtocol:
key = TPLinkSmartHomeProtocol.INITIALIZATION_VECTOR
buffer = []
ciphertext = ciphertext.decode('latin-1')
ciphertext_str = ciphertext.decode('latin-1')
for char in ciphertext:
for char in ciphertext_str:
plain = key ^ ord(char)
key = ord(char)
buffer.append(chr(plain))

View File

@ -1,4 +1,5 @@
from pyHS100 import SmartDevice
from typing import Any, Dict, Optional, Tuple
class SmartBulb(SmartDevice):
@ -40,13 +41,15 @@ class SmartBulb(SmartDevice):
BULB_STATE_ON = 'ON'
BULB_STATE_OFF = 'OFF'
def __init__(self, ip_address, protocol=None):
def __init__(self,
ip_address: str,
protocol: 'TPLinkSmartHomeProtocol' = None) -> None:
SmartDevice.__init__(self, ip_address, protocol)
self.emeter_type = "smartlife.iot.common.emeter"
self.emeter_units = True
@property
def is_color(self):
def is_color(self) -> bool:
"""
Whether the bulb supports color changes
@ -56,7 +59,7 @@ class SmartBulb(SmartDevice):
return bool(self.sys_info['is_color'])
@property
def is_dimmable(self):
def is_dimmable(self) -> bool:
"""
Whether the bulb supports brightness changes
@ -66,7 +69,7 @@ class SmartBulb(SmartDevice):
return bool(self.sys_info['is_dimmable'])
@property
def is_variable_color_temp(self):
def is_variable_color_temp(self) -> bool:
"""
Whether the bulb supports color temperature changes
@ -76,16 +79,16 @@ class SmartBulb(SmartDevice):
"""
return bool(self.sys_info['is_variable_color_temp'])
def get_light_state(self):
def get_light_state(self) -> Dict:
return self._query_helper("smartlife.iot.smartbulb.lightingservice",
"get_light_state")
def set_light_state(self, state):
def set_light_state(self, state: Dict) -> Dict:
return self._query_helper("smartlife.iot.smartbulb.lightingservice",
"transition_light_state", state)
@property
def hsv(self):
def hsv(self) -> Optional[Tuple[int, int, int]]:
"""
Returns the current HSV state of the bulb, if supported
@ -109,7 +112,7 @@ class SmartBulb(SmartDevice):
return hue, saturation, value
@hsv.setter
def hsv(self, state):
def hsv(self, state: Tuple[int, int, int]):
"""
Sets new HSV, if supported
@ -124,10 +127,10 @@ class SmartBulb(SmartDevice):
"brightness": int(state[2] * 100 / 255),
"color_temp": 0
}
return self.set_light_state(light_state)
self.set_light_state(light_state)
@property
def color_temp(self):
def color_temp(self) -> Optional[int]:
"""
Color temperature of the device, if supported
@ -139,12 +142,12 @@ class SmartBulb(SmartDevice):
light_state = self.get_light_state()
if not self.is_on:
return light_state['dft_on_state']['color_temp']
return int(light_state['dft_on_state']['color_temp'])
else:
return light_state['color_temp']
return int(light_state['color_temp'])
@color_temp.setter
def color_temp(self, temp):
def color_temp(self, temp: int) -> None:
"""
Set the color temperature of the device, if supported
@ -156,10 +159,10 @@ class SmartBulb(SmartDevice):
light_state = {
"color_temp": temp,
}
return self.set_light_state(light_state)
self.set_light_state(light_state)
@property
def brightness(self):
def brightness(self) -> Optional[int]:
"""
Current brightness of the device, if supported
@ -171,12 +174,12 @@ class SmartBulb(SmartDevice):
light_state = self.get_light_state()
if not self.is_on:
return light_state['dft_on_state']['brightness']
return int(light_state['dft_on_state']['brightness'])
else:
return light_state['brightness']
return int(light_state['brightness'])
@brightness.setter
def brightness(self, brightness):
def brightness(self, brightness: int) -> None:
"""
Set the current brightness of the device, if supported
@ -188,10 +191,10 @@ class SmartBulb(SmartDevice):
light_state = {
"brightness": brightness,
}
return self.set_light_state(light_state)
self.set_light_state(light_state)
@property
def state(self):
def state(self) -> str:
"""
Retrieve the bulb state
@ -205,42 +208,8 @@ class SmartBulb(SmartDevice):
return self.BULB_STATE_ON
return self.BULB_STATE_OFF
@property
def state_information(self):
"""
Return bulb-specific state information.
:return: Bulb information dict, keys in user-presentable form.
:rtype: dict
"""
info = {
'Brightness': self.brightness,
'Is dimmable': self.is_dimmable,
}
if self.is_variable_color_temp:
info["Color temperature"] = self.color_temp
if self.is_color:
info["HSV"] = self.hsv
return info
@property
def is_on(self):
return self.state == self.BULB_STATE_ON
def turn_off(self):
"""
Turn the bulb off.
"""
self.state = self.BULB_STATE_OFF
def turn_on(self):
"""
Turn the bulb on.
"""
self.state = self.BULB_STATE_ON
@state.setter
def state(self, bulb_state):
def state(self, bulb_state: str) -> None:
"""
Set the new bulb state
@ -249,17 +218,51 @@ class SmartBulb(SmartDevice):
BULB_STATE_OFF
"""
if bulb_state == self.BULB_STATE_ON:
bulb_state = 1
new_state = 1
elif bulb_state == self.BULB_STATE_OFF:
bulb_state = 0
new_state = 0
else:
raise ValueError
light_state = {
"on_off": bulb_state,
"on_off": new_state,
}
return self.set_light_state(light_state)
self.set_light_state(light_state)
@property
def has_emeter(self):
def state_information(self) -> Dict[str, Any]:
"""
Return bulb-specific state information.
:return: Bulb information dict, keys in user-presentable form.
:rtype: dict
"""
info = {
'Brightness': self.brightness,
'Is dimmable': self.is_dimmable,
} # type: Dict[str, Any]
if self.is_variable_color_temp:
info["Color temperature"] = self.color_temp
if self.is_color:
info["HSV"] = self.hsv
return info
@property
def is_on(self) -> bool:
return bool(self.state == self.BULB_STATE_ON)
def turn_off(self) -> None:
"""
Turn the bulb off.
"""
self.state = self.BULB_STATE_OFF
def turn_on(self) -> None:
"""
Turn the bulb on.
"""
self.state = self.BULB_STATE_ON
@property
def has_emeter(self) -> bool:
return True

View File

@ -18,6 +18,7 @@ import logging
import socket
import warnings
from collections import defaultdict
from typing import Any, Dict, List, Tuple, Optional
from .types import SmartDeviceException
from .protocol import TPLinkSmartHomeProtocol
@ -32,7 +33,9 @@ class SmartDevice(object):
ALL_FEATURES = (FEATURE_ENERGY_METER, FEATURE_TIMER)
def __init__(self, ip_address, protocol=None):
def __init__(self,
ip_address: str,
protocol: Optional[TPLinkSmartHomeProtocol] = None) -> None:
"""
Create a new SmartDevice instance, identified through its IP address.
@ -43,8 +46,13 @@ class SmartDevice(object):
if not protocol:
protocol = TPLinkSmartHomeProtocol()
self.protocol = protocol
self.emeter_type = "emeter" # type: str
self.emeter_units = False
def _query_helper(self, target, cmd, arg=None):
def _query_helper(self,
target: str,
cmd: str,
arg: Optional[Dict] = None) -> Any:
"""
Helper returning unwrapped result object and doing error handling.
@ -80,7 +88,7 @@ class SmartDevice(object):
return result
@property
def features(self):
def features(self) -> List[str]:
"""
Returns features of the devices
@ -95,7 +103,7 @@ class SmartDevice(object):
)
warnings.simplefilter('default', DeprecationWarning)
if "feature" not in self.sys_info:
return None
return []
features = self.sys_info['feature'].split(':')
@ -107,7 +115,7 @@ class SmartDevice(object):
return features
@property
def has_emeter(self):
def has_emeter(self) -> bool:
"""
Checks feature list for energey meter support.
@ -117,7 +125,7 @@ class SmartDevice(object):
return SmartDevice.FEATURE_ENERGY_METER in self.features
@property
def sys_info(self):
def sys_info(self) -> Dict[str, Any]:
"""
Returns the complete system information from the device.
@ -126,7 +134,7 @@ class SmartDevice(object):
"""
return defaultdict(lambda: None, self.get_sysinfo())
def get_sysinfo(self):
def get_sysinfo(self) -> Dict:
"""
Retrieve system information.
@ -136,7 +144,7 @@ class SmartDevice(object):
"""
return self._query_helper("system", "get_sysinfo")
def identify(self):
def identify(self) -> Tuple[str, str, Any]:
"""
Query device information to identify model and featureset
@ -157,7 +165,7 @@ class SmartDevice(object):
return info["alias"], info["model"], self.features
@property
def model(self):
def model(self) -> str:
"""
Get model of the device
@ -165,20 +173,20 @@ class SmartDevice(object):
:rtype: str
:raises SmartDeviceException: on error
"""
return self.sys_info['model']
return str(self.sys_info['model'])
@property
def alias(self):
def alias(self) -> str:
"""
Get current device alias (name)
:return: Device name aka alias.
:rtype: str
"""
return self.sys_info['alias']
return str(self.sys_info['alias'])
@alias.setter
def alias(self, alias):
def alias(self, alias: str) -> None:
"""
Sets the device name aka alias.
@ -188,7 +196,7 @@ class SmartDevice(object):
self._query_helper("system", "set_dev_alias", {"alias": alias})
@property
def icon(self):
def icon(self) -> Dict:
"""
Returns device icon
@ -201,7 +209,7 @@ class SmartDevice(object):
return self._query_helper("system", "get_dev_icon")
@icon.setter
def icon(self, icon):
def icon(self, icon: str) -> None:
"""
Content for hash and icon are unknown.
@ -216,7 +224,7 @@ class SmartDevice(object):
# self.initialize()
@property
def time(self):
def time(self) -> Optional[datetime.datetime]:
"""
Returns current time from the device.
@ -232,7 +240,7 @@ class SmartDevice(object):
return None
@time.setter
def time(self, ts):
def time(self, ts: datetime.datetime) -> None:
"""
Sets time based on datetime object.
Note: this calls set_timezone() for setting.
@ -266,7 +274,7 @@ class SmartDevice(object):
"""
@property
def timezone(self):
def timezone(self) -> Dict:
"""
Returns timezone information
@ -277,7 +285,7 @@ class SmartDevice(object):
return self._query_helper("time", "get_timezone")
@property
def hw_info(self):
def hw_info(self) -> Dict:
"""
Returns information about hardware
@ -290,7 +298,7 @@ class SmartDevice(object):
return {key: info[key] for key in keys if key in info}
@property
def location(self):
def location(self) -> Dict:
"""
Location of the device, as read from sysinfo
@ -313,17 +321,19 @@ class SmartDevice(object):
return loc
@property
def rssi(self):
def rssi(self) -> Optional[int]:
"""
Returns WiFi signal strenth (rssi)
:return: rssi
:rtype: int
"""
return self.sys_info["rssi"]
if "rssi" in self.sys_info:
return int(self.sys_info["rssi"])
return None
@property
def mac(self):
def mac(self) -> str:
"""
Returns mac address
@ -333,15 +343,15 @@ class SmartDevice(object):
info = self.sys_info
if 'mac' in info:
return info["mac"]
return str(info["mac"])
elif 'mic_mac' in info:
return info['mic_mac']
return str(info['mic_mac'])
else:
raise SmartDeviceException("Unknown mac, please submit a bug"
"with sysinfo output.")
@mac.setter
def mac(self, mac):
def mac(self, mac: str) -> None:
"""
Sets new mac address
@ -350,7 +360,7 @@ class SmartDevice(object):
"""
self._query_helper("system", "set_mac_addr", {"mac": mac})
def get_emeter_realtime(self):
def get_emeter_realtime(self) -> Optional[Dict]:
"""
Retrive current energy readings from device.
@ -364,7 +374,9 @@ class SmartDevice(object):
return self._query_helper(self.emeter_type, "get_realtime")
def get_emeter_daily(self, year=None, month=None):
def get_emeter_daily(self,
year: int = None,
month: int = None) -> Optional[Dict]:
"""
Retrieve daily statistics for a given month
@ -395,7 +407,7 @@ class SmartDevice(object):
return {entry['day']: entry[key]
for entry in response['day_list']}
def get_emeter_monthly(self, year=datetime.datetime.now().year):
def get_emeter_monthly(self, year=None) -> Optional[Dict]:
"""
Retrieve monthly statistics for a given year.
@ -408,6 +420,9 @@ class SmartDevice(object):
if not self.has_emeter:
return None
if year is None:
year = datetime.datetime.now().year
response = self._query_helper(self.emeter_type, "get_monthstat",
{'year': year})
@ -419,7 +434,7 @@ class SmartDevice(object):
return {entry['month']: entry[key]
for entry in response['month_list']}
def erase_emeter_stats(self):
def erase_emeter_stats(self) -> bool:
"""
Erase energy meter statistics
@ -429,7 +444,7 @@ class SmartDevice(object):
:raises SmartDeviceException: on error
"""
if not self.has_emeter:
return None
return False
self._query_helper(self.emeter_type, "erase_emeter_stat", None)
@ -437,7 +452,7 @@ class SmartDevice(object):
# succeeded when we are this far.
return True
def current_consumption(self):
def current_consumption(self) -> Optional[float]:
"""
Get the current power consumption in Watt.
@ -450,18 +465,18 @@ class SmartDevice(object):
response = self.get_emeter_realtime()
if self.emeter_units:
return response['power_mw']
return float(response['power_mw'])
else:
return response['power']
return float(response['power'])
def turn_off(self):
def turn_off(self) -> None:
"""
Turns the device off.
"""
raise NotImplementedError("Device subclass needs to implement this.")
@property
def is_off(self):
def is_off(self) -> bool:
"""
Returns whether device is off.
@ -470,14 +485,14 @@ class SmartDevice(object):
"""
return not self.is_on
def turn_on(self):
def turn_on(self) -> None:
"""
Turns the device on.
"""
raise NotImplementedError("Device subclass needs to implement this.")
@property
def is_on(self):
def is_on(self) -> bool:
"""
Returns whether the device is on.
@ -488,7 +503,7 @@ class SmartDevice(object):
raise NotImplementedError("Device subclass needs to implement this.")
@property
def state_information(self):
def state_information(self) -> Dict[str, Any]:
"""
Returns device-type specific, end-user friendly state information.
:return: dict with state information.

View File

@ -1,5 +1,6 @@
import datetime
import logging
from typing import Any, Dict
from pyHS100 import SmartDevice
@ -30,13 +31,15 @@ class SmartPlug(SmartDevice):
SWITCH_STATE_OFF = 'OFF'
SWITCH_STATE_UNKNOWN = 'UNKNOWN'
def __init__(self, ip_address, protocol=None):
def __init__(self,
ip_address: str,
protocol: 'TPLinkSmartHomeProtocol' = None) -> None:
SmartDevice.__init__(self, ip_address, protocol)
self.emeter_type = "emeter"
self.emeter_units = False
@property
def state(self):
def state(self) -> str:
"""
Retrieve the switch state
@ -57,7 +60,7 @@ class SmartPlug(SmartDevice):
return SmartPlug.SWITCH_STATE_UNKNOWN
@state.setter
def state(self, value):
def state(self, value: str):
"""
Set the new switch state
@ -78,7 +81,7 @@ class SmartPlug(SmartDevice):
raise ValueError("State %s is not valid.", value)
@property
def is_on(self):
def is_on(self) -> bool:
"""
Returns whether device is on.
@ -103,7 +106,7 @@ class SmartPlug(SmartDevice):
self._query_helper("system", "set_relay_state", {"state": 0})
@property
def led(self):
def led(self) -> bool:
"""
Returns the state of the led.
@ -112,15 +115,8 @@ class SmartPlug(SmartDevice):
"""
return bool(1 - self.sys_info["led_off"])
@property
def state_information(self):
return {
'LED state': self.led,
'On since': self.on_since
}
@led.setter
def led(self, state):
def led(self, state: bool):
"""
Sets the state of the led (night mode)
@ -130,7 +126,7 @@ class SmartPlug(SmartDevice):
self._query_helper("system", "set_led_off", {"off": int(not state)})
@property
def on_since(self):
def on_since(self) -> datetime.datetime:
"""
Returns pretty-printed on-time
@ -139,3 +135,10 @@ class SmartPlug(SmartDevice):
"""
return datetime.datetime.now() - \
datetime.timedelta(seconds=self.sys_info["on_time"])
@property
def state_information(self) -> Dict[str, Any]:
return {
'LED state': self.led,
'On since': self.on_since
}

View File

@ -1,11 +1,13 @@
from unittest import TestCase, skip, skipIf
from voluptuous import Schema, Invalid, All, Range
from functools import partial
from typing import Any, Dict # noqa: F401
from .. import SmartBulb, SmartDeviceException
from .fakes import (FakeTransportProtocol,
sysinfo_lb100, sysinfo_lb110,
sysinfo_lb120, sysinfo_lb130)
BULB_IP = '192.168.250.186'
SKIP_STATE_TESTS = False
@ -24,7 +26,7 @@ def check_mode(x):
class TestSmartBulb(TestCase):
SYSINFO = sysinfo_lb130
SYSINFO = sysinfo_lb130 # type: Dict[str, Any]
# these schemas should go to the mainlib as
# they can be useful when adding support for new features/devices
# as well as to check that faked devices are operating properly.
@ -172,7 +174,7 @@ class TestSmartBulb(TestCase):
def test_current_consumption(self):
x = self.bulb.current_consumption()
self.assertTrue(isinstance(x, int))
self.assertTrue(isinstance(x, float))
self.assertTrue(x >= 0.0)
def test_alias(self):

View File

@ -3,6 +3,7 @@ from voluptuous import Schema, Invalid, All, Any, Range
from functools import partial
import datetime
import re
from typing import Dict # noqa: F401
from .. import SmartPlug, SmartDeviceException
from .fakes import (FakeTransportProtocol,
@ -35,7 +36,7 @@ def check_mode(x):
class TestSmartPlugHS110(TestCase):
SYSINFO = sysinfo_hs110
SYSINFO = sysinfo_hs110 # type: Dict
# these schemas should go to the mainlib as
# they can be useful when adding support for new features/devices
# as well as to check that faked devices are operating properly.

View File

@ -1,15 +1,5 @@
import enum
class SmartDeviceException(Exception):
"""
SmartDeviceException gets raised for errors reported by the plug.
"""
pass
class DeviceType(enum.Enum):
Unknown = -1,
Plug = 0,
Switch = 1
Bulb = 2

View File

@ -8,7 +8,7 @@ setup(name='pyHS100',
author_email='sean@gadgetreactor.com',
license='GPLv3',
packages=['pyHS100'],
install_requires=['click', 'click-datetime'],
install_requires=['click', 'click-datetime', 'typing'],
entry_points={
'console_scripts': [
'pyhs100=pyHS100.cli:cli',

View File

@ -1,8 +1,9 @@
[tox]
envlist=py34,py35,py36,flake8
envlist=py33,py34,py35,py36,flake8
[tox:travis]
2.7 = py27
3.3 = py34
3.4 = py34
3.5 = py35
3.6 = py36
@ -13,6 +14,7 @@ deps=
pytest
pytest-cov
voluptuous
typing
commands=
py.test --cov pyHS100
@ -20,5 +22,9 @@ commands=
deps=flake8
commands=flake8 pyHS100
[testenv:typing]
deps=mypy
commands=mypy --silent-imports pyHS100
[flake8]
exclude = .git,.tox,__pycache__,pyHS100/tests/fakes.py