diff --git a/.travis.yml b/.travis.yml index e226c8d3..52f8bdf3 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,6 +1,7 @@ sudo: false language: python python: + - "3.3" - "3.4" - "3.5" - "3.6" diff --git a/pyHS100/discover.py b/pyHS100/discover.py index a2803ece..be97dec9 100644 --- a/pyHS100/discover.py +++ b/pyHS100/discover.py @@ -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 diff --git a/pyHS100/protocol.py b/pyHS100/protocol.py index 51271e3e..2fcff7a7 100644 --- a/pyHS100/protocol.py +++ b/pyHS100/protocol.py @@ -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)) diff --git a/pyHS100/smartbulb.py b/pyHS100/smartbulb.py index f1781d33..43c16728 100644 --- a/pyHS100/smartbulb.py +++ b/pyHS100/smartbulb.py @@ -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 diff --git a/pyHS100/smartdevice.py b/pyHS100/smartdevice.py index d6112628..df8b5152 100644 --- a/pyHS100/smartdevice.py +++ b/pyHS100/smartdevice.py @@ -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. diff --git a/pyHS100/smartplug.py b/pyHS100/smartplug.py index b916fc8f..02a36a8c 100644 --- a/pyHS100/smartplug.py +++ b/pyHS100/smartplug.py @@ -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 + } diff --git a/pyHS100/tests/test_bulb.py b/pyHS100/tests/test_bulb.py index 6dcf0ee4..5260ee8c 100644 --- a/pyHS100/tests/test_bulb.py +++ b/pyHS100/tests/test_bulb.py @@ -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): diff --git a/pyHS100/tests/test_pyHS100.py b/pyHS100/tests/test_pyHS100.py index 0d7721a2..adb8dfa1 100644 --- a/pyHS100/tests/test_pyHS100.py +++ b/pyHS100/tests/test_pyHS100.py @@ -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. diff --git a/pyHS100/types.py b/pyHS100/types.py index 80272507..8152d32e 100644 --- a/pyHS100/types.py +++ b/pyHS100/types.py @@ -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 diff --git a/setup.py b/setup.py index 6c370861..05d1576b 100644 --- a/setup.py +++ b/setup.py @@ -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', diff --git a/tox.ini b/tox.ini index 6d3f3107..324a3f9d 100644 --- a/tox.ini +++ b/tox.ini @@ -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