Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2024-12-22 19:23:34 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Add precommit hooks & tox env to enforce code structure, add azure pipelines

Browse Source 
Instead of leveraging hound & travis, add an option for azure pipelines,
which may replace the former in the future.

This also streamlines the contributing guidelines by:
* Adding pre-commit hooks to run isort, black, flake8, mypy
* Adding lint environment to allow checks to be run `tox -e lint`

This also contains a major cleanup to the SmartStrip handling which was due.
After seeing #184 I thought I should push this in as it is for comments before the codebase diverges too much.
This commit is contained in:
Teemu Rytilahti 2019-11-11 22:14:34 +01:00
parent 59424d2738
commit 8a131e1eeb
17 changed files with 354 additions and 549 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.flake8
View File

@ -1,5 +0,0 @@
[flake8]
ignore = E203, E266, E501, W503, F403, F401
max-line-length = 79
max-complexity = 18
select = B,C,E,F,W,T4,B9

28
.pre-commit-config.yaml
View File

@ -1,6 +1,24 @@
repos:
- repo: https://github.com/python/black
rev: stable
hooks:
- id: black
language_version: python3.7
- repo: https://github.com/python/black
rev: stable
hooks:
- id: black
language_version: python3.7
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v2.3.0
hooks:
- id: flake8
additional_dependencies: [flake8-docstrings]
- repo: https://github.com/pre-commit/mirrors-isort
rev: v4.3.21
hooks:
- id: isort
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v0.740
hooks:
- id: mypy
# args: [--no-strict-optional, --ignore-missing-imports]

47
azure-pipelines.yml Normal file
View File

@ -0,0 +1,47 @@
trigger:
- master
pr:
- master
pool:
vmImage: 'ubuntu-latest'
strategy:
matrix:
Python36:
python.version: '3.6'
Python37:
python.version: '3.7'
# Python38:
# python.version: '3.8'
steps:
- task: UsePythonVersion@0
inputs:
versionSpec: '$(python.version)'
displayName: 'Use Python $(python.version)'
- script: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install pytest pytest-azurepipelines pytest-cov
displayName: 'Install dependencies'
- script: |
pre-commit run black --all-files
displayName: 'Code formating (black)'
- script: |
pre-commit run flake8 --all-files
displayName: 'Code formating (flake8)'
- script: |
pre-commit run mypy --all-files
displayName: 'Typing checks (mypy)'
- script: |
pre-commit run isort --all-files
displayName: 'Order of imports (isort)'
- script: |
pytest --cov pyHS100 --cov-report html
displayName: 'Tests'

26
pyHS100/__init__.py
View File

@ -1,22 +1,24 @@
"""
This module provides a way to interface with TP-Link's smart home devices,
such as smart plugs (HS1xx), wall switches (HS2xx), and light bulbs (LB1xx).
"""Python interface for TP-Link's smart home devices.
All common, shared functionalities are available through `SmartDevice` class::
x = SmartDevice("192.168.1.1")
print(x.sys_info)
For device type specific actions `SmartBulb` or `SmartPlug` must be used instead.
For device type specific actions `SmartBulb`, `SmartPlug`, or `SmartStrip`
should be used instead.
Module-specific errors are raised as `SmartDeviceException` and are expected
to be handled by the user of the library.
"""
# flake8: noqa
from .smartdevice import SmartDevice, SmartDeviceException, EmeterStatus
from .smartdevice import SmartDevice, SmartDeviceException, EmeterStatus, DeviceType
from .smartplug import SmartPlug
from .smartbulb import SmartBulb
from .smartstrip import SmartStrip, SmartStripException
from .protocol import TPLinkSmartHomeProtocol
from .discover import Discover
from .discover import Discover # noqa
from .protocol import TPLinkSmartHomeProtocol # noqa
from .smartbulb import SmartBulb # noqa
from .smartdevice import ( # noqa
DeviceType,
EmeterStatus,
SmartDevice,
SmartDeviceException,
)
from .smartplug import SmartPlug # noqa
from .smartstrip import SmartStrip # noqa

41
pyHS100/cli.py
View File

@ -1,20 +1,17 @@
"""pyHS100 cli tool."""
import sys
import click
import logging
import sys
from pprint import pformat as pf
if sys.version_info < (3, 4):
print("To use this script you need python 3.4 or newer! got %s" % sys.version_info)
import click
from pyHS100 import SmartPlug # noqa: E402
from pyHS100 import Discover, SmartBulb, SmartDevice, SmartStrip
if sys.version_info < (3, 6):
print("To use this script you need python 3.6 or newer! got %s" % sys.version_info)
sys.exit(1)
from pyHS100 import (
SmartDevice,
SmartPlug,
SmartBulb,
SmartStrip,
Discover,
) # noqa: E402
pass_dev = click.make_pass_decorator(SmartDevice)
@ -50,9 +47,10 @@ pass_dev = click.make_pass_decorator(SmartDevice)
@click.option("--bulb", default=False, is_flag=True)
@click.option("--plug", default=False, is_flag=True)
@click.option("--strip", default=False, is_flag=True)
@click.version_option()
@click.pass_context
def cli(ctx, ip, host, alias, target, debug, bulb, plug, strip):
"""A cli tool for controlling TP-Link smart home plugs."""
"""A cli tool for controlling TP-Link smart home plugs.""" # noqa
if debug:
logging.basicConfig(level=logging.DEBUG)
else:
@ -100,6 +98,10 @@ def cli(ctx, ip, host, alias, target, debug, bulb, plug, strip):
@click.option("--save")
@click.pass_context
def dump_discover(ctx, save):
"""Dump discovery information.
Useful for dumping into a file with `--save` to be added to the test suite.
"""
target = ctx.parent.params["target"]
for dev in Discover.discover(target=target, return_raw=True).values():
model = dev["system"]["get_sysinfo"]["model"]
@ -164,7 +166,7 @@ def sysinfo(dev):
@cli.command()
@pass_dev
@click.pass_context
def state(ctx, dev):
def state(ctx, dev: SmartDevice):
"""Print out device state and versions."""
click.echo(click.style("== %s - %s ==" % (dev.alias, dev.model), bold=True))
@ -174,15 +176,12 @@ def state(ctx, dev):
fg="green" if dev.is_on else "red",
)
)
if dev.num_children > 0:
is_on = dev.get_is_on()
aliases = dev.get_alias()
for child in range(dev.num_children):
if dev.is_strip:
for plug in dev.plugs: # type: ignore
click.echo(
click.style(
" * %s state: %s"
% (aliases[child], ("ON" if is_on[child] else "OFF")),
fg="green" if is_on[child] else "red",
" * %s state: %s" % (plug.alias, ("ON" if plug.is_on else "OFF")),
fg="green" if plug.is_on else "red",
)
)
@ -303,7 +302,7 @@ def temperature(dev: SmartBulb, temperature):
@click.pass_context
@pass_dev
def hsv(dev, ctx, h, s, v):
"""Get or set color in HSV. (Bulb only)"""
"""Get or set color in HSV. (Bulb only)."""
if h is None or s is None or v is None:
click.echo("Current HSV: %s %s %s" % dev.hsv)
elif s is None or v is None:

24
pyHS100/discover.py
View File

@ -1,16 +1,14 @@
import socket
import logging
"""Discovery module for TP-Link Smart Home devices."""
import json
from typing import Dict, Type, Optional
import logging
import socket
from typing import Dict, Optional, Type
from pyHS100 import (
TPLinkSmartHomeProtocol,
SmartDevice,
SmartPlug,
SmartBulb,
SmartStrip,
SmartDeviceException,
)
from .protocol import TPLinkSmartHomeProtocol
from .smartbulb import SmartBulb
from .smartdevice import SmartDevice, SmartDeviceException
from .smartplug import SmartPlug
from .smartstrip import SmartStrip
_LOGGER = logging.getLogger(__name__)
@ -28,8 +26,6 @@ class Discover:
you can initialize the corresponding device class directly without this.
The protocol uses UDP broadcast datagrams on port 9999 for discovery.
"""
DISCOVERY_QUERY = {
@ -49,8 +45,8 @@ class Discover:
discovery_packets=3,
return_raw=False,
) -> Dict[str, SmartDevice]:
"""Discover devices.
"""
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.

