diff --git a/custom_components/vesync/__init__.py b/custom_components/vesync/__init__.py index fa66779..8ae086a 100644 --- a/custom_components/vesync/__init__.py +++ b/custom_components/vesync/__init__.py @@ -6,6 +6,7 @@ from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, Platform from homeassistant.core import HomeAssistant, ServiceCall from homeassistant.helpers import config_validation as cv from homeassistant.helpers.dispatcher import async_dispatcher_send +from pyvesync.vesync import VeSync from .common import async_process_devices from .const import ( @@ -21,7 +22,6 @@ from .const import ( VS_SENSORS, VS_SWITCHES, ) -from .pyvesync.vesync import VeSync PLATFORMS = { Platform.SWITCH: VS_SWITCHES, diff --git a/custom_components/vesync/common.py b/custom_components/vesync/common.py index b20f60d..be0e154 100644 --- a/custom_components/vesync/common.py +++ b/custom_components/vesync/common.py @@ -2,6 +2,7 @@ import logging from homeassistant.helpers.entity import Entity, ToggleEntity +from pyvesync.vesyncfan import model_features from .const import ( DOMAIN, @@ -13,7 +14,6 @@ from .const import ( VS_SENSORS, VS_SWITCHES, ) -from .pyvesync.vesyncfan import model_features _LOGGER = logging.getLogger(__name__) diff --git a/custom_components/vesync/config_flow.py b/custom_components/vesync/config_flow.py index 6fb299c..03c3c88 100644 --- a/custom_components/vesync/config_flow.py +++ b/custom_components/vesync/config_flow.py @@ -8,9 +8,9 @@ from homeassistant.components import dhcp from homeassistant.const import CONF_PASSWORD, CONF_USERNAME from homeassistant.core import callback from homeassistant.data_entry_flow import FlowResult +from pyvesync.vesync import VeSync from .const import DOMAIN -from .pyvesync.vesync import VeSync _LOGGER = logging.getLogger(__name__) diff --git a/custom_components/vesync/const.py b/custom_components/vesync/const.py index fd9a8ce..820d29f 100644 --- a/custom_components/vesync/const.py +++ b/custom_components/vesync/const.py @@ -16,6 +16,8 @@ VS_MANAGER = "manager" VS_MODE_AUTO = "auto" VS_MODE_MANUAL = "manual" +VS_TO_HA_ATTRIBUTES = {"humidity": "current_humidity"} + DEV_TYPE_TO_HA = { "Core200S": "fan", "Core300S": "fan", diff --git a/custom_components/vesync/fan.py b/custom_components/vesync/fan.py index 1915257..367accd 100644 --- a/custom_components/vesync/fan.py +++ b/custom_components/vesync/fan.py @@ -14,7 +14,7 @@ from homeassistant.util.percentage import ( ) from .common import VeSyncDevice -from .const import DEV_TYPE_TO_HA, DOMAIN, VS_DISCOVERY, VS_FANS +from .const import DEV_TYPE_TO_HA, DOMAIN, VS_DISCOVERY, VS_FANS, VS_TO_HA_ATTRIBUTES _LOGGER = logging.getLogger(__name__) @@ -114,28 +114,13 @@ class VeSyncFanHA(VeSyncDevice, FanEntity): def extra_state_attributes(self): """Return the state attributes of the fan.""" attr = {} - - if hasattr(self.smartfan, "active_time"): - attr["active_time"] = self.smartfan.active_time - - if hasattr(self.smartfan, "screen_status"): - attr["screen_status"] = self.smartfan.screen_status - - if hasattr(self.smartfan, "child_lock"): - attr["child_lock"] = self.smartfan.child_lock - - if hasattr(self.smartfan, "night_light"): - attr["night_light"] = self.smartfan.night_light - - if hasattr(self.smartfan, "air_quality"): - attr["air_quality"] = self.smartfan.air_quality - - if hasattr(self.smartfan, "mode"): - attr["mode"] = self.smartfan.mode - - if hasattr(self.smartfan, "filter_life"): - attr["filter_life"] = self.smartfan.filter_life - + for k, v in self.smarthumidifier.details.items(): + if k in VS_TO_HA_ATTRIBUTES: + attr[VS_TO_HA_ATTRIBUTES[k]] = v + elif k in self.state_attributes: + attr[f"vs_{k}"] = v + else: + attr[k] = v return attr def set_percentage(self, percentage): diff --git a/custom_components/vesync/humidifier.py b/custom_components/vesync/humidifier.py index c1e96d5..259452c 100644 --- a/custom_components/vesync/humidifier.py +++ b/custom_components/vesync/humidifier.py @@ -14,7 +14,14 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity_platform import AddEntitiesCallback from .common import VeSyncDevice, is_humidifier -from .const import DOMAIN, VS_DISCOVERY, VS_HUMIDIFIERS, VS_MODE_AUTO, VS_MODE_MANUAL +from .const import ( + DOMAIN, + VS_DISCOVERY, + VS_HUMIDIFIERS, + VS_MODE_AUTO, + VS_MODE_MANUAL, + VS_TO_HA_ATTRIBUTES, +) _LOGGER = logging.getLogger(__name__) @@ -22,7 +29,6 @@ MAX_HUMIDITY = 80 MIN_HUMIDITY = 30 MODES = [MODE_AUTO, MODE_NORMAL, MODE_SLEEP] -VESYNC_TO_HA_ATTRIBUTES = {"humidity": "current_humidity"} async def async_setup_entry( @@ -117,9 +123,11 @@ class VeSyncHumidifierHA(VeSyncDevice, HumidifierEntity): attr = {} for k, v in self.smarthumidifier.details.items(): - if k in VESYNC_TO_HA_ATTRIBUTES: - attr[VESYNC_TO_HA_ATTRIBUTES[k]] = v - elif k not in self.state_attributes: + if k in VS_TO_HA_ATTRIBUTES: + attr[VS_TO_HA_ATTRIBUTES[k]] = v + elif k in self.state_attributes: + attr[f"vs_{k}"] = v + else: attr[k] = v return attr diff --git a/custom_components/vesync/pyvesync/__init__.py b/custom_components/vesync/pyvesync/__init__.py deleted file mode 100755 index 3fb317f..0000000 --- a/custom_components/vesync/pyvesync/__init__.py +++ /dev/null @@ -1,6 +0,0 @@ -"""VeSync API Library.""" -import logging - -logging.basicConfig( - level=logging.INFO, format="%(asctime)s - %(levelname)5s - %(message)s" -) diff --git a/custom_components/vesync/pyvesync/__pycache__/__init__.cpython-39.pyc b/custom_components/vesync/pyvesync/__pycache__/__init__.cpython-39.pyc deleted file mode 100644 index 1c67091..0000000 Binary files a/custom_components/vesync/pyvesync/__pycache__/__init__.cpython-39.pyc and /dev/null differ diff --git a/custom_components/vesync/pyvesync/__pycache__/helpers.cpython-39.pyc b/custom_components/vesync/pyvesync/__pycache__/helpers.cpython-39.pyc deleted file mode 100644 index d6ae29e..0000000 Binary files a/custom_components/vesync/pyvesync/__pycache__/helpers.cpython-39.pyc and /dev/null differ diff --git a/custom_components/vesync/pyvesync/__pycache__/vesync.cpython-39.pyc b/custom_components/vesync/pyvesync/__pycache__/vesync.cpython-39.pyc deleted file mode 100644 index 38eea68..0000000 Binary files a/custom_components/vesync/pyvesync/__pycache__/vesync.cpython-39.pyc and /dev/null differ diff --git a/custom_components/vesync/pyvesync/__pycache__/vesyncbasedevice.cpython-39.pyc b/custom_components/vesync/pyvesync/__pycache__/vesyncbasedevice.cpython-39.pyc deleted file mode 100644 index 2f15ab3..0000000 Binary files a/custom_components/vesync/pyvesync/__pycache__/vesyncbasedevice.cpython-39.pyc and /dev/null differ diff --git a/custom_components/vesync/pyvesync/__pycache__/vesyncbulb.cpython-39.pyc b/custom_components/vesync/pyvesync/__pycache__/vesyncbulb.cpython-39.pyc deleted file mode 100644 index ad23ca9..0000000 Binary files a/custom_components/vesync/pyvesync/__pycache__/vesyncbulb.cpython-39.pyc and /dev/null differ diff --git a/custom_components/vesync/pyvesync/__pycache__/vesyncfan.cpython-39.pyc b/custom_components/vesync/pyvesync/__pycache__/vesyncfan.cpython-39.pyc deleted file mode 100644 index 987a5d8..0000000 Binary files a/custom_components/vesync/pyvesync/__pycache__/vesyncfan.cpython-39.pyc and /dev/null differ diff --git a/custom_components/vesync/pyvesync/__pycache__/vesyncoutlet.cpython-39.pyc b/custom_components/vesync/pyvesync/__pycache__/vesyncoutlet.cpython-39.pyc deleted file mode 100644 index a9865e2..0000000 Binary files a/custom_components/vesync/pyvesync/__pycache__/vesyncoutlet.cpython-39.pyc and /dev/null differ diff --git a/custom_components/vesync/pyvesync/__pycache__/vesyncswitch.cpython-39.pyc b/custom_components/vesync/pyvesync/__pycache__/vesyncswitch.cpython-39.pyc deleted file mode 100644 index 5cca964..0000000 Binary files a/custom_components/vesync/pyvesync/__pycache__/vesyncswitch.cpython-39.pyc and /dev/null differ diff --git a/custom_components/vesync/pyvesync/helpers.py b/custom_components/vesync/pyvesync/helpers.py deleted file mode 100644 index 6bc6bf0..0000000 --- a/custom_components/vesync/pyvesync/helpers.py +++ /dev/null @@ -1,241 +0,0 @@ -"""Helper functions for VeSync API.""" - -import hashlib -import logging -import time - -import requests - -logger = logging.getLogger(__name__) - -API_BASE_URL = "https://smartapi.vesync.com" -API_RATE_LIMIT = 30 -API_TIMEOUT = 5 - -DEFAULT_TZ = "America/New_York" -DEFAULT_REGION = "US" - -APP_VERSION = "2.8.6" -PHONE_BRAND = "SM N9005" -PHONE_OS = "Android" -MOBILE_ID = "1234567890123456" -USER_TYPE = "1" -BYPASS_APP_V = "VeSync 3.0.51" - - -class Helpers: - """VeSync Helper Functions.""" - - @staticmethod - def req_headers(manager) -> dict: - """Build header for api requests.""" - return { - "accept-language": "en", - "accountId": manager.account_id, - "appVersion": APP_VERSION, - "content-type": "application/json", - "tk": manager.token, - "tz": manager.time_zone, - } - - @staticmethod - def req_body_base(manager) -> dict: - """Return universal keys for body of api requests.""" - return {"timeZone": manager.time_zone, "acceptLanguage": "en"} - - @staticmethod - def req_body_auth(manager) -> dict: - """Keys for authenticating api requests.""" - return {"accountID": manager.account_id, "token": manager.token} - - @staticmethod - def req_body_details() -> dict: - """Detail keys for api requests.""" - return { - "appVersion": APP_VERSION, - "phoneBrand": PHONE_BRAND, - "phoneOS": PHONE_OS, - "traceId": str(int(time.time())), - } - - @classmethod - def req_body(cls, manager, type_) -> dict: - """Builder for body of api requests.""" - body = {} - - if type_ == "login": - body = {**cls.req_body_base(manager), **cls.req_body_details()} - body["email"] = manager.username - body["password"] = cls.hash_password(manager.password) - body["devToken"] = "" - body["userType"] = USER_TYPE - body["method"] = "login" - elif type_ == "devicedetail": - body = { - **cls.req_body_base(manager), - **cls.req_body_auth(manager), - **cls.req_body_details(), - } - body["method"] = "devicedetail" - body["mobileId"] = MOBILE_ID - elif type_ == "devicelist": - body = { - **cls.req_body_base(manager), - **cls.req_body_auth(manager), - **cls.req_body_details(), - } - body["method"] = "devices" - body["pageNo"] = "1" - body["pageSize"] = "100" - elif type_ == "devicestatus": - body = {**cls.req_body_base(manager), **cls.req_body_auth(manager)} - elif type_ == "energy_week": - body = { - **cls.req_body_base(manager), - **cls.req_body_auth(manager), - **cls.req_body_details(), - } - body["method"] = "energyweek" - body["mobileId"] = MOBILE_ID - elif type_ == "energy_month": - body = { - **cls.req_body_base(manager), - **cls.req_body_auth(manager), - **cls.req_body_details(), - } - body["method"] = "energymonth" - body["mobileId"] = MOBILE_ID - elif type_ == "energy_year": - body = { - **cls.req_body_base(manager), - **cls.req_body_auth(manager), - **cls.req_body_details(), - } - body["method"] = "energyyear" - body["mobileId"] = MOBILE_ID - elif type_ == "bypass": - body = { - **cls.req_body_base(manager), - **cls.req_body_auth(manager), - **cls.req_body_details(), - } - body["method"] = "bypass" - elif type_ == "bypass_config": - body = { - **cls.req_body_base(manager), - **cls.req_body_auth(manager), - **cls.req_body_details(), - } - body["method"] = "firmwareUpdateInfo" - - return body - - @staticmethod - def calculate_hex(hex_string) -> float: - """Credit for conversion to itsnotlupus/vesync_wsproxy.""" - hex_conv = hex_string.split(":") - return (int(hex_conv[0], 16) + int(hex_conv[1], 16)) / 8192 - - @staticmethod - def hash_password(string) -> str: - """Encode password.""" - return hashlib.md5(string.encode("utf-8")).hexdigest() - - @staticmethod - def call_api( - api: str, method: str, json: dict = None, headers: dict = None - ) -> tuple: - """Make API calls by passing endpoint, header and body.""" - response = None - status_code = None - - try: - logger.debug("[%s] calling '%s' api", method, api) - if method.lower() == "get": - r = requests.get( - API_BASE_URL + api, json=json, headers=headers, timeout=API_TIMEOUT - ) - elif method.lower() == "post": - r = requests.post( - API_BASE_URL + api, json=json, headers=headers, timeout=API_TIMEOUT - ) - elif method.lower() == "put": - r = requests.put( - API_BASE_URL + api, json=json, headers=headers, timeout=API_TIMEOUT - ) - except requests.exceptions.RequestException as e: - logger.debug(e) - except Exception as e: - logger.debug(e) - else: - if r.status_code == 200: - status_code = 200 - if r.content: - response = r.json() - else: - logger.debug("Unable to fetch %s%s", API_BASE_URL, api) - return response, status_code - - @staticmethod - def code_check(r: dict) -> bool: - """Test if code == 0 for successful API call.""" - return isinstance(r, dict) and r.get("code") == 0 - - @staticmethod - def build_details_dict(r: dict) -> dict: - """Build details dictionary from API response.""" - return { - "active_time": r.get("activeTime", 0), - "energy": r.get("energy", 0), - "night_light_status": r.get("nightLightStatus", None), - "night_light_brightness": r.get("nightLightBrightness", None), - "night_light_automode": r.get("nightLightAutomode", None), - "power": r.get("power", 0), - "voltage": r.get("voltage", 0), - } - - @staticmethod - def build_energy_dict(r: dict) -> dict: - """Build energy dictionary from API response.""" - return { - "energy_consumption_of_today": r.get("energyConsumptionOfToday", 0), - "cost_per_kwh": r.get("costPerKWH", 0), - "max_energy": r.get("maxEnergy", 0), - "total_energy": r.get("totalEnergy", 0), - "currency": r.get("currency", 0), - "data": r.get("data", 0), - } - - @staticmethod - def build_config_dict(r: dict) -> dict: - """Build configuration dictionary from API response.""" - if r.get("threshold") is not None: - threshold = r.get("threshold") - else: - threshold = r.get("thresHold") - return { - "current_firmware_version": r.get("currentFirmVersion"), - "latest_firmware_version": r.get("latestFirmVersion"), - "maxPower": r.get("maxPower"), - "threshold": threshold, - "power_protection": r.get("powerProtectionStatus"), - "energy_saving_status": r.get("energySavingStatus"), - } - - @classmethod - def bypass_body_v2(cls, manager): - """Build body dict for bypass calls.""" - bdy = {} - bdy.update(**cls.req_body(manager, "bypass")) - bdy["method"] = "bypassV2" - bdy["debugMode"] = False - bdy["deviceRegion"] = DEFAULT_REGION - return bdy - - @staticmethod - def bypass_header(): - """Build bypass header dict.""" - return { - "Content-Type": "application/json; charset=UTF-8", - "User-Agent": "okhttp/3.12.1", - } diff --git a/custom_components/vesync/pyvesync/vesync.py b/custom_components/vesync/pyvesync/vesync.py deleted file mode 100755 index 0b6f814..0000000 --- a/custom_components/vesync/pyvesync/vesync.py +++ /dev/null @@ -1,300 +0,0 @@ -"""VeSync API Device Library.""" - -import logging -import re -import time -from itertools import chain -from typing import Tuple - -from . import vesyncbulb as bulb_mods -from . import vesyncfan as fan_mods -from . import vesyncoutlet as outlet_mods -from . import vesyncswitch as switch_mods -from .helpers import Helpers -from .vesyncbasedevice import VeSyncBaseDevice - -logger = logging.getLogger(__name__) - -API_RATE_LIMIT: int = 30 -DEFAULT_TZ: str = "America/New_York" - -DEFAULT_ENER_UP_INT: int = 21600 - - -def object_factory(dev_type, config, manager) -> Tuple[str, VeSyncBaseDevice]: - """Get device type and instantiate class.""" - - def fans(dev_type, config, manager): - fan_cls = fan_mods.fan_modules[dev_type] - fan_obj = getattr(fan_mods, fan_cls) - return "fans", fan_obj(config, manager) - - def outlets(dev_type, config, manager): - outlet_cls = outlet_mods.outlet_modules[dev_type] - outlet_obj = getattr(outlet_mods, outlet_cls) - return "outlets", outlet_obj(config, manager) - - def switches(dev_type, config, manager): - switch_cls = switch_mods.switch_modules[dev_type] - switch_obj = getattr(switch_mods, switch_cls) - return "switches", switch_obj(config, manager) - - def bulbs(dev_type, config, manager): - bulb_cls = bulb_mods.bulb_modules[dev_type] - bulb_obj = getattr(bulb_mods, bulb_cls) - return "bulbs", bulb_obj(config, manager) - - if dev_type in fan_mods.fan_modules: - type_str, dev_obj = fans(dev_type, config, manager) - elif dev_type in outlet_mods.outlet_modules: - type_str, dev_obj = outlets(dev_type, config, manager) - elif dev_type in switch_mods.switch_modules: - type_str, dev_obj = switches(dev_type, config, manager) - elif dev_type in bulb_mods.bulb_modules: - type_str, dev_obj = bulbs(dev_type, config, manager) - else: - logger.debug( - "Unknown device named %s model %s", - config.get("deviceName", ""), - config.get("deviceType", ""), - ) - type_str = "unknown" - dev_obj = None - return type_str, dev_obj - - -class VeSync: - """VeSync API functions.""" - - def __init__(self, username, password, time_zone=DEFAULT_TZ, debug=False): - """Initialize VeSync class with username, password and time zone.""" - self.debug = debug - if debug: - logger.setLevel(logging.DEBUG) - self.username = username - self.password = password - self.token = None - self.account_id = None - self.devices = None - self.enabled = False - self.update_interval = API_RATE_LIMIT - self.last_update_ts = None - self.in_process = False - self._energy_update_interval = DEFAULT_ENER_UP_INT - self._energy_check = True - self._dev_list = {} - self.outlets = [] - self.switches = [] - self.fans = [] - self.bulbs = [] - self.scales = [] - - self._dev_list = { - "fans": self.fans, - "outlets": self.outlets, - "switches": self.switches, - "bulbs": self.bulbs, - } - - if isinstance(time_zone, str) and time_zone: - reg_test = r"[^a-zA-Z/_]" - if bool(re.search(reg_test, time_zone)): - self.time_zone = DEFAULT_TZ - logger.debug("Invalid characters in time zone - %s", time_zone) - else: - self.time_zone = time_zone - else: - self.time_zone = DEFAULT_TZ - logger.debug("Time zone is not a string") - - @property - def energy_update_interval(self) -> int: - """Return energy update interval.""" - return self._energy_update_interval - - @energy_update_interval.setter - def energy_update_interval(self, new_energy_update: int) -> None: - """Set energy update interval in seconds.""" - if new_energy_update > 0: - self._energy_update_interval = new_energy_update - - @staticmethod - def remove_dev_test(device, new_list: list) -> bool: - """Test if device should be removed - False = Remove.""" - if isinstance(new_list, list) and device.cid: - for item in new_list: - device_found = False - if "cid" in item: - if device.cid == item["cid"]: - device_found = True - break - else: - logger.debug("No cid found in - %s", str(item)) - if not device_found: - logger.debug( - "Device removed - %s - %s", device.device_name, device.device_type - ) - return False - return True - - def add_dev_test(self, new_dev: dict) -> bool: - """Test if new device should be added - True = Add.""" - if "cid" in new_dev: - for _, v in self._dev_list.items(): - for dev in v: - if ( - dev.cid == new_dev.get("cid") - and new_dev.get("subDeviceNo", 0) == dev.sub_device_no - ): - return False - return True - - def remove_old_devices(self, devices: list) -> bool: - """Remove devices not found in device list return.""" - for k, v in self._dev_list.items(): - before = len(v) - v[:] = [x for x in v if self.remove_dev_test(x, devices)] - after = len(v) - if before != after: - logger.debug("%s %s removed", str(before - after), k) - return True - - @staticmethod - def set_dev_id(devices: list) -> list: - """Correct devices without cid or uuid.""" - dev_rem = [] - for dev_num, dev in enumerate(devices): - if dev.get("cid") is None: - if dev.get("macID") is not None: - dev["cid"] = dev["macID"] - elif dev.get("uuid") is not None: - dev["cid"] = dev["uuid"] - else: - dev_rem.append(dev_num) - logger.warning("Device with no ID - %s", dev.get("deviceName")) - if dev_rem: - devices = [i for j, i in enumerate(devices) if j not in dev_rem] - return devices - - def process_devices(self, dev_list: list) -> bool: - """Instantiate Device Objects.""" - devices = VeSync.set_dev_id(dev_list) - - num_devices = sum( - len(v) if isinstance(v, list) else 1 for _, v in self._dev_list.items() - ) - - if not devices: - logger.warning("No devices found in api return") - return False - if num_devices == 0: - logger.debug("New device list initialized") - else: - self.remove_old_devices(devices) - - devices[:] = [x for x in devices if self.add_dev_test(x)] - - detail_keys = ["deviceType", "deviceName", "deviceStatus"] - for dev in devices: - if any(k not in dev for k in detail_keys): - logger.debug("Error adding device") - continue - dev_type = dev.get("deviceType") - try: - device_str, device_obj = object_factory(dev_type, dev, self) - device_list = getattr(self, device_str) - device_list.append(device_obj) - except AttributeError as err: - logger.debug("Error - %s", err) - logger.debug("%s device not added", dev_type) - continue - - return True - - def get_devices(self) -> bool: - """Return tuple listing outlets, switches, and fans of devices.""" - if not self.enabled: - return False - - self.in_process = True - proc_return = False - response, _ = Helpers.call_api( - "/cloud/v1/deviceManaged/devices", - "post", - headers=Helpers.req_headers(self), - json=Helpers.req_body(self, "devicelist"), - ) - - if response and Helpers.code_check(response): - if "result" in response and "list" in response["result"]: - device_list = response["result"]["list"] - if self.debug: - logger.debug(str(device_list)) - proc_return = self.process_devices(device_list) - else: - logger.error("Device list in response not found") - else: - logger.warning("Error retrieving device list") - - self.in_process = False - - return proc_return - - def login(self) -> bool: - """Return True if log in request succeeds.""" - user_check = isinstance(self.username, str) and len(self.username) > 0 - pass_check = isinstance(self.password, str) and len(self.password) > 0 - if not user_check: - logger.error("Username invalid") - return False - if not pass_check: - logger.error("Password invalid") - return False - - response, _ = Helpers.call_api( - "/cloud/v1/user/login", "post", json=Helpers.req_body(self, "login") - ) - - if Helpers.code_check(response) and "result" in response: - self.token = response.get("result").get("token") - self.account_id = response.get("result").get("accountID") - self.enabled = True - - return True - logger.error("Error logging in with username and password") - return False - - def device_time_check(self) -> bool: - """Test if update interval has been exceeded.""" - return ( - self.last_update_ts is None - or (time.time() - self.last_update_ts) > self.update_interval - ) - - def update(self) -> None: - """Fetch updated information about devices.""" - if self.device_time_check(): - - if not self.enabled: - logger.error("Not logged in to VeSync") - return - self.get_devices() - - devices = list(self._dev_list.values()) - - for device in chain(*devices): - device.update() - - self.last_update_ts = time.time() - - def update_energy(self, bypass_check=False) -> None: - """Fetch updated energy information about devices.""" - if self.outlets: - for outlet in self.outlets: - outlet.update_energy(bypass_check) - - def update_all_devices(self) -> None: - """Run get_details() for each device.""" - devices = list(self._dev_list.keys()) - for dev in chain(*devices): - dev.get_details() diff --git a/custom_components/vesync/pyvesync/vesyncbasedevice.py b/custom_components/vesync/pyvesync/vesyncbasedevice.py deleted file mode 100644 index f9799e4..0000000 --- a/custom_components/vesync/pyvesync/vesyncbasedevice.py +++ /dev/null @@ -1,117 +0,0 @@ -"""Base class for all VeSync devices.""" - -import collections -import json -import logging -from typing import Optional, Union - -logger = logging.getLogger(__name__) - - -class VeSyncBaseDevice: - """Properties shared across all VeSync devices.""" - - def __init__(self, details: dict, manager): - """Initialize VeSync device base class.""" - self.manager = manager - if "cid" in details and details["cid"] is not None: - self.device_name: str = details.get("deviceName", None) - self.device_image: Optional[str] = details.get("deviceImg", None) - self.cid: str = details.get("cid", None) - self.connection_status: str = details.get("connectionStatus", None) - self.connection_type: Optional[str] = details.get("connectionType", None) - self.device_type: str = details.get("deviceType", None) - self.type: str = details.get("type", None) - self.uuid: Optional[str] = details.get("uuid", None) - self.config_module: str = details.get("configModule", None) - self.mac_id: Optional[str] = details.get("macID", None) - self.mode: Optional[str] = details.get("mode", None) - self.speed: Union[str, int, None] = details.get("speed", None) - self.extension = details.get("extension", None) - self.current_firm_version = details.get("currentFirmVersion", None) - self.sub_device_no = details.get("subDeviceNo", 0) - self.config: dict = {} - if isinstance(details.get("extension"), dict): - ext = details["extension"] - self.speed = ext.get("fanSpeedLevel") - self.mode = ext.get("mode") - if self.connection_status != "online": - self.device_status = "off" - else: - self.device_status = details.get("deviceStatus", None) - - else: - logger.error("No cid found for %s", self.__class__.__name__) - - def __eq__(self, other): - """Use device CID and subdevice number to test equality.""" - return bool(other.cid == self.cid and other.sub_device_no == self.sub_device_no) - - def __hash__(self): - """Use CID and sub-device number to make device hash.""" - if isinstance(self.sub_device_no, int) and self.sub_device_no > 0: - return hash(self.cid + str(self.sub_device_no)) - return hash(self.cid) - - def __str__(self): - """Use device info for string represtation of class.""" - return f"Device Name: {self.device_name}, \ - Device Type: {self.device_type},\ - SubDevice No.: {self.sub_device_no},\ - Status: {self.device_status}" - - def __repr__(self): - """Representation of device details.""" - return f"DevClass: {self.__class__.__name__},\ - Name:{self.device_name}, Device No: {self.sub_device_no},\ - DevStatus: {self.device_status}, CID: {self.cid}" - - @property - def is_on(self) -> bool: - """Return true if device is on.""" - if self.device_status == "on": - return True - return False - - @property - def firmware_update(self) -> bool: - """Return True if firmware update available.""" - cfv = self.config.get("current_firmware_version") - lfv = self.config.get("latest_firmware_version") - if cfv is not None and lfv is not None: - if cfv != lfv: - return True - else: - logger.debug("Call device.get_config() to get firmware versions") - return False - - def display(self) -> None: - """Print formatted device info to stdout.""" - disp = [ - ("Device Name:", self.device_name), - ("Model: ", self.device_type), - ("Subdevice No: ", str(self.sub_device_no)), - ("Status: ", self.device_status), - ("Online: ", self.connection_status), - ("Type: ", self.type), - ("CID: ", self.cid), - ] - if self.uuid is not None: - disp.append(("UUID: ", self.uuid)) - disp1 = collections.OrderedDict(disp) - for k, v in disp1.items(): - print(f"{k:.<15} {v:<15}") - - def displayJSON(self) -> str: - """JSON API for device details.""" - return json.dumps( - { - "Device Name": self.device_name, - "Model": self.device_type, - "Subdevice No": str(self.sub_device_no), - "Status": self.device_status, - "Online": self.connection_status, - "Type": self.type, - "CID": self.cid, - } - ) diff --git a/custom_components/vesync/pyvesync/vesyncbulb.py b/custom_components/vesync/pyvesync/vesyncbulb.py deleted file mode 100644 index 19fb115..0000000 --- a/custom_components/vesync/pyvesync/vesyncbulb.py +++ /dev/null @@ -1,378 +0,0 @@ -"""Etekcity Smart Light Bulb.""" - -import json -import logging -from abc import ABCMeta, abstractmethod -from typing import Dict, Union - -from .helpers import Helpers as helpers -from .vesyncbasedevice import VeSyncBaseDevice - -logger = logging.getLogger(__name__) - - -# Possible features - dimmable, color_temp, rgb_shift -feature_dict: dict = { - "ESL100": {"module": "VeSyncBulbESL100", "features": ["dimmable"]}, - "ESL100CW": { - "module": "VeSyncBulbESL100CW", - "features": ["dimmable", "color_temp"], - }, -} - - -bulb_modules: dict = {k: v["module"] for k, v in feature_dict.items()} - -__all__: list = list(bulb_modules.values()) + ["bulb_modules"] - - -def pct_to_kelvin(pct: float, max_k: int = 6500, min_k: int = 2700) -> float: - """Convert percent to kelvin.""" - return ((max_k - min_k) * pct / 100) + min_k - - -class VeSyncBulb(VeSyncBaseDevice): - """Base class for VeSync Bulbs.""" - - __metaclass__ = ABCMeta - - def __init__(self, details: Dict[str, Union[str, list]], manager): - """Initialize VeSync smart bulb base class.""" - super().__init__(details, manager) - self._brightness = 0 - self._color_temp = 0 - self.features = feature_dict.get(self.device_type, {}).get("features") - if self.features is None: - logger.error("No configuration set for - %s", self.device_name) - raise RuntimeError(f"No configuration set for - {self.device_name}") - - @property - def brightness(self) -> int: - """Return brightness of vesync bulb.""" - if self.dimmable_feature and self._brightness is not None: - return self._brightness - return 0 - - @property - def color_temp_kelvin(self) -> int: - """Return Color Temp of bulb if supported in Kelvin.""" - if self.color_temp_feature and self._color_temp is not None: - return int(pct_to_kelvin(self._color_temp)) - return 0 - - @property - def color_temp_pct(self) -> int: - """Return color temperature of bulb in percent.""" - if self.color_temp_feature and self._color_temp is not None: - return int(self._color_temp) - return 0 - - @property - def dimmable_feature(self) -> bool: - """Return true if dimmable bulb.""" - return "dimmable" in self.features - - @property - def color_temp_feature(self) -> bool: - """Return true if bulb supports color temperature changes.""" - return "color_temp" in feature_dict[self.device_type] - - @property - def rgb_shift_feature(self) -> bool: - """Return True if bulb supports changing color.""" - return "rgb_shift" in feature_dict[self.device_type] - - @abstractmethod - def get_details(self) -> None: - """Get vesync bulb details.""" - - @abstractmethod - def toggle(self, status: str) -> bool: - """Toggle vesync lightbulb.""" - - @abstractmethod - def get_config(self) -> None: - """Call api to get configuration details and firmware.""" - - def turn_on(self) -> bool: - """Turn on vesync bulbs.""" - return self._toggle("on") - - def turn_off(self) -> bool: - """Turn off vesync bulbs.""" - return self._toggle("off") - - def _toggle(self, state: str): - if self.toggle(state): - self.device_status = state - return True - logger.warning("Error turning %s %s", self.device_name, state) - return False - - def update(self) -> None: - """Update bulb details.""" - self.get_details() - - def display(self) -> None: - """Return formatted bulb info to stdout.""" - super().display() - if self.connection_status == "online" and self.dimmable_feature: - disp1 = [("Brightness: ", self.brightness, "%")] - for line in disp1: - print(f"{line[0]:.<17} {line[1]} {line[2]}") - - def displayJSON(self) -> str: - """Return bulb device info in JSON format.""" - sup = super().displayJSON() - sup_val = json.loads(sup) - if self.connection_status == "online": - if self.dimmable_feature: - sup_val.update({"Brightness": str(self.brightness)}) - if self.color_temp_feature: - sup_val.update({"Kelvin": str(self.color_temp_kelvin)}) - return sup_val - - -class VeSyncBulbESL100(VeSyncBulb): - """Object to hold VeSync ESL100 light bulb.""" - - def __init__(self, details, manager): - """Initialize Etekcity ESL100 Dimmable Bulb.""" - super().__init__(details, manager) - self.details: dict = {} - - def get_details(self) -> None: - """Get details of dimmable bulb.""" - body = helpers.req_body(self.manager, "devicedetail") - body["uuid"] = self.uuid - r, _ = helpers.call_api( - "/SmartBulb/v1/device/devicedetail", - "post", - headers=helpers.req_headers(self.manager), - json=body, - ) - if helpers.code_check(r): - self.connection_status = r.get("connectionStatus") - self.device_status = r.get("deviceStatus") - if self.dimmable_feature: - self._brightness = int(r.get("brightNess")) - else: - logger.debug("Error getting %s details", self.device_name) - - def get_config(self) -> None: - """Get configuration of dimmable bulb.""" - body = helpers.req_body(self.manager, "devicedetail") - body["method"] = "configurations" - body["uuid"] = self.uuid - - r, _ = helpers.call_api( - "/SmartBulb/v1/device/configurations", - "post", - headers=helpers.req_headers(self.manager), - json=body, - ) - - if helpers.code_check(r): - self.config = helpers.build_config_dict(r) - else: - logger.warning("Error getting %s config info", self.device_name) - - def toggle(self, status) -> bool: - """Toggle dimmable bulb.""" - body = self._get_body(status) - r, _ = helpers.call_api( - "/SmartBulb/v1/device/devicestatus", - "put", - headers=helpers.req_headers(self.manager), - json=body, - ) - if helpers.code_check(r): - self.device_status = status - return True - return False - - def set_brightness(self, brightness: int) -> bool: - """Set brightness of dimmable bulb.""" - if not self.dimmable_feature: - logger.debug("%s is not dimmable", self.device_name) - return False - if isinstance(brightness, int) and (brightness <= 0 or brightness > 100): - - logger.warning("Invalid brightness") - return False - - body = self._get_body("on") - body["brightNess"] = str(brightness) - r, _ = helpers.call_api( - "/SmartBulb/v1/device/updateBrightness", - "put", - headers=helpers.req_headers(self.manager), - json=body, - ) - - if helpers.code_check(r): - self._brightness = brightness - return True - - logger.debug("Error setting brightness for %s", self.device_name) - return False - - def _get_body(self, status: str): - body = helpers.req_body(self.manager, "devicestatus") - body["uuid"] = self.uuid - body["status"] = status - return body - - -class VeSyncBulbESL100CW(VeSyncBulb): - """VeSync Tunable and Dimmable White Bulb.""" - - def __init__(self, details, manager): - """Initialize Etekcity Tunable white bulb.""" - super().__init__(details, manager) - - def get_details(self) -> None: - """Get details of tunable bulb.""" - body = helpers.req_body(self.manager, "bypass") - body["cid"] = self.cid - body["jsonCmd"] = {"getLightStatus": "get"} - body["configModule"] = self.config_module - r, _ = helpers.call_api( - "/cloud/v1/deviceManaged/bypass", - "post", - headers=helpers.req_headers(self.manager), - json=body, - ) - if not isinstance(r, dict) or not helpers.code_check(r): - logger.debug("Error calling %s", self.device_name) - return - response = r - if response.get("result", {}).get("light") is not None: - light = response.get("result", {}).get("light") - self.connection_status = "online" - self.device_status = light.get("action", "off") - if self.dimmable_feature: - self._brightness = light.get("brightness") - if self.color_temp_feature: - self._color_temp = light.get("colorTempe") - elif response.get("code") == -11300027: - logger.debug("%s device offline", self.device_name) - self.connection_status = "offline" - self.device_status = "off" - else: - logger.debug( - "%s - Unknown return code - %d with message %s", - self.device_name, - response.get("code"), - response.get("msg"), - ) - - def get_config(self) -> None: - """Get configuration and firmware info of tunable bulb.""" - body = helpers.req_body(self.manager, "bypass_config") - body["uuid"] = self.uuid - - r, _ = helpers.call_api( - "/cloud/v1/deviceManaged/configurations", - "post", - headers=helpers.req_headers(self.manager), - json=body, - ) - - if helpers.code_check(r): - self.config = helpers.build_config_dict(r) - else: - logger.debug("Error getting %s config info", self.device_name) - - def toggle(self, status) -> bool: - """Toggle tunable bulb.""" - body = helpers.req_body(self.manager, "bypass") - body["cid"] = self.cid - body["configModule"] = self.config_module - body["jsonCmd"] = {"light": {"action": status}} - r, _ = helpers.call_api( - "/cloud/v1/deviceManaged/bypass", - "post", - headers=helpers.req_headers(self.manager), - json=body, - ) - if helpers.code_check(r) == 0: - self.device_status = status - return True - logger.debug("%s offline", self.device_name) - self.device_status = "off" - self.connection_status = "offline" - return False - - def set_brightness(self, brightness: int) -> bool: - """Set brightness of tunable bulb.""" - if not self.dimmable_feature: - logger.debug("%s is not dimmable", self.device_name) - return False - if brightness <= 0 or brightness > 100: - logger.debug("Invalid brightness") - return False - - body = helpers.req_body(self.manager, "bypass") - body["cid"] = self.cid - body["configModule"] = self.config_module - if self.device_status == "off": - light_dict = {"action": "on", "brightness": brightness} - else: - light_dict = {"brightness": brightness} - body["jsonCmd"] = {"light": light_dict} - r, _ = helpers.call_api( - "/cloud/v1/deviceManaged/bypass", - "post", - headers=helpers.req_headers(self.manager), - json=body, - ) - - if helpers.code_check(r): - self._brightness = brightness - return True - self.device_status = "off" - self.connection_status = "offline" - logger.debug("%s offline", self.device_name) - - return False - - def set_color_temp(self, color_temp: int) -> bool: - """Set Color Temperature of Bulb in pct (1 - 100).""" - if color_temp < 0 or color_temp > 100: - logger.debug("Invalid color temperature - only between 0 and 100") - return False - body = helpers.req_body(self.manager, "bypass") - body["cid"] = self.cid - body["jsonCmd"] = {"light": {}} - if self.device_status == "off": - light_dict = {"action": "on", "colorTempe": color_temp} - else: - light_dict = {"colorTempe": color_temp} - body["jsonCmd"]["light"] = light_dict - r, _ = helpers.call_api( - "/cloud/v1/deviceManaged/bypass", - "post", - headers=helpers.req_headers(self.manager), - json=body, - ) - - if not helpers.code_check(r): - return False - - if r.get("code") == -11300027: - logger.debug("%s device offline", self.device_name) - self.connection_status = "offline" - self.device_status = "off" - return False - if r.get("code") == 0: - self.device_status = "on" - self._color_temp = color_temp - return True - logger.debug( - "%s - Unknown return code - %d with message %s", - self.device_name, - r.get("code"), - r.get("msg"), - ) - return False diff --git a/custom_components/vesync/pyvesync/vesyncfan.py b/custom_components/vesync/pyvesync/vesyncfan.py deleted file mode 100644 index e4e25b5..0000000 --- a/custom_components/vesync/pyvesync/vesyncfan.py +++ /dev/null @@ -1,1299 +0,0 @@ -"""VeSync API for controlling fans and purifiers.""" - -import json -import logging -from typing import Dict, Tuple, Union - -from .helpers import Helpers -from .vesyncbasedevice import VeSyncBaseDevice - -humid_features: dict = { - "Classic300S": { - "module": "VeSyncHumid200300S", - "models": ["Classic300S", "LUH-A601S-WUSB"], - "features": ["nightlight"], - "mist_modes": ["auto", "sleep", "manual"], - "mist_levels": list(range(1, 10)), - }, - "Classic200S": { - "module": "VeSyncHumid200S", - "models": ["Classic200S"], - "features": [], - "mist_modes": ["auto", "manual"], - "mist_levels": list(range(1, 10)), - }, - "Dual200S": { - "module": "VeSyncHumid200300S", - "models": ["Dual200S", "LUH-D301S-WUSR", "LUH-D301S-WJP", "LUH-D301S-WEU"], - "features": [], - "mist_modes": ["auto", "manual"], - "mist_levels": list(range(1, 3)), - }, - "LV600S": { - "module": "VeSyncHumid200300S", - "models": [ - "LUH-A602S-WUSR", - "LUH-A602S-WUS", - "LUH-A602S-WEUR", - "LUH-A602S-WEU", - "LUH-A602S-WJP", - ], - "features": ["warm_mist", "nightlight"], - "mist_modes": ["humidity", "sleep", "manual"], - "mist_levels": list(range(1, 10)), - "warm_mist_levels": [0, 1, 2, 3], - }, -} - - -air_features: dict = { - "Core200S": { - "module": "VeSyncAirBypass", - "models": ["Core200S", "LAP-C201S-AUSR", "LAP-C202S-WUSR"], - "modes": ["sleep", "off"], - "features": [], - "levels": list(range(1, 4)), - }, - "Core300S": { - "module": "VeSyncAirBypass", - "models": ["Core300S", "LAP-C301S-WJP"], - "modes": ["sleep", "off", "auto"], - "features": ["air_quality"], - "levels": list(range(1, 5)), - }, - "Core400S": { - "module": "VeSyncAirBypass", - "models": ["Core400S", "LAP-C401S-WJP", "LAP-C401S-WUSR", "LAP-C401S-WAAA"], - "modes": ["sleep", "off", "auto"], - "features": ["air_quality"], - "levels": list(range(1, 5)), - }, - "Core600S": { - "module": "VeSyncAirBypass", - "models": ["Core600S", "LAP-C601S-WUS", "LAP-C601S-WUSR", "LAP-C601S-WEU"], - "modes": ["sleep", "off", "auto"], - "features": ["air_quality"], - "levels": list(range(1, 5)), - }, - "LV-PUR131S": { - "module": "VeSyncAir131", - "models": ["LV-PUR131S", "LV-RH131S"], - "features": ["air_quality"], - }, -} - - -logger = logging.getLogger(__name__) - - -def model_dict() -> dict: - """Build purifier and humidifier model dictionary.""" - model_modules = {} - for dev_dict in {**air_features, **humid_features}.values(): - for model in dev_dict["models"]: - model_modules[model] = dev_dict["module"] - return model_modules - - -def model_features(dev_type: str) -> dict: - """Get features from device type.""" - for dev_dict in {**air_features, **humid_features}.values(): - if dev_type in dev_dict["models"]: - return dev_dict - raise ValueError("Device not configured") - - -fan_classes: set = {v["module"] for k, v in {**air_features, **humid_features}.items()} - -fan_modules: dict = model_dict() - -__all__: list = list(fan_classes) + ["fan_modules"] - - -class VeSyncAirBypass(VeSyncBaseDevice): - """Base class for Levoit Purifier Bypass API Calls.""" - - def __init__(self, details: Dict[str, list], manager): - """Initialize air devices.""" - super().__init__(details, manager) - self.enabled = True - self.config_dict = model_features(self.device_type) - self.features = self.config_dict.get("features", []) - if not isinstance(self.config_dict.get("modes"), list): - logger.error( - "Please set modes for %s in the configuration", self.device_type - ) - raise RuntimeError( - "Please set modes for %s in the configuration", self.device_type - ) - self.modes = self.config_dict["modes"] - self.air_quality_feature = "air_quality" in self.features - self.details: Dict[str, Union[str, int, float, bool]] = { - "filter_life": 0, - "mode": "manual", - "level": 0, - "display": False, - "child_lock": False, - "night_light": "off", - } - if self.air_quality_feature: - self.details["ait_quality"] = 0 - self.config: Dict[str, Union[str, int, float, bool]] = { - "display": False, - "display_forever": False, - } - - def build_api_dict(self, method: str) -> Tuple[Dict, Dict]: - """Build device api body dictionary. - - standard modes are: ['getPurifierStatus', 'setSwitch', - 'setNightLight', - 'setLevel', 'setPurifierMode', 'setDisplay', - 'setChildLock'] - """ - modes = [ - "getPurifierStatus", - "setSwitch", - "setNightLight", - "setLevel", - "setPurifierMode", - "setDisplay", - "setChildLock", - "setIndicatorLight", - ] - if method not in modes: - logger.debug("Invalid mode - %s", method) - return {}, {} - head = Helpers.bypass_header() - body = Helpers.bypass_body_v2(self.manager) - body["cid"] = self.cid - body["configModule"] = self.config_module - body["payload"] = {"method": method, "source": "APP"} - return head, body - - def build_purifier_dict(self, dev_dict: dict) -> None: - """Build Bypass purifier status dictionary.""" - self.enabled = dev_dict.get("enabled", False) - self.device_status = "on" if self.enabled else "off" - self.details["filter_life"] = dev_dict.get("filter_life", 0) - self.mode = dev_dict.get("mode", "manual") - self.speed = dev_dict.get("level", 0) - self.details["display"] = dev_dict.get("display", False) - self.details["child_lock"] = dev_dict.get("child_lock", False) - self.details["night_light"] = dev_dict.get("night_light", "off") - self.details["display"] = dev_dict.get("display", False) - self.details["display_forever"] = dev_dict.get("display_forever", False) - if self.air_quality_feature: - self.details["air_quality"] = dev_dict.get("air_quality", 0) - - def build_config_dict(self, conf_dict: Dict[str, str]) -> None: - """Build configuration dict for Bypass purifier.""" - self.config["display"] = conf_dict.get("display", False) - self.config["display_forever"] = conf_dict.get("display_forever", False) - - def get_details(self) -> None: - """Build Bypass Purifier details dictionary.""" - head = Helpers.bypass_header() - body = Helpers.bypass_body_v2(self.manager) - body["cid"] = self.cid - body["configModule"] = self.config_module - body["payload"] = {"method": "getPurifierStatus", "source": "APP", "data": {}} - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - if not isinstance(r, dict): - logger.debug("Error in purifier response") - return - if not isinstance(r.get("result"), dict): - logger.debug("Error in purifier response") - return - outer_result = r.get("result", {}) - inner_result = r.get("result", {}).get("result") if outer_result else None - if inner_result is not None and Helpers.code_check(r): - if outer_result.get("code") == 0: - self.build_purifier_dict(inner_result) - else: - logger.debug("error in inner result dict from purifier") - if inner_result.get("configuration", {}): - self.build_config_dict(inner_result.get("configuration", {})) - else: - logger.debug("No configuration found in purifier status") - else: - logger.debug("Error in purifier response") - - def update(self): - """Update Purifier details.""" - self.get_details() - - def change_fan_speed(self, speed=None) -> bool: - """Change fan speed based on levels in configuration dict.""" - speeds: list = self.config_dict.get("levels", []) - current_speed = self.speed - - if speed is not None: - if speed not in speeds: - logger.debug( - "%s is invalid speed - valid speeds are %s", speed, str(speeds) - ) - return False - new_speed = speed - if current_speed == new_speed: - return True - elif current_speed == speeds[-1]: - new_speed = speeds[0] - else: - current_index = speeds.index(current_speed) - new_speed = speeds[current_index + 1] - - body = Helpers.req_body(self.manager, "devicestatus") - body["uuid"] = self.uuid - - head, body = self.build_api_dict("setLevel") - if not head and not body: - return False - - body["payload"]["data"] = { - "id": 0, - "level": new_speed, - "type": "wind", - "mode": "manual", - } - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - self.speed = new_speed - return True - logger.debug("Error changing %s speed", self.device_name) - return False - - def child_lock_on(self) -> bool: - """Turn Bypass child lock on.""" - return self.set_child_lock(True) - - def child_lock_off(self) -> bool: - """Turn Bypass child lock off.""" - return self.set_child_lock(False) - - def set_child_lock(self, mode: bool) -> bool: - """Set Bypass child lock.""" - if mode not in (True, False): - logger.debug("Invalid mode passed to set_child_lock - %s", mode) - return False - - head, body = self.build_api_dict("setChildLock") - if not head and not body: - return False - - body["payload"]["data"] = {"child_lock": mode} - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - self.details["child_lock"] = mode - return True - if isinstance(r, dict): - logger.debug("Error toggling child lock") - else: - logger.debug("Error in api return json for %s", self.device_name) - return False - - def mode_toggle(self, mode: str) -> bool: - """Set purifier mode - sleep or manual.""" - if mode.lower() not in self.modes: - logger.debug("Invalid purifier mode used - %s", mode) - return False - head, body = self.build_api_dict("setPurifierMode") - if not head and not body: - return False - - body["payload"]["data"] = {"mode": mode.lower()} - if mode == "manual": - body["payload"] = { - "data": {"id": 0, "level": 1, "type": "wind"}, - "method": "setLevel", - "type": "APP", - } - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if Helpers.code_check(r): - return True - logger.debug("Error setting purifier mode") - return False - - def manual_mode(self) -> bool: - """Set mode to manual.""" - if "manual" not in self.modes: - logger.debug("%s does not have manual mode", self.device_name) - return False - return self.mode_toggle("manual") - - def sleep_mode(self) -> bool: - """Set sleep mode to on.""" - if "sleep" not in self.modes: - logger.debug("%s does not have sleep mode", self.device_name) - return False - return self.mode_toggle("sleep") - - def auto_mode(self) -> bool: - """Set mode to auto.""" - if "auto" not in self.modes: - logger.debug("%s does not have auto mode", self.device_name) - return False - return self.mode_toggle("auto") - - def toggle_switch(self, toggle: bool) -> bool: - """Toggle purifier on/off.""" - if not isinstance(toggle, bool): - logger.debug("Invalid toggle value for purifier switch") - return False - - head = Helpers.bypass_header() - body = Helpers.bypass_body_v2(self.manager) - body["cid"] = self.cid - body["configModule"] = self.config_module - body["payload"] = { - "data": {"enabled": toggle, "id": 0}, - "method": "setSwitch", - "source": "APP", - } - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - self.device_status = "on" if toggle else "off" - return True - logger.debug("Error toggling purifier - %s", self.device_name) - return False - - def turn_on(self) -> bool: - """Turn bypass Purifier on.""" - return self.toggle_switch(True) - - def turn_off(self): - """Turn Bypass Purifier off.""" - return self.toggle_switch(False) - - def set_display(self, mode: bool) -> bool: - """Toggle display on/off.""" - if not isinstance(mode, bool): - logger.debug("Mode must be True or False") - return False - - head, body = self.build_api_dict("setDisplay") - - body["payload"]["data"] = {"state": mode} - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - return True - logger.debug("Error toggling purifier display - %s", self.device_name) - return False - - def turn_on_display(self) -> bool: - """Turn Display on.""" - return self.set_display(True) - - def turn_off_display(self): - """Turn Display off.""" - return self.set_display(False) - - def set_night_light(self, mode: str) -> bool: - """Set night list - on, off or dim.""" - if mode.lower() not in ["on", "off", "dim"]: - logger.debug("Invalid nightlight mode used (on, off or dim)- %s", mode) - return False - head, body = self.build_api_dict("setNightLight") - if not head and not body: - return False - body["payload"]["data"] = {"night_light": mode.lower()} - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - self.details["night_light"] = mode.lower() - return True - logger.debug("Error setting nightlight mode") - return False - - @property - def air_quality(self): - """Get air quality value (ug/m3).""" - if self.air_quality_feature is not True: - logger.debug("%s does not have air quality sensor", self.device_type) - try: - return int(self.details["air_quality"]) - except KeyError: - return 0 - - @property - def fan_level(self): - """Get current fan level (1-3).""" - try: - speed = int(self.speed) - except ValueError: - speed = self.speed - return speed - - @property - def filter_life(self) -> int: - """Get percentage of filter life remaining.""" - try: - return int(self.details["filter_life"]) - except KeyError: - return 0 - - @property - def display_state(self) -> bool: - """Get display state.""" - return bool(self.details["display"]) - - @property - def screen_status(self) -> bool: - """Get display status.""" - return bool(self.details["display"]) - - @property - def child_lock(self) -> bool: - """Get child lock state.""" - return bool(self.details["child_lock"]) - - @property - def night_light(self) -> str: - """Get night light state (on/dim/off).""" - return str(self.details["night_light"]) - - def display(self) -> None: - """Return formatted device info to stdout.""" - super().display() - disp1 = [ - ("Mode: ", self.mode, ""), - ("Filter Life: ", self.details["filter_life"], "percent"), - ("Fan Level: ", self.speed, ""), - ("Display: ", self.details["display"], ""), - ("Child Lock: ", self.details["child_lock"], ""), - ("Night Light: ", self.details["night_light"], ""), - ("Display Config: ", self.config["display"], ""), - ("Display_Forever Config: ", self.config["display_forever"], ""), - ] - if self.air_quality_feature: - disp1.append(("Air Quality: ", self.details["air_quality"], "ug/m3")) - for line in disp1: - print(f"{line[0]:.<20} {line[1]} {line[2]}") - - def displayJSON(self) -> str: - """Return air purifier status and properties in JSON output.""" - sup = super().displayJSON() - sup_val = json.loads(sup) - sup_val.update( - { - "Mode": self.mode, - "Filter Life": str(self.details["filter_life"]), - "Fan Level": str(self.speed), - "Display": self.details["display"], - "Child Lock": self.details["child_lock"], - "Night Light": str(self.details["night_light"]), - "Display Config": self.config["display"], - "Display_Forever Config": self.config["display_forever"], - } - ) - if self.air_quality_feature: - sup_val.update({"Air Quality": str(self.details["air_quality"])}) - return json.dumps(sup_val) - - -class VeSyncAir131(VeSyncBaseDevice): - """Levoit Air Purifier Class.""" - - def __init__(self, details, manager): - """Initialize air purifier class.""" - super().__init__(details, manager) - - self.details: Dict = {} - - def get_details(self) -> None: - """Build Air Purifier details dictionary.""" - body = Helpers.req_body(self.manager, "devicedetail") - body["uuid"] = self.uuid - head = Helpers.req_headers(self.manager) - - r, _ = Helpers.call_api( - "/131airPurifier/v1/device/deviceDetail", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - self.device_status = r.get("deviceStatus", "unknown") - self.connection_status = r.get("connectionStatus", "unknown") - self.details["active_time"] = r.get("activeTime", 0) - self.details["filter_life"] = r.get("filterLife", {}) - self.details["screen_status"] = r.get("screenStatus", "unknown") - self.mode = r.get("mode", self.mode) - self.details["level"] = r.get("level", 0) - self.details["air_quality"] = r.get("airQuality", "unknown") - else: - logger.debug("Error getting %s details", self.device_name) - - def get_config(self) -> None: - """Get configuration info for air purifier.""" - body = Helpers.req_body(self.manager, "devicedetail") - body["method"] = "configurations" - body["uuid"] = self.uuid - - r, _ = Helpers.call_api( - "/131airpurifier/v1/device/configurations", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if r is not None and Helpers.code_check(r): - self.config = Helpers.build_config_dict(r) - else: - logger.debug("Unable to get config info for %s", self.device_name) - - @property - def active_time(self) -> int: - """Return total time active in minutes.""" - return self.details.get("active_time", 0) - - @property - def fan_level(self) -> int: - """Get current fan level (1-3).""" - return self.details.get("level", 0) - - @property - def filter_life(self) -> int: - """Get percentage of filter life remaining.""" - try: - return self.details["filter_life"].get("percent", 0) - except KeyError: - return 0 - - @property - def air_quality(self) -> str: - """Get Air Quality.""" - return self.details.get("air_quality", "unknown") - - @property - def screen_status(self) -> str: - """Return Screen status (on/off).""" - return self.details.get("screen_status", "unknown") - - def turn_on(self) -> bool: - """Turn Air Purifier on.""" - if self.device_status == "on": - return False - body = Helpers.req_body(self.manager, "devicestatus") - body["uuid"] = self.uuid - body["status"] = "on" - head = Helpers.req_headers(self.manager) - - r, _ = Helpers.call_api( - "/131airPurifier/v1/device/deviceStatus", "put", json=body, headers=head - ) - - if r is not None and Helpers.code_check(r): - self.device_status = "on" - return True - logger.debug("Error turning %s on", self.device_name) - return False - - def turn_off(self) -> bool: - """Turn Air Purifier Off.""" - if self.device_status != "on": - return True - body = Helpers.req_body(self.manager, "devicestatus") - body["uuid"] = self.uuid - body["status"] = "off" - head = Helpers.req_headers(self.manager) - - r, _ = Helpers.call_api( - "/131airPurifier/v1/device/deviceStatus", "put", json=body, headers=head - ) - - if r is not None and Helpers.code_check(r): - self.device_status = "off" - return True - logger.debug("Error turning %s off", self.device_name) - return False - - def auto_mode(self) -> bool: - """Set mode to auto.""" - return self.mode_toggle("auto") - - def manual_mode(self) -> bool: - """Set mode to manual.""" - return self.mode_toggle("manual") - - def sleep_mode(self) -> bool: - """Set sleep mode to on.""" - return self.mode_toggle("sleep") - - def change_fan_speed(self, speed: int = None) -> bool: - """Adjust Fan Speed for air purifier. - - Specifying 1,2,3 as argument or call without argument to cycle - through speeds increasing by one. - """ - if self.mode != "manual": - logger.debug("%s not in manual mode, cannot change speed", self.device_name) - return False - - try: - level = self.details["level"] - except KeyError: - logger.debug( - "Cannot change fan speed, no level set for %s", self.device_name - ) - return False - - body = Helpers.req_body(self.manager, "devicestatus") - body["uuid"] = self.uuid - head = Helpers.req_headers(self.manager) - if speed is not None: - if speed == level: - return True - if speed in {1, 2, 3}: - body["level"] = speed - else: - logger.debug("Invalid fan speed for %s", self.device_name) - return False - else: - body["level"] = 1 if level > 2 else int(level + 1) - r, _ = Helpers.call_api( - "/131airPurifier/v1/device/updateSpeed", "put", json=body, headers=head - ) - - if r is not None and Helpers.code_check(r): - self.details["level"] = body["level"] - return True - logger.debug("Error changing %s speed", self.device_name) - return False - - def mode_toggle(self, mode: str) -> bool: - """Set mode to manual, auto or sleep.""" - head = Helpers.req_headers(self.manager) - body = Helpers.req_body(self.manager, "devicestatus") - body["uuid"] = self.uuid - if mode != self.mode and mode in {"sleep", "auto", "manual"}: - body["mode"] = mode - if mode == "manual": - body["level"] = 1 - - r, _ = Helpers.call_api( - "/131airPurifier/v1/device/updateMode", "put", json=body, headers=head - ) - - if r is not None and Helpers.code_check(r): - self.mode = mode - return True - - logger.debug("Error setting %s mode - %s", self.device_name, mode) - return False - - def update(self) -> None: - """Run function to get device details.""" - self.get_details() - - def display(self) -> None: - """Return formatted device info to stdout.""" - super().display() - disp1 = [ - ("Active Time : ", self.active_time, " minutes"), - ("Fan Level: ", self.fan_level, ""), - ("Air Quality: ", self.air_quality, ""), - ("Mode: ", self.mode, ""), - ("Screen Status: ", self.screen_status, ""), - ("Filter Life: ", self.filter_life, " percent"), - ] - for line in disp1: - print(f"{line[0]:.<15} {line[1]} {line[2]}") - - def displayJSON(self) -> str: - """Return air purifier status and properties in JSON output.""" - sup = super().displayJSON() - sup_val = json.loads(sup) - sup_val.update( - { - "Active Time": str(self.active_time), - "Fan Level": self.fan_level, - "Air Quality": self.air_quality, - "Mode": self.mode, - "Screen Status": self.screen_status, - "Filter Life": str(self.filter_life), - } - ) - return sup_val - - -class VeSyncHumid200300S(VeSyncBaseDevice): - """200S/300S Humidifier Class.""" - - def __init__(self, details, manager): - """Initialize 200S/300S Humidifier class.""" - super().__init__(details, manager) - self.enabled = True - self.config_dict = model_features(self.device_type) - self.mist_levels = self.config_dict.get("mist_levels") - self.mist_modes = self.config_dict.get("mist_modes") - self.features = self.config_dict.get("features") - if "warm_mist" in self.features: - self.warm_mist_levels = self.config_dict.get("warm_mist_levels", []) - self.warm_mist_feature = True - else: - self.warm_mist_feature = False - self.warm_mist_levels = [] - self.night_light = "nightlight" in self.config_dict.get("features", []) - self.details: Dict[str, Union[str, int, float]] = { - "humidity": 0, - "mist_virtual_level": 0, - "mist_level": 0, - "mode": "manual", - "water_lacks": False, - "humidity_high": False, - "water_tank_lifted": False, - "display": False, - "automatic_stop_reach_target": False, - } - if self.night_light: - self.details["night_light_brightness"] = 0 - self.config: Dict[str, Union[str, int, float]] = { - "auto_target_humidity": 0, - "display": False, - "automatic_stop": True, - } - self._api_modes = [ - "getHumidifierStatus", - "setAutomaticStop", - "setSwitch", - "setNightLightBrightness", - "setVirtualLevel", - "setTargetHumidity", - "setHumidityMode", - "setDisplay", - "setLevel", - ] - - def build_api_dict(self, method: str) -> Tuple[Dict, Dict]: - """Build humidifier api call header and body. - - Available methods are: 'getHumidifierStatus', 'setAutomaticStop', - 'setSwitch', 'setNightLightBrightness', 'setVirtualLevel', - 'setTargetHumidity', 'setHumidityMode' - """ - if method not in self._api_modes: - logger.debug("Invalid mode - %s", method) - raise ValueError - head = Helpers.bypass_header() - body = Helpers.bypass_body_v2(self.manager) - body["cid"] = self.cid - body["configModule"] = self.config_module - body["payload"] = {"method": method, "source": "APP"} - return head, body - - def build_humid_dict(self, dev_dict: Dict[str, str]) -> None: - """Build humidifier status dictionary.""" - self.enabled = dev_dict.get("enabled") - self.details["humidity"] = dev_dict.get("humidity", 0) - self.details["mist_virtual_level"] = dev_dict.get("mist_virtual_level", 0) - self.details["mist_level"] = dev_dict.get("mist_level", 0) - self.details["mode"] = dev_dict.get("mode", "manual") - self.details["water_lacks"] = dev_dict.get("water_lacks", False) - self.details["humidity_high"] = dev_dict.get("humidity_high", False) - self.details["water_tank_lifted"] = dev_dict.get("water_tank_lifted", False) - self.details["automatic_stop_reach_target"] = dev_dict.get( - "automatic_stop_reach_target", True - ) - if self.night_light: - self.details["night_light_brightness"] = dev_dict.get( - "night_light_brightness", 0 - ) - if self.warm_mist_feature: - self.details["warm_mist_level"] = dev_dict.get("warm_level", 0) - self.details["warm_mist_enabled"] = dev_dict.get("warm_enabled", False) - try: - self.details["display"] = dev_dict["display"] - except KeyError: - self.details["display"] = dev_dict.get("indicator_light_switch", False) - - def build_config_dict(self, conf_dict): - """Build configuration dict for 300s humidifier.""" - self.config["auto_target_humidity"] = conf_dict.get("auto_target_humidity", 0) - self.config["display"] = conf_dict.get("display", False) - self.config["automatic_stop"] = conf_dict.get("automatic_stop", True) - - def get_details(self) -> None: - """Build 200S/300S Humidifier details dictionary.""" - head = Helpers.bypass_header() - body = Helpers.bypass_body_v2(self.manager) - body["cid"] = self.cid - body["configModule"] = self.config_module - body["payload"] = {"method": "getHumidifierStatus", "source": "APP", "data": {}} - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - if r is None or not isinstance(r, dict): - logger.debug("Error getting status of %s ", self.device_name) - outer_result = None - else: - outer_result = r.get("result", {}) - inner_result = None - if outer_result is not None: - inner_result = r.get("result", {}).get("result") - if inner_result is not None and Helpers.code_check(r): - if outer_result.get("code") == 0: - self.build_humid_dict(inner_result) - else: - logger.debug("error in inner result dict from humidifier") - if inner_result.get("configuration", {}): - self.build_config_dict(inner_result.get("configuration", {})) - else: - logger.debug("No configuration found in humidifier status") - else: - logger.debug("Error in humidifier response") - - def update(self): - """Update 200S/300S Humidifier details.""" - self.get_details() - - def toggle_switch(self, toggle: bool) -> bool: - """Toggle humidifier on/off.""" - if not isinstance(toggle, bool): - logger.debug("Invalid toggle value for humidifier switch") - return False - - head = Helpers.bypass_header() - body = Helpers.bypass_body_v2(self.manager) - body["cid"] = self.cid - body["configModule"] = self.config_module - body["payload"] = { - "data": {"enabled": toggle, "id": 0}, - "method": "setSwitch", - "source": "APP", - } - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - self.device_status = "on" if toggle else "off" - return True - logger.debug("Error toggling 300S humidifier - %s", self.device_name) - return False - - def turn_on(self) -> bool: - """Turn 200S/300S Humidifier on.""" - return self.toggle_switch(True) - - def turn_off(self): - """Turn 200S/300S Humidifier off.""" - return self.toggle_switch(False) - - def automatic_stop_on(self) -> bool: - """Turn 200S/300S Humidifier automatic stop on.""" - return self.set_automatic_stop(True) - - def automatic_stop_off(self) -> bool: - """Turn 200S/300S Humidifier automatic stop on.""" - return self.set_automatic_stop(False) - - def set_automatic_stop(self, mode: bool) -> bool: - """Set 200S/300S Humidifier to automatic stop.""" - if mode not in (True, False): - logger.debug("Invalid mode passed to set_automatic_stop - %s", mode) - return False - - head, body = self.build_api_dict("setAutomaticStop") - if not head and not body: - return False - - body["payload"]["data"] = {"enabled": mode} - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - return True - if isinstance(r, dict): - logger.debug("Error toggling automatic stop") - else: - logger.debug("Error in api return json for %s", self.device_name) - return False - - def set_display(self, mode: bool) -> bool: - """Toggle display on/off.""" - if not isinstance(mode, bool): - logger.debug("Mode must be True or False") - return False - - head, body = self.build_api_dict("setDisplay") - - body["payload"]["data"] = {"state": mode} - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - return True - logger.debug("Error toggling 300S display - %s", self.device_name) - return False - - def turn_on_display(self) -> bool: - """Turn 200S/300S Humidifier on.""" - return self.set_display(True) - - def turn_off_display(self): - """Turn 200S/300S Humidifier off.""" - return self.set_display(False) - - def set_humidity(self, humidity: int) -> bool: - """Set target 200S/300S Humidifier humidity.""" - if humidity < 30 or humidity > 80: - logger.debug("Humidity value must be set between 30 and 80") - return False - head, body = self.build_api_dict("setTargetHumidity") - - if not head and not body: - return False - - body["payload"]["data"] = {"target_humidity": humidity} - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - return True - logger.debug("Error setting humidity") - return False - - def set_night_light_brightness(self, brightness: int) -> bool: - """Set target 200S/300S Humidifier night light brightness.""" - if not self.night_light: - logger.debug( - "%s is a %s does not have a nightlight", - self.device_name, - self.device_type, - ) - return False - if brightness < 0 or brightness > 100: - logger.debug("Brightness value must be set between 0 and 100") - return False - head, body = self.build_api_dict("setNightLightBrightness") - - if not head and not body: - return False - - body["payload"]["data"] = {"night_light_brightness": brightness} - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - return True - logger.debug("Error setting humidity") - return False - - def set_humidity_mode(self, mode: str) -> bool: - """Set humidifier mode - sleep or auto.""" - if mode.lower() not in self.mist_modes: - logger.debug("Invalid humidity mode used - %s", mode) - logger.debug("Proper modes for this device are - %s", str(self.mist_modes)) - return False - head, body = self.build_api_dict("setHumidityMode") - if not head and not body: - return False - body["payload"]["data"] = {"mode": mode.lower()} - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - return True - logger.debug("Error setting humidity mode") - return False - - def set_warm_level(self, warm_level) -> bool: - """Set target 600S Humidifier mist warmth.""" - if not self.warm_mist_feature: - logger.debug( - "%s is a %s does not have a mist warmer", - self.device_name, - self.device_type, - ) - return False - if not isinstance(warm_level, int): - try: - warm_level = int(warm_level) - except ValueError: - logger.debug("Error converting warm mist level to a integer") - if warm_level not in self.warm_mist_levels: - logger.debug("warm_level value must be - %s", str(self.warm_mist_levels)) - return False - head, body = self.build_api_dict("setLevel") - - if not head and not body: - return False - - body["payload"]["data"] = { - "type": "warm", - "level": warm_level, - "id": 0, - } - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - return True - logger.debug("Error setting warm") - return False - - def set_auto_mode(self): - """Set auto mode for humidifiers.""" - if "auto" in self.mist_modes: - call_str = "auto" - elif "humidity" in self.mist_modes: - call_str = "humidity" - else: - logger.debug( - "Trying auto mode, mode not set for this model, " - "please ensure %s model " - "is in configuration dictionary", - self.device_type, - ) - call_str = "auto" - return self.set_humidity_mode(call_str) - - def set_manual_mode(self): - """Set humifier to manual mode with 1 mist level.""" - return self.set_humidity_mode("manual") - - def set_mist_level(self, level) -> bool: - """Set humidifier mist level with int between 0 - 9.""" - try: - level = int(level) - except ValueError: - level = str(level) - if level not in self.mist_levels: - logger.debug("Humidifier mist level must be between 0 and 9") - return False - - head, body = self.build_api_dict("setVirtualLevel") - if not head and not body: - return False - - body["payload"]["data"] = {"id": 0, "level": level, "type": "mist"} - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - return True - logger.debug("Error setting mist level") - return False - - @property - def humidity(self): - """Get Humidity level.""" - return self.details["humidity"] - - @property - def mist_level(self): - """Get current mist level.""" - return self.details["virtual_mist_level"] - - @property - def water_lacks(self): - """If tank is empty return true.""" - return self.details["water_lacks"] - - @property - def auto_humidity(self): - """Auto target humidity.""" - return self.config["auto_target_humidity"] - - @property - def auto_enabled(self): - """Auto mode is enabled.""" - return self.details.get("mode") in ["auto", "humidity"] - - @property - def warm_mist_enabled(self): - """Warm mist feature enabled.""" - return self.details["warm_mist_enabled"] if self.warm_mist_feature else False - - def display(self) -> None: - """Return formatted device info to stdout.""" - super().display() - disp1 = [ - ("Mode: ", self.details["mode"], ""), - ("Humidity: ", self.details["humidity"], "percent"), - ("Mist Virtual Level: ", self.details["mist_virtual_level"], ""), - ("Mist Level: ", self.details["mist_level"], ""), - ("Water Lacks: ", self.details["water_lacks"], ""), - ("Humidity High: ", self.details["humidity_high"], ""), - ("Water Tank Lifted: ", self.details["water_tank_lifted"], ""), - ("Display: ", self.details["display"], ""), - ( - "Automatic Stop Reach Target: ", - self.details["automatic_stop_reach_target"], - "", - ), - ("Auto Target Humidity: ", self.config["auto_target_humidity"], "percent"), - ("Automatic Stop: ", self.config["automatic_stop"], ""), - ] - if self.night_light: - disp1.append( - ( - "Night Light Brightness: ", - self.details["night_light_brightness"], - "percent", - ) - ) - if self.warm_mist_feature: - disp1.append(("Warm mist enabled: ", self.details["warm_mist_enabled"], "")) - disp1.append(("Warm mist level: ", self.details["warm_mist_level"], "")) - for line in disp1: - print(f"{line[0]:.<29} {line[1]} {line[2]}") - - def displayJSON(self) -> str: - """Return air purifier status and properties in JSON output.""" - sup = super().displayJSON() - sup_val = json.loads(sup) - sup_val.update( - { - "Mode": self.details["mode"], - "Humidity": str(self.details["humidity"]), - "Mist Virtual Level": str(self.details["mist_virtual_level"]), - "Mist Level": str(self.details["mist_level"]), - "Water Lacks": self.details["water_lacks"], - "Humidity High": self.details["humidity_high"], - "Water Tank Lifted": self.details["water_tank_lifted"], - "Display": self.details["display"], - "Automatic Stop Reach Target": self.details[ - "automatic_stop_reach_target" - ], - "Auto Target Humidity": str(self.config["auto_target_humidity"]), - "Automatic Stop": self.config["automatic_stop"], - } - ) - if self.night_light: - sup_val["Night Light Brightness"] = self.details["night_light_brightness"] - if self.warm_mist_feature: - sup_val["Warm mist enabled"] = self.details["warm_mist_enabled"] - sup_val["Warm mist level"] = self.details["warm_mist_level"] - return json.dumps(sup_val) - - -class VeSyncHumid200S(VeSyncHumid200300S): - """Levoit Classic 200S Specific class.""" - - def __init__(self, details, manager): - """Initialize levoit 200S device class.""" - super().__init__(details, manager) - self._api_modes = [ - "getHumidifierStatus", - "setAutomaticStop", - "setSwitch", - "setVirtualLevel", - "setTargetHumidity", - "setHumidityMode", - "setIndicatorLightSwitch", - ] - - def set_display(self, mode: bool) -> bool: - """Toggle display on/off.""" - if not isinstance(mode, bool): - logger.debug("Mode must be True or False") - return False - - head, body = self.build_api_dict("setIndicatorLightSwitch") - - body["payload"]["data"] = {"enabled": mode, "id": 0} - - r, _ = Helpers.call_api( - "/cloud/v2/deviceManaged/bypassV2", - method="post", - headers=head, - json=body, - ) - - if r is not None and Helpers.code_check(r): - return True - logger.debug("Error toggling 300S display - %s", self.device_name) - return False diff --git a/custom_components/vesync/pyvesync/vesyncoutlet.py b/custom_components/vesync/pyvesync/vesyncoutlet.py deleted file mode 100644 index d9f45f8..0000000 --- a/custom_components/vesync/pyvesync/vesyncoutlet.py +++ /dev/null @@ -1,703 +0,0 @@ -"""Etekcity Outlets.""" - -import json -import logging -import time -from abc import ABCMeta, abstractmethod - -from .helpers import Helpers -from .vesyncbasedevice import VeSyncBaseDevice - -logger = logging.getLogger(__name__) - -outlet_config = { - "wifi-switch-1.3": {"module": "VeSyncOutlet7A"}, - "ESW03-USA": {"module": "VeSyncOutlet10A"}, - "ESW01-EU": {"module": "VeSyncOutlet10A"}, - "ESW15-USA": {"module": "VeSyncOutlet15A"}, - "ESO15-TB": {"module": "VeSyncOutdoorPlug"}, -} - -outlet_modules = {k: v["module"] for k, v in outlet_config.items()} - -__all__ = list(outlet_modules.values()) + ["outlet_modules"] - - -class VeSyncOutlet(VeSyncBaseDevice): - """Base class for Etekcity Outlets.""" - - __metaclass__ = ABCMeta - - def __init__(self, details, manager): - """Initialize VeSync Outlet base class.""" - super().__init__(details, manager) - - self.details = {} - self.energy = {} - self.update_energy_ts = None - self._energy_update_interval = manager.energy_update_interval - - @property - def update_time_check(self) -> bool: - """Test if energy update interval has been exceeded.""" - if self.update_energy_ts is None: - return True - - if (time.time() - self.update_energy_ts) > self._energy_update_interval: - return True - return False - - @abstractmethod - def turn_on(self) -> bool: - """Return True if device has beeeen turned on.""" - - @abstractmethod - def turn_off(self) -> bool: - """Return True if device has beeeen turned off.""" - - @abstractmethod - def get_details(self) -> None: - """Build details dictionary.""" - - @abstractmethod - def get_weekly_energy(self) -> None: - """Build weekly energy history dictionary.""" - - @abstractmethod - def get_monthly_energy(self) -> None: - """Build Monthly Energy History Dictionary.""" - - @abstractmethod - def get_yearly_energy(self): - """Build Yearly Energy Dictionary.""" - - @abstractmethod - def get_config(self): - """Get configuration and firmware details.""" - - def update(self): - """Get Device Energy and Status.""" - self.get_details() - - def update_energy(self, bypass_check: bool = False): - """Build weekly, monthly and yearly dictionaries.""" - if bypass_check or self.update_time_check: - self.update_energy_ts = time.time() - self.get_weekly_energy() - if "week" in self.energy: - self.get_monthly_energy() - self.get_yearly_energy() - if not bypass_check: - self.update_energy_ts = time.time() - - @property - def active_time(self) -> int: - """Return active time of a device in minutes.""" - return self.details.get("active_time", 0) - - @property - def energy_today(self) -> float: - """Return energy.""" - return self.details.get("energy", 0) - - @property - def power(self) -> float: - """Return current power in watts.""" - return float(self.details.get("power", 0)) - - @property - def voltage(self) -> float: - """Return current voltage.""" - return float(self.details.get("voltage", 0)) - - @property - def monthly_energy_total(self) -> float: - """Return total energy usage over the month.""" - return self.energy.get("month", {}).get("total_energy", 0) - - @property - def weekly_energy_total(self) -> float: - """Return total energy usage over the week.""" - return self.energy.get("week", {}).get("total_energy", 0) - - @property - def yearly_energy_total(self) -> float: - """Return total energy usage over the year.""" - return self.energy.get("year", {}).get("total_energy", 0) - - def display(self): - """Return formatted device info to stdout.""" - super().display() - disp1 = [ - ("Active Time : ", self.active_time, " minutes"), - ("Energy: ", self.energy_today, " kWh"), - ("Power: ", self.power, " Watts"), - ("Voltage: ", self.voltage, " Volts"), - ("Energy Week: ", self.weekly_energy_total, " kWh"), - ("Energy Month: ", self.monthly_energy_total, " kWh"), - ("Energy Year: ", self.yearly_energy_total, " kWh"), - ] - for line in disp1: - print(f"{line[0]:.<15} {line[1]} {line[2]}") - - def displayJSON(self): - """Return JSON details for outlet.""" - sup = super().displayJSON() - sup_val = json.loads(sup) - sup_val.update( - { - "Active Time": str(self.active_time), - "Energy": str(self.energy_today), - "Power": str(self.power), - "Voltage": str(self.voltage), - "Energy Week": str(self.weekly_energy_total), - "Energy Month": str(self.monthly_energy_total), - "Energy Year": str(self.yearly_energy_total), - } - ) - - return sup_val - - -class VeSyncOutlet7A(VeSyncOutlet): - """Etekcity 7A Round Outlet Class.""" - - def __init__(self, details, manager): - """Initialize Etekcity 7A round outlet class.""" - super().__init__(details, manager) - self.det_keys = ["deviceStatus", "activeTime", "energy", "power", "voltage"] - self.energy_keys = ["energyConsumptionOfToday", "maxEnergy", "totalEnergy"] - - def get_details(self) -> None: - """Get 7A outlet details.""" - r, _ = Helpers.call_api( - f"/v1/device/{self.cid}/detail", - "get", - headers=Helpers.req_headers(self.manager), - ) - - if r is not None and all(x in r for x in self.det_keys): - self.device_status = r.get("deviceStatus", self.device_status) - self.details["active_time"] = r.get("activeTime", 0) - self.details["energy"] = r.get("energy", 0) - power = r.get("power", "0:0") - power = round(float(Helpers.calculate_hex(power)), 2) - self.details["power"] = power - voltage = r.get("voltage", "0:0") - voltage = round(float(Helpers.calculate_hex(voltage)), 2) - self.details["voltage"] = voltage - else: - logger.debug("Unable to get %s details", self.device_name) - - def get_weekly_energy(self) -> None: - """Get 7A outlet weekly energy info and build weekly energy dict.""" - r, _ = Helpers.call_api( - f"/v1/device/{self.cid}/energy/week", - "get", - headers=Helpers.req_headers(self.manager), - ) - - if r is not None and all(x in r for x in self.energy_keys): - self.energy["week"] = Helpers.build_energy_dict(r) - else: - logger.debug("Unable to get %s weekly data", self.device_name) - - def get_monthly_energy(self) -> None: - """Get 7A outlet monthly energy info and build monthly energy dict.""" - r, _ = Helpers.call_api( - f"/v1/device/{self.cid}/energy/month", - "get", - headers=Helpers.req_headers(self.manager), - ) - - if r is not None and all(x in r for x in self.energy_keys): - self.energy["month"] = Helpers.build_energy_dict(r) - else: - logger.warning("Unable to get %s monthly data", self.device_name) - - def get_yearly_energy(self) -> None: - """Get 7A outlet yearly energy info and build yearly energy dict.""" - r, _ = Helpers.call_api( - f"/v1/device/{self.cid}/energy/year", - "get", - headers=Helpers.req_headers(self.manager), - ) - - if r is not None and all(x in r for x in self.energy_keys): - self.energy["year"] = Helpers.build_energy_dict(r) - else: - logger.debug("Unable to get %s yearly data", self.device_name) - - def turn_on(self) -> bool: - """Turn 7A outlet on - return True if successful.""" - return self._toggle("on") - - def turn_off(self) -> bool: - """Turn 7A outlet off - return True if successful.""" - return self._toggle("off") - - def _toggle(self, state): - _, status_code = Helpers.call_api( - f"/v1/wifi-switch-1.3/{self.cid}/status/{state}", - "put", - headers=Helpers.req_headers(self.manager), - ) - - if status_code is not None and status_code == 200: - self.device_status = state - return True - logger.warning("Error turning %s %s", self.device_name, state) - return False - - def get_config(self) -> None: - """Get 7A outlet configuration info.""" - r, _ = Helpers.call_api( - f"/v1/device/{self.cid}/configurations", - "get", - headers=Helpers.req_headers(self.manager), - ) - - if "currentFirmVersion" in r: - self.config = Helpers.build_config_dict(r) - else: - logger.debug("Error getting configuration info for %s", self.device_name) - - -class VeSyncOutlet10A(VeSyncOutlet): - """Etekcity 10A Round Outlets.""" - - def __init__(self, details, manager): - """Initialize 10A outlet class.""" - super().__init__(details, manager) - - def get_details(self) -> None: - """Get 10A outlet details.""" - body = Helpers.req_body(self.manager, "devicedetail") - body["uuid"] = self.uuid - - r, _ = Helpers.call_api( - "/10a/v1/device/devicedetail", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(r): - self.device_status = r.get("deviceStatus", self.device_status) - self.connection_status = r.get("connectionStatus", self.connection_status) - self.details = Helpers.build_details_dict(r) - else: - logger.debug("Unable to get %s details", self.device_name) - - def get_config(self) -> None: - """Get 10A outlet configuration info.""" - body = Helpers.req_body(self.manager, "devicedetail") - body["method"] = "configurations" - body["uuid"] = self.uuid - - r, _ = Helpers.call_api( - "/10a/v1/device/configurations", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(r): - self.config = Helpers.build_config_dict(r) - else: - logger.debug("Error getting %s config info", self.device_name) - - def get_weekly_energy(self) -> None: - """Get 10A outlet weekly energy info and populate energy dict.""" - body = Helpers.req_body(self.manager, "energy_week") - body["uuid"] = self.uuid - - response, _ = Helpers.call_api( - "/10a/v1/device/energyweek", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - self.energy["week"] = Helpers.build_energy_dict(response) - else: - logger.debug("Unable to get %s weekly data", self.device_name) - - def get_monthly_energy(self) -> None: - """Get 10A outlet monthly energy info and populate energy dict.""" - body = Helpers.req_body(self.manager, "energy_month") - body["uuid"] = self.uuid - - response, _ = Helpers.call_api( - "/10a/v1/device/energymonth", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - self.energy["month"] = Helpers.build_energy_dict(response) - else: - logger.debug("Unable to get %s monthly data", self.device_name) - - def get_yearly_energy(self) -> None: - """Get 10A outlet yearly energy info and populate energy dict.""" - body = Helpers.req_body(self.manager, "energy_year") - body["uuid"] = self.uuid - - response, _ = Helpers.call_api( - "/10a/v1/device/energyyear", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - self.energy["year"] = Helpers.build_energy_dict(response) - else: - logger.debug("Unable to get %s yearly data", self.device_name) - - def turn_on(self) -> bool: - """Turn 10A outlet on - return True if successful.""" - body = Helpers.req_body(self.manager, "devicestatus") - body["uuid"] = self.uuid - body["status"] = "on" - - response, _ = Helpers.call_api( - "/10a/v1/device/devicestatus", - "put", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - self.device_status = "on" - return True - logger.warning("Error turning %s on", self.device_name) - return False - - def turn_off(self) -> bool: - """Turn 10A outlet off - return True if successful.""" - body = Helpers.req_body(self.manager, "devicestatus") - body["uuid"] = self.uuid - body["status"] = "off" - - response, _ = Helpers.call_api( - "/10a/v1/device/devicestatus", - "put", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - self.device_status = "off" - return True - logger.warning("Error turning %s off", self.device_name) - return False - - -class VeSyncOutlet15A(VeSyncOutlet): - """Class for Etekcity 15A Rectangular Outlets.""" - - def __init__(self, details, manager): - """Initialize 15A rectangular outlets.""" - super().__init__(details, manager) - self.nightlight_status = "off" - self.nightlight_brightness = 0 - - def get_details(self) -> None: - """Get 15A outlet details.""" - body = Helpers.req_body(self.manager, "devicedetail") - body["uuid"] = self.uuid - - r, _ = Helpers.call_api( - "/15a/v1/device/devicedetail", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - attr_list = ( - "deviceStatus", - "activeTime", - "energy", - "power", - "voltage", - "nightLightStatus", - "nightLightAutomode", - "nightLightBrightness", - ) - - if Helpers.code_check(r) and all(k in r for k in attr_list): - - self.device_status = r.get("deviceStatus") - self.connection_status = r.get("connectionStatus") - self.nightlight_status = r.get("nightLightStatus") - self.nightlight_brightness = r.get("nightLightBrightness") - self.details = Helpers.build_details_dict(r) - else: - logger.debug("Unable to get %s details", self.device_name) - - def get_config(self) -> None: - """Get 15A outlet configuration info.""" - body = Helpers.req_body(self.manager, "devicedetail") - body["method"] = "configurations" - body["uuid"] = self.uuid - - r, _ = Helpers.call_api( - "/15a/v1/device/configurations", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(r): - self.config = Helpers.build_config_dict(r) - else: - logger.debug("Unable to get %s config info", self.device_name) - - def get_weekly_energy(self) -> None: - """Get 15A outlet weekly energy info and populate energy dict.""" - body = Helpers.req_body(self.manager, "energy_week") - body["uuid"] = self.uuid - - response, _ = Helpers.call_api( - "/15a/v1/device/energyweek", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - self.energy["week"] = Helpers.build_energy_dict(response) - else: - logger.debug("Unable to get %s weekly data", self.device_name) - - def get_monthly_energy(self) -> None: - """Get 15A outlet monthly energy info and populate energy dict.""" - body = Helpers.req_body(self.manager, "energy_month") - body["uuid"] = self.uuid - - response, _ = Helpers.call_api( - "/15a/v1/device/energymonth", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - self.energy["month"] = Helpers.build_energy_dict(response) - else: - logger.debug("Unable to get %s monthly data", self.device_name) - - def get_yearly_energy(self) -> None: - """Get 15A outlet yearly energy info and populate energy dict.""" - body = Helpers.req_body(self.manager, "energy_year") - body["uuid"] = self.uuid - - response, _ = Helpers.call_api( - "/15a/v1/device/energyyear", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - self.energy["year"] = Helpers.build_energy_dict(response) - else: - logger.debug("Unable to get %s yearly data", self.device_name) - - def turn_on(self) -> bool: - """Turn 15A outlet on - return True if successful.""" - body = Helpers.req_body(self.manager, "devicestatus") - body["uuid"] = self.uuid - body["status"] = "on" - - response, _ = Helpers.call_api( - "/15a/v1/device/devicestatus", - "put", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - self.device_status = "on" - return True - logger.warning("Error turning %s on", self.device_name) - return False - - def turn_off(self) -> bool: - """Turn 15A outlet off - return True if successful.""" - body = Helpers.req_body(self.manager, "devicestatus") - body["uuid"] = self.uuid - body["status"] = "off" - - response, _ = Helpers.call_api( - "/15a/v1/device/devicestatus", - "put", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - self.device_status = "off" - return True - - logger.warning("Error turning %s off", self.device_name) - return False - - def turn_on_nightlight(self) -> bool: - """Turn on nightlight.""" - return self._extracted_from_turn_off_nightlight_3("auto") - - def turn_off_nightlight(self) -> bool: - """Turn Off Nightlight.""" - return self._extracted_from_turn_off_nightlight_3("manual") - - def _extracted_from_turn_off_nightlight_3(self, mode): - body = Helpers.req_body(self.manager, "devicestatus") - body["uuid"] = self.uuid - body["mode"] = mode - response, _ = Helpers.call_api( - "/15a/v1/device/nightlightstatus", - "put", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - return True - logger.debug( - "Error turning %s %s nightlight", - "on" if mode == "auto" else "off", - self.device_name, - ) - return False - - -class VeSyncOutdoorPlug(VeSyncOutlet): - """Class to hold Etekcity outdoor outlets.""" - - def __init__(self, details, manager): - """Initialize Etekcity Outdoor Plug class.""" - super().__init__(details, manager) - - def get_details(self) -> None: - """Get details for outdoor outlet.""" - body = Helpers.req_body(self.manager, "devicedetail") - body["uuid"] = self.uuid - r, _ = Helpers.call_api( - "/outdoorsocket15a/v1/device/devicedetail", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(r): - self.details = Helpers.build_details_dict(r) - self.connection_status = r.get("connectionStatus") - - dev_no = self.sub_device_no - sub_device_list = r.get("subDevices") - if sub_device_list and dev_no <= len(sub_device_list): - self.device_status = sub_device_list[(dev_no + -1)].get( - "subDeviceStatus" - ) - return - logger.debug("Unable to get %s details", self.device_name) - - def get_config(self) -> None: - """Get configuration info for outdoor outlet.""" - body = Helpers.req_body(self.manager, "devicedetail") - body["method"] = "configurations" - body["uuid"] = self.uuid - - r, _ = Helpers.call_api( - "/outdoorsocket15a/v1/device/configurations", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(r): - self.config = Helpers.build_config_dict(r) - logger.debug("Error getting %s config info", self.device_name) - - def get_weekly_energy(self) -> None: - """Get outdoor outlet weekly energy info and populate energy dict.""" - body = Helpers.req_body(self.manager, "energy_week") - body["uuid"] = self.uuid - - response, _ = Helpers.call_api( - "/outdoorsocket15a/v1/device/energyweek", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - self.energy["week"] = Helpers.build_energy_dict(response) - else: - logger.debug("Unable to get %s weekly data", self.device_name) - - def get_monthly_energy(self) -> None: - """Get outdoor outlet monthly energy info and populate energy dict.""" - body = Helpers.req_body(self.manager, "energy_month") - body["uuid"] = self.uuid - - response, _ = Helpers.call_api( - "/outdoorsocket15a/v1/device/energymonth", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - self.energy["month"] = Helpers.build_energy_dict(response) - logger.debug("Unable to get %s monthly data", self.device_name) - - def get_yearly_energy(self) -> None: - """Get outdoor outlet yearly energy info and populate energy dict.""" - body = Helpers.req_body(self.manager, "energy_year") - body["uuid"] = self.uuid - - response, _ = Helpers.call_api( - "/outdoorsocket15a/v1/device/energyyear", - "post", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - self.energy["year"] = Helpers.build_energy_dict(response) - else: - logger.debug("Unable to get %s yearly data", self.device_name) - - def toggle(self, status) -> bool: - """Toggle power for outdoor outlet.""" - body = Helpers.req_body(self.manager, "devicestatus") - body["uuid"] = self.uuid - body["status"] = status - body["switchNo"] = self.sub_device_no - - response, _ = Helpers.call_api( - "/outdoorsocket15a/v1/device/devicestatus", - "put", - headers=Helpers.req_headers(self.manager), - json=body, - ) - - if Helpers.code_check(response): - self.device_status = status - return True - logger.warning("Error turning %s %s", self.device_name, status) - return False - - def turn_on(self) -> bool: - """Turn outdoor outlet on and return True if successful.""" - return bool(self.toggle("on")) - - def turn_off(self) -> bool: - """Turn outdoor outlet off and return True if successful.""" - return bool(self.toggle("off")) diff --git a/custom_components/vesync/pyvesync/vesyncswitch.py b/custom_components/vesync/pyvesync/vesyncswitch.py deleted file mode 100644 index 5cca257..0000000 --- a/custom_components/vesync/pyvesync/vesyncswitch.py +++ /dev/null @@ -1,347 +0,0 @@ -"""Classes for VeSync Switch Devices.""" - -import json -import logging -from abc import ABCMeta, abstractmethod -from typing import Dict, Union - -from .helpers import Helpers as helpers -from .vesyncbasedevice import VeSyncBaseDevice - -logger = logging.getLogger(__name__) - -feature_dict: Dict[str, Dict[str, Union[list, str]]] = { - "ESWL01": {"module": "VeSyncWallSwitch", "features": []}, - "ESWD16": {"module": "VeSyncDimmerSwitch", "features": ["dimmable"]}, - "ESWL03": {"module": "VeSyncWallSwitch", "features": []}, -} - -switch_modules: dict = {k: v["module"] for k, v in feature_dict.items()} - -__all__: list = list(switch_modules.values()) + ["switch_modules"] - - -class VeSyncSwitch(VeSyncBaseDevice): - """Etekcity Switch Base Class.""" - - __metaclasss__ = ABCMeta - - def __init__(self, details, manager): - """Initialize Switch Base Class.""" - super().__init__(details, manager) - self.features = feature_dict.get(self.device_type, {}).get("features") - if self.features is None: - logger.error("No configuration set for - %s", self.device_name) - raise RuntimeError(f"No configuration set for - {self.device_name}") - self.details = {} - - def is_dimmable(self) -> bool: - """Return True if switch is dimmable.""" - return bool("dimmable" in self.features) - - @abstractmethod - def get_details(self) -> None: - """Get Device Details.""" - - @abstractmethod - def turn_on(self) -> bool: - """Turn Switch On.""" - - @abstractmethod - def turn_off(self) -> bool: - """Turn switch off.""" - - @abstractmethod - def get_config(self) -> None: - """Get configuration and firmware deatils.""" - - @property - def active_time(self) -> int: - """Get active time of switch.""" - return self.details.get("active_time", 0) - - def update(self) -> None: - """Update device details.""" - self.get_details() - - -class VeSyncWallSwitch(VeSyncSwitch): - """Etekcity standard wall switch class.""" - - def __init__(self, details, manager): - """Initialize standard etekcity wall switch class.""" - super().__init__(details, manager) - - def get_details(self) -> None: - """Get switch device details.""" - body = helpers.req_body(self.manager, "devicedetail") - body["uuid"] = self.uuid - head = helpers.req_headers(self.manager) - - r, _ = helpers.call_api( - "/inwallswitch/v1/device/devicedetail", "post", headers=head, json=body - ) - - if r is not None and helpers.code_check(r): - self.device_status = r.get("deviceStatus", self.device_status) - self.details["active_time"] = r.get("activeTime", 0) - self.connection_status = r.get("connectionStatus", self.connection_status) - else: - logger.debug("Error getting %s details", self.device_name) - - def get_config(self) -> None: - """Get switch device configuration info.""" - body = helpers.req_body(self.manager, "devicedetail") - body["method"] = "configurations" - body["uuid"] = self.uuid - - r, _ = helpers.call_api( - "/inwallswitch/v1/device/configurations", - "post", - headers=helpers.req_headers(self.manager), - json=body, - ) - - if helpers.code_check(r): - self.config = helpers.build_config_dict(r) - else: - logger.warning("Unable to get %s config info", self.device_name) - - def turn_off(self) -> bool: - """Turn off switch device.""" - body = helpers.req_body(self.manager, "devicestatus") - body["status"] = "off" - body["uuid"] = self.uuid - head = helpers.req_headers(self.manager) - - r, _ = helpers.call_api( - "/inwallswitch/v1/device/devicestatus", "put", headers=head, json=body - ) - - if r is not None and helpers.code_check(r): - self.device_status = "off" - return True - logger.warning("Error turning %s off", self.device_name) - return False - - def turn_on(self) -> bool: - """Turn on switch device.""" - body = helpers.req_body(self.manager, "devicestatus") - body["status"] = "on" - body["uuid"] = self.uuid - head = helpers.req_headers(self.manager) - - r, _ = helpers.call_api( - "/inwallswitch/v1/device/devicestatus", "put", headers=head, json=body - ) - - if r is not None and helpers.code_check(r): - self.device_status = "on" - return True - logger.warning("Error turning %s on", self.device_name) - return False - - -class VeSyncDimmerSwitch(VeSyncSwitch): - """Vesync Dimmer Switch Class with RGB Faceplate.""" - - def __init__(self, details, manager): - """Initialize dimmer switch class.""" - super().__init__(details, manager) - self._brightness = 0 - self._rgb_value = {"red": 0, "blue": 0, "green": 0} - self._rgb_status = "unknown" - self._indicator_light = "unknown" - - def get_details(self) -> None: - """Get dimmer switch details.""" - body = helpers.req_body(self.manager, "devicedetail") - body["uuid"] = self.uuid - head = helpers.req_headers(self.manager) - - r, _ = helpers.call_api( - "/dimmer/v1/device/devicedetail", "post", headers=head, json=body - ) - - if r is not None and helpers.code_check(r): - self.device_status = r.get("deviceStatus", self.device_status) - self.details["active_time"] = r.get("activeTime", 0) - self.connection_status = r.get("connectionStatus", self.connection_status) - self._brightness = r.get("brightness") - self._rgb_status = r.get("rgbStatus") - self._rgb_value = r.get("rgbValue") - self._indicator_light = r.get("indicatorlightStatus") - else: - logger.debug("Error getting %s details", self.device_name) - - @property - def brightness(self) -> float: - """Return brightness in percent.""" - return self._brightness - - @property - def indicator_light_status(self) -> str: - """Faceplate brightness light status.""" - return self._indicator_light - - @property - def rgb_light_status(self) -> str: - """RGB Faceplate light status.""" - return self._rgb_status - - @property - def rgb_light_value(self) -> dict: - """RGB Light Values.""" - return self._rgb_value - - def switch_toggle(self, status: str) -> bool: - """Toggle switch status.""" - if status not in ["on", "off"]: - logger.debug("Invalid status passed to wall switch") - return False - body = helpers.req_body(self.manager, "devicestatus") - body["status"] = status - body["uuid"] = self.uuid - head = helpers.req_headers(self.manager) - - r, _ = helpers.call_api( - "/dimmer/v1/device/devicestatus", "put", headers=head, json=body - ) - - if r is not None and helpers.code_check(r): - self.device_status = status - return True - - logger.warning("Error turning %s %s", self.device_name, status) - return False - - def turn_on(self) -> bool: - """Turn switch on.""" - return self.switch_toggle("on") - - def turn_off(self) -> bool: - """Turn switch off.""" - return self.switch_toggle("off") - - def indicator_light_toggle(self, status: str) -> bool: - """Toggle indicator light.""" - if status not in ["on", "off"]: - logger.debug("Invalid status for wall switch") - return False - body = helpers.req_body(self.manager, "devicestatus") - body["status"] = status - body["uuid"] = self.uuid - head = helpers.req_headers(self.manager) - - r, _ = helpers.call_api( - "/dimmer/v1/device/indicatorlightstatus", "put", headers=head, json=body - ) - - if r is not None and helpers.code_check(r): - self.device_status = status - return True - - logger.warning("Error turning %s indicator light %s", self.device_name, status) - return False - - def indicator_light_on(self) -> bool: - """Turn Indicator light on.""" - return self.indicator_light_toggle("on") - - def indicator_light_off(self) -> bool: - """Turn indicator light off.""" - return self.indicator_light_toggle("off") - - def rgb_color_status( - self, status: str, red: int = None, blue: int = None, green: int = None - ) -> bool: - """Set faceplate RGB color.""" - body = helpers.req_body(self.manager, "devicestatus") - body["status"] = status - body["uuid"] = self.uuid - head = helpers.req_headers(self.manager) - if red is not None and blue is not None and green is not None: - body["rgbValue"] = {"red": red, "blue": blue, "green": green} - - r, _ = helpers.call_api( - "/dimmer/v1/device/devicergbstatus", "put", headers=head, json=body - ) - - if r is not None and helpers.code_check(r): - self._rgb_status = status - if body.get("rgbValue") is not None: - self._rgb_value = {"red": red, "blue": blue, "green": green} - return True - logger.warning("Error turning %s off", self.device_name) - return False - - def rgb_color_off(self) -> bool: - """Turn RGB Color Off.""" - return self.rgb_color_status("off") - - def rgb_color_on(self) -> bool: - """Turn RGB Color Off.""" - return self.rgb_color_status("on") - - def rgb_color_set(self, red: int, green: int, blue: int) -> bool: - """Set RGB color of faceplate.""" - if isinstance(red, int) and isinstance(green, int) and isinstance(blue, int): - for color in [red, green, blue]: - if color < 0 or color > 255: - logger.warning("Invalid RGB value") - return False - - return bool(self.rgb_color_status("on", red, green, blue)) - return False - - def set_brightness(self, brightness: int) -> bool: - """Set brightness of dimmer - 1 - 100.""" - if isinstance(brightness, int) and (brightness > 0 or brightness <= 100): - - body = helpers.req_body(self.manager, "devicestatus") - body["brightness"] = brightness - body["uuid"] = self.uuid - head = helpers.req_headers(self.manager) - - r, _ = helpers.call_api( - "/dimmer/v1/device/updatebrightness", "put", headers=head, json=body - ) - - if r is not None and helpers.code_check(r): - self._brightness = brightness - return True - logger.warning("Error setting %s brightness", self.device_name) - else: - logger.warning("Invalid brightness") - return False - - def displayJSON(self) -> str: - """JSON API for dimmer switch.""" - sup_val = json.loads(super().displayJSON()) - if self.is_dimmable: - sup_val.update( - { - "Indicator Light": str(self.active_time), - "Brightness": str(self._brightness), - "RGB Light": str(self._rgb_status), - } - ) - return sup_val - - def get_config(self) -> None: - """Get dimmable switch device configuration info.""" - body = helpers.req_body(self.manager, "devicedetail") - body["method"] = "configurations" - body["uuid"] = self.uuid - - r, _ = helpers.call_api( - "/dimmer/v1/device/configurations", - "post", - headers=helpers.req_headers(self.manager), - json=body, - ) - - if helpers.code_check(r): - self.config = helpers.build_config_dict(r) - else: - logger.warning("Unable to get %s config info", self.device_name) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..032e9e5 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +pyvesync==2.0.3 \ No newline at end of file diff --git a/requirements_dev.txt b/requirements_dev.txt index b5b86bf..d7b47f6 100644 --- a/requirements_dev.txt +++ b/requirements_dev.txt @@ -1,4 +1,5 @@ -homeassistant +-r requirements.txt +homeassistant==2022.6.4 black isort flake8