From e5c805c8a4fa51ba32bc8c9db9aa86d519c2c2b1 Mon Sep 17 00:00:00 2001 From: Franck Nijhof Date: Mon, 18 Oct 2021 09:07:38 +0200 Subject: [PATCH] Styling, formatting & bugfixes --- setup.py | 48 ++++++++++++------------- tuya_iot/__init__.py | 10 +++--- tuya_iot/asset.py | 10 +++--- tuya_iot/device.py | 85 +++++++++++++++++++++----------------------- tuya_iot/home.py | 49 +++++++++++++------------ tuya_iot/openapi.py | 10 ++---- tuya_iot/openmq.py | 10 ++---- 7 files changed, 102 insertions(+), 120 deletions(-) diff --git a/setup.py b/setup.py index 273bfd1..69a69b2 100644 --- a/setup.py +++ b/setup.py @@ -1,9 +1,10 @@ -from setuptools import setup, find_packages +from setuptools import find_packages, setup + from tuya_iot import __version__ def requirements(): - with open('requirements.txt') as fileobj: + with open("requirements.txt") as fileobj: return [line.strip() for line in fileobj] @@ -12,37 +13,36 @@ def requirements(): setup( - name='tuya-iot-py-sdk', - url='https://github.com/tuya/tuya-iot-app-sdk-python', + name="tuya-iot-py-sdk", + url="https://github.com/tuya/tuya-iot-app-sdk-python", author="Tuya Inc.", - author_email='developer@tuya.com', - keywords='tuya iot app sdk python', + author_email="developer@tuya.com", + keywords="tuya iot app sdk python", long_description=doc_long_description, long_description_content_type="text/markdown", - description='A Python sdk for Tuya Open API, which provides IoT capabilities, maintained by Tuya official', - license='MIT', + description="A Python sdk for Tuya Open API, which provides IoT capabilities, maintained by Tuya official", + license="MIT", project_urls={ "Bug Tracker": "https://github.com/tuya/tuya-iot-app-sdk-python/issues", - "Changes": "https://github.com/tuya/tuya-iot-python-sdk/wiki/Tuya-IoT-Python-SDK-Release-Notes" + "Changes": "https://github.com/tuya/tuya-iot-python-sdk/wiki/Tuya-IoT-Python-SDK-Release-Notes", }, classifiers=[ - 'Development Status :: 4 - Beta', - 'License :: OSI Approved :: MIT License', - 'Operating System :: OS Independent', - 'Topic :: Software Development', - 'Programming Language :: Python :: 3', - 'Programming Language :: Python :: 3.6', - 'Programming Language :: Python :: 3.7', - 'Programming Language :: Python :: 3.8', - 'Programming Language :: Python :: 3.9', - 'Programming Language :: Python :: 3.10', - 'Programming Language :: Python :: Implementation :: PyPy' + "Development Status :: 4 - Beta", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Topic :: Software Development", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.6", + "Programming Language :: Python :: 3.7", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: Implementation :: PyPy", ], - version=__version__, install_requires=requirements(), - test_suite='runtests.runtests', - entry_points={'nose.plugins': []}, + test_suite="runtests.runtests", + entry_points={"nose.plugins": []}, packages=find_packages(), - python_requires='>=3.6', + python_requires=">=3.6", ) diff --git a/tuya_iot/__init__.py b/tuya_iot/__init__.py index 8b6a94d..79b9f2d 100644 --- a/tuya_iot/__init__.py +++ b/tuya_iot/__init__.py @@ -1,11 +1,11 @@ -from .openapi import TuyaOpenAPI, TuyaTokenInfo -from .openmq import TuyaOpenMQ from .asset import TuyaAssetManager -from .device import TuyaDeviceManager, TuyaDevice, TuyaDeviceListener -from .tuya_enums import AuthType, TuyaCloudOpenAPIEndpoint +from .device import TuyaDevice, TuyaDeviceListener, TuyaDeviceManager from .home import TuyaHomeManager, TuyaScene from .infrared import TuyaRemote +from .openapi import TuyaOpenAPI, TuyaTokenInfo from .openlogging import TUYA_LOGGER +from .openmq import TuyaOpenMQ +from .tuya_enums import AuthType, TuyaCloudOpenAPIEndpoint from .version import VERSION __all__ = [ @@ -20,6 +20,6 @@ "TuyaCloudOpenAPIEndpoint", "TuyaHomeManager", "TuyaScene", - "TUYA_LOGGER" + "TUYA_LOGGER", ] __version__ = VERSION diff --git a/tuya_iot/asset.py b/tuya_iot/asset.py index e9b8c5a..a65de81 100644 --- a/tuya_iot/asset.py +++ b/tuya_iot/asset.py @@ -10,8 +10,6 @@ class TuyaAssetManager: Attributes: api: tuya openapi - - """ def __init__(self, api: TuyaOpenAPI) -> None: @@ -31,7 +29,7 @@ def get_device_list(self, asset_id: str) -> list[str]: Returns: A list of device ids. """ - device_id_list = [] + device_id_list: list[str] = [] has_next = True last_row_key = "" @@ -43,9 +41,9 @@ def get_device_list(self, asset_id: str) -> list[str]: result = response.get("result", {}) has_next = result.get("has_next", False) last_row_key = result.get("last_row_key", "") - totalSize = result.get("total_size", 0) + total_size = result.get("total_size", 0) - if len(device_id_list) > totalSize: # Error + if len(device_id_list) > total_size: # Error raise Exception("get_device_list error, too many devices.") for item in result.get("list", []): @@ -62,7 +60,7 @@ def get_asset_info(self, asset_id: str) -> dict[str, Any]: Returns: asset's info """ - return self.api.get("/v1.0/iot-02/assets/{}".format(asset_id)) + return self.api.get(f"/v1.0/iot-02/assets/{asset_id}") def get_asset_list(self, parent_asset_id: str = "-1") -> list: """Get under-nodes unser the current node. diff --git a/tuya_iot/device.py b/tuya_iot/device.py index 6ea5a4c..429d60e 100644 --- a/tuya_iot/device.py +++ b/tuya_iot/device.py @@ -19,7 +19,6 @@ BIZCODE_DPNAME_UPDATE = "dpNameUpdate" BIZCODE_BIND_USER = "bindUser" BIZCODE_DELETE = "delete" -BIZCODE_P2P_SIGNAL = "p2pSignal" class TuyaDeviceFunction(SimpleNamespace): @@ -172,8 +171,6 @@ def on_message(self, msg: str): self._on_device_report(data["devId"], data["status"]) elif protocol == PROTOCOL_OTHER: self._on_device_other(data["devId"], data["bizCode"], data) - else: - pass def __update_device(self, device: TuyaDevice): for listener in self.device_listeners: @@ -198,14 +195,14 @@ def _on_device_other(self, device_id: str, biz_code: str, data: dict[str, Any]): # bind device to user if biz_code == BIZCODE_BIND_USER: device_id = data["devId"] - devIds = [device_id] + device_ids = [device_id] # wait for es sync time.sleep(1) - self._update_device_list_info_cache(devIds) - self._update_device_list_status_cache(devIds) + self._update_device_list_info_cache(device_ids) + self._update_device_list_status_cache(device_ids) - self.update_device_function_cache(devIds) + self.update_device_function_cache(device_ids) if device_id in self.device_map.keys(): device = self.device_map.get(device_id) @@ -232,19 +229,13 @@ def _on_device_other(self, device_id: str, biz_code: str, data: dict[str, Any]): del self.device_map[device_id] for listener in self.device_listeners: listener.remove_device(device.id) - elif biz_code == BIZCODE_P2P_SIGNAL: - pass - else: - pass ############################## # Memory Cache def update_device_list_in_smart_home(self): """Update devices status in project type SmartHome.""" - response = self.api.get( - "/v1.0/users/{}/devices".format(self.api.token_info.uid) - ) + response = self.api.get(f"/v1.0/users/{self.api.token_info.uid}/devices") if response["success"]: for item in response["result"]: device = TuyaDevice(**item) @@ -296,7 +287,7 @@ def update_device_function_cache(self, devIds: list = []): """Update device function cache.""" device_map = ( filter(lambda d: d.id in devIds, self.device_map.values()) - if len(devIds) > 0 + if devIds else self.device_map.values() ) @@ -304,17 +295,17 @@ def update_device_function_cache(self, devIds: list = []): response = self.get_device_specification(device.id) if response.get("success"): result = response.get("result", {}) - functionMap = {} + function_map = {} for function in result["functions"]: code = function["code"] - functionMap[code] = TuyaDeviceFunction(**function) + function_map[code] = TuyaDeviceFunction(**function) status_range = {} for status in result["status"]: code = status["code"] status_range[code] = TuyaDeviceStatusRange(**status) - device.function = functionMap + device.function = function_map device.status_range = status_range def add_device_listener(self, listener: TuyaDeviceListener): @@ -507,7 +498,7 @@ def get_device_stream_allocate( Args: device_id(str): device id stream_type(str): type of stream - + Returns: None or URL to the requested stream """ @@ -584,12 +575,12 @@ def update_device_caches(self, devIds: list[str]): pass def get_device_info(self, device_id: str) -> dict[str, Any]: - response = self.api.get("/v1.0/devices/{}".format(device_id)) + response = self.api.get(f"/v1.0/devices/{device_id}") response["result"].pop("status") return response - def get_device_list_info(self, devIds: list[str]) -> dict[str, Any]: - response = self.api.get("/v1.0/devices/", {"device_ids": ",".join(devIds)}) + def get_device_list_info(self, device_ids: list[str]) -> dict[str, Any]: + response = self.api.get("/v1.0/devices/", {"device_ids": ",".join(device_ids)}) if response["success"]: for info in response["result"]["devices"]: info.pop("status") @@ -597,12 +588,12 @@ def get_device_list_info(self, devIds: list[str]) -> dict[str, Any]: return response def get_device_status(self, device_id: str) -> dict[str, Any]: - response = self.api.get("/v1.0/devices/{}".format(device_id)) + response = self.api.get(f"/v1.0/devices/{device_id}") response["result"] = response["result"]["status"] return response - def get_device_list_status(self, devIds: list[str]) -> dict[str, Any]: - response = self.api.get("/v1.0/devices/", {"device_ids": ",".join(devIds)}) + def get_device_list_status(self, device_ids: list[str]) -> dict[str, Any]: + response = self.api.get("/v1.0/devices/", {"device_ids": ",".join(device_ids)}) status_list = [] if response["success"]: for info in response["result"]["devices"]: @@ -617,23 +608,23 @@ def get_factory_info(self, devIds: str) -> dict[str, Any]: ) def factory_reset(self, device_id: str) -> dict[str, Any]: - return self.api.post("/v1.0/devices/{}/reset-factory".format(device_id)) + return self.api.post(f"/v1.0/devices/{device_id}/reset-factory") def remove_device(self, device_id: str) -> dict[str, Any]: - return self.api.delete("/v1.0/devices/{}".format(device_id)) + return self.api.delete(f"/v1.0/devices/{device_id}") def remove_device_list(self, devIds: list[str]) -> dict[str, Any]: raise Exception("Api not support.") def get_device_functions(self, device_id: str) -> dict[str, Any]: - return self.api.get("/v1.0/devices/{}/functions".format(device_id)) + return self.api.get(f"/v1.0/devices/{device_id}/functions") def get_category_functions(self, categoryId: str) -> dict[str, Any]: - return self.api.get("/v1.0/functions/{}".format(categoryId)) + return self.api.get(f"/v1.0/functions/{categoryId}") # https://developer.tuya.com/en/docs/cloud/device-control?id=K95zu01ksols7#title-27-Get%20the%20specifications%20and%20properties%20of%20the%20device%2C%20including%20the%20instruction%20set%20and%20status%20set def get_device_specification(self, device_id: str) -> dict[str, str]: - return self.api.get("/v1.0/devices/{}/specifications".format(device_id)) + return self.api.get(f"/v1.0/devices/{device_id}/specifications") def get_device_stream_allocate( self, device_id: str, stream_type: Literal["flv", "hls", "rtmp", "rtsp"] @@ -664,13 +655,15 @@ def update_device_caches(self, devIds: list[str]): pass def get_device_info(self, device_id: str) -> dict[str, Any]: - return self.api.get("/v1.0/iot-03/devices/{}".format(device_id)) + return self.api.get(f"/v1.0/iot-03/devices/{device_id}") - def get_device_list_info(self, devIds: list[str]) -> dict[str, Any]: - return self.api.get("/v1.0/iot-03/devices", {"device_ids": ",".join(devIds)}) + def get_device_list_info(self, device_ids: list[str]) -> dict[str, Any]: + return self.api.get( + "/v1.0/iot-03/devices", {"device_ids": ",".join(device_ids)} + ) def get_device_status(self, device_id: str) -> dict[str, Any]: - return self.api.get("/v1.0/iot-03/devices/{}/status".format(device_id)) + return self.api.get(f"/v1.0/iot-03/devices/{device_id}/status") def get_device_list_status(self, devIds: list[str]) -> dict[str, Any]: return self.api.get( @@ -681,25 +674,25 @@ def get_factory_info(self, device_id: str) -> dict[str, Any]: return self.api.get("/v1.0/iot-03/devices/factory-infos", device_id) def factory_reset(self, device_id: str) -> dict[str, Any]: - return self.api.delete( - "/v1.0/iot-03/devices/{}/actions/reset".format(device_id) - ) + return self.api.delete(f"/v1.0/iot-03/devices/{device_id}/actions/reset") def remove_device(self, device_id: str) -> dict[str, Any]: - return self.api.delete("/v1.0/iot-03/devices/{}".format(device_id)) + return self.api.delete(f"/v1.0/iot-03/devices/{device_id}") - def remove_device_list(self, devIds: list[str]) -> dict[str, Any]: - return self.api.delete("/v1.0/iot-03/devices", {"device_ids": ",".join(devIds)}) + def remove_device_list(self, device_ids: list[str]) -> dict[str, Any]: + return self.api.delete( + "/v1.0/iot-03/devices", {"device_ids": ",".join(device_ids)} + ) def get_device_functions(self, device_id: str) -> dict[str, Any]: - return self.api.get("/v1.0/iot-03/devices/{}/functions".format(device_id)) + return self.api.get(f"/v1.0/iot-03/devices/{device_id}/functions") def get_category_functions(self, categoryId: str) -> dict[str, Any]: - return self.api.get("/v1.0/iot-03/categories/{}/functions".format(categoryId)) + return self.api.get(f"/v1.0/iot-03/categories/{categoryId}/functions") # https://developer.tuya.com/en/docs/cloud/device-control?id=K95zu01ksols7#title-27-Get%20the%20specifications%20and%20properties%20of%20the%20device%2C%20including%20the%20instruction%20set%20and%20status%20set def get_device_specification(self, device_id: str) -> dict[str, str]: - return self.api.get("/v1.0/iot-03/devices/{}/specification".format(device_id)) + return self.api.get(f"/v1.0/iot-03/devices/{device_id}/specification") def get_device_stream_allocate( self, device_id: str, stream_type: Literal["flv", "hls", "rtmp", "rtsp"] @@ -710,8 +703,10 @@ def get_device_stream_allocate( """ return None - def send_commands(self, device_id: str, commands: list[dict[str, Any]]) -> dict[str, Any]: + def send_commands( + self, device_id: str, commands: list[dict[str, Any]] + ) -> dict[str, Any]: return self.api.post( - "/v1.0/iot-03/devices/{}/commands".format(device_id), {"commands": commands} + f"/v1.0/iot-03/devices/{device_id}/commands", {"commands": commands} ) diff --git a/tuya_iot/home.py b/tuya_iot/home.py index 53beaaf..b086f18 100644 --- a/tuya_iot/home.py +++ b/tuya_iot/home.py @@ -1,14 +1,14 @@ """Tuya home's api base on asset and device api.""" -from typing import Any from types import SimpleNamespace +from typing import Any -from .openapi import TuyaOpenAPI -from .openmq import TuyaOpenMQ -from .tuya_enums import AuthType from .asset import TuyaAssetManager from .device import TuyaDeviceManager from .infrared import TuyaRemote, TuyaRemoteDevice, TuyaRemoteDeviceKey +from .openapi import TuyaOpenAPI +from .openmq import TuyaOpenMQ +from .tuya_enums import AuthType class TuyaScene(SimpleNamespace): @@ -52,7 +52,7 @@ def update_device_cache(self): # for asset in assets: # asset_id = asset["asset_id"] # device_ids += asset_manager.get_device_list(asset_id) - if len(device_ids) > 0: + if device_ids: self.device_manager.update_device_caches(device_ids) elif self.api.auth_type == AuthType.SMART_HOME: self.device_manager.update_device_list_in_smart_home() @@ -74,14 +74,14 @@ def query_scenes(self) -> list: response = self.api.get(f"/v1.0/users/{self.api.token_info.uid}/homes") if response.get("success", False): - homes = response.get("result") + homes = response.get("result", []) scenes = [] for home in homes: home_id = home["home_id"] scenes_response = self.api.get(f"/v1.0/homes/{home_id}/scenes") if scenes_response.get("success", False): - for scene in scenes_response.get("result"): + for scene in scenes_response.get("result", []): __tuya_scene = TuyaScene(**scene) __tuya_scene.home_id = home_id scenes.append(__tuya_scene) @@ -102,10 +102,11 @@ def query_infrared_devices(self) -> list: if self.api.auth_type == AuthType.CUSTOM: return [] - remote_ids = [] - for (device_id, device) in self.device_manager.device_map.items(): - if device.category == "qt": - remote_ids.append(device_id) + remote_ids = [ + device_id + for (device_id, device) in self.device_manager.device_map.items() + if device.category == "qt" + ] remotes = [] for remote_id in remote_ids: @@ -113,7 +114,7 @@ def query_infrared_devices(self) -> list: if not remotes_response.get("success", False): continue - remote_device_response = remotes_response.get("result") + remote_device_response = remotes_response.get("result", []) remote_devices = [] for remote_device in remote_device_response: @@ -129,19 +130,17 @@ def query_infrared_devices(self) -> list: continue keys_result = keys_response.get("result") - key_values = keys_result.get("key_list") - - tuya_remote_device_keys = [] - for tuya_key in key_values: - tuya_remote_device_keys.append( - TuyaRemoteDeviceKey( - tuya_key["key"], - tuya_key["key_id"], - tuya_key["key_name"], - tuya_key["standard_key"], - ) + key_values = keys_result.get("key_list", []) + + tuya_remote_device_keys = [ + TuyaRemoteDeviceKey( + tuya_key["key"], + tuya_key["key_id"], + tuya_key["key_name"], + tuya_key["standard_key"], ) - + for tuya_key in key_values + ] remote_devices.append( TuyaRemoteDevice(remote_device, tuya_remote_device_keys) ) @@ -157,6 +156,6 @@ def trigger_infrared_commands(self, remote_id, device_id, key) -> None: return [] self.api.post( - "/v1.0/infrareds/{}/remotes/{}/command".format(remote_id, device_id), + f"/v1.0/infrareds/{remote_id}/remotes/{device_id}/command", {"key": key}, ) diff --git a/tuya_iot/openapi.py b/tuya_iot/openapi.py index 12303c4..1674add 100644 --- a/tuya_iot/openapi.py +++ b/tuya_iot/openapi.py @@ -118,11 +118,8 @@ def _calculate_sign( if params is not None and len(params.keys()) > 0: str_to_sign += "?" - query_builder = "" params_keys = sorted(params.keys()) - - for key in params_keys: - query_builder += f"{key}={params[key]}&" + query_builder = "".join(f"{key}={params[key]}&" for key in params_keys) str_to_sign += query_builder[:-1] # Sign @@ -234,10 +231,7 @@ def __request( self.__refresh_access_token_if_need(path) - access_token = "" - if self.token_info: - access_token = self.token_info.access_token - + access_token = self.token_info.access_token if self.token_info else "" sign, t = self._calculate_sign(method, path, params, body) headers = { "client_id": self.access_id, diff --git a/tuya_iot/openmq.py b/tuya_iot/openmq.py index 560caa6..ab4fbec 100644 --- a/tuya_iot/openmq.py +++ b/tuya_iot/openmq.py @@ -7,6 +7,7 @@ import uuid from typing import Any, Callable from urllib.parse import urlsplit +from typing import Optional from Crypto.Cipher import AES from paho.mqtt import client as mqtt @@ -15,7 +16,7 @@ from .openlogging import logger from .tuya_enums import AuthType -LINK_ID = "tuya-iot-app-sdk-python.{}".format(uuid.uuid1()) +LINK_ID = f"tuya-iot-app-sdk-python.{uuid.uuid1()}" GCM_TAG_LENGTH = 16 CONNECT_FAILED_NOT_AUTHORISED = 5 @@ -53,7 +54,7 @@ def __init__(self, api: TuyaOpenAPI) -> None: self.mq_config = None self.message_listeners = set() - def _get_mqtt_config(self) -> TuyaMQConfig: + def _get_mqtt_config(self) -> Optional[TuyaMQConfig]: response = self.api.post( "/v1.0/iot-03/open-hub/access-config", { @@ -124,11 +125,6 @@ def _on_message(self, mqttc: mqtt.Client, user_data: Any, msg: mqtt.MQTTMessage) msg_dict = json.loads(msg.payload.decode("utf8")) - # topic = msg.topic - # protocol = msg_dict.get("protocol", 0) - # pv = msg_dict.get("pv", "") - # data = msg_dict.get("data", "") - t = msg_dict.get("t", "") mq_config = user_data["mqConfig"]