From 1cf4f7b392f30073d152012da9a7ba8c7e7771ab Mon Sep 17 00:00:00 2001 From: Joshua Mulliken Date: Sun, 16 May 2021 13:17:25 -0400 Subject: [PATCH] feat: Add support for the wyze thermostat --- setup.cfg | 2 +- src/wyzeapy/base_client.py | 274 ++++++++++----------------------- src/wyzeapy/client.py | 119 +++++--------- src/wyzeapy/const.py | 21 +++ src/wyzeapy/crypto.py | 39 +++++ src/wyzeapy/exceptions.py | 28 ++++ src/wyzeapy/payload_factory.py | 43 ++++++ src/wyzeapy/types.py | 147 ++++++++++++++++++ 8 files changed, 401 insertions(+), 272 deletions(-) create mode 100644 src/wyzeapy/const.py create mode 100644 src/wyzeapy/crypto.py create mode 100644 src/wyzeapy/exceptions.py create mode 100644 src/wyzeapy/payload_factory.py create mode 100644 src/wyzeapy/types.py diff --git a/setup.cfg b/setup.cfg index 745249c..fdb435b 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,7 +1,7 @@ [metadata] # replace with your username: name = wyzeapy -version = 0.0.22 +version = 0.0.23-beta.1 author = Mulliken LLC author_email = joshua@mulliken.net description = Python client for private Wyze API diff --git a/src/wyzeapy/base_client.py b/src/wyzeapy/base_client.py index b484aef..7e6436f 100644 --- a/src/wyzeapy/base_client.py +++ b/src/wyzeapy/base_client.py @@ -5,118 +5,27 @@ # joshua@mulliken.net to receive a copy import datetime import hashlib +import json import time -import uuid -from enum import Enum -from typing import List, Dict +from typing import Any import requests -PHONE_SYSTEM_TYPE = "1" -API_KEY = "WMXHYf79Nr5gIlt3r0r7p9Tcw5bvs6BB4U8O8nGJ" -APP_VERSION = "2.18.43" -APP_VER = "com.hualai.WyzeCam___2.18.43" -APP_NAME = "com.hualai.WyzeCam" -PHONE_ID = str(uuid.uuid4()) +from .const import * +from .payload_factory import ford_create_payload, olive_create_get_payload, olive_create_post_payload +from .crypto import olive_create_signature +from .exceptions import * +from .types import ResponseCodes, Device, DeviceTypes, ThermostatProps -class DeviceTypes(Enum): - LIGHT = "Light" - PLUG = "Plug" - OUTDOOR_PLUG = "OutdoorPlug" - MESH_LIGHT = "MeshLight" - CAMERA = "Camera" - CHIME_SENSOR = "ChimeSensor" - CONTACT_SENSOR = "ContactSensor" - MOTION_SENSOR = "MotionSensor" - WRIST = "Wrist" - BASE_STATION = "BaseStation" - SCALE = "WyzeScale" - LOCK = "Lock" - GATEWAY = "gateway" - COMMON = "Common" - VACUUM = "JA_RO2" - HEADPHONES = "JA.SC" - THERMOSTAT = "Thermostat" - GATEWAY_V2 = "GateWay" - KEYPAD = "Keypad" - SENSOR_GATEWAY = "S1Gateway" - - -class PropertyIDs(Enum): - ON = "P3" - AVAILABLE = "P5" - BRIGHTNESS = "P1501" # From 0-100 - COLOR_TEMP = "P1502" # In Kelvin - COLOR = "P1507" # As a hex string RrGgBb - DOOR_OPEN = "P2001" # 0 if the door is closed - - -class ResponseCodes(Enum): - SUCCESS = "1" - PARAMETER_ERROR = "1001" - ACCESS_TOKEN_ERROR = "2001" - - -class ResponseCodesLock(Enum): - SUCCESS = 0 - - -SWITCHABLE_DEVICES = [DeviceTypes.LIGHT, DeviceTypes.MESH_LIGHT, DeviceTypes.PLUG] - - -class ActionNotSupported(Exception): - def __init__(self, device_type): - message = "The action specified is not supported by device type: {}".format(device_type) - - super().__init__(message) - - -class ParameterError(Exception): - pass - - -class AccessTokenError(Exception): - pass - -class LoginError(Exception): - pass - - -class UnknownApiError(Exception): - def __init__(self, response_json): - super(UnknownApiError, self).__init__(str(response_json)) - - -class Device: - product_type: str - product_model: str - mac: str - nickname: str - - def __init__(self, dictionary): - for k, v in dictionary.items(): - setattr(self, k, v) - - def __repr__(self): - return "".format(DeviceTypes(self.product_type), self.mac) - -class Group: - group_id: str - group_name: str - - def __init__(self, dictionary): - for k, v in dictionary.items(): - setattr(self, k, v) - - def __repr__(self): - return "".format(self.group_id, self.group_name) - class BaseClient: access_token = "" refresh_token = "" def login(self, email, password) -> bool: + email = email + password = password + login_payload = { "email": email, "password": self.create_password(password) @@ -129,9 +38,6 @@ def login(self, email, password) -> bool: response_json = requests.post("https://auth-prod.api.wyze.com/user/login", headers=headers, json=login_payload).json() - # response_json = requests.post("https://auth-prod.api.wyze.com/user/login", - # json=login_payload).json() - try: self.access_token = response_json['access_token'] self.refresh_token = response_json['refresh_token'] @@ -169,41 +75,6 @@ def check_for_errors_lock(response_json): else: raise UnknownApiError(response_json) - app_key = "275965684684dbdaf29a0ed9" - app_secret = "4deekof1ba311c5c33a9cb8e12787e8c" - - def postMethod(self, str, str2, map): - str4 = str - str3 = str2 - map2 = map - - return self.getTransferParam(map2, str3, str4, "post") - - def getTransferParam(self, map, str4, str2, str3): - map["accessToken"] = self.access_token - map["key"] = self.app_key - map["timestamp"] = str(datetime.datetime.now().timestamp()).split(".")[0] + "000" - map["sign"] = self.getSignStr(str4, str3, map) - import json - print(json.dumps(map)) - return map - - def getSignStr(self, str, str2, map: Dict): - string_buf = str2 + str - for entry in sorted(map.keys()): - print(entry) - print(map[entry]) - string_buf += entry + "=" + map[entry] + "&" - - string_buf = string_buf[:-1] - print(string_buf) - string_buf += self.app_secret - import urllib.parse - urlencoded = urllib.parse.quote_plus(string_buf) - sign_str = hashlib.md5(urlencoded.encode()).hexdigest() - print(sign_str) - return sign_str - def get_object_list(self): payload = { "phone_system_type": PHONE_SYSTEM_TYPE, @@ -220,7 +91,6 @@ def get_object_list(self): response_json = requests.post("https://api.wyzecam.com/app/v2/home_page/get_object_list", json=payload).json() - print(payload) self.check_for_errors(response_json) return response_json @@ -247,27 +117,6 @@ def get_property_list(self, device: Device): self.check_for_errors(response_json) return response_json - - def get_auto_group_list(self): - payload = { - "access_token": self.access_token, - "app_name": APP_NAME, - "app_ver": APP_VER, - "app_version": APP_VERSION, - "group_type": "0", - "phone_id": PHONE_ID, - "phone_system_type": PHONE_SYSTEM_TYPE, - "sc": "9f275790cab94a72bd206c8876429f3c", - "sv": "9d74946e652647e9b6c9d59326aef104", - "ts": int(time.time()), - } - - response_json = requests.post("https://api.wyzecam.com/app/v2/auto_group/get_list", - json=payload).json() - - self.check_for_errors(response_json) - - return response_json def run_action_list(self, device: Device, plist): if DeviceTypes(device.product_type) not in [ @@ -306,28 +155,7 @@ def run_action_list(self, device: Device, plist): self.check_for_errors(response_json) - def auto_group_run(self, group: Group): - #if DeviceTypes(device.product_type) not in [ - # DeviceTypes.CAMERA - #]: - # raise ActionNotSupported(device.product_type) - - payload = { - "access_token": self.access_token, - "app_name": APP_NAME, - "app_ver": APP_VER, - "app_version": APP_VERSION, - "group_id": group.group_id, - "phone_id": PHONE_ID, - "phone_system_type": PHONE_SYSTEM_TYPE, - "sc": "9f275790cab94a72bd206c8876429f3c", - "sv": "9d74946e652647e9b6c9d59326aef104", - "ts": int(time.time()), - } - - response_json = requests.post("https://api.wyzecam.com/app/v2/auto_group/run", json=payload).json() - - self.check_for_errors(response_json) + return response_json def run_action(self, device: Device, action: str): if DeviceTypes(device.product_type) not in [ @@ -356,6 +184,8 @@ def run_action(self, device: Device, action: str): self.check_for_errors(response_json) + return response_json + def set_property_list(self, device: Device, plist): if DeviceTypes(device.product_type) not in [ DeviceTypes.LIGHT @@ -380,6 +210,8 @@ def set_property_list(self, device: Device, plist): self.check_for_errors(response_json) + return response_json + def set_property(self, device: Device, pid, pvalue): """ Sets a single property on the selected device. @@ -416,6 +248,8 @@ def set_property(self, device: Device, pid, pvalue): self.check_for_errors(response_json) + return response_json + def get_event_list(self, device: Device, count: int) -> dict: """ Gets motion events from the event listing endpoint. @@ -426,7 +260,7 @@ def get_event_list(self, device: Device, count: int) -> dict: """ payload = { "phone_id": PHONE_ID, - "begin_time": int(str(int(time.time() - (24 * 60 * 60))) + "000"), + "begin_time": int(str(datetime.date.today().strftime("%s")) + "000"), "event_type": "", "app_name": APP_NAME, "count": count, @@ -459,19 +293,75 @@ def get_event_list(self, device: Device, count: int) -> dict: return response_json def lock_control(self, device: Device, action: str): - sb2 = "https://yd-saas-toc.wyzecam.com/openapi/lock/v1/control" - str3 = "/openapi/lock/v1/control" + url_path = "/openapi/lock/v1/control" - uuid = device.mac.split(".")[2] + device_uuid = device.mac.split(".")[2] - hash_map = { - "uuid": uuid, + payload = { + "uuid": device_uuid, "action": action # "remoteLock" or "remoteUnlock" } - - payload = self.postMethod(sb2, str3, hash_map) + payload = ford_create_payload(self.access_token, payload, url_path, "post") url = "https://yd-saas-toc.wyzecam.com/openapi/lock/v1/control" response_json = requests.post(url, json=payload).json() - print(response_json) + + self.check_for_errors_lock(response_json) + + return response_json + + def thermostat_get_iot_prop(self, device: Device): + payload = olive_create_get_payload(device.mac) + signature = olive_create_signature(payload, self.access_token) + headers = { + 'Accept-Encoding': 'gzip', + 'User-Agent': 'myapp', + 'appid': OLIVE_APP_ID, + 'appinfo': APP_INFO, + 'phoneid': PHONE_ID, + 'access_token': self.access_token, + 'signature2': signature + } + + url = 'https://wyze-earth-service.wyzecam.com/plugin/earth/get_iot_prop' + response_json = requests.get(url, headers=headers, params=payload).json() + + self.check_for_errors_thermostat(response_json) + + return response_json + + def thermostat_set_iot_prop(self, device: Device, prop: ThermostatProps, value: Any): + url = "https://wyze-earth-service.wyzecam.com/plugin/earth/set_iot_prop_by_topic" + payload = olive_create_post_payload(device.mac, device.product_model, prop, value) + signature = olive_create_signature(json.dumps(payload, separators=(',', ':')), self.access_token) + headers = { + 'Accept-Encoding': 'gzip', + 'User-Agent': 'myapp', + 'appid': OLIVE_APP_ID, + 'appinfo': APP_INFO, + 'phoneid': PHONE_ID, + 'access_token': self.access_token, + 'signature2': signature + } + + with requests.Session() as session: + session.headers.update(headers) + + req = session.prepare_request(requests.Request('POST', url, json=payload)) + + payload = json.dumps(payload, separators=(',', ':')) + + req.body = payload.encode('utf-8') + req.prepare_content_length(req.body) + + response_json = session.send(req).json() + + self.check_for_errors_thermostat(response_json) + + return response_json + + @staticmethod + def check_for_errors_thermostat(response_json): + if response_json['code'] != 1: + raise UnknownApiError(response_json) diff --git a/src/wyzeapy/client.py b/src/wyzeapy/client.py index c8fdc8f..db62478 100644 --- a/src/wyzeapy/client.py +++ b/src/wyzeapy/client.py @@ -6,63 +6,10 @@ import re -from typing import List, Any -from .base_client import BaseClient, Device, DeviceTypes, ActionNotSupported, PropertyIDs, Group - - -class File: - file_id: str - type: Any - url: str - status: int - en_algorithm: int - en_password: str - is_ai: int - ai_tag_list: List - ai_url: str - file_params: dict - - def __init__(self, dictionary): - for k, v in dictionary.items(): - setattr(self, k, v) - - if self.type == 1: - self.type = "Image" - else: - self.type = "Video" - - def __repr__(self): - return "".format(self.file_id, self.type) - - -class Event: - event_id: str - device_mac: str - device_model: str - event_category: int - event_value: str - event_ts: int - event_ack_result: int - is_feedback_correct: int - is_feedback_face: int - is_feedback_person: int - file_list: List[File] - event_params: dict - recognized_instance_list: List - tag_list: List - read_state: int - - def __init__(self, dictionary): - for k, v in dictionary.items(): - setattr(self, k, v) - temp_file_list = [] - if len(self.file_list) > 0: - for file in self.file_list: - temp_file_list.append(File(file)) - self.file_list = temp_file_list - - def __repr__(self): - return "".format(self.event_id, self.event_ts) +from typing import Any, Optional, List, Tuple +from .base_client import BaseClient +from .exceptions import ActionNotSupported +from .types import ThermostatProps, Device, DeviceTypes, PropertyIDs, Event class Client: @@ -73,14 +20,14 @@ def __init__(self, email, password): self.client = BaseClient() self.client.login(self.email, self.password) - def reauthenticate(self): + def reauthenticate(self) -> None: self.client.login(self.email, self.password) @staticmethod - def create_pid_pair(pid_enum: PropertyIDs, value): + def create_pid_pair(pid_enum: PropertyIDs, value) -> dict: return {"pid": pid_enum.value, "pvalue": value} - def get_devices(self): + def get_devices(self) -> List[Device]: object_list = self.client.get_object_list() devices = [] @@ -89,19 +36,7 @@ def get_devices(self): return devices - def get_groups(self): - object_list = self.client.get_auto_group_list() - - groups = [] - for group in object_list['data']['auto_group_list']: - groups.append(Group(group)) - - return groups - - def activate_group(self, group: Group): - self.client.auto_group_run(group) - - def turn_on(self, device: Device, extra_pids=None): + def turn_on(self, device: Device, extra_pids=None) -> None: device_type: DeviceTypes = DeviceTypes(device.product_type) if device_type in [ @@ -140,7 +75,7 @@ def turn_on(self, device: Device, extra_pids=None): else: raise ActionNotSupported(device_type.value) - def turn_off(self, device: Device, extra_pids=None): + def turn_off(self, device: Device, extra_pids=None) -> None: device_type: DeviceTypes = DeviceTypes(device.product_type) if device_type in [ @@ -179,7 +114,7 @@ def turn_off(self, device: Device, extra_pids=None): else: raise ActionNotSupported(device_type.value) - def set_brightness(self, device: Device, brightness: int): + def set_brightness(self, device: Device, brightness: int) -> None: if DeviceTypes(device.product_type) not in [ DeviceTypes.LIGHT, DeviceTypes.MESH_LIGHT @@ -193,7 +128,7 @@ def set_brightness(self, device: Device, brightness: int): self.create_pid_pair(PropertyIDs.BRIGHTNESS, str(brightness)) ]) - def set_color(self, device, rgb_hex_string): + def set_color(self, device, rgb_hex_string) -> None: if DeviceTypes(device.product_type) not in [ DeviceTypes.MESH_LIGHT ]: @@ -206,7 +141,7 @@ def set_color(self, device, rgb_hex_string): self.create_pid_pair(PropertyIDs.COLOR, rgb_hex_string) ]) - def get_info(self, device): + def get_info(self, device) -> List[Tuple[PropertyIDs, Any]]: properties = self.client.get_property_list(device)['data']['property_list'] property_list = [] @@ -222,7 +157,7 @@ def get_info(self, device): return property_list - def get_events(self, device): + def get_events(self, device) -> List[Event]: raw_events = self.client.get_event_list(device, 10)['data']['event_list'] events = [] @@ -233,10 +168,36 @@ def get_events(self, device): return events - def get_latest_event(self, device): + def get_latest_event(self, device) -> Optional[Event]: raw_events = self.client.get_event_list(device, 10)['data']['event_list'] if len(raw_events) > 0: return Event(raw_events[0]) return None + + def get_thermostat_info(self, device) -> List[Tuple[ThermostatProps, Any]]: + if DeviceTypes(device.product_type) not in [ + DeviceTypes.THERMOSTAT + ]: + raise ActionNotSupported(device.product_type) + + properties = self.client.thermostat_get_iot_prop(device)['data']['props'] + + device_props = [] + for property in properties: + try: + prop = ThermostatProps(property) + device_props.append((prop, properties[property])) + except AttributeError as e: + print(e) + + return device_props + + def set_thermostat_prop(self, device: Device, prop: ThermostatProps, value: Any) -> None: + if DeviceTypes(device.product_type) not in [ + DeviceTypes.THERMOSTAT + ]: + raise ActionNotSupported(device.product_type) + + self.client.thermostat_set_iot_prop(device, prop, value) diff --git a/src/wyzeapy/const.py b/src/wyzeapy/const.py new file mode 100644 index 0000000..173a085 --- /dev/null +++ b/src/wyzeapy/const.py @@ -0,0 +1,21 @@ +# Copyright (c) 2021. Mulliken, LLC - All Rights Reserved +# You may use, distribute and modify this code under the terms +# of the attached license. You should have received a copy of +# the license with this file. If not, please write to: +# joshua@mulliken.net to receive a copy +import uuid + +# Here is where all the *magic* lives +PHONE_SYSTEM_TYPE = "1" +API_KEY = "WMXHYf79Nr5gIlt3r0r7p9Tcw5bvs6BB4U8O8nGJ" +APP_VERSION = "2.18.43" +APP_VER = "com.hualai.WyzeCam___2.18.43" +APP_NAME = "com.hualai.WyzeCam" +PHONE_ID = str(uuid.uuid4()) +APP_INFO = 'wyze_android_2.19.14' # Required for the thermostat + +# Crypto secrets +OLIVE_SIGNING_SECRET = 'wyze_app_secret_key_132' # Required for the thermostat +OLIVE_APP_ID = '9319141212m2ik' # Required for the thermostat +FORD_APP_KEY = "275965684684dbdaf29a0ed9" # Required for the locks +FORD_APP_SECRET = "4deekof1ba311c5c33a9cb8e12787e8c" # Required for the locks diff --git a/src/wyzeapy/crypto.py b/src/wyzeapy/crypto.py new file mode 100644 index 0000000..4e1c783 --- /dev/null +++ b/src/wyzeapy/crypto.py @@ -0,0 +1,39 @@ +# Copyright (c) 2021. Mulliken, LLC - All Rights Reserved +# You may use, distribute and modify this code under the terms +# of the attached license. You should have received a copy of +# the license with this file. If not, please write to: +# joshua@mulliken.net to receive a copy +import hashlib +import hmac +import urllib.parse +from typing import Dict, Union + +from src.wyzeapy.const import FORD_APP_SECRET, OLIVE_SIGNING_SECRET + + +def olive_create_signature(payload: Union[dict, str], access_token: str) -> str: + if isinstance(payload, dict): + body = "" + for item in sorted(payload): + body += item + "=" + str(payload[item]) + "&" + + body = body[:-1] + + else: + body = payload + + access_key = "{}{}".format(access_token, OLIVE_SIGNING_SECRET) + + secret = hashlib.md5(access_key.encode()).hexdigest() + return hmac.new(secret.encode(), body.encode(), hashlib.md5).hexdigest() + + +def ford_create_signature(url_path, request_method, payload: Dict): + string_buf = request_method + url_path + for entry in sorted(payload.keys()): + string_buf += entry + "=" + payload[entry] + "&" + + string_buf = string_buf[:-1] + string_buf += FORD_APP_SECRET + urlencoded = urllib.parse.quote_plus(string_buf) + return hashlib.md5(urlencoded.encode()).hexdigest() diff --git a/src/wyzeapy/exceptions.py b/src/wyzeapy/exceptions.py new file mode 100644 index 0000000..93de29a --- /dev/null +++ b/src/wyzeapy/exceptions.py @@ -0,0 +1,28 @@ +# Copyright (c) 2021. Mulliken, LLC - All Rights Reserved +# You may use, distribute and modify this code under the terms +# of the attached license. You should have received a copy of +# the license with this file. If not, please write to: +# joshua@mulliken.net to receive a copy + +class ActionNotSupported(Exception): + def __init__(self, device_type): + message = "The action specified is not supported by device type: {}".format(device_type) + + super().__init__(message) + + +class ParameterError(Exception): + pass + + +class AccessTokenError(Exception): + pass + + +class LoginError(Exception): + pass + + +class UnknownApiError(Exception): + def __init__(self, response_json): + super(UnknownApiError, self).__init__(str(response_json)) diff --git a/src/wyzeapy/payload_factory.py b/src/wyzeapy/payload_factory.py new file mode 100644 index 0000000..a87fa4b --- /dev/null +++ b/src/wyzeapy/payload_factory.py @@ -0,0 +1,43 @@ +# Copyright (c) 2021. Mulliken, LLC - All Rights Reserved +# You may use, distribute and modify this code under the terms +# of the attached license. You should have received a copy of +# the license with this file. If not, please write to: +# joshua@mulliken.net to receive a copy +import time +from typing import Any + +from src.wyzeapy.const import FORD_APP_KEY +from src.wyzeapy.types import ThermostatProps +from src.wyzeapy.crypto import ford_create_signature + + +def ford_create_payload(access_token, payload, url_path, request_method): + payload["accessToken"] = access_token + payload["key"] = FORD_APP_KEY + payload["timestamp"] = str(int(time.time() * 1000)) + payload["sign"] = ford_create_signature(url_path, request_method, payload) + return payload + + +def olive_create_get_payload(device_mac): + nonce = int(time.time() * 1000) + + return { + 'keys': 'trigger_off_val,emheat,temperature,humidity,time2temp_val,protect_time,mode_sys,heat_sp,cool_sp,current_scenario,config_scenario,temp_unit,fan_mode,iot_state,w_city_id,w_lat,w_lon,working_state,dev_hold,dev_holdtime,asw_hold,app_version,setup_state,wiring_logic_id,save_comfort_balance,kid_lock,calibrate_humidity,calibrate_temperature,fancirc_time,query_schedule', + 'did': device_mac, + 'nonce': nonce + } + + +def olive_create_post_payload(device_mac, device_model, prop: ThermostatProps, value: Any): + nonce = int(time.time() * 1000) + + return { + "did": device_mac, + "model": device_model, + "props": { + prop.value: str(value) + }, + "is_sub_device": 0, + "nonce": str(nonce) + } diff --git a/src/wyzeapy/types.py b/src/wyzeapy/types.py new file mode 100644 index 0000000..bbec7cd --- /dev/null +++ b/src/wyzeapy/types.py @@ -0,0 +1,147 @@ +# Copyright (c) 2021. Mulliken, LLC - All Rights Reserved +# You may use, distribute and modify this code under the terms +# of the attached license. You should have received a copy of +# the license with this file. If not, please write to: +# joshua@mulliken.net to receive a copy +from enum import Enum +from typing import Union, List + + +class Device: + product_type: str + product_model: str + mac: str + nickname: str + + def __init__(self, dictionary): + for k, v in dictionary.items(): + setattr(self, k, v) + + def __repr__(self): + return "".format(DeviceTypes(self.product_type), self.mac) + + +class DeviceTypes(Enum): + LIGHT = "Light" + PLUG = "Plug" + OUTDOOR_PLUG = "OutdoorPlug" + MESH_LIGHT = "MeshLight" + CAMERA = "Camera" + CHIME_SENSOR = "ChimeSensor" + CONTACT_SENSOR = "ContactSensor" + MOTION_SENSOR = "MotionSensor" + WRIST = "Wrist" + BASE_STATION = "BaseStation" + SCALE = "WyzeScale" + LOCK = "Lock" + GATEWAY = "gateway" + COMMON = "Common" + VACUUM = "JA_RO2" + HEADPHONES = "JA.SC" + THERMOSTAT = "Thermostat" + GATEWAY_V2 = "GateWay" + + +class PropertyIDs(Enum): + ON = "P3" + AVAILABLE = "P5" + BRIGHTNESS = "P1501" # From 0-100 + COLOR_TEMP = "P1502" # In Kelvin + COLOR = "P1507" # As a hex string RrGgBb + DOOR_OPEN = "P2001" # 0 if the door is closed + + +class ThermostatProps(Enum): + APP_VERSION = "app_version" + IOT_STATE = "iot_state" # Connection state: connected, disconnected + SETUP_STATE = "setup_state" + CURRENT_SCENARIO = "current_scenario" # home, away + PROTECT_TIME = "protect_time" + COOL_SP = "cool_sp" # Cool stop point + EMHEAT = "emheat" + TIME2TEMP_VAL = "time2temp_val" + SAVE_COMFORT_BALANCE = "save_comfort_balance" # savings, comfort, or balance value + QUERY_SCHEDULE = "query_schedule" + WORKING_STATE = "working_state" # idle, etc. + WIRING_LOGIC_ID = "wiring_logic_id" + W_CITY_ID = "w_city_id" + FAN_MODE = "fan_mode" # auto, on, off + TEMPERATURE = "temperature" # current temp + HUMIDITY = "humidity" # current humidity + KID_LOCK = "kid_lock" + CALIBRATE_HUMIDITY = "calibrate_humidity" + HEAT_SP = "heat_sp" # heat stop point + CALIBRATE_TEMPERATURE = "calibrate_temperature" + MODE_SYS = "mode_sys" # auto, heat, cool + W_LAT = "w_lat" + CONFIG_SCENARIO = "config_scenario" + FANCIRC_TIME = "fancirc_time" + W_LON = "w_lon" + DEV_HOLD = "dev_hold" + TEMP_UNIT = "temp_unit" + + +class ResponseCodes(Enum): + SUCCESS = "1" + PARAMETER_ERROR = "1001" + ACCESS_TOKEN_ERROR = "2001" + + +class ResponseCodesLock(Enum): + SUCCESS = 0 + + +class File: + file_id: str + type: Union[int, str] + url: str + status: int + en_algorithm: int + en_password: str + is_ai: int + ai_tag_list: List + ai_url: str + file_params: dict + + def __init__(self, dictionary): + for k, v in dictionary.items(): + setattr(self, k, v) + + if self.type == 1: + self.type = "Image" + else: + self.type = "Video" + + def __repr__(self): + return "".format(self.file_id, self.type) + + +class Event: + event_id: str + device_mac: str + device_model: str + event_category: int + event_value: str + event_ts: int + event_ack_result: int + is_feedback_correct: int + is_feedback_face: int + is_feedback_person: int + file_list: List[File] + event_params: dict + recognized_instance_list: List + tag_list: List + read_state: int + + def __init__(self, dictionary): + for k, v in dictionary.items(): + setattr(self, k, v) + temp_file_list = [] + if len(self.file_list) > 0: + for file in self.file_list: + # noinspection PyTypeChecker + temp_file_list.append(File(file)) + self.file_list = temp_file_list + + def __repr__(self): + return "".format(self.event_id, self.event_ts)