diff --git a/LICENSE b/LICENSE index 99b2af1..b1ab6a7 100644 --- a/LICENSE +++ b/LICENSE @@ -1,6 +1,6 @@ MIT License -Copyright (c) 2024 Thomas Luther +Copyright (c) 2024 thomluther Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal diff --git a/README.md b/README.md new file mode 100644 index 0000000..8bc404f --- /dev/null +++ b/README.md @@ -0,0 +1,122 @@ +# Anker Solix API + +[![License](https://img.shields.io/pypi/l/python-eufy-security.svg)](https://github.com/fuzzymistborn/python-eufy-security/blob/master/LICENSE) +[![Code Coverage](https://codecov.io/gh/fuzzymistborn/python-eufy-security/branch/master/graph/badge.svg)](https://codecov.io/gh/fuzzymistborn/python-eufy-security) +[![Say Thanks](https://img.shields.io/badge/SayThanks-!-1EAEDB.svg)](https://saythanks.io/to/fuzzymistborn) + +This is an experimental Python library for Anker Solix Power devices (Solarbank, Inverter etc). +🚨 This is by no means an official Anker API. It can break at any time, +or API request can be removed/added/changed and break some of the endpoint methods used in this API.🚨 + +# Python Versions + +The library is currently supported on + +* Python 3.11 +* Python 3.12 + +# Required libraries + +```bash +pip install cryptography +pip install aiohttp +``` + +# Anker Account Information + +Because of the way the Anker Solix API works, one account with email/password combo cannot be used for the Anker mobile App work and this API in parallel. +The Anker Cloud allows only one account token at a time, each new authentication request will create a new token and drop a previous token. +It is recommended to create a second Anker account and share your Power System(s) with the second account. +Attention: A shared account is only a member of the shared site, and as such currently has no permissions to access or query device details. However, a shared account +can receive the data as provided in the scene/site details, which is equivalent to what is displayed in the mobile App on the Home screen for the selected system. + +# Usage + +Everything starts with an: +[aiohttp](https://aiohttp.readthedocs.io/en/stable/) `ClientSession`: + +```python +import asyncio +from aiohttp import ClientSession +import logging, json + +_LOGGER: logging.Logger = logging.getLogger(__name__) + +async def main() -> None: + """Create the aiohttp session and run the example.""" + async with ClientSession() as websession: + """put your code here, example""" + myapi = api.API("username@domain.com","password","de",websession, _LOGGER) + await myapi.update_sites() + await myapi.update_device_details() + print("System Overview:") + print(json.dumps(myapi.sites, indent=2)) + print("Device Overview:") + print(json.dumps(myapi.devices, indent=2)) + +"""run async main""" +if __name__ == '__main__': + try: + asyncio.run(main()) + except Exception as err: + print(f'{type(err)}: {err}') +``` + +Check out `test_api.py` and other python executable tools that may help to leverage and explore the API. +The subfolder examples contains json files with anonymized responses of the export_system.py module giving you an idea of how various API responses look like. +Those json files can also be used to develop/debug the API for system constellations not available to the developper. + +# API Tools + +## test_api.py + +Example exec module that can be used to explore and test API methods or direct enpoint requests with parameters. + +## export_system.py + +Example exec module to use the Anker API for export of defined system data and device details. +This module will prompt for the Anker account details if not pre-set in the header. +Upon successfull authentication, you can specify a subfolder for the exported JSON files received as API query response, defaulting to your nick name +Optionally you can specify whether personalized information in the response data should be randomized in the files, like SNs, Site IDs, Trace IDs etc. +You can review the response files afterwards. They can be used as examples for dedicated data extraction from the devices. +Optionally the API class can use the json files for debugging and testing on various system outputs. + +## solarbank_monitor.py + +Example exec module to use the Anker API for continously querying and displaying important solarbank parameters +This module will prompt for the Anker account details if not pre-set in the header. +Upon successfull authentication, you will see the solarbank parameters displayed and refreshed at reqular interval. +Note: When the system owning account is used, more details for the solarbank can be queried and displayed. +Attention: During executiion of this module, the used account cannot be used in the Anker App since it will be kicked out on each refresh. + +## energy_csv.py + +Example exec module to use the Anker API for export of daily Solarbank Energy Data. +This method will prompt for the Anker account details if not pre-set in the header. +Then you can specify a start day and the number of days for data extraction from the Anker Cloud. +Note: The Solar production and Solarbank discharge can be queried across the full range. The solarbank +charge however can be queried only as total for an interval (e.g. day). Therefore when solarbank charge +data is also selected for export, an additional API query per day is required. +The received daily values will be exported into a csv file. + + +# Contributing + +1. [Check for open features/bugs](https://github.com/FuzzyMistborn/python-eufy-security/issues) + or [initiate a discussion on one](https://github.com/FuzzyMistborn/python-eufy-security/issues/new). +2. [Fork the repository](https://github.com/FuzzyMistborn/python-eufy-security/fork). +3. Install the dev environment: `make init`. +4. Enter the virtual environment: `source ./venv/bin/activate` +5. Code your new feature or bug fix. +6. Write a test that covers your new functionality. +7. Update `README.md` with any new documentation. +8. Run tests and ensure 100% code coverage: `make coverage` +9. Ensure you have no linting errors: `make lint` +10. Ensure you have typed your code correctly: `make typing` +11. Submit a pull request! + + +# Ackknowledgements + +[python-eufy-security](https://github.com/FuzzyMistborn/python-eufy-security) +[solix2mqtt](https://github.com/tomquist/solix2mqtt) \ No newline at end of file diff --git a/api/__init__.py b/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/api/__pycache__/__init__.cpython-310.pyc b/api/__pycache__/__init__.cpython-310.pyc new file mode 100644 index 0000000..65dab49 Binary files /dev/null and b/api/__pycache__/__init__.cpython-310.pyc differ diff --git a/api/__pycache__/__init__.cpython-311.pyc b/api/__pycache__/__init__.cpython-311.pyc new file mode 100644 index 0000000..848983c Binary files /dev/null and b/api/__pycache__/__init__.cpython-311.pyc differ diff --git a/api/__pycache__/api.cpython-310.pyc b/api/__pycache__/api.cpython-310.pyc new file mode 100644 index 0000000..2b27e96 Binary files /dev/null and b/api/__pycache__/api.cpython-310.pyc differ diff --git a/api/__pycache__/api.cpython-311.pyc b/api/__pycache__/api.cpython-311.pyc new file mode 100644 index 0000000..0576f3c Binary files /dev/null and b/api/__pycache__/api.cpython-311.pyc differ diff --git a/api/__pycache__/errors.cpython-310.pyc b/api/__pycache__/errors.cpython-310.pyc new file mode 100644 index 0000000..7c77e13 Binary files /dev/null and b/api/__pycache__/errors.cpython-310.pyc differ diff --git a/api/__pycache__/errors.cpython-311.pyc b/api/__pycache__/errors.cpython-311.pyc new file mode 100644 index 0000000..fbef8b8 Binary files /dev/null and b/api/__pycache__/errors.cpython-311.pyc differ diff --git a/api/api.py b/api/api.py new file mode 100644 index 0000000..94028d4 --- /dev/null +++ b/api/api.py @@ -0,0 +1,833 @@ +""" +Class for interacting with the Anker Power / Solix API. +Required Python modules: +pip install cryptography +pip install aiohttp +""" + +from datetime import datetime, timedelta +from typing import Dict, Optional +import time +import logging +import sys, os, json + +from aiohttp import ClientSession +from aiohttp.client_exceptions import ClientError +from cryptography.hazmat.primitives import hashes, serialization, padding +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from base64 import b64encode + +from . import errors + +_LOGGER: logging.Logger = logging.getLogger(__name__) + +"""Default definitions required for the Anker Power/Solix Cloud API""" +_API_BASE: str = "https://ankerpower-api-eu.anker.com" +_API_LOGIN="passport/login" +_API_HEADERS={ + 'Content-Type': 'application/json', + 'Model-Type': 'DESKTOP', + 'App-Name': 'anker_power', + 'Os-Type': 'android' + } + +"""Following are the Anker Power/Solix Cloud API endpoints known so far""" +_API_ENDPOINTS={ + 'homepage': 'power_service/v1/site/get_site_homepage', # Scene info for configured site(s), content as preseneted on App Home Page (works only for site owners) + 'site_list': 'power_service/v1/site/get_site_list', # List of available site ids for the user, will also show sites user is only member of + 'site_detail': 'power_service/v1/site/get_site_detail', # Information for given site_id, can also be used by site members + 'scene_info': 'power_service/v1/site/get_scen_info', # Scene info for provided site id (does not contain all device details) + 'user_devices': 'power_service/v1/site/list_user_devices', # List Device details of owned devices, not all device information included + 'charging_devices': 'power_service/v1/site/get_charging_device', # List of Power units? + 'get_device_parm': 'power_service/v1/site/get_site_device_param', # Get settings of a device for the provided site id and param type (e.g. Schedules) + 'set_device_parm': 'power_service/v1/site/set_site_device_param', # Apply provided settings to a device for the provided site id and param type (e.g. Schedules), NOT IMPLEMENTED YET + 'wifi_list': 'power_service/v1/site/get_wifi_info_list', # List of available networks for provided site id + 'get_site_price': 'power_service/v1/site/get_site_price', # List defined power price and CO2 for given site, works only for site owners + 'update_site_price': 'power_service/v1/site/update_site_price', # Update power price for given site, REQUIRED PARAMETERS UNKNOWN + 'get_auto_upgrade': 'power_service/v1/app/get_auto_upgrade', # List of Auto-Upgrade configuration and enabled devices + 'set_auto_upgrade': 'power_service/v1/app/set_auto_upgrade', # Set/Enable Auto-Upgrade configuration, not implemented yet, REQUIRED PARAMETERS UNKNOWN + 'bind_devices': 'power_service/v1/app/get_relate_and_bind_devices', # List with details of locally connected/bound devices, includes firmware version + 'get_device_load': 'power_service/v1/app/device/get_device_home_load', # Get defined device schedule (similar to data in device param) + 'set_device_load': 'power_service/v1/app/device/set_device_home_load', # Set defined device schedule (Not implemented yet REQUIRED PARAMETERS UNKNOWN) + 'get_ota_info': 'power_service/v1/app/compatible/get_ota_info', # Not implemented (List of available firmware updates?) REQUIRED PARAMETERS UNKNOWN + 'get_ota_update': 'power_service/v1/app/compatible/get_ota_update', # Not implemented (Perform local firmware update?) REQUIRED PARAMETERS UNKNOWN + 'solar_info': 'power_service/v1/app/compatible/get_compatible_solar_info', # Solar Panel details with Anker Inverters? REQUIRED PARAMETERS UNKNOWN + 'get_cutoff': 'power_service/v1/app/compatible/get_power_cutoff', # Get Power Cutoff settings (Min SOC) for provided site id and device sn + 'set_cutoff': 'power_service/v1/app/compatible/set_power_cutoff', # Set Min SOC for device, not implemented yet REQUIRED PARAMETERS UNKNOWN + 'compatible_process': 'power_service/v1/app/compatible/get_compatible_process', # Not implemented (What is this used for?) REQUIRED PARAMETERS UNKNOWN + 'get_device_fittings': 'power_service/v1/app/get_relate_device_fittings', # Device fittings for given site id and device sn. Solarbank response does not contain info + 'energy_analysis': 'power_service/v1/site/energy_analysis', # Fetch energy data for given time frames + 'home_load_chart': 'power_service/v1/site/get_home_load_chart', # Fetch data as displayed in home load chart for given site_id and optional device SN (empty if solarbank not connected) + } +""" Other endpoints neither implemented nor explored + 'power_service/v1/site/can_create_site', + 'power_service/v1/site/create_site', + 'power_service/v1/site/update_site', + 'power_service/v1/site/delete_site', + 'power_service/v1/site/add_charging_device', + 'power_service/v1/site/update_charging_device', + 'power_service/v1/site/reset_charging_device', + 'power_service/v1/site/delete_charging_device', + 'power_service/v1/site/update_site_device', + 'power_service/v1/site/delete_site_devices', + 'power_service/v1/site/add_site_devicesDatagram', + 'power_service/v1/app/compatible/set_ota_update', + 'power_service/v1/app/compatible/save_ota_complete_status', + 'power_service/v1/app/compatible/check_third_sn', + 'power_service/v1/app/compatible/save_compatible_solar', + 'power_service/v1/app/after_sale/check_popup', + 'power_service/v1/app/after_sale/check_sn', + 'power_service/v1/app/after_sale/mark_sn', + 'power_service/v1/app/share_site/delete_site_member', + 'power_service/v1/app/share_site/invite_member', + 'power_service/v1/app/share_site/delete_inviting_member', + 'power_service/v1/app/share_site/get_invited_list', + 'power_service/v1/app/share_site/join_site', + 'power_service/v1/app/upgrade_event_report', + 'power_service/v1/app/check_upgrade_record', + 'power_service/v1/app/get_upgrade_record', + 'power_service/v1/app/get_phonecode_list', + 'power_service/v1/app/compatible/get_confirm_permissions', + 'power_service/v1/app/compatible/confirm_permissions_settings', + 'power_service/v1/message_not_disturb', + 'power_service/v1/get_message_not_disturb', + 'power_service/v1/read_message', + 'power_service/v1/get_message', + 'power_service/v1/del_message', + 'power_service/v1/product_categories', + 'power_service/v1/product_accessories', +""" + + +"""Structure of the JSON response for an API Login Request +An unexpired token_id must be used for API request, along with the gtoken which is an MD5 hash of the returned(encrypted) user_id. +The combination of the provided token and MD5 hashed user_id authenticate the client to the server. +The Login Response it cached in a JSON file per email user account and can be reused by this API object without further login request. +ATTENTION: Anker allows only 1 active token on the server per user account. Any login for the same account (e.g. via Anker APP) will kickoff the token used in this API object and further requests are no longer authorized +Currently, the API will re-authenticate automatically and therefore may kick off the other user that obtained the actual access token (e.g. kick out the App user again when used for regular API requests) +NOTES: Parallel API objects should use different user accounts. They may work in parallel when all using the same cached data. The first API object with failed authorization will restart a new Login request and updates +the cached data file. Other objects should recognize an update of the cached login file will refresh their login credentials in the object for the token and gtoken. +""" +LOGIN_RESPONSE: Dict = { + "user_id": str, + "email": str, + "nick_name": str, + "auth_token": str, + "token_expires_at": int, + "avatar": str, + "mac_addr": str, + "domain": str, + "ab_code": str, + "token_id": int, + "geo_key": str, + "privilege": int, + "phone": str, + "phone_number": str, + "phone_code": str, + "server_secret_info": { + "public_key": str + }, + "params": list, + "trust_list": list, + "fa_info": { + "step": int, + "info": str + }, + "country_code": str +} + + +class API: + """Define the API class to handle server authentication and API requests, along with the last state of queried Homepage and Device information""" + + def __init__(self, email: str, password: str, countryId: str, websession: ClientSession, logger = None) -> None: + """Initialize.""" + self._api_base: str = _API_BASE + self._email: str = email + self._password: str = password + self._countryId: str = countryId.upper() + self._session: ClientSession = websession + self._loggedIn: bool = False + self._testdir: str = 'test' + self._retry_attempt: bool = False # Flag for retry after any token error + os.makedirs(os.path.join(".","authcache"), exist_ok=True) # ensure folder for authentication caching exists + self._authFile: str = os.path.join(".","authcache",f"{email}.json") # filename for authentication cache + self._authFileTime: float = 0 + """initialize logger for object""" + if logger: + self._logger = logger + else: + self._logger = _LOGGER + self._logger.setLevel(logging.WARNING) + if not self._logger.hasHandlers(): + self._logger.addHandler(logging.StreamHandler(sys.stdout)) + + + self._timezone: str = self._getTimezoneGMTString() #Timezone format: 'GMT+01:00' + self._gtoken: Optional[str] = None + self._token: Optional[str] = None + self._token_expiration: Optional[datetime] = None + self._login_response: Optional[LOGIN_RESPONSE] = {} + + """Define Encryption for password, using ECDH assymetric key exchange for shared secret calculation, which must be used to encrypt the password using AES-256-CBC with seed of 16""" + # uncompressed public key from Anker server in the format 04 [32 byte x vlaue] [32 byte y value] + self._api_public_key_hex = '04c5c00c4f8d1197cc7c3167c52bf7acb054d722f0ef08dcd7e0883236e0d72a3868d9750cb47fa4619248f3d83f0f662671dadc6e2d31c2f41db0161651c7c076' + self._curve = ec.SECP256R1() # Encryption curve SECP256R1 (identical to prime256v1) + self._ecdh = ec.generate_private_key(self._curve, default_backend()) # returns EllipticCurvePrivateKey + self._public_key = self._ecdh.public_key() # returns EllipticCurvePublicKey + self._shared_key = self._ecdh.exchange(ec.ECDH(), ec.EllipticCurvePublicKey.from_encoded_point(self._curve, bytes.fromhex(self._api_public_key_hex))) # returns bytes of shared secret + + """Define class variables saving the most recent site and device data""" + self.nickname: str = "" + self.sites: Dict = {} + self.devices: Dict = {} + + + def _md5(self,text: str) -> str: + h = hashes.Hash(hashes.MD5()) + h.update(text.encode('utf-8')) + return h.finalize().hex() + + def _getTimezoneGMTString(self) -> str: + """Construct timezone GMT string with offset, e.g. GMT+01:00""" + tzo = datetime.now().astimezone().strftime('%z') + return f"GMT{tzo[:3]}:{tzo[3:5]}" + + def _encryptApiData(self, raw: str) -> str: + """Password must be UTF-8 encoded and AES-256-CBC encrypted with block size of 16 + Return Base64 encoded secret as utf-8 decoded string using the shared secret with seed of 16 for the encryption""" + aes = Cipher(algorithms.AES(self._shared_key), modes.CBC(self._shared_key[0:16]), backend=default_backend()) + encryptor = aes.encryptor() + """Use default PKCS7 padding for incomplete AES blocks""" + padder = padding.PKCS7(128).padder() + raw_padded = padder.update(raw.encode("utf-8")) + padder.finalize() + return (b64encode(encryptor.update(raw_padded) + encryptor.finalize())).decode("utf-8") + + def _loadFromFile(self, filename: str) -> dict: + """Load json data from given file for testing""" + try: + if os.path.isfile(filename): + with open(filename, 'r') as file: + data = json.load(file) + self._logger.debug(f"Loaded JSON from file {filename}:") + self._logger.debug(f"{data}:") + return data + return {} + except Exception as err: + self._logger.error(f"ERROR: Failed to load JSON from file {filename}") + self._logger.error(err) + return {} + + def _saveToFile(self, filename: str, json: dict = {}) -> bool: + """Save json data to given file for testing""" + try: + with open(filename, 'w') as file: + json.dump(json, file, indent=2) + self._logger.debug(f"Saved JSON to file {filename}") + return True + except Exception as err: + self._logger.error(f"ERROR: Failed to save JSON to file {filename}") + self._logger.error(err) + return False + + def _update_dev(self, devData: dict, devType: str = None, siteId: str = None, isAdmin: bool = None) -> None: + """Update the internal device details dictionary with the given data. The device_sn key must be set in the data dict for the update to be applied. + This method is used to consolidate various device related key values from various requests under a common set of device keys. + TODO: Add more relevent keys for Solarbank or other devices once known/required""" + sn = devData.get("device_sn", None) + if sn: + device = self.devices.get(sn,{}) # lookup old device info if any + device.update({"sn": str(sn)}) + if devType: + device.update({"type": devType.lower()}) + if siteId: + device.update({"site_id": str(siteId)}) + if isAdmin: + device.update({"is_admin": True}) + elif isAdmin == False and device.get("is_admin") == None: + device.update({"is_admin": False}) + for key, value in devData.items(): + if key in ["product_code","device_pn"]: + device.update({"pn": str(value)}) + elif key in ["device_name"]: + device.update({"name": str(value)}) + elif key in ["device_sw_version"]: + device.update({"sw_version": str(value)}) + elif key in ["wifi_online"]: + device.update({"wifi_online": bool(value)}) + elif key in ["bt_ble_mac"]: + device.update({"bt_ble_mac": str(value)}) + elif key in ["battery_power"]: + device.update({"battery_soc": str(value)}) # This is a perventage value, not power + elif key in ["charging_power"]: + device.update({"charging_power": str(value)}) + elif key in ["photovoltaic_power"]: + device.update({"input_power": str(value)}) + elif key in ["output_power"]: + device.update({"output_power": str(value)}) + elif key in ["set_load_power"]: + device.update({"set_output_power": str(value)}) + elif key in ["power_unit"]: + device.update({"power_unit": str(value)}) + elif key in ["charging_status"]: + device.update({"charging_status": str(value)}) + elif key in ["bws_surplus"]: + device.update({"bws_surplus": str(value)}) + elif key in ["charge"]: + device.update({"charge": bool(value)}) + elif key in ["auto_upgrade"]: + device.update({"auto_upgrade": bool(value)}) + + self.devices.update({str(sn): device}) + return + + def testDir(self, subfolder: str = None) -> str: + """get or set the subfolder for local API test files""" + if not subfolder: + return self._testdir + else: + if not os.path.isdir(subfolder): + self._logger.error(f"Specified test folder does not exist: {subfolder}") + else: + self._testdir = subfolder + self._logger.debug(f"Set test folder to: {subfolder}") + return self._testdir + + def logLevel(self, level: int = None) -> int: + """get or set the logger log level""" + if level: + self._logger.setLevel(level) + self._logger.info(f"Set log level to: {level}") + return self._logger.getEffectiveLevel() + + async def async_authenticate(self, restart: bool = False) -> bool: + """Authenticate with server and get an access token. If restart is not enforced, cached login data may be used to obtain previous token""" + if restart: + self._token = None + self._token_expiration = None + self._gtoken = None + self._loggedIn= False + self._authFileTime = 0 + self.nickname = "" + if os.path.isfile(self._authFile): + try: + os.remove(self._authFile) + except: + pass + # First check if cached login response is availble and login params can be filled, otherwise query server for new login tokens + if os.path.isfile(self._authFile): + data = self._loadFromFile(self._authFile) + self._authFileTime = os.path.getmtime(self._authFile) + self._logger.debug(f"Cached Login for {self._email} from {datetime.fromtimestamp(self._authFileTime).isoformat()}:") + self._logger.debug(f"{data}") + self._retry_attempt = False # set first attempt to allow retry for authentication refresh + else: + self._logger.debug("Fetching new Login credentials from server...") + now = datetime.now().astimezone() + self._retry_attempt = True # set last attempt to avoid retry on failed authentication + auth_resp = await self.request("post", _API_LOGIN, json = { + "ab": self._countryId, + "client_secret_info": { + "public_key": self._public_key.public_bytes(serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint).hex() # Uncompressed format of points in hex (0x04 + 32 Byte + 32 Byte) + }, + "enc": 0, + "email": self._email, + "password": self._encryptApiData(self._password), # AES-256-CBC encrypted by the ECDH shared key derived from server public key and local private key + "time_zone": round(datetime.utcoffset(now).total_seconds()*1000), # timezone offset in ms, e.g. 'GMT+01:00' => 3600000 + "transaction": str(int(time.mktime(now.timetuple())*1000)) # Unix Timestamp in ms as string + }) + data = auth_resp.get("data",{}) + self._logger.debug(f"Login Response: {data}") + self._loggedIn = True + # Cache login response in file for reuse + with open(self._authFile, 'w') as authfile: + json.dump(data, authfile, indent=2, skipkeys=True) + self._logger.debug(f"Response cached in file: {self._authFile}") + self._authFileTime = os.path.getmtime(self._authFile) + + #Update the login params + self._login_response = {} + self._login_response.update(data) + self._token = data.get("auth_token") + self.nickname = data.get("nick_name") + if data.get("token_expires_at"): + self._token_expiration = datetime.fromtimestamp(data.get("token_expires_at")) + else: + self._token_expiration = None + self._loggedIn = False + if data.get("user_id"): + self._gtoken = self._md5(data["user_id"]) # gtoken is MD5 hash of user_id from login response + else: + self._gtoken = None + self._loggedIn = False + return self._loggedIn + + + async def request(self, method: str, endpoint: str, *, headers: Optional[dict] = {}, json: Optional[dict] = {}) -> dict: + """This method handles all requests to the API, this is also called recursively by login requests if necessary""" + if self._token_expiration and (self._token_expiration - datetime.now()).total_seconds() < 60: + self._logger.warning("WARNING: Access token expired, fetching a new one...") + await self.async_authenticate(restart=True) + """For non-Login requests, ensure authentication will be updated if not logged in yet or cached file was refreshed""" + if endpoint != _API_LOGIN and (not self._loggedIn or (os.path.isfile(self._authFile) and self._authFileTime != os.path.getmtime(self._authFile))): + await self.async_authenticate() + + url: str = f"{self._api_base}/{endpoint}" + mergedHeaders = _API_HEADERS + mergedHeaders.update(headers) + if self._countryId: + mergedHeaders["Country"] = self._countryId + if self._timezone: + mergedHeaders["Timezone"] = self._timezone + if self._token: + mergedHeaders["x-auth-token"] = self._token + mergedHeaders["gtoken"] = self._gtoken + + self._logger.debug(f"Request Url: {method} {url}") + self._logger.debug(f"Request Headers: {mergedHeaders}") + self._logger.debug(f"Request Body: {json}") + async with self._session.request(method, url, headers=mergedHeaders, json=json) as resp: + try: + resp.raise_for_status() + data: dict = await resp.json(content_type=None) + self._logger.debug(f"Request Response: {data}") + + if not data: + raise ClientError(f"No response while requesting {endpoint}") + + errors.raise_error(data) # check the response code in the data + if endpoint != _API_LOGIN: + self._retry_attempt = False # reset retry flag only when valid token received and not another login request + + # valid response at this point, mark login and return data + self._loggedIn = True + return data + + except ClientError as err: # Exception from ClientSession based on standard response codes + self._logger.error(f"Request Error: {err}") + if "401" in str(err): + #Unauthorized request + if self._retry_attempt: + raise errors.AnkerSolixError(f"Login failed for user {self._email}") + self._logger.warning("Login failed, retrying authentication...") + if await self.async_authenticate(restart=True): + return await self.request(method, endpoint, headers=headers, json=json) + else: + self._logger.error(f"Failed to login with token {self._token} and gtoken {self._gtoken}") + raise err("Failed to login") + raise ClientError(f"There was an unknown error while requesting {endpoint}: {err}") from None + except (errors.InvalidCredentialsError, errors.TokenKickedOutError) as err: # Exception for API specific response codes + self._logger.error(f"API ERROR: {err}") + if self._retry_attempt: + raise errors.AnkerSolixError(f"Login failed for user {self._email}") + self._logger.warning("Login failed, retrying authentication...") + if await self.async_authenticate(restart=True): + return await self.request(method, endpoint, headers=headers, json=json) + else: + self._logger.error(f"Failed to login with token {self._token} and gtoken {self._gtoken}") + raise err("Failed to login") + except errors.AnkerSolixError as err: # Other Exception from API + self._logger.error(f"ANKER API ERROR: {err}") + raise err + + + async def update_sites(self, fromFile: bool = False) -> Dict: + """Get the latest info for all accessible sites and update class site and device variables + Example data: + {'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c': + {'site_info': {'site_id': 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c', 'site_name': 'BKW', 'site_img': '', 'device_type_list': [3], 'ms_type': 1, 'power_site_type': 2, 'is_allow_delete': True}, + 'site_admin': True, + 'home_info': {'home_name': 'Home', 'home_img': '', 'charging_power': '0.00', 'power_unit': 'W'}, + 'solar_list': [], + 'pps_info': {'pps_list': [], 'total_charging_power': '0.00', 'power_unit': 'W', 'total_battery_power': '0.00', 'updated_time': '', 'pps_status': 0}, + 'statistics': [{'type': '1', 'total': '89.75', 'unit': 'kwh'}, {'type': '2', 'total': '89.48', 'unit': 'kg'}, {'type': '3', 'total': '35.90', 'unit': '€'}], + 'topology_type': '1', + 'solarbank_info': {'solarbank_list': [{'device_pn': 'A17C0', 'device_sn': '9JVB42LJK8J0P5RY', 'device_name': 'Solarbank E1600', + 'device_img': 'https://public-aiot-fra-prod.s3.dualstack.eu-central-1.amazonaws.com/anker-power/public/product/anker-power/e9478c2d-e665-4d84-95d7-dd4844f82055/20230719-144818.png', + 'battery_power': '75', 'bind_site_status': '', 'charging_power': '0', 'power_unit': 'W', 'charging_status': '2', 'status': '0', 'wireless_type': '1', 'main_version': '', 'photovoltaic_power': '0', + 'output_power': '0', 'create_time': 1695392386, 'set_load_power': ''}], + 'total_charging_power': '0', 'power_unit': 'W', 'charging_status': '0', 'total_battery_power': '0.00', 'updated_time': '2023-12-28 18:53:27', 'total_photovoltaic_power': '0', 'total_output_power': '0.00', + 'display_set_power': False}, + 'retain_load': '0W', + 'updated_time': '01-01-0001 00:00:00', + 'power_site_type': 2, + 'site_id': 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c', + 'powerpanel_list': []}} + """ + self._logger.info("Updating Sites data...") + self.sites = {} + self._logger.debug("Getting site list...") + sites = await self.get_site_list(fromFile=fromFile) + for site in sites.get("site_list",[]): + if site.get("site_id"): + """Update site info""" + myid = site.get("site_id") + mysite = self.sites.get(myid,{}) + siteInfo = mysite.get("site_info",{}) + siteInfo.update(site) + mysite.update({"site_info": siteInfo}) + admin = siteInfo.get("ms_type",0) in [0,1] # add boolean key to indicate whether user is site admin (ms_type 1 or not known) and can query device details + mysite.update({"site_admin": admin}) + """Update scene info for site""" + self._logger.debug("Getting scene info for site...") + scene = await self.get_scene_info(myid,fromFile=fromFile) + mysite.update(scene) + self.sites.update({myid: mysite}) + """Update device details from scene info""" + for solarbank in (mysite.get("solarbank_info",{})).get("solarbank_list",[]): + self._update_dev(solarbank,devType="solarbank",siteId=myid,isAdmin=admin) + for pps in (mysite.get("pps_info",{})).get("pps_list",[]): + self._update_dev(pps,devType="pps",siteId=myid,isAdmin=admin) + for solar in mysite.get("solar_list",[]): + self._update_dev(solar,devType="solar",siteId=myid,isAdmin=admin) + for powerpanel in mysite.get("powerpanel_list",[]): + self._update_dev(powerpanel,devType="powerpanel",siteId=myid,isAdmin=admin) + return self.sites + + + async def update_device_details(self, fromFile: bool = False) -> Dict: + """Get the latest updates for additional device info updated less frequently. + Most of theses requests return data only when user has admin rights for sites owning the devices. + To limit API requests, this update device details method should be called less frequently than update site method, which will also update most device details as found in the site data response. + """ + self._logger.info("Updating Device Details...") + """Fetch firmware version of device""" + self._logger.debug("Getting bind devices...") + data = await self.get_bind_devices(fromFile=fromFile) + for device in data.get("data",[]): + self._update_dev(device) + """Get the setting for effective automated FW upgrades""" + self._logger.debug("Getting OTA settings...") + data = await self.get_auto_upgrade(fromFile=fromFile) + main = data.get("main_switch",None) + devicelist = data.get("device_list",[]) # could be null for non owning account + if not devicelist: + devicelist = [] + for device in devicelist: + ota = device.get("auto_upgrade",None) + if isinstance(ota,bool): + """update device setting based on main setting if available""" + if isinstance(main,bool): + device.update({"auto_upgrade": main and ota}) + self._update_dev(device) + """Fetch other relevant device information that requires site id and/or SN""" + for sn, device in self.devices.items(): + """Fetch active Power Cutoff setting for solarbanks""" + if device.get("is_admin",False) and device.get("site_id","") != "" and device.get("type","") in ["solarbank"]: + self._logger.debug(f"Getting Power Cutoff settings for device...") + data = await self.get_power_cutoff(siteId=device.get("site_id"),deviceSn=sn,fromFile=fromFile) + for setting in data.get("power_cutoff_data",[]): + if int(setting.get("is_selected",0)) > 0 and int(setting.get("output_cutoff_data",0)) > 0: + device.update({"power_cutoff": int(setting.get("output_cutoff_data"))}) + self.devices.update({sn: device}) + """TODO: Fetch other details of specific device types as known and relevant""" + return self.devices + + + async def get_site_list(self, fromFile: bool = False) -> Dict: + """Get the site list. + Example data: + {'site_list': [{'site_id': 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c', 'site_name': 'BKW', 'site_img': '', 'device_type_list': [3], 'ms_type': 2, 'power_site_type': 2, 'is_allow_delete': True}]} + """ + if fromFile: + resp = self._loadFromFile(os.path.join(self._testdir,f"site_list.json")) + else: + resp = await self.request("post", _API_ENDPOINTS["site_list"]) + return resp.get("data",{}) + + + async def get_scene_info(self, siteId: str, fromFile: bool = False) -> Dict: + """Get scene info. This can be queried for each siteId listed in the homepage info site_list. It shows also data for accounts that are only site members. + Example data for provided site_id: + {"home_info":{"home_name":"Home","home_img":"","charging_power":"0.00","power_unit":"W"}, + "solar_list":[], + "pps_info":{"pps_list":[],"total_charging_power":"0.00","power_unit":"W","total_battery_power":"0.00","updated_time":"","pps_status":0}, + "statistics":[{"type":"1","total":"89.75","unit":"kwh"},{"type":"2","total":"89.48","unit":"kg"},{"type":"3","total":"35.90","unit":"€"}], + "topology_type":"1","solarbank_info":{"solarbank_list": + [{"device_pn":"A17C0","device_sn":"9JVB42LJK8J0P5RY","device_name":"Solarbank E1600", + "device_img":"https://public-aiot-fra-prod.s3.dualstack.eu-central-1.amazonaws.com/anker-power/public/product/anker-power/e9478c2d-e665-4d84-95d7-dd4844f82055/20230719-144818.png", + "battery_power":"75","bind_site_status":"","charging_power":"0","power_unit":"W","charging_status":"2","status":"0","wireless_type":"1","main_version":"","photovoltaic_power":"0", + "output_power":"0","create_time":1695392386}], + "total_charging_power":"0","power_unit":"W","charging_status":"0","total_battery_power":"0.00","updated_time":"2023-12-28 18:53:27","total_photovoltaic_power":"0","total_output_power":"0.00"}, + "retain_load":"0W","updated_time":"01-01-0001 00:00:00","power_site_type":2,"site_id":"efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c"} + """ + data = {"site_id": siteId} + if fromFile: + resp = self._loadFromFile(os.path.join(self._testdir,f"scene_{siteId}.json")) + else: + resp = await self.request("post", _API_ENDPOINTS["scene_info"], json = data) + return resp.get("data",{}) + + + async def get_homepage(self, fromFile: bool = False) -> Dict: + """Get the latest homepage info. + NOTE: This returns only data if the site is owned by the account. No data returned for site member accounts + Example data: + {"site_list":[{"site_id":"efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c","site_name":"BKW","site_img":"","device_type_list":[3],"ms_type":0,"power_site_type":0,"is_allow_delete":false}], + "solar_list":[],"pps_list":[], + "solarbank_list":[{"device_pn":"","device_sn":"9JVB42LJK8J0P5RY","device_name":"Solarbank E1600", + "device_img":"https://public-aiot-fra-prod.s3.dualstack.eu-central-1.amazonaws.com/anker-power/public/product/anker-power/e9478c2d-e665-4d84-95d7-dd4844f82055/20230719-144818.png", + "battery_power":"75","bind_site_status":"1","charging_power":"","power_unit":"","charging_status":"","status":"","wireless_type":"","main_version":"","photovoltaic_power":"","output_power":"","create_time":0}], + "powerpanel_list":[]} + """ + if fromFile: + resp = self._loadFromFile(os.path.join(self._testdir,f"homepage.json")) + else: + resp = await self.request("post", _API_ENDPOINTS["homepage"]) + return resp.get("data",{}) + + + async def get_bind_devices(self, fromFile: bool = False) -> Dict: + """Get the bind device information, contains firmware level of devices + Example data: + {"data": [{"device_sn":"9JVB42LJK8J0P5RY","product_code":"A17C0","bt_ble_id":"BC:A2:AF:C7:55:F9","bt_ble_mac":"BCA2AFC755F9","device_name":"Solarbank E1600","alias_name":"Solarbank E1600", + "img_url":"https://public-aiot-fra-prod.s3.dualstack.eu-central-1.amazonaws.com/anker-power/public/product/anker-power/e9478c2d-e665-4d84-95d7-dd4844f82055/20230719-144818.png", + "link_time":1695392302068,"wifi_online":false,"wifi_name":"","relate_type":["ble","wifi"],"charge":false,"bws_surplus":0,"device_sw_version":"v1.4.4","has_manual":false}]} + """ + if fromFile: + resp = self._loadFromFile(os.path.join(self._testdir,f"bind_devices.json")) + else: + resp = await self.request("post", _API_ENDPOINTS["bind_devices"]) + return resp.get("data",{}) + + + async def get_user_devices(self, fromFile: bool = False) -> Dict: + """Get device details of all devices owned by user. + Example data: + {'solar_list': [], 'pps_list': [], 'solarbank_list': [{'device_pn': 'A17C0', 'device_sn': '9JVB42LJK8J0P5RY', 'device_name': 'Solarbank E1600', + 'device_img': 'https://public-aiot-fra-prod.s3.dualstack.eu-central-1.amazonaws.com/anker-power/public/product/anker-power/e9478c2d-e665-4d84-95d7-dd4844f82055/20230719-144818.png', + 'battery_power': '', 'bind_site_status': '1', 'charging_power': '', 'power_unit': '', 'charging_status': '', 'status': '', 'wireless_type': '1', 'main_version': '', + 'photovoltaic_power': '', 'output_power': '', 'create_time': 0}]} + """ + if fromFile: + resp = self._loadFromFile(os.path.join(self._testdir,f"user_devices.json")) + else: + resp = await self.request("post", _API_ENDPOINTS["user_devices"]) + return resp.get("data",{}) + + + async def get_charging_devices(self, fromFile: bool = False) -> Dict: + """Get the charging devices (Power stations?) + Example data: + {'device_list': None, 'guide_txt': ''} + """ + if fromFile: + resp = self._loadFromFile(os.path.join(self._testdir,f"charging_devices.json")) + else: + resp = await self.request("post", _API_ENDPOINTS["charging_devices"]) + return resp.get("data",{}) + + + async def get_solar_info(self, siteId: str, fromFile: bool = False) -> Dict: + """Get the solar info, likely requires also Anker Inverters attached) + TODO: Need example output + """ + data = {"site_id": siteId} #TODO: Required data parameters UNKNOWN, only site_id does not work, may need device SN for Anker Inverter? + if fromFile: + resp = self._loadFromFile(os.path.join(self._testdir,f"solar_info_{siteId}.json")) + else: + resp = await self.request("post", _API_ENDPOINTS["solar_info"], json = data) + return resp.get("data",{}) + + + async def get_auto_upgrade(self, fromFile: bool = False) -> Dict: + """Get auto upgrade settings and devices enabled for auto upgrade + Example data: + {'main_switch': True, 'device_list': [{'device_sn': '9JVB42LJK8J0P5RY', 'device_name': 'Solarbank E1600', 'auto_upgrade': True, 'alias_name': 'Solarbank E1600', + 'icon': 'https://public-aiot-fra-prod.s3.dualstack.eu-central-1.amazonaws.com/anker-power/public/product/anker-power/e9478c2d-e665-4d84-95d7-dd4844f82055/20230719-144818.png'}]} + """ + if fromFile: + resp = self._loadFromFile(os.path.join(self._testdir,f"auto_upgrade.json")) + else: + resp = await self.request("post", _API_ENDPOINTS["get_auto_upgrade"]) + return resp.get("data",{}) + + + async def get_wifi_list(self, siteId: str, fromFile: bool = False) -> Dict: + """Get the wifi list. + Example data: + {'wifi_info_list': [{'wifi_name': '1und1-HomeServer', 'wifi_signal': '100'}]} + """ + data = {"site_id": siteId} + if fromFile: + resp = self._loadFromFile(os.path.join(self._testdir,f"wifi_list_{siteId}.json")) + else: + resp = await self.request("post", _API_ENDPOINTS["wifi_list"], json = data) + return resp.get("data",{}) + + + async def get_power_cutoff(self, siteId: str, deviceSn: str, fromFile: bool = False) -> Dict: + """Get power cut off settings. + Example data: + {'power_cutoff_data': [ + {'id': 1, 'is_selected': 1, 'output_cutoff_data': 10, 'lowpower_input_data': 5, 'input_cutoff_data': 10}, + {'id': 2, 'is_selected': 0, 'output_cutoff_data': 5, 'lowpower_input_data': 4, 'input_cutoff_data': 5}]} + """ + data = {"site_id": siteId, "device_sn": deviceSn} + if fromFile: + resp = self._loadFromFile(os.path.join(self._testdir,f"power_cutoff_{deviceSn}.json")) + else: + resp = await self.request("post", _API_ENDPOINTS["get_cutoff"], json = data) + return resp.get("data",{}) + + + async def set_power_cutoff(self, siteId: str, deviceSn: str, setId: int , toFile: bool = False) -> Dict: + """Set power cut off settings. + TODO: This still must be validated. + """ + data = {"site_id": siteId, "device_sn": deviceSn, "id": setId} #TODO: No idea which parameters to pass to select or pass whole list? + if toFile: + resp = self._saveToFile(os.path.join(self._testdir,f"set_power_cutoff_{deviceSn}.json"), json = data) + else: + resp = await self.request("post", _API_ENDPOINTS["set_cutoff"], json = data) + return resp.get("data",{}) + + + async def get_device_load(self, siteId: str, deviceSn: str, fromFile: bool = False) -> Dict: + """Get device load settings. + Example data: + {"site_id": "efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c", + "home_load_data": "{\"ranges\":[{\"id\":0,\"start_time\":\"00:00\",\"end_time\":\"08:30\",\"turn_on\":true,\"appliance_loads\":[{\"id\":0,\"name\":\"Benutzerdefiniert\",\"power\":300,\"number\":1}],\"charge_priority\":80},{\"id\":0,\"start_time\":\"08:30\",\"end_time\":\"17:00\",\"turn_on\":false,\"appliance_loads\":[{\"id\":0,\"name\":\"Benutzerdefiniert\",\"power\":100,\"number\":1}],\"charge_priority\":80},{\"id\":0,\"start_time\":\"17:00\",\"end_time\":\"24:00\",\"turn_on\":true,\"appliance_loads\":[{\"id\":0,\"name\":\"Benutzerdefiniert\",\"power\":300,\"number\":1}],\"charge_priority\":0}],\"min_load\":100,\"max_load\":800,\"step\":0,\"is_charge_priority\":0,\"default_charge_priority\":0,\"is_zero_output_tips\":1}", + "current_home_load": "300W","parallel_home_load": "","parallel_display": false} + """ + data = {"site_id": siteId, "device_sn": deviceSn} + if fromFile: + resp = self._loadFromFile(os.path.join(self._testdir,f"device_load_{deviceSn}.json")) + else: + resp = await self.request("post", _API_ENDPOINTS["get_device_load"], json = data) + """ API Bug? home_load_data provided as string instead of object...Convert into object for proper handling """ + string_data = (resp.get("data",{})).get("home_load_data",{}) + if isinstance(string_data, str): + resp["data"].update({"home_load_data": json.loads(string_data)}) + return resp.get("data",{}) + + + async def get_device_parm(self, siteId: str, paramType: str = "4", fromFile: bool = False) -> Dict: + """Get device parameters (e.g. solarbank schedule). This can be queried for each siteId listed in the homepage info site_list. The paramType is always 4, but can be modified if necessary + Example data for provided site_id: + {"param_data": "{\"ranges\":[ + {\"id\":0,\"start_time\":\"00:00\",\"end_time\":\"08:30\",\"turn_on\":true,\"appliance_loads\":[{\"id\":0,\"name\":\"Benutzerdefiniert\",\"power\":300,\"number\":1}],\"charge_priority\":80}, + {\"id\":0,\"start_time\":\"08:30\",\"end_time\":\"17:00\",\"turn_on\":false,\"appliance_loads\":[{\"id\":0,\"name\":\"Benutzerdefiniert\",\"power\":100,\"number\":1}],\"charge_priority\":80}, + {\"id\":0,\"start_time\":\"17:00\",\"end_time\":\"24:00\",\"turn_on\":true,\"appliance_loads\":[{\"id\":0,\"name\":\"Benutzerdefiniert\",\"power\":300,\"number\":1}],\"charge_priority\":0}], + \"min_load\":100,\"max_load\":800,\"step\":0,\"is_charge_priority\":0,\"default_charge_priority\":0,\"is_zero_output_tips\":1}"} + """ + data = {"site_id": siteId, "param_type": paramType} + if fromFile: + resp = self._loadFromFile(os.path.join(self._testdir,f"device_parm_{siteId}.json")) + else: + resp = await self.request("post", _API_ENDPOINTS["get_device_parm"], json = data) + """ API Bug? param_data provided as string instead of object...Convert into object for proper handling """ + string_data = (resp.get("data",{})).get("param_data",{}) + if isinstance(string_data, str): + resp["data"].update({"param_data": json.loads(string_data)}) + return resp.get("data",{}) + + + async def set_device_parm(self, siteId: str, paramData: dict, paramType: str = "4", command: int = 17, toFile: bool = False) -> dict: + """Set device parameters (e.g. solarbank schedule). + command: Must be 17 for solarbank schedule. + paramType: was always string "4" + Example paramData: + {"param_data":{"ranges":[ + {"id":0,"start_time":"00:00","end_time":"08:30","turn_on":true,"appliance_loads":[{"id":0,"name":"Benutzerdefiniert","power":300,"number":1}],"charge_priority":80}, + {"id":0,"start_time":"08:30","end_time":"17:00","turn_on":false,"appliance_loads":[{"id":0,"name":"Benutzerdefiniert","power":100,"number":1}],"charge_priority":80}, + {"id":0,"start_time":"17:00","end_time":"24:00","turn_on":true,"appliance_loads":[{"id":0,"name":"Benutzerdefiniert","power":300,"number":1}],"charge_priority":0}], + "min_load":100,"max_load":800,"step":0,"is_charge_priority":0,default_charge_priority":0}} + """ + data = {"site_id": siteId, "param_type": paramType, "cmd": command, "param_data": json.dumps(paramData)} + if toFile: + resp = self._saveToFile(os.path.join(self._testdir,f"set_device_parm_{siteId}.json"), json = data) + else: + resp = await self.request("post", _API_ENDPOINTS["set_device_parm"], json = data) + return resp.get("data",{}) + + + async def energy_analysis(self, siteId: str, deviceSn: str, rangeType: str = None, startDay: datetime = None, endDay: datetime = None, devType: str = None) -> dict: + """Fetch Energy data for given device and optional time frame. + siteId: site ID of device + deviceSn: Device to fetch data + rangeType: "day" | "week" | "year" + startTime: optional start Date and time + endTime: optional end Date and time + devType: "solar_production" | "solarbank" + Example Data for solar_production: + {'power': [{'time': '2023-10-01', 'value': '3.67'}, {'time': '2023-10-02', 'value': '3.29'}, {'time': '2023-10-03', 'value': '0.55'}], + 'charge_trend': None, 'charge_level': [], 'power_unit': 'wh', 'charge_total': '3.67', 'charge_unit': 'kwh', 'discharge_total': '3.11', 'discharge_unit': 'kwh', + 'charging_pre': '0.49', 'electricity_pre': '0.51', 'others_pre': '0', + 'statistics': [{'type': '1', 'total': '7.51', 'unit': 'kwh'}, {'type': '2', 'total': '7.49', 'unit': 'kg'}, {'type': '3', 'total': '3.00', 'unit': '€'}]} + """ + data = { + "site_id": siteId, + "device_sn": deviceSn, + "type": rangeType if rangeType in ["day", "week", "year"] else "day", + "start_time": startDay.strftime("%Y-%m-%d") if startDay else datetime.today().strftime("%Y-%m-%d"), + "device_type": devType if devType in ["solar_production","solarbank"] else "solar_production", + "end_time": endDay.strftime("%Y-%m-%d") if endDay else "", + } + resp = await self.request("post", _API_ENDPOINTS["energy_analysis"], json = data) + return resp.get("data",{}) + + + async def energy_daily(self, siteId: str, deviceSn: str, startDay: datetime = datetime.today(), numDays: int = 1, dayTotals: bool = False) -> dict: + """Fetch daily Energy data for given interval and provide it in a table format dictionary. + Example: + {"2023-09-29": {"date": "2023-09-29", "solar_production": "1.21", "solarbank_discharge": "0.47", "solarbank_charge": "0.56"}, + "2023-09-30": {"date": "2023-09-30", "solar_production": "3.07", "solarbank_discharge": "1.06", "solarbank_charge": "1.39"}} + """ + table = {} + today = datetime.today() + """check daily range and limit to 1 year max and avoid future days""" + if startDay > today: + startDay = today + numDays = 1 + elif (startDay + timedelta(days=numDays)) > today: + numDays = (today-startDay).days + 1 + numDays = min(366,max(1,numDays)) + """first get solar production""" + resp = await self.energy_analysis(siteId=siteId, deviceSn=deviceSn, rangeType = "week", startDay=startDay, endDay=startDay+timedelta(days=numDays-1), devType="solar_production") + for item in resp.get("power",[]): + daystr = item.get("time", None) + if daystr: + table.update({daystr: {"date": daystr, "solar_production": item.get("value", "")}}) + """Add solarbank discharge""" + resp = await self.energy_analysis(siteId=siteId, deviceSn=deviceSn, rangeType = "week", startDay=startDay, endDay=startDay+timedelta(days=numDays-1), devType="solarbank") + for item in resp.get("power",[]): + daystr = item.get("time", None) + if daystr: + entry = table.get(daystr,{}) + entry.update({"date": daystr, "solarbank_discharge": item.get("value", "")}) + table.update({daystr: entry}) + """Solarbank charge is only received as total value for given interval. If requested, make daily queries for given interval with some delay""" + if dayTotals: + if numDays == 1: + daystr = startDay.strftime("%Y-%m-%d") + entry = table.get(daystr,{}) + entry.update({"date": daystr, "solarbank_charge": resp.get("charge_total", "")}) + table.update({daystr: entry}) + else: + daylist = [startDay + timedelta(days=x) for x in range(numDays)] + for day in daylist: + daystr = day.strftime("%Y-%m-%d") + if day != daylist[0]: + time.sleep(1) # delay to avoid hammering API + resp = await self.energy_analysis(siteId=siteId, deviceSn=deviceSn, rangeType = "week", startDay=day, endDay=day, devType="solarbank") + entry = table.get(daystr,{}) + entry.update({"date": daystr, "solarbank_charge": resp.get("charge_total", "")}) + table.update({daystr: entry}) + return table + + + async def home_load_chart(self, siteId: str, deviceSn: str = None) -> Dict: + """Get home load chart data. + Example data: + {"data": []} + """ + data = {"site_id": siteId} + if deviceSn: + data.update({"device_sn": deviceSn}) + resp = await self.request("post", _API_ENDPOINTS["home_load_chart"], json = data) + return resp.get("data",{}) + + diff --git a/api/errors.py b/api/errors.py new file mode 100644 index 0000000..4ab2ca0 --- /dev/null +++ b/api/errors.py @@ -0,0 +1,96 @@ +"""Define package errors.""" +from typing import Dict, Type + + +class AnkerSolixError(Exception): + """Define a base error.""" + pass + +class AuthorizationError(AnkerSolixError): + """Authorization error.""" + pass + +class ConnectError(AnkerSolixError): + """Connection error.""" + pass + +class NetworkError(AnkerSolixError): + """Network error.""" + pass + +class ServerError(AnkerSolixError): + """Server error.""" + pass + +class RequestError(AnkerSolixError): + """Request error.""" + pass + +class VerifyCodeError(AnkerSolixError): + """Verify code error.""" + pass + +class VerifyCodeExpiredError(AnkerSolixError): + """Verification code has expired.""" + pass + +class NeedVerifyCodeError(AnkerSolixError): + """Need verification code error.""" + pass + +class VerifyCodeMaxError(AnkerSolixError): + """Maximum attempts of verications error.""" + pass + +class VerifyCodeNoneMatchError(AnkerSolixError): + """Verify code none match error.""" + pass + +class VerifyCodePasswordError(AnkerSolixError): + """Verify code password error.""" + pass + +class ClientPublicKeyError(AnkerSolixError): + """Define an error for client public key error.""" + pass + +class TokenKickedOutError(AnkerSolixError): + """Define an error for token does not exist because it was kicked out.""" + pass + +class InvalidCredentialsError(AnkerSolixError): + """Define an error for unauthenticated accounts.""" + pass + +class RetryExceeded(AnkerSolixError): + """Define an error for exceeded retry attempts. Please try again in 24 hours.""" + pass + +ERRORS: Dict[int, Type[AnkerSolixError]] = { + 401: AuthorizationError, + 997: ConnectError, + 998: NetworkError, + 999: ServerError, + 10000: RequestError, + 10003: RequestError, + 10007: RequestError, + 26050: VerifyCodeError, + 26051: VerifyCodeExpiredError, + 26052: NeedVerifyCodeError, + 26053: VerifyCodeMaxError, + 26054: VerifyCodeNoneMatchError, + 26055: VerifyCodePasswordError, + 26070: ClientPublicKeyError, + 26084: TokenKickedOutError, + 26108: InvalidCredentialsError, + 100053: RetryExceeded, +} + + +def raise_error(data: dict) -> None: + """Raise the appropriate error based upon a response code.""" + code = data.get("code", -1) + if code in [0]: + return + cls = ERRORS.get(code, AnkerSolixError) + raise cls(f'({code}) {data.get("msg","Error msg not found")}') diff --git a/energy_csv.py b/energy_csv.py new file mode 100644 index 0000000..b389adf --- /dev/null +++ b/energy_csv.py @@ -0,0 +1,106 @@ +""" +Example exec module to use the Anker API for export of daily Solarbank Energy Data. +This method will prompt for the Anker account details if not pre-set in the header. +Then you can specify a start day and the number of days for data extraction from the Anker Cloud. +Note: The Solar production and Solarbank discharge can be queried across the full range. The solarbank +charge however can be queried only as total for an interval (e.g. day). Therefore when solarbank charge +data is also selected for export, an additional API query per day is required. +The received daily values will be exported into a csv file. +""" + +import asyncio +from aiohttp import ClientSession +from datetime import datetime +from api import api +from getpass import getpass +import json, logging, sys, csv + +_LOGGER: logging.Logger = logging.getLogger(__name__) +_LOGGER.addHandler(logging.StreamHandler(sys.stdout)) +#_LOGGER.setLevel(logging.DEBUG) # enable for debug output + +# Optional default Anker Account credentials to be used +USER = "" +PASSWORD = "" +COUNTRY = "" + + +async def main() -> None: + global USER, PASSWORD, COUNTRY + print("Exporting daily Energy data for Anker Solarbank:") + if USER == "": + print("\nEnter Anker Account credentials:") + USER = input("Username (email): ") + if USER == "": + return False + PASSWORD = getpass("Password: ") + if PASSWORD == "": + return False + COUNTRY = input("Country ID (e.g. DE): ") + if COUNTRY == "": + return False + try: + async with ClientSession() as websession: + print("\nTrying authentication...",end="") + myapi = api.API(USER,PASSWORD,COUNTRY,websession, _LOGGER) + if await myapi.async_authenticate(): + print("OK") + else: + print("CACHED") # Login validation will be done during first API call + # Refresh the site and device info of the API + print("\nUpdating Site Info...", end="") + if (await myapi.update_sites()) == {}: + print("NO INFO") + return False + print("OK") + print(f"\nDevices: {len(myapi.devices)}") + _LOGGER.debug(json.dumps(myapi.devices, indent=2)) + + for sn, device in myapi.devices.items(): + if device.get("type") == "solarbank": + print(f"Found {device.get('name')} SN: {sn}") + try: + daystr = input("\nEnter start day for daily energy data (yyyy-mm-dd) or enter to skip: ") + if daystr == "": + print(f"Skipped SN: {sn}, checking for next Solarbank...") + continue + startday = datetime.fromisoformat(daystr) + numdays = int(input("How many days to query (1-366): ")) + daytotals = input("Do you want to include daily total data (e.g. solarbank charge) which require API query per day? (Y/N): ") + daytotals = daytotals.upper() in ["Y","YES","TRUE",1] + filename = input(f"CSV filename for export (daily_energy_{daystr}.csv): ") + if filename == "": + filename = f"daily_energy_{daystr}.csv" + except ValueError: + return False + print(f"Queries may take up to {numdays*daytotals + 2} seconds...please wait...") + data = await myapi.energy_daily(siteId=device.get("site_id"),deviceSn=sn,startDay=startday,numDays=numdays,dayTotals=daytotals) + _LOGGER.debug(json.dumps(data, indent=2)) + # Write csv file + if len(data) > 0: + with open(filename, 'w', newline='') as csvfile: + fieldnames = (next(iter(data.values()))).keys() + writer = csv.DictWriter(csvfile, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(data.values()) + print(f"\nCompleted: Successfully exported data to {filename}") + return True + + print("No data received for device") + return False + print("No accepted Solarbank device found.") + return False + + except Exception as err: + print(f'{type(err)}: {err}') + return False + + +"""run async main""" +if __name__ == '__main__': + try: + if not asyncio.run(main()): + print("Aborted!") + except Exception as err: + print(f'{type(err)}: {err}') + diff --git a/examples/auto_upgrade.json b/examples/auto_upgrade.json new file mode 100644 index 0000000..5557b47 --- /dev/null +++ b/examples/auto_upgrade.json @@ -0,0 +1,17 @@ +{ + "code": 0, + "msg": "success!", + "data": { + "main_switch": true, + "device_list": [ + { + "device_sn": "9JVB42LJK8J0P5RY", + "device_name": "Solarbank E1600", + "auto_upgrade": true, + "alias_name": "Solarbank E1600", + "icon": "https://public-aiot-fra-prod.s3.dualstack.eu-central-1.amazonaws.com/anker-power/public/product/anker-power/e9478c2d-e665-4d84-95d7-dd4844f82055/20230719-144818.png" + } + ] + }, + "trace_id": "f7e8b441db711a3daf0c2f56fdd5d657" +} \ No newline at end of file diff --git a/examples/bind_devices.json b/examples/bind_devices.json new file mode 100644 index 0000000..1b180f4 --- /dev/null +++ b/examples/bind_devices.json @@ -0,0 +1,30 @@ +{ + "code": 0, + "msg": "success!", + "data": { + "data": [ + { + "device_sn": "9JVB42LJK8J0P5RY", + "product_code": "A17C0", + "bt_ble_id": "BC:A2:AF:C7:55:F9", + "bt_ble_mac": "BCA2AFC755F9", + "device_name": "Solarbank E1600", + "alias_name": "Solarbank E1600", + "img_url": "https://public-aiot-fra-prod.s3.dualstack.eu-central-1.amazonaws.com/anker-power/public/product/anker-power/e9478c2d-e665-4d84-95d7-dd4844f82055/20230719-144818.png", + "link_time": 1695392302068, + "wifi_online": false, + "wifi_name": "", + "relate_type": [ + "ble", + "wifi" + ], + "charge": false, + "bws_surplus": 0, + "device_sw_version": "v1.4.4", + "has_manual": false, + "hes_data": null + } + ] + }, + "trace_id": "1badaafff0da3ca9fbebebc5f0011788" +} \ No newline at end of file diff --git a/examples/charging_devices.json b/examples/charging_devices.json new file mode 100644 index 0000000..0d65b5a --- /dev/null +++ b/examples/charging_devices.json @@ -0,0 +1,9 @@ +{ + "code": 0, + "msg": "success!", + "data": { + "device_list": null, + "guide_txt": "" + }, + "trace_id": "ddac5ef6d15d6a79790f3614e5dbb40e" +} \ No newline at end of file diff --git a/examples/device_fittings_9JVB42LJK8J0P5RY.json b/examples/device_fittings_9JVB42LJK8J0P5RY.json new file mode 100644 index 0000000..fcd3aa9 --- /dev/null +++ b/examples/device_fittings_9JVB42LJK8J0P5RY.json @@ -0,0 +1,8 @@ +{ + "code": 0, + "msg": "success!", + "data": { + "data": [] + }, + "trace_id": "806ef1c3fab6e90d30feeee4fbddeb33" +} \ No newline at end of file diff --git a/examples/device_load_9JVB42LJK8J0P5RY.json b/examples/device_load_9JVB42LJK8J0P5RY.json new file mode 100644 index 0000000..b89d775 --- /dev/null +++ b/examples/device_load_9JVB42LJK8J0P5RY.json @@ -0,0 +1,12 @@ +{ + "code": 0, + "msg": "success!", + "data": { + "site_id": "efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c", + "home_load_data": "{\"ranges\":[{\"id\":0,\"start_time\":\"00:00\",\"end_time\":\"08:30\",\"turn_on\":true,\"appliance_loads\":[{\"id\":0,\"name\":\"Benutzerdefiniert\",\"power\":300,\"number\":1}],\"charge_priority\":80},{\"id\":0,\"start_time\":\"08:30\",\"end_time\":\"17:00\",\"turn_on\":false,\"appliance_loads\":[{\"id\":0,\"name\":\"Benutzerdefiniert\",\"power\":100,\"number\":1}],\"charge_priority\":80},{\"id\":0,\"start_time\":\"17:00\",\"end_time\":\"24:00\",\"turn_on\":true,\"appliance_loads\":[{\"id\":0,\"name\":\"Benutzerdefiniert\",\"power\":300,\"number\":1}],\"charge_priority\":0}],\"min_load\":100,\"max_load\":800,\"step\":0,\"is_charge_priority\":0,\"default_charge_priority\":0,\"is_zero_output_tips\":1}", + "current_home_load": "300W", + "parallel_home_load": "", + "parallel_display": false + }, + "trace_id": "f1adf477a5ef9ef0bb095bfe5d3bdd1e" +} \ No newline at end of file diff --git a/examples/device_parm_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json b/examples/device_parm_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json new file mode 100644 index 0000000..d1bb76d --- /dev/null +++ b/examples/device_parm_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json @@ -0,0 +1,8 @@ +{ + "code": 0, + "msg": "success!", + "data": { + "param_data": "{\"ranges\":[{\"id\":0,\"start_time\":\"00:00\",\"end_time\":\"08:30\",\"turn_on\":true,\"appliance_loads\":[{\"id\":0,\"name\":\"Benutzerdefiniert\",\"power\":300,\"number\":1}],\"charge_priority\":80},{\"id\":0,\"start_time\":\"08:30\",\"end_time\":\"17:00\",\"turn_on\":false,\"appliance_loads\":[{\"id\":0,\"name\":\"Benutzerdefiniert\",\"power\":100,\"number\":1}],\"charge_priority\":80},{\"id\":0,\"start_time\":\"17:00\",\"end_time\":\"24:00\",\"turn_on\":true,\"appliance_loads\":[{\"id\":0,\"name\":\"Benutzerdefiniert\",\"power\":300,\"number\":1}],\"charge_priority\":0}],\"min_load\":100,\"max_load\":800,\"step\":0,\"is_charge_priority\":0,\"default_charge_priority\":0,\"is_zero_output_tips\":1}" + }, + "trace_id": "cef9143cbda7ab7faeefcdd58df903c0" +} \ No newline at end of file diff --git a/examples/homepage.json b/examples/homepage.json new file mode 100644 index 0000000..c4ad2c5 --- /dev/null +++ b/examples/homepage.json @@ -0,0 +1,43 @@ +{ + "code": 0, + "msg": "success!", + "data": { + "site_list": [ + { + "site_id": "efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c", + "site_name": "BKW", + "site_img": "", + "device_type_list": [ + 3 + ], + "ms_type": 0, + "power_site_type": 0, + "is_allow_delete": false + } + ], + "solar_list": [], + "pps_list": [], + "solarbank_list": [ + { + "device_pn": "", + "device_sn": "9JVB42LJK8J0P5RY", + "device_name": "Solarbank E1600", + "device_img": "https://public-aiot-fra-prod.s3.dualstack.eu-central-1.amazonaws.com/anker-power/public/product/anker-power/e9478c2d-e665-4d84-95d7-dd4844f82055/20230719-144818.png", + "battery_power": "75", + "bind_site_status": "1", + "charging_power": "", + "power_unit": "", + "charging_status": "", + "status": "", + "wireless_type": "", + "main_version": "", + "photovoltaic_power": "", + "output_power": "", + "create_time": 0, + "set_load_power": "" + } + ], + "powerpanel_list": [] + }, + "trace_id": "224d9e6f6b05aca200062cebb58a6deb" +} \ No newline at end of file diff --git a/examples/power_cutoff_9JVB42LJK8J0P5RY.json b/examples/power_cutoff_9JVB42LJK8J0P5RY.json new file mode 100644 index 0000000..d55241b --- /dev/null +++ b/examples/power_cutoff_9JVB42LJK8J0P5RY.json @@ -0,0 +1,23 @@ +{ + "code": 0, + "msg": "success!", + "data": { + "power_cutoff_data": [ + { + "id": 1, + "is_selected": 1, + "output_cutoff_data": 10, + "lowpower_input_data": 5, + "input_cutoff_data": 10 + }, + { + "id": 2, + "is_selected": 0, + "output_cutoff_data": 5, + "lowpower_input_data": 4, + "input_cutoff_data": 5 + } + ] + }, + "trace_id": "39e58be56087ee5b4f4003b6efe90be2" +} \ No newline at end of file diff --git a/examples/price_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json b/examples/price_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json new file mode 100644 index 0000000..9f35ded --- /dev/null +++ b/examples/price_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json @@ -0,0 +1,11 @@ +{ + "code": 0, + "msg": "success!", + "data": { + "site_id": "efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c", + "price": 0.4, + "site_co2": 0, + "site_price_unit": "\u20ac" + }, + "trace_id": "9ab93ef8eeab3b25687368cb51364e10" +} \ No newline at end of file diff --git a/examples/scene_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json b/examples/scene_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json new file mode 100644 index 0000000..6a6b8ae --- /dev/null +++ b/examples/scene_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json @@ -0,0 +1,75 @@ +{ + "code": 0, + "msg": "success!", + "data": { + "home_info": { + "home_name": "Home", + "home_img": "", + "charging_power": "0.00", + "power_unit": "W" + }, + "solar_list": [], + "pps_info": { + "pps_list": [], + "total_charging_power": "0.00", + "power_unit": "W", + "total_battery_power": "0.00", + "updated_time": "", + "pps_status": 0 + }, + "statistics": [ + { + "type": "1", + "total": "89.75", + "unit": "kwh" + }, + { + "type": "2", + "total": "89.48", + "unit": "kg" + }, + { + "type": "3", + "total": "35.90", + "unit": "\u20ac" + } + ], + "topology_type": "1", + "solarbank_info": { + "solarbank_list": [ + { + "device_pn": "A17C0", + "device_sn": "9JVB42LJK8J0P5RY", + "device_name": "Solarbank E1600", + "device_img": "https://public-aiot-fra-prod.s3.dualstack.eu-central-1.amazonaws.com/anker-power/public/product/anker-power/e9478c2d-e665-4d84-95d7-dd4844f82055/20230719-144818.png", + "battery_power": "75", + "bind_site_status": "", + "charging_power": "0", + "power_unit": "W", + "charging_status": "2", + "status": "0", + "wireless_type": "1", + "main_version": "", + "photovoltaic_power": "0", + "output_power": "0", + "create_time": 1695392386, + "set_load_power": "" + } + ], + "total_charging_power": "0", + "power_unit": "W", + "charging_status": "0", + "total_battery_power": "0.00", + "updated_time": "2023-12-28 18:53:27", + "total_photovoltaic_power": "0", + "total_output_power": "0.00", + "display_set_power": false + }, + "retain_load": "300W", + "updated_time": "01-01-0001 00:00:00", + "power_site_type": 2, + "site_id": "efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c", + "powerpanel_list": [] + }, + "trace_id": "d4b0c8fb0dfff5fe716f6d439a4be9a2" +} \ No newline at end of file diff --git a/examples/site_detail_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json b/examples/site_detail_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json new file mode 100644 index 0000000..766b230 --- /dev/null +++ b/examples/site_detail_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json @@ -0,0 +1,39 @@ +{ + "code": 0, + "msg": "success!", + "data": { + "site_info": { + "site_id": "efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c", + "site_name": "BKW", + "site_img": "", + "device_type_list": null, + "ms_type": 0, + "power_site_type": 0, + "is_allow_delete": false + }, + "solar_list": [], + "pps_list": [], + "solarbank_list": [ + { + "device_pn": "A17C0", + "device_sn": "9JVB42LJK8J0P5RY", + "device_name": "Solarbank E1600", + "device_img": "https://public-aiot-fra-prod.s3.dualstack.eu-central-1.amazonaws.com/anker-power/public/product/anker-power/e9478c2d-e665-4d84-95d7-dd4844f82055/20230719-144818.png", + "battery_power": "", + "bind_site_status": "", + "charging_power": "", + "power_unit": "", + "charging_status": "", + "status": "", + "wireless_type": "", + "main_version": "", + "photovoltaic_power": "", + "output_power": "", + "create_time": 0, + "set_load_power": "" + } + ], + "powerpanel_list": [] + }, + "trace_id": "d4f3ecfae6a7a085e5ea93feeadb3ced" +} \ No newline at end of file diff --git a/examples/site_list.json b/examples/site_list.json new file mode 100644 index 0000000..c16cf4f --- /dev/null +++ b/examples/site_list.json @@ -0,0 +1,20 @@ +{ + "code": 0, + "msg": "success!", + "data": { + "site_list": [ + { + "site_id": "efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c", + "site_name": "BKW", + "site_img": "", + "device_type_list": [ + 3 + ], + "ms_type": 1, + "power_site_type": 2, + "is_allow_delete": true + } + ] + }, + "trace_id": "b7cbcdb5ddc7dccdb73f87fdeea97a98" +} \ No newline at end of file diff --git a/examples/user_devices.json b/examples/user_devices.json new file mode 100644 index 0000000..304d177 --- /dev/null +++ b/examples/user_devices.json @@ -0,0 +1,29 @@ +{ + "code": 0, + "msg": "success!", + "data": { + "solar_list": [], + "pps_list": [], + "solarbank_list": [ + { + "device_pn": "A17C0", + "device_sn": "9JVB42LJK8J0P5RY", + "device_name": "Solarbank E1600", + "device_img": "https://public-aiot-fra-prod.s3.dualstack.eu-central-1.amazonaws.com/anker-power/public/product/anker-power/e9478c2d-e665-4d84-95d7-dd4844f82055/20230719-144818.png", + "battery_power": "", + "bind_site_status": "1", + "charging_power": "", + "power_unit": "", + "charging_status": "", + "status": "0", + "wireless_type": "1", + "main_version": "", + "photovoltaic_power": "", + "output_power": "", + "create_time": 0, + "set_load_power": "" + } + ] + }, + "trace_id": "bca735ddc8cdab6fa89bd2d4de31ab8f" +} \ No newline at end of file diff --git a/examples/wifi_list_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json b/examples/wifi_list_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json new file mode 100644 index 0000000..ce8c3d3 --- /dev/null +++ b/examples/wifi_list_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json @@ -0,0 +1,13 @@ +{ + "code": 0, + "msg": "success!", + "data": { + "wifi_info_list": [ + { + "wifi_name": "1und1-HomeServer", + "wifi_signal": "100" + } + ] + }, + "trace_id": "1fb13293474e50d24c4d6d93bd6d2bbe" +} \ No newline at end of file diff --git a/export_system.py b/export_system.py new file mode 100644 index 0000000..ca8b15c --- /dev/null +++ b/export_system.py @@ -0,0 +1,226 @@ +""" +Example exec module to use the Anker API for export of defined system data and device details. +This module will prompt for the Anker account details if not pre-set in the header. +Upon successfull authentication, you can specify a subfolder for the exported JSON files received as API query response, defaulting to your nick name +Optionally you can specify whether personalized information in the response data should be randomized in the files, like SNs, Site IDs, Trace IDs etc. +You can review the response files afterwards. They can be used as examples for dedicated data extraction from the devices. +Optionally the API class can use the json files for debugging and testing on various system outputs. +""" + +import asyncio +from aiohttp import ClientSession +from datetime import datetime +from getpass import getpass +from contextlib import suppress +import json, logging, sys, csv, os, time, string, random +from api import api + +_LOGGER: logging.Logger = logging.getLogger(__name__) +_LOGGER.addHandler(logging.StreamHandler(sys.stdout)) +#_LOGGER.setLevel(logging.DEBUG) # enable for debug output + +# Optional default Anker Account credentials to be used +USER = "" +PASSWORD = "" +COUNTRY = "" + +RANDOMIZE = True # Global flag to save randomize decission +RANDOMDATA = {} # Global dict for randomized data, printed at the end + + +def randomize(val, key: str = "") -> str: + """Randomize a given string while maintaining its format if format is known for given key name. + Reuse same randomization if value was already randomized""" + global RANDOMDATA + if not RANDOMIZE: + return str(val) + randomstr = RANDOMDATA.get(val,"") + if not randomstr: + if "_sn" in key: + randomstr = "".join(random.choices(string.ascii_uppercase+string.digits, k=len(val))) + elif "bt_ble_" in key: + """Handle values with and without : """ + temp = val.replace(":","") + randomstr = RANDOMDATA.get(temp) # retry existing randomized value without : + if not randomstr: + randomstr = "".join(random.choices(string.hexdigits.upper(), k=len(temp))) + if ":" in val: + RANDOMDATA.update({temp: randomstr}) # save also key value without : + randomstr = ':'.join(a+b for a,b in zip(randomstr[::2], randomstr[1::2])) + elif "_id" in key: + for part in val.split("-"): + if randomstr: + randomstr = "-".join([randomstr,"".join(random.choices(string.hexdigits.lower(), k=len(part)))]) + else: + randomstr = "".join(random.choices(string.hexdigits.lower(), k=len(part))) + else: + # default randomize format + randomstr = "".join(random.choices(string.ascii_letters, k=len(val))) + RANDOMDATA.update({val: randomstr}) + return randomstr + + +def check_keys(data): + """Recursive traversal of complex nested objects to randomize value for certain keys""" + if isinstance(data, str) or isinstance(data, int): + return data + for k, v in data.copy().items(): + if isinstance(v, dict): + v = check_keys(v) + if isinstance(v, list): + v = [check_keys(i) for i in v] + """Randomize value for certain keys""" + if any(x in k for x in ["_sn","site_id","trace_id","bt_ble_"]): + data[k] = randomize(v,k) + return data + + +def export(filename: str, d: dict = {}) -> None: + """Save dict data to given file""" + time.sleep(1) # central delay between multiple requests + if len(d) == 0: + print(f"WARNING: File {filename} not saved because JSON is empty") + return + elif RANDOMIZE: + d = check_keys(d) + try: + with open(filename, 'w') as file: + json.dump(d, file, indent=2) + print(f"Saved JSON to file {filename}") + except Exception as err: + print(f"ERROR: Failed to save JSON to file {filename}") + return + + +async def main() -> None: + global USER, PASSWORD, COUNTRY, RANDOMIZE + print("Exporting found Anker Solix system data for all assigned sites:") + if USER == "": + print("\nEnter Anker Account credentials:") + USER = input("Username (email): ") + if USER == "": + return False + PASSWORD = getpass("Password: ") + if PASSWORD == "": + return False + COUNTRY = input("Country ID (e.g. DE): ") + if COUNTRY == "": + return False + try: + async with ClientSession() as websession: + print("\nTrying authentication...",end="") + myapi = api.API(USER,PASSWORD,COUNTRY,websession, _LOGGER) + if await myapi.async_authenticate(): + print("OK") + else: + print("CACHED") # Login validation will be done during first API call + + random = input(f"\nDo you want to randomize unique IDs and SNs in exported files? (default: {'YES' if RANDOMIZE else 'NO'}) (Y/N): ") + if random != "" or not isinstance(RANDOMIZE,bool): + RANDOMIZE = random.upper() in ["Y","YES","TRUE",1] + nickname = myapi.nickname.replace("*","#") # avoid filesystem problems with * in user nicknames + folder = input(f"Subfolder for export (default: {nickname}): ") + if folder == "": + if nickname == "": + return False + else: + folder = nickname + os.makedirs(folder, exist_ok=True) + + # first update sites in API object + print("\nQuerying site information...") + await myapi.update_sites() + print(f"Sites: {len(myapi.sites)}, Devices: {len(myapi.devices)}") + _LOGGER.debug(json.dumps(myapi.devices, indent=2)) + + # Query API using direct endpoints to save full response of each query in json files + print("\nExporting homepage...") + export(os.path.join(folder,f"homepage.json"), await myapi.request("post", api._API_ENDPOINTS["homepage"],json={})) + print("Exporting site list...") + export(os.path.join(folder,f"site_list.json"), await myapi.request("post", api._API_ENDPOINTS["site_list"],json={})) + print("Exporting bind devices...") + export(os.path.join(folder,f"bind_devices.json"), await myapi.request("post", api._API_ENDPOINTS["bind_devices"],json={})) # shows only owner devices + print("Exporting user devices...") + export(os.path.join(folder,f"user_devices.json"), await myapi.request("post", api._API_ENDPOINTS["user_devices"],json={})) # shows only owner devices + print("Exporting charging devices...") + export(os.path.join(folder,f"charging_devices.json"), await myapi.request("post", api._API_ENDPOINTS["charging_devices"],json={})) # shows only owner devices + print("Exporting auto upgrade settings...") + export(os.path.join(folder,f"auto_upgrade.json"), await myapi.request("post", api._API_ENDPOINTS["get_auto_upgrade"],json={})) # shows only owner devices + for siteId, site in myapi.sites.items(): + print(f"\nExporting site specific data for site {siteId}...") + print("Exporting scene info...") + export(os.path.join(folder,f"scene_{randomize(siteId,'site_id')}.json"), await myapi.request("post", api._API_ENDPOINTS["scene_info"],json={"site_id": siteId})) + print("Exporting solar info...") + with suppress(Exception): + export(os.path.join(folder,f"solar_info_{randomize(siteId,'site_id')}.json"), await myapi.request("post", api._API_ENDPOINTS["solar_info"],json={"site_id": siteId})) # PARAMETERS UNKNOWN, site id not sufficient + print("Exporting site detail...") + admin = site.get("site_admin") + try: + export(os.path.join(folder,f"site_detail_{randomize(siteId,'site_id')}.json"), await myapi.request("post", api._API_ENDPOINTS["site_detail"],json={"site_id": siteId})) + except Exception as err: + if not admin: + print("Query requires account of site owner!") + print("Exporting wifi list...") + try: + export(os.path.join(folder,f"wifi_list_{randomize(siteId,'site_id')}.json"), await myapi.request("post", api._API_ENDPOINTS["wifi_list"],json={"site_id": siteId})) # works only for site owners + except Exception as err: + if not admin: + print("Query requires account of site owner!") + print("Exporting site price...") + try: + export(os.path.join(folder,f"price_{randomize(siteId,'site_id')}.json"), await myapi.request("post", api._API_ENDPOINTS["get_site_price"],json={"site_id": siteId})) # works only for site owners + except Exception as err: + if not admin: + print("Query requires account of site owner!") + print("Exporting device parameter settings...") + try: + export(os.path.join(folder,f"device_parm_{randomize(siteId,'site_id')}.json"), await myapi.request("post", api._API_ENDPOINTS["get_device_parm"],json={"site_id": siteId, "param_type": "4"})) # works only for site owners + except Exception as err: + if not admin: + print("Query requires account of site owner!") + for sn, device in myapi.devices.items(): + print(f"\nExporting device specific data for device {device.get('name','')} SN {sn}...") + siteId = device.get('site_id','') + admin = site.get('is_admin') + print("Exporting power cutoff settings...") + try: + export(os.path.join(folder,f"power_cutoff_{randomize(sn,'_sn')}.json"), await myapi.request("post", api._API_ENDPOINTS["get_cutoff"],json={"site_id": siteId, "device_sn": sn})) # works only for site owners + except Exception as err: + if not admin: + print("Query requires account of site owner!") + print("Exporting fittings...") + try: + export(os.path.join(folder,f"device_fittings_{randomize(sn,'_sn')}.json"), await myapi.request("post", api._API_ENDPOINTS["get_device_fittings"],json={"site_id": siteId, "device_sn": sn})) # works only for site owners + except Exception as err: + if not admin: + print("Query requires account of site owner!") + print("Exporting load...") + try: + export(os.path.join(folder,f"device_load_{randomize(sn,'_sn')}.json"), await myapi.request("post", api._API_ENDPOINTS["get_device_load"],json={"site_id": siteId, "device_sn": sn})) # works only for site owners + except Exception as err: + if not admin: + print("Query requires account of site owner!") + + print(f"\nCompleted export of Anker Solix system data for user {USER}") + if RANDOMIZE: + print(f"Folder {os.path.abspath(folder)} contains the randomized JSON files. Pls check and update fields that may contain unrecognized personalized data.") + print(f"Following trace or site IDs, SNs and MAC addresses have been randomized in files (from -> to):") + print(json.dumps(RANDOMDATA, indent=2)) + else: + print(f"Folder {os.path.abspath(folder)} contains the JSON files.") + return True + + except Exception as err: + print(f'{type(err)}: {err}') + return False + + +"""run async main""" +if __name__ == '__main__': + try: + if not asyncio.run(main()): + print("Aborted!") + except KeyboardInterrupt: + print("Aborted!") + except Exception as err: + print(f'{type(err)}: {err}') diff --git a/solarbank_monitor.py b/solarbank_monitor.py new file mode 100644 index 0000000..4728294 --- /dev/null +++ b/solarbank_monitor.py @@ -0,0 +1,154 @@ +""" +Example exec module to use the Anker API for continously querying and displaying important solarbank parameters +This module will prompt for the Anker account details if not pre-set in the header. +Upon successfull authentication, you will see the solarbank parameters displayed and refreshed at reqular interval. +Note: When the system owning account is used, more details for the solarbank can be queried and displayed. +Attention: During executiion of this module, the used account cannot be used in the Anker App since it will be kicked out on each refresh. +""" + +import asyncio +from aiohttp import ClientSession +from datetime import datetime, timedelta +from getpass import getpass +import json, logging, sys, time, os +from api import api + +_LOGGER: logging.Logger = logging.getLogger(__name__) +_LOGGER.addHandler(logging.StreamHandler(sys.stdout)) +#_LOGGER.setLevel(logging.DEBUG) # enable for debug output + +# Optional default Anker Account credentials to be used +USER = "" +PASSWORD = "" +COUNTRY = "" +REFRESH = 30 # refresh interval in seconds + + +def clearscreen(): + if sys.stdin is sys.__stdin__: # check if not in IDLE shell + os.system("cls") if os.name == "nt" else os.system("clear") + #print("\033[H\033[2J", end="") # ESC characters to clear screen, system independent? + + +async def main() -> None: + global USER, PASSWORD, COUNTRY, REFRESH + print("Solarbank Monitor:") + if USER == "": + print("\nEnter Anker Account credentials:") + USER = input("Username (email): ") + if USER == "": + return False + PASSWORD = getpass("Password: ") + if PASSWORD == "": + return False + COUNTRY = input("Country ID (e.g. DE): ") + if COUNTRY == "": + return False + try: + async with ClientSession() as websession: + print("\nTrying authentication...",end="") + myapi = api.API(USER,PASSWORD,COUNTRY,websession, _LOGGER) + if await myapi.async_authenticate(): + print("OK") + else: + print("CACHED") # Login validation will be done during first API call + + while True: + resp = input(f"\nHow many seconds refresh interval should be used? (10-600, default: {REFRESH}): ") + if not resp: + break + elif resp.isdigit() and 10 <= int(resp) <= 600: + REFRESH = int(resp) + break + + # Run loop to update Solarbank parameters + now = datetime.now().astimezone() + next_refr = now + next_dev_refr = now + col1 = 15 + col2 = 20 + t1 = 2 + t2 = 5 + t3 = 5 + t4 = 9 + t5 = 6 + t6 = 10 + while True: + print("\n") + now = datetime.now().astimezone() + if next_refr <= now: + print("Running site refresh...") + await myapi.update_sites() + next_refr = now + timedelta(seconds=REFRESH) + if next_dev_refr <= now: + print("Running device details refresh...") + await myapi.update_device_details() + next_dev_refr = next_refr + timedelta(seconds=REFRESH*9) + schedules = {} + clearscreen() + print(f"Solarbank Monitor (refresh {REFRESH} s, details refresh {10*REFRESH} s):") + print(f"Sites: {len(myapi.sites)}, Devices: {len(myapi.devices)}") + for sn, dev in myapi.devices.items(): + devtype = dev.get('type','unknown') + admin = dev.get('is_admin',False) + print(f"{'Device':<{col1}}: {(dev.get('name','NoName')):<{col2}} (Admin: {'YES' if admin else 'NO'})") + print(f"{'SN':<{col1}}: {sn}") + print(f"{'PN':<{col1}}: {dev.get('pn','')}") + print(f"{'Type':<{col1}}: {devtype.capitalize()}") + if devtype == "solarbank": + siteid = dev.get('site_id','') + print(f"{'Site ID':<{col1}}: {siteid}") + online = dev.get('wifi_online') + print(f"{'Wifi state':<{col1}}: {('Unknown' if online == None else 'Online' if online else 'Offline'):<{col2}} (Charging Status: {dev.get('charging_status','')})") + upgrade = dev.get('auto_upgrade') + print(f"{'SW Version':<{col1}}: {dev.get('sw_version','Unknown'):<{col2}} (Auto-Upgrade: {'Unknown' if upgrade == None else 'Enabled' if upgrade else 'Disabled'})") + soc = f"{dev.get('battery_soc','---'):>3} %" + print(f"{'State Of Charge':<{col1}}: {soc:<{col2}} (Min SOC: {str(dev.get('power_cutoff','--'))+' %'})") + unit = dev.get('power_unit','W') + print(f"{'Input Power':<{col1}}: {dev.get('input_power',''):>3} {unit}") + print(f"{'Charge Power':<{col1}}: {dev.get('charging_power',''):>3} {unit}") + print(f"{'Output Power':<{col1}}: {dev.get('output_power',''):>3} {unit}") + preset = dev.get('set_output_power') + if not preset: + preset = '---' + print(f"{'Output Preset':<{col1}}: {preset:>3} {unit}") + """update schedule with device details refresh and print it""" + if admin: + if not schedules.get(sn) and siteid: + schedules.update({sn: await myapi.get_device_load(siteId=siteid,deviceSn=sn)}) + data = schedules.get(sn,{}) + print(f"{'Schedule':<{col1}}: {now.strftime('%H:%M UTC %z'):<{col2}} (Current Preset: {data.get('current_home_load','')})") + print(f"{'ID':<{t1}} {'Start':<{t2}} {'End':<{t3}} {'Discharge':<{t4}} {'Output':<{t5}} {'ChargePrio':<{t6}}") + for slot in (data.get("home_load_data",{})).get("ranges",[]): + enabled = slot.get('turn_on') + load = slot.get('appliance_loads',[]) + load = load[0] if len(load) > 0 else {} + print(f"{str(slot.get('id','')):>{t1}} {slot.get('start_time',''):<{t2}} {slot.get('end_time',''):<{t3}} {('---' if enabled == None else 'YES' if enabled else 'NO'):^{t4}} {str(load.get('power',''))+' W':>{t5}} {str(slot.get('charge_priority',''))+' %':>{t6}}") + else: + print(f"Not a Solarbank device, further details skipped") + print("") + #print(json.dumps(myapi.devices, indent=2)) + for sec in range(0,REFRESH): + now = datetime.now() + if sys.stdin is sys.__stdin__: + print(f"Site refresh: {int((next_refr-now).total_seconds()):>3} sec, Device details refresh: {int((next_dev_refr-now).total_seconds()):>3} sec (CTRL-C to abort)", end = "\r", flush=True) + elif sec == 0: + # IDLE may be used and does not support cursor placement, skip time progress display + print(f"Site refresh: {int((next_refr-now).total_seconds()):>3} sec, Device details refresh: {int((next_dev_refr-now).total_seconds()):>3} sec (CTRL-C to abort)", end = "", flush=True) + time.sleep(1) + return False + + except Exception as err: + print(f'{type(err)}: {err}') + return False + + +"""run async main""" +if __name__ == '__main__': + try: + if not asyncio.run(main()): + print("\nAborted!") + except KeyboardInterrupt: + print("\nAborted!") + except Exception as err: + print(f'{type(err)}: {err}') diff --git a/test_api.py b/test_api.py new file mode 100644 index 0000000..569b174 --- /dev/null +++ b/test_api.py @@ -0,0 +1,103 @@ +""" +Example exec module to test the Anker API for various methods or direct endpoint requests with various parameters. +""" + +import asyncio +from aiohttp import ClientSession +from datetime import datetime +from api import api, errors +import json, logging, sys + +_LOGGER: logging.Logger = logging.getLogger(__name__) +_LOGGER.addHandler(logging.StreamHandler(sys.stdout)) +#_LOGGER.setLevel(logging.DEBUG) # enable for detailed API output + +async def main() -> None: + """Create the aiohttp session and run the example.""" + print("Testing Solix API:") + try: + async with ClientSession() as websession: + myapi = api.API("username@domain.com","password","de",websession, _LOGGER) + + #show login response + ''' + #new = await myapi.async_authenticate(restart=True) # enforce new login data from server + new = await myapi.async_authenticate() # receive new or load cached login data + if new: + print("Received Login response:") + else: + print("Cached Login response:") + print(json.dumps(myapi._login_response, indent=2)) # show used login response for API reqests + ''' + + # test site api methods + + await myapi.update_sites() + await myapi.update_device_details() + print("System Overview:") + print(json.dumps(myapi.sites, indent=2)) + print("Device Overview:") + print(json.dumps(myapi.devices, indent=2)) + + + # test api methods + ''' + print(json.dumps(await myapi.get_site_list(), indent=2)) + print(json.dumps(await myapi.get_homepage(), indent=2)) + print(json.dumps(await myapi.get_bind_devices(), indent=2)) + print(json.dumps(await myapi.get_user_devices(), indent=2)) + print(json.dumps(await myapi.get_charging_devices(), indent=2)) + print(json.dumps(await myapi.get_auto_upgrade(), indent=2)) + print(json.dumps(await myapi.get_scene_info(siteId='efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'), indent=2)) + print(json.dumps(await myapi.get_wifi_list(siteId='efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'), indent=2)) + print(json.dumps(await myapi.get_solar_info(siteId='efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'), indent=2)) # json parameters unknown: site_id not sifficient, or works only with Anker Inverters? + print(json.dumps(await myapi.get_device_parm(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",paramType="4"), indent=2)) + print(json.dumps(await myapi.get_power_cutoff(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",deviceSn="9JVB42LJK8J0P5RY"), indent=2)) + print(json.dumps(await myapi.get_device_load(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",deviceSn="9JVB42LJK8J0P5RY"), indent=2)) + + print(json.dumps(await myapi.energy_analysis(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",deviceSn="9JVB42LJK8J0P5RY",rangeType="week",startDay=datetime.fromisoformat("2023-10-10"),endDay=datetime.fromisoformat("2023-10-10"),devType="solar_production"), indent=2)) + print(json.dumps(await myapi.energy_analysis(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",deviceSn="9JVB42LJK8J0P5RY",rangeType="week",startDay=datetime.fromisoformat("2023-10-10"),endDay=datetime.fromisoformat("2023-10-10"),devType="solarbank"), indent=2)) + print(json.dumps(await myapi.energy_daily(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",deviceSn="9JVB42LJK8J0P5RY",startDay=datetime.fromisoformat("2024-01-10"),numDays=10), indent=2)) + print(json.dumps(await myapi.home_load_chart(siteId='efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'), indent=2)) + ''' + + # test api endpoints directly + ''' + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["homepage"],json={})), indent=2)) + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["site_list"],json={})), indent=2)) + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["bind_devices"],json={})), indent=2)) + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["user_devices"],json={})), indent=2)) + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["charging_devices"],json={})), indent=2)) + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_auto_upgrade"],json={})), indent=2)) + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["site_detail"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'})), indent=2)) + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["wifi_list"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'})), indent=2)) + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_site_price"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'})), indent=2)) + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["solar_info"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'})), indent=2)) # json parameters unknown: site_id not sifficient, or works only with Anker Inverters? + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_cutoff"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c', "device_sn": "9JVB42LJK8J0P5RY"})), indent=2)) + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_device_fittings"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c', "device_sn": "9JVB42LJK8J0P5RY"})), indent=2)) + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_device_load"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c', "device_sn": "9JVB42LJK8J0P5RY"})), indent=2)) + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_device_parm"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c', "param_type": "4"})), indent=2)) + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["compatible_process"],json={})), indent=2)) # json parameters unknown + print(json.dumps((await myapi.request("post", api._API_ENDPOINTS["home_load_chart"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'})), indent=2)) + ''' + + # test api from json files + ''' + myapi.testDir("examples") + await myapi.update_sites(fromFile=True) + await myapi.update_device_details(fromFile=True) + print(json.dumps(myapi.sites,indent=2)) + print(json.dumps(myapi.devices,indent=2)) + ''' + + except Exception as err: + print(f'{type(err)}: {err}') + + + +"""run async main""" +if __name__ == '__main__': + try: + asyncio.run(main()) + except Exception as err: + print(f'{type(err)}: {err}')