Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2025-10-22 15:28:01 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Add klap protocol (#509)

Browse Source 
* Add support for the new encryption protocol

This adds support for the new TP-Link discovery and encryption
protocols. It is currently incomplete - only devices without
username and password are current supported, and single device
discovery is not implemented.

Discovery should find both old and new devices. When accessing
a device by IP the --klap option can be specified on the command
line to active the new connection protocol.

sdb9696 - This commit also contains 16 later commits from Simon Wilkinson
squashed into the original

* Update klap changes 2023 to fix encryption, deal with kasa credential switching and work with new discovery changes

* Move from aiohttp to httpx

* Changes following review comments

---------

Co-authored-by: Simon Wilkinson <simon@sxw.org.uk>
This commit is contained in:
sdb9696
2023-11-20 13:17:10 +00:00
committed by GitHub
parent bde07d117f
commit 30f217b8ab
10 changed files with 1297 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
kasa/smartdevice.py
View File

@@ -24,7 +24,7 @@ from .credentials import Credentials
from .emeterstatus import EmeterStatus
from .exceptions import SmartDeviceException
from .modules import Emeter, Module
from .protocol import TPLinkSmartHomeProtocol
from .protocol import TPLinkProtocol, TPLinkSmartHomeProtocol
_LOGGER = logging.getLogger(__name__)
@@ -71,7 +71,7 @@ def requires_update(f):
@functools.wraps(f)
async def wrapped(*args, **kwargs):
self = args[0]
if self._last_update is None:
if self._last_update is None and f.__name__ not in self._sys_info:
raise SmartDeviceException(
"You need to await update() to access the data"
)
@@ -82,7 +82,7 @@ def requires_update(f):
@functools.wraps(f)
def wrapped(*args, **kwargs):
self = args[0]
if self._last_update is None:
if self._last_update is None and f.__name__ not in self._sys_info:
raise SmartDeviceException(
"You need to await update() to access the data"
)
@@ -213,8 +213,9 @@ class SmartDevice:
"""
self.host = host
self.port = port
self.protocol = TPLinkSmartHomeProtocol(host, port=port, timeout=timeout)
self.protocol: TPLinkProtocol = TPLinkSmartHomeProtocol(
host, port=port, timeout=timeout
)
self.credentials = credentials
_LOGGER.debug("Initializing %s of type %s", self.host, type(self))
self._device_type = DeviceType.Unknown
@@ -222,6 +223,7 @@ class SmartDevice:
# checks in accessors. the @updated_required decorator does not ensure
# mypy that these are not accessed incorrectly.
self._last_update: Any = None
self._sys_info: Any = None # TODO: this is here to avoid changing tests
self._features: Set[str] = set()
self.modules: Dict[str, Any] = {}
@@ -374,8 +376,14 @@ class SmartDevice:
def update_from_discover_info(self, info: Dict[str, Any]) -> None:
"""Update state from info from the discover call."""
self._last_update = info
self._set_sys_info(info["system"]["get_sysinfo"])
if "system" in info and (sys_info := info["system"].get("get_sysinfo")):
self._last_update = info
self._set_sys_info(sys_info)
else:
# This allows setting of some info properties directly
# from partial discovery info that will then be found
# by the requires_update decorator
self._set_sys_info(info)
def _set_sys_info(self, sys_info: Dict[str, Any]) -> None:
"""Set sys_info."""
@@ -388,21 +396,26 @@ class SmartDevice:
@property # type: ignore
@requires_update
def sys_info(self) -> Dict[str, Any]:
"""Return system information."""
"""
Return system information.
Do not call this function from within the SmartDevice
class itself as @requires_update will be affected for other properties.
"""
return self._sys_info # type: ignore
@property # type: ignore
@requires_update
def model(self) -> str:
"""Return device model."""
sys_info = self.sys_info
sys_info = self._sys_info
return str(sys_info["model"])
@property # type: ignore
@requires_update
def alias(self) -> str:
"""Return device name (alias)."""
sys_info = self.sys_info
sys_info = self._sys_info
return str(sys_info["alias"])
async def set_alias(self, alias: str) -> None:
@@ -454,14 +467,14 @@ class SmartDevice:
"oemId",
"dev_name",
]
sys_info = self.sys_info
sys_info = self._sys_info
return {key: sys_info[key] for key in keys if key in sys_info}
@property # type: ignore
@requires_update
def location(self) -> Dict:
"""Return geographical location."""
sys_info = self.sys_info
sys_info = self._sys_info
loc = {"latitude": None, "longitude": None}
if "latitude" in sys_info and "longitude" in sys_info:
@@ -479,7 +492,7 @@ class SmartDevice:
@requires_update
def rssi(self) -> Optional[int]:
"""Return WiFi signal strength (rssi)."""
rssi = self.sys_info.get("rssi")
rssi = self._sys_info.get("rssi")
return None if rssi is None else int(rssi)
@property # type: ignore
@@ -489,14 +502,14 @@ class SmartDevice:
:return: mac address in hexadecimal with colons, e.g. 01:23:45:67:89:ab
"""
sys_info = self.sys_info
sys_info = self._sys_info
mac = sys_info.get("mac", sys_info.get("mic_mac"))
if not mac:
raise SmartDeviceException(
"Unknown mac, please submit a bug report with sys_info output."
)
mac = mac.replace("-", ":")
# Format a mac that has no colons (usually from mic_mac field)
if ":" not in mac:
mac = ":".join(format(s, "02x") for s in bytes.fromhex(mac))
@@ -607,13 +620,13 @@ class SmartDevice:
@requires_update
def on_since(self) -> Optional[datetime]:
"""Return pretty-printed on-time, or None if not available."""
if "on_time" not in self.sys_info:
if "on_time" not in self._sys_info:
return None
if self.is_off:
return None
on_time = self.sys_info["on_time"]
on_time = self._sys_info["on_time"]
return datetime.now().replace(microsecond=0) - timedelta(seconds=on_time)