mirror of
https://github.com/micahqcade/custom_vesync.git
synced 2025-02-10 09:08:58 +01:00
Remove local pyvesync lib and migrate to public one. (#25)
* move to pip pyvesync library * remove pyvesync * improve extra attributes * fix style
This commit is contained in:
Vincent Le Bourlot
GitHub
parent
4f3c7fd36f
commit
d10928ecca
@ -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,
|
||||
|
||||
@ -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__)
|
||||
|
||||
|
||||
@ -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__)
|
||||
|
||||
|
||||
@ -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",
|
||||
|
||||
@ -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):
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
@ -1,6 +0,0 @@
|
||||
"""VeSync API Library."""
|
||||
import logging
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO, format="%(asctime)s - %(levelname)5s - %(message)s"
|
||||
)
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@ -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",
|
||||
}
|
||||
@ -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()
|
||||
@ -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,
|
||||
}
|
||||
)
|
||||
@ -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
|
||||
File diff suppressed because it is too large
Load Diff
@ -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"))
|
||||
@ -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)
|
||||
1
requirements.txt
Normal file
1
requirements.txt
Normal file
@ -0,0 +1 @@
|
||||
pyvesync==2.0.3
|
||||
@ -1,4 +1,5 @@
|
||||
homeassistant
|
||||
-r requirements.txt
|
||||
homeassistant==2022.6.4
|
||||
black
|
||||
isort
|
||||
flake8
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user