27
pyHS100/protocol.py
View File

@ -1,24 +1,25 @@
"""Implementation of the TP-Link Smart Home Protocol.
Encryption/Decryption methods based on the works of
Lubomir Stroetmann and Tobias Esser
https://www.softscheck.com/en/reverse-engineering-tp-link-hs110/
https://github.com/softScheck/tplink-smartplug/
which are licensed under the Apache License, Version 2.0
http://www.apache.org/licenses/LICENSE-2.0
"""
import json
import logging
import socket
import struct
import logging
from typing import Any, Dict, Union
_LOGGER = logging.getLogger(__name__)
class TPLinkSmartHomeProtocol:
"""Implementation of the TP-Link Smart Home Protocol.
Encryption/Decryption methods based on the works of
Lubomir Stroetmann and Tobias Esser
https://www.softscheck.com/en/reverse-engineering-tp-link-hs110/
https://github.com/softScheck/tplink-smartplug/
which are licensed under the Apache License, Version 2.0
http://www.apache.org/licenses/LICENSE-2.0
"""
"""Implementation of the TP-Link Smart Home protocol."""
INITIALIZATION_VECTOR = 171
DEFAULT_PORT = 9999
@ -78,7 +79,7 @@ class TPLinkSmartHomeProtocol:
return json.loads(response)
@staticmethod
def encrypt(request: str) -> bytearray:
def encrypt(request: str) -> bytes:
"""
Encrypt a request for a TP-Link Smart Home Device.

27
pyHS100/smartbulb.py
View File

@ -1,10 +1,11 @@
from pyHS100 import DeviceType, SmartDevice, SmartDeviceException
from .protocol import TPLinkSmartHomeProtocol
from deprecation import deprecated
"""Module for bulbs."""
import re
from datetime import datetime
from typing import Any, Dict, Tuple
from deprecation import deprecated
from .protocol import TPLinkSmartHomeProtocol
from .smartdevice import DeviceType, SmartDevice, SmartDeviceException
TPLINK_KELVIN = {
"LB130": (2500, 9000),
@ -12,8 +13,8 @@ TPLINK_KELVIN = {
"LB230": (2500, 9000),
"KB130": (2500, 9000),
"KL130": (2500, 9000),
"KL120\(EU\)": (2700, 6500),
"KL120\(US\)": (2700, 5000),
r"KL120\(EU\)": (2700, 6500),
r"KL120\(US\)": (2700, 5000),
}
@ -131,7 +132,6 @@ class SmartBulb(SmartDevice):
:return: hue, saturation and value (degrees, %, %)
:rtype: tuple
"""
if not self.is_color:
raise SmartDeviceException("Bulb does not support color.")
@ -161,7 +161,9 @@ class SmartBulb(SmartDevice):
def set_hsv(self, hue: int, saturation: int, value: int):
"""Set new HSV.
:param tuple state: hue, saturation and value (degrees, %, %)
:param int hue: hue in degrees
:param int saturation: saturation in percentage [0,100]
:param int value: value in percentage [0, 100]
"""
if not self.is_color:
raise SmartDeviceException("Bulb does not support color.")
@ -230,7 +232,7 @@ class SmartBulb(SmartDevice):
@property
def brightness(self) -> int:
"""Current brightness of the device.
"""Return the current brightness.
:return: brightness in percent
:rtype: int
@ -250,7 +252,7 @@ class SmartBulb(SmartDevice):
self.set_brightness(brightness)
def set_brightness(self, brightness: int) -> None:
"""Set the current brightness of the device.
"""Set the brightness.
:param int brightness: brightness in percent
"""
@ -303,10 +305,10 @@ class SmartBulb(SmartDevice):
:return: Bulb information dict, keys in user-presentable form.
:rtype: dict
"""
info = {
info: Dict[str, Any] = {
"Brightness": self.brightness,
"Is dimmable": self.is_dimmable,
} # type: Dict[str, Any]
}
if self.is_variable_color_temp:
info["Color temperature"] = self.color_temp
info["Valid temperature range"] = self.valid_temperature_range
@ -331,4 +333,5 @@ class SmartBulb(SmartDevice):
@property
def has_emeter(self) -> bool:
"""Return that the bulb has an emeter."""
return True

