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

make SmartDevice use asyncio

Browse Source
This commit is contained in:
Bas Nijholt
2019-11-11 17:21:23 +01:00
parent af0a1d61b1
commit 904bbe5c1a
1 changed files with 74 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

171
pyHS100/smartdevice.py
View File

@@ -13,11 +13,12 @@ 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 asyncio
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
@@ -150,7 +151,9 @@ class SmartDevice:
self.cache[target][cmd] = response.copy()
self.cache[target][cmd]["last_updated"] = datetime.utcnow()
async def _query_helper(self, target: str, cmd: str, arg: Optional[Dict] = None) -> Any:
async def _query_helper(
self, target: str, cmd: str, arg: Optional[Dict] = None
) -> Any:
"""Handle result unwrapping and error handling.
:param target: Target system {system, time, emeter, ..}
@@ -196,8 +199,7 @@ class SmartDevice:
return result
@property
def has_emeter(self) -> bool:
async def get_has_emeter(self) -> bool:
"""Return if device has an energy meter.
:return: True if energey meter is available
@@ -213,54 +215,45 @@ class SmartDevice:
:rtype: dict
"""
return self.get_sysinfo()
return asyncio.run(self.get_sys_info())
def get_sysinfo(self) -> Dict:
async def get_sys_info(self) -> Dict:
"""Retrieve system information.
:return: sysinfo
:rtype dict
:raises SmartDeviceException: on error
"""
return self._query_helper("system", "get_sysinfo")
return await self._query_helper("system", "get_sysinfo")
@property
def model(self) -> str:
async def get_model(self) -> str:
"""Return device model.
:return: device model
:rtype: str
:raises SmartDeviceException: on error
"""
return str(self.sys_info["model"])
sys_info = await self.get_sys_info()
return str(sys_info["model"])
@property
def alias(self) -> str:
async def get_alias(self) -> str:
"""Return device name (alias).
:return: Device name aka alias.
:rtype: str
"""
return str(self.sys_info["alias"])
sys_info = await self.get_sys_info()
return str(sys_info["alias"])
def get_alias(self) -> str:
return self.alias
@alias.setter # type: ignore
@deprecated(details="use set_alias")
def alias(self, alias: str) -> None:
self.set_alias(alias)
def set_alias(self, alias: str) -> None:
async def set_alias(self, alias: str) -> None:
"""Set the device name (alias).
:param alias: New alias (name)
:raises SmartDeviceException: on error
"""
self._query_helper("system", "set_dev_alias", {"alias": alias})
await self._query_helper("system", "set_dev_alias", {"alias": alias})
@property
def icon(self) -> Dict:
async def get_icon(self) -> Dict:
"""Return device icon.
Note: not working on HS110, but is always empty.
@@ -269,10 +262,9 @@ class SmartDevice:
:rtype: dict
:raises SmartDeviceException: on error
"""
return self._query_helper("system", "get_dev_icon")
return await self._query_helper("system", "get_dev_icon")
@icon.setter
def icon(self, icon: str) -> None:
def set_icon(self, icon: str) -> None:
"""Set device icon.
Content for hash and icon are unknown.
@@ -283,12 +275,11 @@ class SmartDevice:
"""
raise NotImplementedError()
# here just for the sake of completeness
# self._query_helper("system",
# await self._query_helper("system",
# "set_dev_icon", {"icon": "", "hash": ""})
# self.initialize()
@property
def time(self) -> Optional[datetime]:
async def get_time(self) -> Optional[datetime]:
"""Return current time from the device.
:return: datetime for device's time
@@ -296,7 +287,7 @@ class SmartDevice:
:raises SmartDeviceException: on error
"""
try:
res = self._query_helper("time", "get_time")
res = await self._query_helper("time", "get_time")
return datetime(
res["year"],
res["month"],
@@ -308,8 +299,7 @@ class SmartDevice:
except SmartDeviceException:
return None
@time.setter
def time(self, ts: datetime) -> None:
async def set_time(self, ts: datetime) -> None:
"""Set the device time.
Note: this calls set_timezone() for setting.
@@ -336,24 +326,22 @@ class SmartDevice:
}
response = self._query_helper("time", "set_timezone", ts_obj)
response = await self._query_helper("time", "set_timezone", ts_obj)
self.initialize()
return response
"""
@property
def timezone(self) -> Dict:
async def get_timezone(self) -> Dict:
"""Return timezone information.
:return: Timezone information
:rtype: dict
:raises SmartDeviceException: on error
"""
return self._query_helper("time", "get_timezone")
return await self._query_helper("time", "get_timezone")
@property
def hw_info(self) -> Dict:
async def get_hw_info(self) -> Dict:
"""Return hardware information.
:return: Information about hardware
@@ -371,17 +359,16 @@ class SmartDevice:
"oemId",
"dev_name",
]
info = self.sys_info
return {key: info[key] for key in keys if key in info}
sys_info = await self.get_sys_info()
return {key: sys_info[key] for key in keys if key in sys_info}
@property
def location(self) -> Dict:
async def get_location(self) -> Dict:
"""Return geographical location.
:return: latitude and longitude
:rtype: dict
"""
info = self.sys_info
info = await self.get_sys_info()
loc = {"latitude": None, "longitude": None}
if "latitude" in info and "longitude" in info:
@@ -395,49 +382,43 @@ class SmartDevice:
return loc
@property
def rssi(self) -> Optional[int]:
async def get_rssi(self) -> Optional[int]:
"""Return WiFi signal strenth (rssi).
:return: rssi
:rtype: int
"""
if "rssi" in self.sys_info:
return int(self.sys_info["rssi"])
sys_info = await self.get_sys_info()
if "rssi" in sys_info:
return int(sys_info["rssi"])
return None
@property
def mac(self) -> str:
async def get_mac(self) -> str:
"""Return mac address.
:return: mac address in hexadecimal with colons, e.g. 01:23:45:67:89:ab
:rtype: str
"""
info = self.sys_info
sys_info = await self.get_sys_info()
if "mac" in info:
return str(info["mac"])
elif "mic_mac" in info:
return ":".join(format(s, "02x") for s in bytes.fromhex(info["mic_mac"]))
if "mac" in sys_info:
return str(sys_info["mac"])
elif "mic_mac" in sys_info:
return ":".join(format(s, "02x") for s in bytes.fromhex(sys_info["mic_mac"]))
raise SmartDeviceException(
"Unknown mac, please submit a bug " "with sysinfo output."
"Unknown mac, please submit a bug report with sys_info output."
)
@mac.setter
@deprecated(details="use set_mac")
def mac(self, mac: str) -> None:
self.set_mac(mac)
def set_mac(self, mac):
async def set_mac(self, mac):
"""Set the mac address.
:param str mac: mac in hexadecimal with colons, e.g. 01:23:45:67:89:ab
:raises SmartDeviceException: on error
"""
self._query_helper("system", "set_mac_addr", {"mac": mac})
await self._query_helper("system", "set_mac_addr", {"mac": mac})
def get_emeter_realtime(self) -> EmeterStatus:
async def get_emeter_realtime(self) -> EmeterStatus:
"""Retrive current energy readings.
:returns: current readings or False
@@ -445,12 +426,12 @@ class SmartDevice:
None if device has no energy meter or error occurred
:raises SmartDeviceException: on error
"""
if not self.has_emeter:
if not await self.get_has_emeter():
raise SmartDeviceException("Device has no emeter")
return EmeterStatus(self._query_helper(self.emeter_type, "get_realtime"))
return EmeterStatus(await self._query_helper(self.emeter_type, "get_realtime"))
def get_emeter_daily(
async def get_emeter_daily(
self, year: int = None, month: int = None, kwh: bool = True
) -> Dict:
"""Retrieve daily statistics for a given month.
@@ -464,7 +445,7 @@ class SmartDevice:
:rtype: dict
:raises SmartDeviceException: on error
"""
if not self.has_emeter:
if not await self.get_has_emeter():
raise SmartDeviceException("Device has no emeter")
if year is None:
@@ -472,7 +453,7 @@ class SmartDevice:
if month is None:
month = datetime.now().month
response = self._query_helper(
response = await self._query_helper(
self.emeter_type, "get_daystat", {"month": month, "year": year}
)
response = [EmeterStatus(**x) for x in response["day_list"]]
@@ -485,7 +466,7 @@ class SmartDevice:
return data
def get_emeter_monthly(self, year: int = None, kwh: bool = True) -> Dict:
async 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)
@@ -495,13 +476,15 @@ class SmartDevice:
:rtype: dict
:raises SmartDeviceException: on error
"""
if not self.has_emeter:
if not await self.get_has_emeter():
raise SmartDeviceException("Device has no emeter")
if year is None:
year = datetime.now().year
response = self._query_helper(self.emeter_type, "get_monthstat", {"year": year})
response = await self._query_helper(
self.emeter_type, "get_monthstat", {"year": year}
)
response = [EmeterStatus(**x) for x in response["month_list"]]
key = "energy_wh"
@@ -510,7 +493,7 @@ class SmartDevice:
return {entry["month"]: entry[key] for entry in response}
def erase_emeter_stats(self) -> bool:
async def erase_emeter_stats(self) -> bool:
"""Erase energy meter statistics.
:return: True if statistics were deleted
@@ -518,29 +501,29 @@ class SmartDevice:
:rtype: bool
:raises SmartDeviceException: on error
"""
if not self.has_emeter:
if not await self.get_has_emeter():
raise SmartDeviceException("Device has no emeter")
self._query_helper(self.emeter_type, "erase_emeter_stat", None)
await 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]:
async def current_consumption(self) -> Optional[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:
if not await self.get_has_emeter():
raise SmartDeviceException("Device has no emeter")
response = EmeterStatus(self.get_emeter_realtime())
return response["power"]
def reboot(self, delay=1) -> None:
async def reboot(self, delay=1) -> None:
"""Reboot the device.
Note that giving a delay of zero causes this to block,
@@ -549,27 +532,25 @@ class SmartDevice:
:param delay: Delay the reboot for `delay` seconds.
:return: None
"""
self._query_helper("system", "reboot", {"delay": delay})
await self._query_helper("system", "reboot", {"delay": delay})
def turn_off(self) -> None:
async def turn_off(self) -> None:
"""Turn off the device."""
raise NotImplementedError("Device subclass needs to implement this.")
@property
def is_off(self) -> bool:
async def is_off(self) -> bool:
"""Return True if device is off.
:return: True if device is off, False otherwise.
:rtype: bool
"""
return not self.is_on
return not await self.is_on()
def turn_on(self) -> None:
async def turn_on(self) -> None:
"""Turn device on."""
raise NotImplementedError("Device subclass needs to implement this.")
@property
def is_on(self) -> bool:
async def is_on(self) -> bool:
"""Return if the device is on.
:return: True if the device is on, False otherwise.
@@ -578,8 +559,7 @@ class SmartDevice:
"""
raise NotImplementedError("Device subclass needs to implement this.")
@property
def state_information(self) -> Dict[str, Any]:
async def get_state_information(self) -> Dict[str, Any]:
"""Return device-type specific, end-user friendly state information.
:return: dict with state information.
@@ -613,14 +593,11 @@ class SmartDevice:
return False
def __repr__(self):
is_on = self.is_on
if callable(is_on):
is_on = is_on()
return "<%s model %s at %s (%s), is_on: %s - dev specific: %s>" % (
self.__class__.__name__,
self.model,
asyncio.run(self.get_model()),
self.host,
self.alias,
is_on,
self.state_information,
asyncio.run(self.get_alias()),
asyncio.run(self.is_on()),
asyncio.run(self.get_state_information()),
)