46
pyHS100/smartdevice.py
View File

@ -1,6 +1,4 @@
"""
pyHS100
Python library supporting TP-Link Smart Plugs/Switches (HS100/HS110/Hs200).
"""Python library supporting TP-Link Smart Home devices.
The communication protocol was reverse engineered by Lubomir Stroetmann and
Tobias Esser in 'Reverse Engineering the TP-Link HS110':
@ -13,11 +11,11 @@ Stroetmann which is licensed under the Apache License, Version 2.0.
You may obtain a copy of the license at
http://www.apache.org/licenses/LICENSE-2.0
"""
from datetime import datetime, timedelta
import logging
from collections import defaultdict
from typing import Any, Dict, Optional
from datetime import datetime, timedelta
from enum import Enum
from typing import Any, Dict, Optional
from deprecation import deprecated
@ -102,7 +100,7 @@ class SmartDevice:
if protocol is None: # pragma: no cover
protocol = TPLinkSmartHomeProtocol()
self.protocol = protocol
self.emeter_type = "emeter" # type: str
self.emeter_type = "emeter"
self.context = context
self.num_children = 0
self.cache_ttl = timedelta(seconds=cache_ttl)
@ -112,11 +110,12 @@ class SmartDevice:
self.context,
self.cache_ttl,
)
self.cache = defaultdict(lambda: defaultdict(lambda: None))
self.cache = defaultdict(lambda: defaultdict(lambda: None)) # type: ignore
self._device_type = DeviceType.Unknown
def _result_from_cache(self, target, cmd) -> Optional[Dict]:
"""Return query result from cache if still fresh.
Only results from commands starting with `get_` are considered cacheable.
:param target: Target system
@ -141,7 +140,7 @@ class SmartDevice:
return None
def _insert_to_cache(self, target: str, cmd: str, response: Dict) -> None:
"""Internal function to add response to cache.
"""Add response for a given command to the cache.
:param target: Target system
:param cmd: Command
@ -160,9 +159,8 @@ class SmartDevice:
:rtype: dict
:raises SmartDeviceException: if command was not executed correctly
"""
if self.context is None:
request = {target: {cmd: arg}}
else:
request: Dict[str, Any] = {target: {cmd: arg}}
if self.context is not None:
request = {"context": {"child_ids": [self.context]}, target: {cmd: arg}}
try:
@ -212,7 +210,6 @@ class SmartDevice:
:return: System information dict.
:rtype: dict
"""
return self.get_sysinfo()
def get_sysinfo(self) -> Dict:
@ -244,11 +241,13 @@ class SmartDevice:
return str(self.sys_info["alias"])
def get_alias(self) -> str:
"""Return the alias."""
return self.alias
@alias.setter # type: ignore
@deprecated(details="use set_alias")
def alias(self, alias: str) -> None:
"""Set the device name, deprecated."""
self.set_alias(alias)
def set_alias(self, alias: str) -> None:
@ -424,7 +423,7 @@ class SmartDevice:
"Unknown mac, please submit a bug " "with sysinfo output."
)
@mac.setter
@mac.setter # type: ignore
@deprecated(details="use set_mac")
def mac(self, mac: str) -> None:
self.set_mac(mac)
@ -438,11 +437,10 @@ class SmartDevice:
self._query_helper("system", "set_mac_addr", {"mac": mac})
def get_emeter_realtime(self) -> EmeterStatus:
"""Retrive current energy readings.
"""Retrieve current energy readings.
:returns: current readings or False
:rtype: dict, None
None if device has no energy meter or error occurred
:raises SmartDeviceException: on error
"""
if not self.has_emeter:
@ -460,7 +458,6 @@ class SmartDevice:
month)
:param kwh: return usage in kWh (default: True)
:return: mapping of day of month to value
None if device has no energy meter or error occurred
:rtype: dict
:raises SmartDeviceException: on error
"""
@ -491,7 +488,6 @@ class SmartDevice:
:param year: year for which to retrieve statistics (default: this year)
:param kwh: return usage in kWh (default: True)
:return: dict: mapping of month to value
None if device has no energy meter
:rtype: dict
:raises SmartDeviceException: on error
"""
@ -510,12 +506,10 @@ class SmartDevice:
return {entry["month"]: entry[key] for entry in response}
def erase_emeter_stats(self) -> bool:
def erase_emeter_stats(self):
"""Erase energy meter statistics.
:return: True if statistics were deleted
False if device has no energy meter.
:rtype: bool
:raises SmartDeviceException: on error
"""
if not self.has_emeter:
@ -523,15 +517,10 @@ class SmartDevice:
self._query_helper(self.emeter_type, "erase_emeter_stat", None)
# As query_helper raises exception in case of failure, we have
# succeeded when we are this far.
return True
def current_consumption(self) -> Optional[float]:
def current_consumption(self) -> float:
"""Get the current power consumption in Watt.
:return: the current power consumption in Watts.
None if device has no energy meter.
:raises SmartDeviceException: on error
"""
if not self.has_emeter:
@ -594,22 +583,27 @@ class SmartDevice:
@property
def is_bulb(self) -> bool:
"""Return True if the device is a bulb."""
return self._device_type == DeviceType.Bulb
@property
def is_plug(self) -> bool:
"""Return True if the device is a plug."""
return self._device_type == DeviceType.Plug
@property
def is_strip(self) -> bool:
"""Return True if the device is a strip."""
return self._device_type == DeviceType.Strip
@property
def is_dimmable(self):
"""Return True if the device is dimmable."""
return False
@property
def is_variable_color_temp(self) -> bool:
"""Return True if the device supports color temperature."""
return False
def __repr__(self):

3
pyHS100/smartplug.py
View File

@ -1,11 +1,12 @@
"""Module for plugs."""
import datetime
import logging
from typing import Any, Dict
from deprecation import deprecated
from pyHS100 import SmartDevice, DeviceType, SmartDeviceException
from .protocol import TPLinkSmartHomeProtocol
from .smartdevice import DeviceType, SmartDevice, SmartDeviceException
_LOGGER = logging.getLogger(__name__)

380
pyHS100/smartstrip.py
View File

@ -1,35 +1,37 @@
"""Module for multi-socket devices (HS300, HS107).
.. todo:: describe how this interfaces with single plugs.
"""
import datetime
import logging
from typing import Any, Dict, Optional, Union
from deprecation import deprecated
from collections import defaultdict
from typing import Any, DefaultDict, Dict, List
from pyHS100 import SmartPlug, SmartDeviceException, EmeterStatus, DeviceType
from .protocol import TPLinkSmartHomeProtocol
from .smartplug import DeviceType, SmartPlug
_LOGGER = logging.getLogger(__name__)
class SmartStripException(SmartDeviceException):
"""SmartStripException gets raised for errors specific to the smart strip."""
pass
class SmartStrip(SmartPlug):
"""Representation of a TP-Link Smart Power Strip.
Usage example when used as library:
p = SmartStrip("192.168.1.105")
# query the state of the strip
print(p.is_on)
# change state of all outlets
p.turn_on()
p.turn_off()
# change state of a single outlet
p.turn_on(index=1)
# individual outlets are accessible through plugs variable
for plug in p.plugs:
print("%s: %s" % (p, p.is_on))
# query and print current state of all outlets
print(p.get_state())
# change state of a single outlet
p.plugs[0].turn_on()
Errors reported by the device are raised as SmartDeviceExceptions,
and should be handled by the user of the library.
@ -41,366 +43,122 @@ class SmartStrip(SmartPlug):
SmartPlug.__init__(self, host=host, protocol=protocol, cache_ttl=cache_ttl)
self.emeter_type = "emeter"
self._device_type = DeviceType.Strip
self.plugs = {}
self.plugs: List[SmartPlug] = []
children = self.sys_info["children"]
self.num_children = len(children)
for plug in range(self.num_children):
self.plugs[plug] = SmartPlug(
host, protocol, context=children[plug]["id"], cache_ttl=cache_ttl
self.plugs.append(
SmartPlug(
host, protocol, context=children[plug]["id"], cache_ttl=cache_ttl
)
)
def raise_for_index(self, index: int):
"""
Raises SmartStripException if the plug index is out of bounds
:param index: plug index to check
:raises SmartStripException: index out of bounds
"""
if index not in range(self.num_children):
raise SmartStripException("plug index of %d " "is out of bounds" % index)
@property
@deprecated(details="use is_on, get_is_on()")
def state(self) -> bool:
if self.is_on:
return self.STATE_ON
return self.STATE_OFF
def get_state(self, *, index=-1) -> Dict[int, str]:
"""Retrieve the switch state
:returns: list with the state of each child plug
STATE_ON
STATE_OFF
:rtype: dict
"""
def _state_for_bool(b):
return SmartPlug.STATE_ON if b else SmartPlug.STATE_OFF
is_on = self.get_is_on(index=index)
if isinstance(is_on, bool):
return _state_for_bool(is_on)
print(is_on)
return {k: _state_for_bool(v) for k, v in self.get_is_on().items()}
@state.setter
@deprecated(details="use turn_on(), turn_off()")
def state(self, value: str):
"""Sets the state of all plugs in the strip
:param value: one of
STATE_ON
STATE_OFF
:raises ValueError: on invalid state
:raises SmartDeviceException: on error
"""
if not isinstance(value, str):
raise ValueError("State must be str, not of %s.", type(value))
elif value.upper() == SmartPlug.STATE_ON:
self.turn_on()
elif value.upper() == SmartPlug.STATE_OFF:
self.turn_off()
else:
raise ValueError("State %s is not valid.", value)
def set_state(self, value: str, *, index: int = -1):
"""Sets the state of a plug on the strip
:param value: one of
STATE_ON
STATE_OFF
:param index: plug index (-1 for all)
:raises ValueError: on invalid state
:raises SmartDeviceException: on error
:raises SmartStripException: index out of bounds
"""
if index < 0:
self.state = value
else:
self.raise_for_index(index)
self.plugs[index].state = value
@property
def is_on(self) -> bool:
"""Return if any of the outlets are on"""
return any(state == "ON" for state in self.get_state().values())
"""Return if any of the outlets are on."""
return any(plug.is_on for plug in self.plugs)
def get_is_on(self, *, index: int = -1) -> Any:
"""
Returns whether device is on.
def turn_on(self):
"""Turn the strip on.
:param index: plug index (-1 for all)
:return: True if device is on, False otherwise, Dict without index
:rtype: bool if index is provided
Dict[int, bool] if no index provided
:raises SmartStripException: index out of bounds
"""
children = self.sys_info["children"]
if index < 0:
is_on = {}
for plug in range(self.num_children):
is_on[plug] = bool(children[plug]["state"])
return is_on
else:
self.raise_for_index(index)
return bool(children[index]["state"])
def get_is_off(self, *, index: int = -1) -> Any:
is_on = self.get_is_on(index=index)
if isinstance(is_on, bool):
return not is_on
else:
return {k: not v for k, v in is_on}
def turn_on(self, *, index: int = -1):
"""
Turns outlets on
:param index: plug index (-1 for all)
:raises SmartDeviceException: on error
:raises SmartStripException: index out of bounds
"""
if index < 0:
self._query_helper("system", "set_relay_state", {"state": 1})
else:
self.raise_for_index(index)
self.plugs[index].turn_on()
self._query_helper("system", "set_relay_state", {"state": 1})
def turn_off(self, *, index: int = -1):
"""
Turns outlets off
def turn_off(self):
"""Turn the strip off.
:param index: plug index (-1 for all)
:raises SmartDeviceException: on error
:raises SmartStripException: index out of bounds
"""
if index < 0:
self._query_helper("system", "set_relay_state", {"state": 0})
else:
self.raise_for_index(index)
self.plugs[index].turn_off()
self._query_helper("system", "set_relay_state", {"state": 0})
@property
def on_since(self) -> datetime:
"""Returns the maximum on-time of all outlets."""
return max(v for v in self.get_on_since().values())
def get_on_since(self, *, index: int = -1) -> Any:
"""
Returns pretty-printed on-time
:param index: plug index (-1 for all)
:return: datetime for on since
:rtype: datetime with index
Dict[int, str] without index
:raises SmartStripException: index out of bounds
"""
if index < 0:
on_since = {}
children = self.sys_info["children"]
for plug in range(self.num_children):
child_ontime = children[plug]["on_time"]
on_since[plug] = datetime.datetime.now() - datetime.timedelta(
seconds=child_ontime
)
return on_since
else:
self.raise_for_index(index)
return self.plugs[index].on_since
def on_since(self) -> datetime.datetime:
"""Return the maximum on-time of all outlets."""
return max(plug.on_since for plug in self.plugs)
@property
def state_information(self) -> Dict[str, Any]:
"""
Returns strip-specific state information.
"""Return strip-specific state information.
:return: Strip information dict, keys in user-presentable form.
:rtype: dict
"""
state = {"LED state": self.led}
is_on = self.get_is_on()
on_since = self.get_on_since()
for plug_index in range(self.num_children):
plug_number = plug_index + 1
if is_on[plug_index]:
state["Plug %d on since" % plug_number] = on_since[plug_index]
state: Dict[str, Any] = {"LED state": self.led}
for plug in self.plugs:
if plug.is_on:
state["Plug %s on since" % str(plug)] = plug.on_since
return state
def get_emeter_realtime(self, *, index: int = -1) -> Optional[Any]:
"""
Retrieve current energy readings from device
def current_consumption(self) -> float:
"""Get the current power consumption in watts.
:param index: plug index (-1 for all)
:returns: list of current readings or None
:rtype: Dict, Dict[int, Dict], None
Dict if index is provided
Dict[int, Dict] if no index provided
None if device has no energy meter or error occurred
:return: the current power consumption in watts.
:rtype: float
:raises SmartDeviceException: on error
:raises SmartStripException: index out of bounds
"""
if not self.has_emeter: # pragma: no cover
raise SmartStripException("Device has no emeter")
consumption = sum(plug.current_consumption() for plug in self.plugs)
if index < 0:
emeter_status = {}
for plug in range(self.num_children):
emeter_status[plug] = self.plugs[plug].get_emeter_realtime()
return emeter_status
else:
self.raise_for_index(index)
return self.plugs[index].get_emeter_realtime()
return consumption
def current_consumption(self, *, index: int = -1) -> Optional[Any]:
"""
Get the current power consumption in Watts.
@property # type: ignore # required to avoid mypy error on non-implemented setter
def icon(self) -> Dict:
"""Icon for the device.
:param index: plug index (-1 for all)
:return: the current power consumption in Watts.
None if device has no energy meter.
:rtype: Dict, Dict[int, Dict], None
Dict if index is provided
Dict[int, Dict] if no index provided
None if device has no energy meter or error occurred
:raises SmartDeviceException: on error
:raises SmartStripException: index out of bounds
"""
if not self.has_emeter: # pragma: no cover
raise SmartStripException("Device has no emeter")
if index < 0:
consumption = {}
emeter_reading = self.get_emeter_realtime()
for plug in range(self.num_children):
response = EmeterStatus(emeter_reading[plug])
consumption[plug] = response["power"]
return consumption
else:
self.raise_for_index(index)
response = EmeterStatus(self.get_emeter_realtime(index=index))
return response["power"]
@property
def icon(self):
"""Override for base class icon property, SmartStrip and children do not
have icons so we return dummy strings.
Overriden to keep the API, as the SmartStrip and children do not
have icons, we just return dummy strings.
"""
return {"icon": "SMARTSTRIP-DUMMY", "hash": "SMARTSTRIP-DUMMY"}
def get_alias(self, *, index: int = -1) -> Union[str, Dict[int, str]]:
"""Gets the alias for a plug.
def set_alias(self, alias: str) -> None:
"""Set the alias for the strip.
:param index: plug index (-1 for all)
:return: the current power consumption in Watts.
None if device has no energy meter.
:rtype: str if index is provided
Dict[int, str] if no index provided
:raises SmartStripException: index out of bounds
"""
children = self.sys_info["children"]
if index < 0:
alias = {}
for plug in range(self.num_children):
alias[plug] = children[plug]["alias"]
return alias
else:
self.raise_for_index(index)
return children[index]["alias"]
def set_alias(self, alias: str, *, index: int = -1):
"""Sets the alias for a plug
:param index: plug index
:param alias: new alias
:raises SmartDeviceException: on error
:raises SmartStripException: index out of bounds
"""
# Renaming the whole strip
if index < 0:
return super().set_alias(alias)
self.raise_for_index(index)
self.plugs[index].set_alias(alias)
return super().set_alias(alias)
def get_emeter_daily(
self, year: int = None, month: int = None, kwh: bool = True, *, index: int = -1
self, year: int = None, month: int = None, kwh: bool = True
) -> Dict:
"""Retrieve daily statistics for a given month
"""Retrieve daily statistics for a given month.
:param year: year for which to retrieve statistics (default: this year)
:param month: month for which to retrieve statistics (default: this
month)
:param kwh: return usage in kWh (default: True)
:return: mapping of day of month to value
None if device has no energy meter or error occurred
:rtype: dict
:raises SmartDeviceException: on error
:raises SmartStripException: index out of bounds
"""
if not self.has_emeter: # pragma: no cover
raise SmartStripException("Device has no emeter")
emeter_daily: DefaultDict[int, float] = defaultdict(lambda: 0.0)
for plug in self.plugs:
for day, value in plug.get_emeter_daily(
year=year, month=month, kwh=kwh
).items():
emeter_daily[day] += value
return emeter_daily
emeter_daily = {}
if index < 0:
for plug in range(self.num_children):
emeter_daily = self.plugs[plug].get_emeter_daily(
year=year, month=month, kwh=kwh
)
return emeter_daily
else:
self.raise_for_index(index)
return self.plugs[index].get_emeter_daily(year=year, month=month, kwh=kwh)
def get_emeter_monthly(
self, year: int = None, kwh: bool = True, *, index: int = -1
) -> Dict:
def get_emeter_monthly(self, year: int = None, kwh: bool = True) -> Dict:
"""Retrieve monthly statistics for a given year.
:param year: year for which to retrieve statistics (default: this year)
:param kwh: return usage in kWh (default: True)
:return: dict: mapping of month to value
None if device has no energy meter
:rtype: dict
:raises SmartDeviceException: on error
:raises SmartStripException: index out of bounds
"""
if not self.has_emeter: # pragma: no cover
raise SmartStripException("Device has no emeter")
emeter_monthly: DefaultDict[int, float] = defaultdict(lambda: 0.0)
for plug in self.plugs:
for month, value in plug.get_emeter_monthly(year=year, kwh=kwh):
emeter_monthly[month] += value
return emeter_monthly
emeter_monthly = {}
if index < 0:
for plug in range(self.num_children):
emeter_monthly = self.plugs[plug].get_emeter_monthly(year=year, kwh=kwh)
return emeter_monthly
else:
self.raise_for_index(index)
return self.plugs[index].get_emeter_monthly(year=year, kwh=kwh)
def erase_emeter_stats(self):
"""Erase energy meter statistics for all plugs.
def erase_emeter_stats(self, *, index: int = -1) -> bool:
"""Erase energy meter statistics
:param index: plug index (-1 for all)
:return: True if statistics were deleted
False if device has no energy meter.
:rtype: bool
:raises SmartDeviceException: on error
:raises SmartStripException: index out of bounds
"""
if not self.has_emeter: # pragma: no cover
raise SmartStripException("Device has no emeter")
if index < 0:
for plug in range(self.num_children):
self.plugs[plug].erase_emeter_stats()
else:
self.raise_for_index(index)
self.plugs[index].erase_emeter_stats()
# As query_helper raises exception in case of failure, we have
# succeeded when we are this far.
return True
for plug in self.plugs:
plug.erase_emeter_stats()

9
pyHS100/tests/conftest.py
View File

@ -1,10 +1,13 @@
import pytest
import glob
import json
import os
from .newfakes import FakeTransportProtocol
from os.path import basename
from pyHS100 import SmartPlug, SmartBulb, SmartStrip, Discover
import pytest
from pyHS100 import Discover, SmartBulb, SmartPlug, SmartStrip
from .newfakes import FakeTransportProtocol
SUPPORTED_DEVICES = glob.glob(
os.path.dirname(os.path.abspath(__file__)) + "/fixtures/*.json"

8
pyHS100/tests/newfakes.py
View File

@ -1,8 +1,10 @@
from ..protocol import TPLinkSmartHomeProtocol
from .. import SmartDeviceException
import logging
import re
from voluptuous import Schema, Range, All, Any, Coerce, Invalid, Optional, REMOVE_EXTRA
from voluptuous import REMOVE_EXTRA, All, Any, Coerce, Invalid, Optional, Range, Schema
from .. import SmartDeviceException
from ..protocol import TPLinkSmartHomeProtocol
_LOGGER = logging.getLogger(__name__)

160
pyHS100/tests/test_fixtures.py
View File

@ -1,31 +1,31 @@
import datetime
from unittest.mock import patch
import pytest
from pyHS100 import DeviceType, SmartStripException, SmartDeviceException
from .newfakes import (
BULB_SCHEMA,
PLUG_SCHEMA,
FakeTransportProtocol,
CURRENT_CONSUMPTION_SCHEMA,
TZ_SCHEMA,
)
from pyHS100 import DeviceType, SmartDeviceException
from .conftest import (
turn_on,
handle_turn_on,
plug,
strip,
bulb,
color_bulb,
non_color_bulb,
dimmable,
handle_turn_on,
has_emeter,
no_emeter,
dimmable,
non_color_bulb,
non_dimmable,
variable_temp,
non_variable_temp,
plug,
strip,
turn_on,
variable_temp,
)
from .newfakes import (
BULB_SCHEMA,
CURRENT_CONSUMPTION_SCHEMA,
PLUG_SCHEMA,
TZ_SCHEMA,
FakeTransportProtocol,
)
@ -436,67 +436,44 @@ def test_deprecated_hsv(dev, turn_on):
dev.hsv = (1, 1, 1)
@strip
def test_children_is_on(dev):
is_on = dev.get_is_on()
for i in range(dev.num_children):
assert is_on[i] == dev.get_is_on(index=i)
@strip
@turn_on
def test_children_change_state(dev, turn_on):
handle_turn_on(dev, turn_on)
for i in range(dev.num_children):
orig_state = dev.get_is_on(index=i)
for plug in dev.plugs:
orig_state = plug.is_on
if orig_state:
dev.turn_off(index=i)
assert not dev.get_is_on(index=i)
assert dev.get_is_off(index=i)
plug.turn_off()
assert not plug.is_on
assert plug.is_off
dev.turn_on(index=i)
assert dev.get_is_on(index=i)
assert not dev.get_is_off(index=i)
plug.turn_on()
assert plug.is_on
assert not plug.is_off
else:
dev.turn_on(index=i)
assert dev.get_is_on(index=i)
assert not dev.get_is_off(index=i)
dev.turn_off(index=i)
assert not dev.get_is_on(index=i)
assert dev.get_is_off(index=i)
@strip
def test_children_bounds(dev):
out_of_bounds = dev.num_children + 100
with pytest.raises(SmartDeviceException):
dev.turn_off(index=out_of_bounds)
with pytest.raises(SmartDeviceException):
dev.turn_on(index=out_of_bounds)
with pytest.raises(SmartDeviceException):
dev.get_is_on(index=out_of_bounds)
with pytest.raises(SmartDeviceException):
dev.get_alias(index=out_of_bounds)
with pytest.raises(SmartDeviceException):
dev.get_on_since(index=out_of_bounds)
plug.turn_on()
assert plug.is_on
assert not plug.is_off
plug.turn_off()
assert not plug.is_on
assert plug.is_off
@strip
def test_children_alias(dev):
original = dev.get_alias()
test_alias = "TEST1234"
for idx in range(dev.num_children):
dev.set_alias(alias=test_alias, index=idx)
assert dev.get_alias(index=idx) == test_alias
dev.set_alias(alias=original[idx], index=idx)
assert dev.get_alias(index=idx) == original[idx]
for plug in dev.plugs:
original = plug.get_alias()
plug.set_alias(alias=test_alias)
assert plug.get_alias() == test_alias
plug.set_alias(alias=original)
assert plug.get_alias() == original
@strip
def test_children_on_since(dev):
for idx in range(dev.num_children):
assert dev.get_on_since(index=idx)
for plug in dev.plugs:
assert plug.get_on_since()
@pytest.mark.skip("this test will wear out your relays")
@ -548,78 +525,61 @@ def test_all_binary_states(dev):
def test_children_get_emeter_realtime(dev):
assert dev.has_emeter
# test with index
for plug_index in range(dev.num_children):
emeter = dev.get_emeter_realtime(index=plug_index)
for plug in dev.plugs:
emeter = plug.get_emeter_realtime()
CURRENT_CONSUMPTION_SCHEMA(emeter)
# test without index
for index, emeter in dev.get_emeter_realtime().items():
CURRENT_CONSUMPTION_SCHEMA(emeter)
# TODO test that sum matches the sum of individiaul plugs.
# out of bounds
with pytest.raises(SmartStripException):
dev.get_emeter_realtime(index=dev.num_children + 100)
# for index, emeter in dev.get_emeter_realtime().items():
# CURRENT_CONSUMPTION_SCHEMA(emeter)
@strip
def test_children_get_emeter_daily(dev):
assert dev.has_emeter
# test with index
for plug_index in range(dev.num_children):
emeter = dev.get_emeter_daily(year=1900, month=1, index=plug_index)
# test individual emeters
for plug in dev.plugs:
emeter = plug.get_emeter_daily(year=1900, month=1)
assert emeter == {}
emeter = dev.get_emeter_daily(index=plug_index)
emeter = plug.get_emeter_daily()
assert len(emeter) > 0
k, v = emeter.popitem()
assert isinstance(k, int)
assert isinstance(v, float)
# test without index
# test sum of emeters
all_emeter = dev.get_emeter_daily(year=1900, month=1)
for plug_index, emeter in all_emeter.items():
assert emeter == {}
emeter = dev.get_emeter_daily(index=plug_index)
k, v = emeter.popitem()
assert isinstance(k, int)
assert isinstance(v, float)
# out of bounds
with pytest.raises(SmartStripException):
dev.get_emeter_daily(year=1900, month=1, index=dev.num_children + 100)
k, v = all_emeter.popitem()
assert isinstance(k, int)
assert isinstance(v, float)
@strip
def test_children_get_emeter_monthly(dev):
assert dev.has_emeter
# test with index
for plug_index in range(dev.num_children):
emeter = dev.get_emeter_monthly(year=1900, index=plug_index)
# test individual emeters
for plug in dev.plugs:
emeter = plug.get_emeter_monthly(year=1900)
assert emeter == {}
emeter = dev.get_emeter_monthly()
emeter = plug.get_emeter_monthly()
assert len(emeter) > 0
k, v = emeter.popitem()
assert isinstance(k, int)
assert isinstance(v, float)
# test without index
# test sum of emeters
all_emeter = dev.get_emeter_monthly(year=1900)
for index, emeter in all_emeter.items():
assert emeter == {}
assert len(emeter) > 0
k, v = emeter.popitem()
assert isinstance(k, int)
assert isinstance(v, float)
# out of bounds
with pytest.raises(SmartStripException):
dev.get_emeter_monthly(year=1900, index=dev.num_children + 100)
k, v = all_emeter.popitem()
assert isinstance(k, int)
assert isinstance(v, float)
def test_cache(dev):
@ -658,6 +618,6 @@ def test_cache_invalidates(dev):
def test_representation(dev):
import re
pattern = re.compile("<.* model .* at .* (.*), is_on: .* - dev specific: .*>")
assert pattern.match(str(dev))

5
pyHS100/tests/test_protocol.py
View File

@ -1,6 +1,7 @@
from unittest import TestCase
from ..protocol import TPLinkSmartHomeProtocol
import json
from unittest import TestCase
from ..protocol import TPLinkSmartHomeProtocol
class TestTPLinkSmartHomeProtocol(TestCase):

33
setup.py
View File

@ -1,18 +1,19 @@
from setuptools import setup
setup(name='pyHS100',
version='0.3.5',
description='Interface for TPLink HS1xx plugs, HS2xx wall switches & LB1xx bulbs',
url='https://github.com/GadgetReactor/pyHS100',
author='Sean Seah (GadgetReactor)',
author_email='sean@gadgetreactor.com',
license='GPLv3',
packages=['pyHS100'],
install_requires=['click', 'deprecation'],
python_requires='>=3.5',
entry_points={
'console_scripts': [
'pyhs100=pyHS100.cli:cli',
],
},
zip_safe=False)
with open("pyHS100/version.py") as f:
exec(f.read())
setup(
name="pyHS100",
version=__version__, # type: ignore # noqa: F821
description="Python interface for TPLink KASA-enabled smart home devices",
url="https://github.com/GadgetReactor/pyHS100",
author="Sean Seah (GadgetReactor)",
author_email="sean@gadgetreactor.com",
license="GPLv3",
packages=["pyHS100"],
install_requires=["click", "deprecation"],
python_requires=">=3.6",
entry_points={"console_scripts": ["pyhs100=pyHS100.cli:cli"]},
zip_safe=False,
)

34
tox.ini
View File

@ -1,5 +1,5 @@
[tox]
envlist=py35,py36,py37,flake8
envlist=py35,py36,py37,flake8,linting,typing
skip_missing_interpreters = True
[tox:travis]
@ -20,16 +20,31 @@ commands=
py.test --cov --cov-config=tox.ini pyHS100
[testenv:flake8]
deps=flake8
deps=
flake8
flake8-docstrings
commands=flake8 pyHS100
max-line-length=88
[testenv:typing]
skip_install=true
deps=mypy
commands=mypy --silent-imports pyHS100
commands=mypy --ignore-missing-imports pyHS100
[flake8]
exclude = .git,.tox,__pycache__,pyHS100/tests/fakes.py
exclude = .git,.tox,__pycache__,pyHS100/tests/newfakes.py,pyHS100/tests/test_fixtures.py
max-line-length = 88
per-file-ignores =
pyHS100/tests/*.py:D100,D101,D102,D103,D104
setup.py:D100
ignore = D105, D107, E203, E501, W503
#ignore = E203, E266, E501, W503, F403, F401
#max-complexity = 18
#select = B,C,E,F,W,T4,B9
[testenv:lint]
deps = pre-commit
skip_install = true
commands = pre-commit run --all-files
[coverage:run]
source = pyHS100
@ -43,3 +58,12 @@ exclude_lines =
# ignore abstract methods
raise NotImplementedError
def __repr__
[isort]
multi_line_output=3
include_trailing_comma=True
force_grid_wrap=0
use_parentheses=True
line_length=88
known_first_party=pyHS100
known_third_party=click,deprecation,pytest,setuptools,voluptuous