diff --git a/README.md b/README.md index 81b5c1c..abf0d54 100644 --- a/README.md +++ b/README.md @@ -1,15 +1,15 @@ -Solarbank E1600 Logo +Solarbank E1600 Logo -# Anker Solix API +# Anker Solix Api [![github licence](https://img.shields.io/badge/Licence-MIT-orange)](https://github.com/thomluther/anker-solix-api/blob/main/LICENSE) ![python badge](https://img.shields.io/badge/Made%20with-Python-orange) This is an experimental Python library for Anker Solix Power devices (Solarbank, Inverter etc). -🚨 This is by no means an official Anker API. 🚨 +🚨 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.🚨 +🚨 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 @@ -27,13 +27,13 @@ pip install aiohttp # Anker Account Information -Because of the way the Anker Solix API works, one account with email/password cannot be used for the Anker mobile App and this API in parallel. +Because of the way the Anker Solix Api works, one account with email/password cannot be used for the Anker mobile App and this Api in parallel. The Anker Cloud allows only one request token per account at any time. Each new authentication request by a client will create a new token and drop a previous token. -Therefore usage of this API may kick out your account login in the mobile app. +Therefore usage of this Api may kick out your account login in the mobile app. However, starting with Anker mobile app release 2.0, you can share your defined system(s) with 'family members'. Therefore it is recommended to create a second Anker account with a different email address and share your defined system(s) with the second account. Attention: A shared account is only a member of the shared system, and as such currently has no permissions to access or query device details of the shared system. -Therefore an API homepage query will neither display any data for a shared account. However, a shared account can receive API scene/site details of shared systems (App system = API site), +Therefore an Api homepage query will neither display any data for a shared account. However, a shared account can receive Api scene/site details of shared systems (App system = Api site), which is equivalent to what is displayed in the mobile app on the home screen for the selected system. # Usage @@ -48,13 +48,13 @@ from aiohttp import ClientSession from api import api, errors _LOGGER: logging.Logger = logging.getLogger(__name__) -#_LOGGER.setLevel(logging.DEBUG) # enable for detailed API output +#_LOGGER.setLevel(logging.DEBUG) # enable for detailed Api output 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) + myapi = api.AnkerSolixApi("username@domain.com","password","de",websession, _LOGGER) await myapi.update_sites() await myapi.update_device_details() print("System Overview:") @@ -62,7 +62,7 @@ async def main() -> None: print("Device Overview:") print(json.dumps(myapi.devices, indent=2)) -"""run async main""" +# run async main if __name__ == '__main__': try: asyncio.run(main()) @@ -70,36 +70,36 @@ if __name__ == '__main__': print(f'{type(err)}: {err}') ``` -The API class provides 2 main methods: -- `API.update_sites()` to query overview data for all accessible sites and store data in API dictionaries `API.sites` and `API.devices` for quick access. - This method could be run in regular intervals (30s or more) to fetch new data of the systems -- `API.update_device_details()` to query further settings for the device serials as found in the sites query. - This method should be run less frequently since this will mostly fetch various device configuration settings and needs multiple queries. +The AnkerSolixApi class provides 2 main methods: +- `AnkerSolixApi.update_sites()` to query overview data for all accessible sites and store data in dictionaries `AnkerSolixApi.sites` and `AnkerSolixApi.devices` for quick access. + This method could be run in regular intervals (30sec or more) to fetch new data of the systems +- `AnkerSolixApi.update_device_details()` to query further settings for the device serials as found in the sites query. + This method should be run less frequently since this will mostly fetch various device configuration settings and needs multiple queries. It currently is developped for Solarbank devices only, further device types such as Inverters or Power Stations could be added once example data is available. -Check out `test_api.py` and other python executable tools that may help to leverage and explore the API for your Anker power system. +Check out `test_api.py` and other python executable tools that may help to leverage and explore the Api for your Anker power system. The subfolder [`examples`](https://github.com/thomluther/anker-solix-api/tree/main/examples) contains json files with anonymized responses of the -`export_system.py` module giving you an idea of how various API responses look like. (Note that the Solarbank was switched off when the data were pulled, so some fields may be empty) -Those json files can also be used to develop/debug the API for system constellations not available to the developper. +`export_system.py` module giving you an idea of how various Api responses look like. (Note that the Solarbank was switched off when the data were pulled, so some fields may be empty) +Those json files can also be used to develop/debug the Api for system constellations not available to the developper. -# API Tools +# AnkerSolixApi Tools ## test_api.py -Example exec module that can be used to explore and test API methods or direct enpoint requests with parameters. +Example exec module that can be used to explore and test AnkerSolixApi 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. +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 +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. +Optionally the AnkerSolixApi 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 +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. @@ -107,12 +107,12 @@ Attention: During executiion of this module, the used account cannot be used in ## energy_csv.py -Example exec module to use the Anker API for export of daily Solarbank Energy Data. +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. +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. @@ -122,18 +122,17 @@ The received daily values will be exported into a csv file. ![last commit](https://img.shields.io/github/last-commit/thomluther/anker-solix-api?color=orange) [![Community Discussion](https://img.shields.io/badge/Home%20Assistant%20Community-Discussion-orange)](https://community.home-assistant.io/t/feature-request-integration-or-addon-for-anker-solix-e1600-solarbank/641086) +Github is used to host code, to track issues and feature requests, as well as accept pull requests. + +Pull requests are the best way to propose changes to the codebase. + 1. [Check for open features/bugs](https://github.com/thomluther/anker-solix-api/issues) or [initiate a discussion on one](https://github.com/thomluther/anker-solix-api/issues/new). -2. [Fork the repository](https://github.com/thomluther/anker-solix-api/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! +1. [Fork the repository](https://github.com/thomluther/anker-solix-api/fork). +1. Fork the repo and create your branch from `main`. +1. If you've changed something, update the documentation. +1. Test your contribution. +1. Issue that pull request! # Acknowledgements / Credits @@ -144,4 +143,4 @@ The received daily values will be exported into a csv file. # Showing Your Appreciation -If you like this project, please give it a star on [GitHub](https://github.com/thomluther/anker-solix-api) \ No newline at end of file +If you like this project, please give it a star on [GitHub](https://github.com/thomluther/anker-solix-api) \ No newline at end of file diff --git a/api/api.py b/api/api.py index 94028d4..be9d44f 100644 --- a/api/api.py +++ b/api/api.py @@ -1,67 +1,73 @@ -""" -Class for interacting with the Anker Power / Solix API. +"""Class for interacting with the Anker Power / Solix API. + Required Python modules: pip install cryptography pip install aiohttp """ +from __future__ import annotations + +from base64 import b64encode +import contextlib from datetime import datetime, timedelta -from typing import Dict, Optional -import time +import json import logging -import sys, os, json +import os +import sys +import time +from typing import Optional 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 import hashes, padding, serialization from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes -from base64 import b64encode -from . import errors +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' - } +_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 +_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', @@ -70,9 +76,9 @@ _API_ENDPOINTS={ '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/add_site_devices', 'power_service/v1/site/delete_site_devices', - 'power_service/v1/site/add_site_devicesDatagram', + 'power_service/v1/site/update_site_device', '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', @@ -98,19 +104,20 @@ _API_ENDPOINTS={ '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 +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. +The Login Response is cached in a JSON file per email user account and can be reused by this API class 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 mobile App) will kickoff the token used in this Api instance 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 instances should use different user accounts. They may work in parallel when all using the same cached authentication data. The first API instance with failed authorization will restart a new Login request and updates +the cached JSON file. Other instances should recognize an update of the cached JSON file and will refresh their login credentials in the instance for the actual token and gtoken without new login request. """ -LOGIN_RESPONSE: Dict = { + +LOGIN_RESPONSE: dict = { "user_id": str, "email": str, "nick_name": str, @@ -126,23 +133,25 @@ LOGIN_RESPONSE: Dict = { "phone": str, "phone_number": str, "phone_code": str, - "server_secret_info": { - "public_key": str - }, + "server_secret_info": {"public_key": str}, "params": list, "trust_list": list, - "fa_info": { - "step": int, - "info": str - }, - "country_code": str + "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""" +class AnkerSolixApi: + """Define the API class to handle Anker server authentication and API requests, along with the last state of queried site details and Device information.""" - def __init__(self, email: str, password: str, countryId: str, websession: ClientSession, logger = None) -> None: + 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 @@ -150,12 +159,16 @@ class API: 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._testdir: str = "test" + self._retry_attempt: bool = False # Flag for retry after any token error + os.makedirs( + os.path.join(os.path.dirname(__file__), "authcache"), exist_ok=True + ) # ensure folder for authentication caching exists + self._authFile: str = os.path.join( + os.path.dirname(__file__), "authcache", f"{email}.json" + ) # filename for authentication cache self._authFileTime: float = 0 - """initialize logger for object""" + # initialize logger for object if logger: self._logger = logger else: @@ -164,81 +177,106 @@ class API: if not self._logger.hasHandlers(): self._logger.addHandler(logging.StreamHandler(sys.stdout)) - - self._timezone: str = self._getTimezoneGMTString() #Timezone format: 'GMT+01:00' + 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""" + + # 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._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 + 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""" + # Define class variables saving the most recent site and device data self.nickname: str = "" - self.sites: Dict = {} - self.devices: Dict = {} + self.sites: dict = {} + self.devices: dict = {} - - def _md5(self,text: str) -> str: + def _md5(self, text: str) -> str: h = hashes.Hash(hashes.MD5()) - h.update(text.encode('utf-8')) + 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') + """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()) + """Return Base64 encoded secret as utf-8 decoded string using the shared secret with seed of 16 for the encryption.""" + # Password must be UTF-8 encoded and AES-256-CBC encrypted with block size of 16 + 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""" + # 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") + 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""" + """Load json data from given file for testing.""" try: if os.path.isfile(filename): - with open(filename, 'r') as file: + with open(filename, encoding="utf-8") as file: data = json.load(file) - self._logger.debug(f"Loaded JSON from file {filename}:") - self._logger.debug(f"{data}:") + self._logger.debug("Loaded JSON from file %s:", filename) + self._logger.debug("Data: %s", data) return data return {} - except Exception as err: - self._logger.error(f"ERROR: Failed to load JSON from file {filename}") + except OSError as err: + self._logger.error("ERROR: Failed to load JSON from file %s", filename) self._logger.error(err) return {} - - def _saveToFile(self, filename: str, json: dict = {}) -> bool: - """Save json data to given file for testing""" + + def _saveToFile(self, filename: str, data: dict = None) -> bool: + """Save json data to given file for testing.""" + if not data: + data = {} try: - with open(filename, 'w') as file: - json.dump(json, file, indent=2) - self._logger.debug(f"Saved JSON to file {filename}") + with open(filename, "w", encoding="utf-8") as file: + json.dump(data, file, indent=2) + self._logger.debug("Saved JSON to file %s:", filename) return True - except Exception as err: - self._logger.error(f"ERROR: Failed to save JSON to file {filename}") + except OSError as err: + self._logger.error("ERROR: Failed to save JSON to file %s", filename) self._logger.error(err) return False - def _update_dev(self, devData: dict, devType: str = None, siteId: str = None, isAdmin: bool = None) -> None: + 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""" + 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 = self.devices.get(sn, {}) # lookup old device info if any device.update({"sn": str(sn)}) if devType: device.update({"type": devType.lower()}) @@ -246,10 +284,10 @@ class API: device.update({"site_id": str(siteId)}) if isAdmin: device.update({"is_admin": True}) - elif isAdmin == False and device.get("is_admin") == None: + elif isAdmin is False and device.get("is_admin") is None: device.update({"is_admin": False}) for key, value in devData.items(): - if key in ["product_code","device_pn"]: + if key in ["product_code", "device_pn"]: device.update({"pn": str(value)}) elif key in ["device_name"]: device.update({"name": str(value)}) @@ -260,7 +298,9 @@ class API: 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 + 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"]: @@ -279,162 +319,221 @@ class API: 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""" + """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 + if not os.path.isdir(subfolder): + self._logger.error("Specified test folder does not exist: %s", subfolder) + else: + self._testdir = subfolder + self._logger.info("Set test folder to: %s", subfolder) + return self._testdir def logLevel(self, level: int = None) -> int: - """get or set the logger log level""" + """Get or set the logger log level.""" if level: self._logger.setLevel(level) - self._logger.info(f"Set log level to: {level}") + self._logger.info("Set log level to: %s", 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""" + """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._loggedIn = False self._authFileTime = 0 self.nickname = "" if os.path.isfile(self._authFile): - try: + with contextlib.suppress(Exception): 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 + self._logger.debug( + "Cached Login for %s from %s:", + self._email, + datetime.fromtimestamp(self._authFileTime).isoformat(), + ) + self._logger.debug("%s", data) + self._retry_attempt = ( + False # clear retry 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) + self._retry_attempt = ( + True # set retry 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 }, - "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}") + ) + data = auth_resp.get("data", {}) + self._logger.debug("Login Response: %s", data) self._loggedIn = True # Cache login response in file for reuse - with open(self._authFile, 'w') as authfile: + with open(self._authFile, "w", encoding="utf-8") as authfile: json.dump(data, authfile, indent=2, skipkeys=True) - self._logger.debug(f"Response cached in file: {self._authFile}") + self._logger.debug("Response cached in file: %s", self._authFile) self._authFileTime = os.path.getmtime(self._authFile) - #Update the login params + # 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")) + 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 + self._gtoken = self._md5( + data.get("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: + async def request( + self, + method: str, + endpoint: str, + *, + headers: Optional[dict] = None, + json: Optional[dict] = None, # lint W0621 + ) -> dict: + """Handle all requests to the API. This is also called recursively by login requests if necessary.""" + if not headers: + headers = {} + if not json: + data = {} + 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))): + # 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 + mergedHeaders.update({"Country": self._countryId}) if self._timezone: - mergedHeaders["Timezone"] = self._timezone + mergedHeaders.update({"Timezone": self._timezone}) if self._token: - mergedHeaders["x-auth-token"] = self._token - mergedHeaders["gtoken"] = self._gtoken + mergedHeaders.update({"x-auth-token": self._token}) + mergedHeaders.update({"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: + self._logger.debug("Request Url: %s %s", method.upper(), url) + self._logger.debug("Request Headers: %s", mergedHeaders) + self._logger.debug("Request Body: %s", 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}") + self._logger.debug("Request Response: %s", data) if not data: raise ClientError(f"No response while requesting {endpoint}") - errors.raise_error(data) # check the response code in the data + 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 + 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 + except ( + ClientError + ) as err: # Exception from ClientSession based on standard response codes + self._logger.error("Request Error: %s", err) + if "401" in str(err) or "403" in str(err): + # Unauthorized or forbidden request if self._retry_attempt: - raise errors.AnkerSolixError(f"Login failed for user {self._email}") + raise errors.AuthorizationError( + "Login failed for user %s" % self._email + ) from err 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}") + return await self.request( + method, endpoint, headers=headers, json=json + ) + self._logger.error("Login failed for user %s", self._email) + raise errors.AuthorizationError( + "Login failed for user %s" % self._email + ) from err + raise ClientError( + f"There was an error while requesting {endpoint}: {err}" + ) from err + except ( + errors.InvalidCredentialsError, + errors.TokenKickedOutError, + ) as err: # Exception for API specific response codes + self._logger.error("API ERROR: %s", err) if self._retry_attempt: - raise errors.AnkerSolixError(f"Login failed for user {self._email}") + raise errors.AuthorizationError( + f"Login failed for user {self._email}" + ) from err 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") + return await self.request( + method, endpoint, headers=headers, json=json + ) + self._logger.error("Login failed for user %s", self._email) + raise errors.AuthorizationError( + "Login failed for user %s" % self._email + ) from err except errors.AnkerSolixError as err: # Other Exception from API - self._logger.error(f"ANKER API ERROR: {err}") + self._logger.error("ANKER API ERROR: %s", 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 + 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}, @@ -460,86 +559,105 @@ class API: self.sites = {} self._logger.debug("Getting site list...") sites = await self.get_site_list(fromFile=fromFile) - for site in sites.get("site_list",[]): + for site in sites.get("site_list", []): if site.get("site_id"): - """Update site info""" + # Update site info myid = site.get("site_id") - mysite = self.sites.get(myid,{}) - siteInfo = mysite.get("site_info",{}) + 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""" + 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) + 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) + # 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: + 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""" + # 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",[]): + 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""" + # 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 + 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): + 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""" + # 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""" + # 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("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(#1): Fetch other details of specific device types as known and relevant return self.devices - - async def get_site_list(self, fromFile: bool = False) -> Dict: + 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")) + resp = self._loadFromFile(os.path.join(self._testdir, "site_list.json")) else: resp = await self.request("post", _API_ENDPOINTS["site_list"]) - return resp.get("data",{}) + return resp.get("data", {}) - - async def get_scene_info(self, siteId: str, fromFile: bool = False) -> Dict: + 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":[], @@ -555,14 +673,16 @@ class API: """ data = {"site_id": siteId} if fromFile: - resp = self._loadFromFile(os.path.join(self._testdir,f"scene_{siteId}.json")) + 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",{}) + resp = await self.request("post", _API_ENDPOINTS["scene_info"], json=data) + return resp.get("data", {}) - - async def get_homepage(self, fromFile: bool = False) -> Dict: + 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}], @@ -573,28 +693,28 @@ class API: "powerpanel_list":[]} """ if fromFile: - resp = self._loadFromFile(os.path.join(self._testdir,f"homepage.json")) + resp = self._loadFromFile(os.path.join(self._testdir, "homepage.json")) else: resp = await self.request("post", _API_ENDPOINTS["homepage"]) - return resp.get("data",{}) + return resp.get("data", {}) + async def get_bind_devices(self, fromFile: bool = False) -> dict: + """Get the bind device information, contains firmware level of devices. - 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")) + resp = self._loadFromFile(os.path.join(self._testdir, "bind_devices.json")) else: resp = await self.request("post", _API_ENDPOINTS["bind_devices"]) - return resp.get("data",{}) + return resp.get("data", {}) - - async def get_user_devices(self, fromFile: bool = False) -> Dict: + 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', @@ -602,64 +722,74 @@ class API: 'photovoltaic_power': '', 'output_power': '', 'create_time': 0}]} """ if fromFile: - resp = self._loadFromFile(os.path.join(self._testdir,f"user_devices.json")) + resp = self._loadFromFile(os.path.join(self._testdir, "user_devices.json")) else: resp = await self.request("post", _API_ENDPOINTS["user_devices"]) - return resp.get("data",{}) + return resp.get("data", {}) + async def get_charging_devices(self, fromFile: bool = False) -> dict: + """Get the charging devices (Power stations?). - 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")) + resp = self._loadFromFile( + os.path.join(self._testdir, "charging_devices.json") + ) else: resp = await self.request("post", _API_ENDPOINTS["charging_devices"]) - return resp.get("data",{}) + 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). - 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? + data = { + "site_id": siteId + } # TODO(#2): 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")) + 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",{}) + 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. - 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")) + resp = self._loadFromFile(os.path.join(self._testdir, "auto_upgrade.json")) else: resp = await self.request("post", _API_ENDPOINTS["get_auto_upgrade"]) - return resp.get("data",{}) + return resp.get("data", {}) - - async def get_wifi_list(self, siteId: str, fromFile: bool = False) -> Dict: + 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")) + 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",{}) + 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: + 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}, @@ -667,26 +797,39 @@ class API: """ data = {"site_id": siteId, "device_sn": deviceSn} if fromFile: - resp = self._loadFromFile(os.path.join(self._testdir,f"power_cutoff_{deviceSn}.json")) + 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",{}) + 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: + 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. + + 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? + data = { + "site_id": siteId, + "device_sn": deviceSn, + "id": setId, + } # TODO(#3): 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) + 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",{}) + 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: + r"""Get device load settings. - 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}", @@ -694,18 +837,24 @@ class API: """ data = {"site_id": siteId, "device_sn": deviceSn} if fromFile: - resp = self._loadFromFile(os.path.join(self._testdir,f"device_load_{deviceSn}.json")) + 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",{}) + 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",{}) + return resp.get("data", {}) + async def get_device_parm( + self, siteId: str, paramType: str = "4", fromFile: bool = False + ) -> dict: + r"""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. - 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}, @@ -715,18 +864,29 @@ class API: """ data = {"site_id": siteId, "param_type": paramType} if fromFile: - resp = self._loadFromFile(os.path.join(self._testdir,f"device_parm_{siteId}.json")) + 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",{}) + 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",{}) + return resp.get("data", {}) - - async def set_device_parm(self, siteId: str, paramData: dict, paramType: str = "4", command: int = 17, toFile: bool = False) -> dict: + 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: @@ -736,16 +896,33 @@ class API: {"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)} + 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) + 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",{}) + 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: + 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" @@ -762,72 +939,118 @@ class API: "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", + "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",{}) + } + 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: + 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""" + # 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",[]): + 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",[]): + 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", "")}) + 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""" + # 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", "")}) + 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", "")}) + 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: + 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",{}) - - + 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 index 4ab2ca0..5bcb57c 100644 --- a/api/errors.py +++ b/api/errors.py @@ -1,73 +1,75 @@ -"""Define package errors.""" -from typing import Dict, Type +"""Define Anker Solix API errors.""" + +from __future__ import annotations 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]] = { + +ERRORS: dict[int, type[AnkerSolixError]] = { 401: AuthorizationError, + 403: AuthorizationError, 997: ConnectError, 998: NetworkError, 999: ServerError, @@ -83,6 +85,7 @@ ERRORS: Dict[int, Type[AnkerSolixError]] = { 26070: ClientPublicKeyError, 26084: TokenKickedOutError, 26108: InvalidCredentialsError, + 26156: InvalidCredentialsError, 100053: RetryExceeded, } diff --git a/energy_csv.py b/energy_csv.py index b389adf..bc0bcdb 100644 --- a/energy_csv.py +++ b/energy_csv.py @@ -1,5 +1,5 @@ -""" -Example exec module to use the Anker API for export of daily Solarbank Energy Data. +"""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 @@ -9,15 +9,22 @@ The received daily values will be exported into a csv file. """ import asyncio -from aiohttp import ClientSession +import csv from datetime import datetime -from api import api from getpass import getpass -import json, logging, sys, csv +import json +import logging +import sys + +from aiohttp import ClientSession +from api import api _LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER.addHandler(logging.StreamHandler(sys.stdout)) -#_LOGGER.setLevel(logging.DEBUG) # enable for debug output +# _LOGGER.setLevel(logging.DEBUG) # enable for debug output +CONSOLE: logging.Logger = logging.getLogger("console") +CONSOLE.addHandler(logging.StreamHandler(sys.stdout)) +CONSOLE.setLevel(logging.INFO) # Optional default Anker Account credentials to be used USER = "" @@ -26,14 +33,15 @@ COUNTRY = "" async def main() -> None: - global USER, PASSWORD, COUNTRY - print("Exporting daily Energy data for Anker Solarbank:") + """Run main to export energy history from cloud.""" + global USER, PASSWORD, COUNTRY # noqa: PLW0603 + CONSOLE.info("Exporting daily Energy data for Anker Solarbank:") if USER == "": - print("\nEnter Anker Account credentials:") + CONSOLE.info("\nEnter Anker Account credentials:") USER = input("Username (email): ") if USER == "": return False - PASSWORD = getpass("Password: ") + PASSWORD = getpass("Password: ") if PASSWORD == "": return False COUNTRY = input("Country ID (e.g. DE): ") @@ -41,66 +49,87 @@ async def main() -> None: return False try: async with ClientSession() as websession: - print("\nTrying authentication...",end="") - myapi = api.API(USER,PASSWORD,COUNTRY,websession, _LOGGER) + CONSOLE.info("\nTrying authentication...") + myapi = api.AnkerSolixApi(USER, PASSWORD, COUNTRY, websession, _LOGGER) if await myapi.async_authenticate(): - print("OK") + CONSOLE.info("OK") else: - print("CACHED") # Login validation will be done during first API call + CONSOLE.info( + "CACHED" + ) # Login validation will be done during first API call # Refresh the site and device info of the API - print("\nUpdating Site Info...", end="") + CONSOLE.info("\nUpdating Site Info...") if (await myapi.update_sites()) == {}: - print("NO INFO") + CONSOLE.info("NO INFO") return False - print("OK") - print(f"\nDevices: {len(myapi.devices)}") + CONSOLE.info("OK") + CONSOLE.info(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: ") + CONSOLE.info(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...") + CONSOLE.info( + 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): ") + 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) + CONSOLE.info( + 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: + with open( + filename, "w", newline="", encoding="utf-8" + ) 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}") + CONSOLE.info( + f"\nCompleted: Successfully exported data to {filename}" + ) return True - - print("No data received for device") + + CONSOLE.info("No data received for device") return False - print("No accepted Solarbank device found.") + CONSOLE.info("No accepted Solarbank device found.") return False except Exception as err: - print(f'{type(err)}: {err}') + CONSOLE.info(f"{type(err)}: {err}") return False -"""run async main""" -if __name__ == '__main__': +# run async main +if __name__ == "__main__": try: if not asyncio.run(main()): - print("Aborted!") - except Exception as err: - print(f'{type(err)}: {err}') - + CONSOLE.info("Aborted!") + except Exception as exception: + CONSOLE.info(f"{type(exception)}: {exception}") diff --git a/examples/auto_upgrade.json b/examples/example1/auto_upgrade.json similarity index 100% rename from examples/auto_upgrade.json rename to examples/example1/auto_upgrade.json diff --git a/examples/bind_devices.json b/examples/example1/bind_devices.json similarity index 100% rename from examples/bind_devices.json rename to examples/example1/bind_devices.json diff --git a/examples/charging_devices.json b/examples/example1/charging_devices.json similarity index 100% rename from examples/charging_devices.json rename to examples/example1/charging_devices.json diff --git a/examples/device_fittings_9JVB42LJK8J0P5RY.json b/examples/example1/device_fittings_9JVB42LJK8J0P5RY.json similarity index 100% rename from examples/device_fittings_9JVB42LJK8J0P5RY.json rename to examples/example1/device_fittings_9JVB42LJK8J0P5RY.json diff --git a/examples/device_load_9JVB42LJK8J0P5RY.json b/examples/example1/device_load_9JVB42LJK8J0P5RY.json similarity index 100% rename from examples/device_load_9JVB42LJK8J0P5RY.json rename to examples/example1/device_load_9JVB42LJK8J0P5RY.json diff --git a/examples/device_parm_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json b/examples/example1/device_parm_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json similarity index 100% rename from examples/device_parm_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json rename to examples/example1/device_parm_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json diff --git a/examples/homepage.json b/examples/example1/homepage.json similarity index 100% rename from examples/homepage.json rename to examples/example1/homepage.json diff --git a/examples/power_cutoff_9JVB42LJK8J0P5RY.json b/examples/example1/power_cutoff_9JVB42LJK8J0P5RY.json similarity index 100% rename from examples/power_cutoff_9JVB42LJK8J0P5RY.json rename to examples/example1/power_cutoff_9JVB42LJK8J0P5RY.json diff --git a/examples/price_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json b/examples/example1/price_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json similarity index 100% rename from examples/price_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json rename to examples/example1/price_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json diff --git a/examples/scene_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json b/examples/example1/scene_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json similarity index 100% rename from examples/scene_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json rename to examples/example1/scene_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json diff --git a/examples/example1/setup.txt b/examples/example1/setup.txt new file mode 100644 index 0000000..930de75 --- /dev/null +++ b/examples/example1/setup.txt @@ -0,0 +1,5 @@ +SETUP: +HM 600 Inverter +2x 395 W panels. +1x solarbank e1600 +1 System in Anker App \ No newline at end of file diff --git a/examples/site_detail_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json b/examples/example1/site_detail_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json similarity index 100% rename from examples/site_detail_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json rename to examples/example1/site_detail_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json diff --git a/examples/site_list.json b/examples/example1/site_list.json similarity index 100% rename from examples/site_list.json rename to examples/example1/site_list.json diff --git a/examples/user_devices.json b/examples/example1/user_devices.json similarity index 100% rename from examples/user_devices.json rename to examples/example1/user_devices.json diff --git a/examples/wifi_list_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json b/examples/example1/wifi_list_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json similarity index 100% rename from examples/wifi_list_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json rename to examples/example1/wifi_list_efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c.json diff --git a/export_system.py b/export_system.py index eac4769..a27a7a5 100644 --- a/export_system.py +++ b/export_system.py @@ -1,5 +1,5 @@ -""" -Example exec module to use the Anker API for export of defined system data and device details. +"""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. @@ -8,102 +8,133 @@ Optionally the API class can use the json files for debugging and testing on var """ 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 getpass import getpass +import json +import logging +import os +import random +import string +import sys +import time + +from aiohttp import ClientSession from api import api _LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER.addHandler(logging.StreamHandler(sys.stdout)) -#_LOGGER.setLevel(logging.DEBUG) # enable for debug output +# _LOGGER.setLevel(logging.DEBUG) # enable for debug output +CONSOLE: logging.Logger = logging.getLogger("console") +CONSOLE.addHandler(logging.StreamHandler(sys.stdout)) +CONSOLE.setLevel(logging.INFO) # 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 +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 + + Reuse same randomization if value was already randomized + """ + global RANDOMDATA # noqa: PLW0602 if not RANDOMIZE: return str(val) - randomstr = RANDOMDATA.get(val,"") + randomstr = RANDOMDATA.get(val, "") if not randomstr and val: if "_sn" in key: - randomstr = "".join(random.choices(string.ascii_uppercase+string.digits, k=len(val))) + 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 : + # 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))) + 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])) + 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)))]) + randomstr = "-".join( + [ + randomstr, + "".join( + random.choices(string.hexdigits.lower(), k=len(part)) + ), + ] + ) else: - randomstr = "".join(random.choices(string.hexdigits.lower(), k=len(part))) + randomstr = "".join( + random.choices(string.hexdigits.lower(), k=len(part)) + ) elif "wifi_name" in key: - idx = sum(1 for s in RANDOMDATA.values() if 'wifi-network-' in s) - randomstr = f'wifi-network-{idx+1}' + idx = sum(1 for s in RANDOMDATA.values() if "wifi-network-" in s) + randomstr = f"wifi-network-{idx+1}" 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): + """Recursive traversal of complex nested objects to randomize value for certain keys.""" + if isinstance(data, int | str): 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_","wifi_name"]): - data[k] = randomize(v,k) + # Randomize value for certain keys + if any(x in k for x in ["_sn", "site_id", "trace_id", "bt_ble_", "wifi_name"]): + 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 +def export(filename: str, d: dict = None) -> None: + """Save dict data to given file.""" + if not d: + d = {} + time.sleep(1) # central delay between multiple requests if len(d) == 0: - print(f"WARNING: File {filename} not saved because JSON is empty") + CONSOLE.info(f"WARNING: File {filename} not saved because JSON is empty") return elif RANDOMIZE: d = check_keys(d) try: - with open(filename, 'w') as file: + with open(filename, "w", encoding="utf-8") as file: json.dump(d, file, indent=2) - print(f"Saved JSON to file {filename}") + CONSOLE.info(f"Saved JSON to file {filename}") except Exception as err: - print(f"ERROR: Failed to save JSON to file {filename}") + CONSOLE.error(f"ERROR: Failed to save JSON to file {filename}: {err}") return -async def main() -> bool: - global USER, PASSWORD, COUNTRY, RANDOMIZE - print("Exporting found Anker Solix system data for all assigned sites:") +async def main() -> bool: # noqa: C901 + """Run main function to export config.""" + global USER, PASSWORD, COUNTRY, RANDOMIZE # noqa: PLW0603 + CONSOLE.info("Exporting found Anker Solix system data for all assigned sites:") if USER == "": - print("\nEnter Anker Account credentials:") + CONSOLE.info("\nEnter Anker Account credentials:") USER = input("Username (email): ") if USER == "": return False - PASSWORD = getpass("Password: ") + PASSWORD = getpass("Password: ") if PASSWORD == "": return False COUNTRY = input("Country ID (e.g. DE): ") @@ -111,119 +142,237 @@ async def main() -> bool: return False try: async with ClientSession() as websession: - print("\nTrying authentication...",end="") - myapi = api.API(USER,PASSWORD,COUNTRY,websession, _LOGGER) + CONSOLE.info("\nTrying authentication...") + myapi = api.AnkerSolixApi(USER, PASSWORD, COUNTRY, websession, _LOGGER) if await myapi.async_authenticate(): - print("OK") + CONSOLE.info("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 + CONSOLE.info( + "CACHED" + ) # Login validation will be done during first API call + + resp = input( + f"\nDo you want to randomize unique IDs and SNs in exported files? (default: {'YES' if RANDOMIZE else 'NO'}) (Y/N): " + ) + if resp != "" or not isinstance(RANDOMIZE, bool): + RANDOMIZE = resp.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) - + folder = nickname + os.makedirs(folder, exist_ok=True) + # first update sites in API object - print("\nQuerying site information...") + CONSOLE.info("\nQuerying site information...") await myapi.update_sites() - print(f"Sites: {len(myapi.sites)}, Devices: {len(myapi.devices)}") + CONSOLE.info(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 + CONSOLE.info("\nExporting homepage...") + export( + os.path.join(folder, "homepage.json"), + await myapi.request("post", api._API_ENDPOINTS["homepage"], json={}), + ) + CONSOLE.info("Exporting site list...") + export( + os.path.join(folder, "site_list.json"), + await myapi.request("post", api._API_ENDPOINTS["site_list"], json={}), + ) + CONSOLE.info("Exporting bind devices...") + export( + os.path.join(folder, "bind_devices.json"), + await myapi.request( + "post", api._API_ENDPOINTS["bind_devices"], json={} + ), + ) # shows only owner devices + CONSOLE.info("Exporting user devices...") + export( + os.path.join(folder, "user_devices.json"), + await myapi.request( + "post", api._API_ENDPOINTS["user_devices"], json={} + ), + ) # shows only owner devices + CONSOLE.info("Exporting charging devices...") + export( + os.path.join(folder, "charging_devices.json"), + await myapi.request( + "post", api._API_ENDPOINTS["charging_devices"], json={} + ), + ) # shows only owner devices + CONSOLE.info("Exporting auto upgrade settings...") + export( + os.path.join(folder, "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...") + CONSOLE.info(f"\nExporting site specific data for site {siteId}...") + CONSOLE.info("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}, + ), + ) + CONSOLE.info("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...") + 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 + CONSOLE.info("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: + 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: if not admin: - print("Query requires account of site owner!") - print("Exporting wifi list...") + CONSOLE.warning("Query requires account of site owner!") + CONSOLE.info("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: + 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: if not admin: - print("Query requires account of site owner!") - print("Exporting site price...") + CONSOLE.warning("Query requires account of site owner!") + CONSOLE.info("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: + 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: if not admin: - print("Query requires account of site owner!") - print("Exporting device parameter settings...") + CONSOLE.warning("Query requires account of site owner!") + CONSOLE.info("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: + export( + os.path.join( + folder, "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: if not admin: - print("Query requires account of site owner!") + CONSOLE.warning("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...") + CONSOLE.info( + f"\nExporting device specific data for device {device.get('name','')} SN {sn}..." + ) + siteId = device.get("site_id", "") + admin = site.get("is_admin") + CONSOLE.info("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: + 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: if not admin: - print("Query requires account of site owner!") - print("Exporting fittings...") + CONSOLE.warning("Query requires account of site owner!") + CONSOLE.info("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: + 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: if not admin: - print("Query requires account of site owner!") - print("Exporting load...") + CONSOLE.warning("Query requires account of site owner!") + CONSOLE.info("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: + 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: if not admin: - print("Query requires account of site owner!") + CONSOLE.warning("Query requires account of site owner!") - print(f"\nCompleted export of Anker Solix system data for user {USER}") + CONSOLE.info( + 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)) + CONSOLE.info( + f"Folder {os.path.abspath(folder)} contains the randomized JSON files. Pls check and update fields that may contain unrecognized personalized data." + ) + CONSOLE.info( + "Following trace or site IDs, SNs and MAC addresses have been randomized in files (from -> to):" + ) + CONSOLE.info(json.dumps(RANDOMDATA, indent=2)) else: - print(f"Folder {os.path.abspath(folder)} contains the JSON files.") + CONSOLE.info( + f"Folder {os.path.abspath(folder)} contains the JSON files." + ) return True except Exception as err: - print(f'{type(err)}: {err}') + CONSOLE.info(f"{type(err)}: {err}") return False -"""run async main""" -if __name__ == '__main__': +# run async main +if __name__ == "__main__": try: if not asyncio.run(main()): - print("Aborted!") + CONSOLE.info("Aborted!") except KeyboardInterrupt: - print("Aborted!") - except Exception as err: - print(f'{type(err)}: {err}') + CONSOLE.info("Aborted!") + except Exception as exception: + CONSOLE.info(f"{type(exception)}: {exception}") diff --git a/solarbank_monitor.py b/solarbank_monitor.py index 521dcab..b05a7a4 100644 --- a/solarbank_monitor.py +++ b/solarbank_monitor.py @@ -1,44 +1,55 @@ -""" -Example exec module to use the Anker API for continously querying and displaying important solarbank parameters +"""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. -""" +""" # noqa: D205 import asyncio -from aiohttp import ClientSession from datetime import datetime, timedelta from getpass import getpass -import json, logging, sys, time, os +import logging +import os +import sys +import time + +from aiohttp import ClientSession from api import api _LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER.addHandler(logging.StreamHandler(sys.stdout)) #_LOGGER.setLevel(logging.DEBUG) # enable for debug output +CONSOLE: logging.Logger = logging.getLogger("console") +CONSOLE.addHandler(logging.StreamHandler(sys.stdout)) +CONSOLE.setLevel(logging.INFO) # Optional default Anker Account credentials to be used USER = "" PASSWORD = "" COUNTRY = "" -REFRESH = 30 # refresh interval in seconds +REFRESH = 30 # default refresh interval in seconds def clearscreen(): + """Clear the terminal screen.""" 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? + if os.name == "nt": + os.system("cls") + else: + os.system("clear") + #CONSOLE.info("\033[H\033[2J", end="") # ESC characters to clear terminal screen, system independent? async def main() -> None: - global USER, PASSWORD, COUNTRY, REFRESH - print("Solarbank Monitor:") + """Run Main routine to start Solarbank monitor in a loop.""" + global USER, PASSWORD, COUNTRY, REFRESH # noqa: PLW0603 + CONSOLE.info("Solarbank Monitor:") if USER == "": - print("\nEnter Anker Account credentials:") + CONSOLE.info("\nEnter Anker Account credentials:") USER = input("Username (email): ") if USER == "": return False - PASSWORD = getpass("Password: ") + PASSWORD = getpass("Password: ") if PASSWORD == "": return False COUNTRY = input("Country ID (e.g. DE): ") @@ -46,12 +57,13 @@ async def main() -> None: return False try: async with ClientSession() as websession: - print("\nTrying authentication...",end="") - myapi = api.API(USER,PASSWORD,COUNTRY,websession, _LOGGER) + CONSOLE.info("\nTrying authentication...") + myapi = api.AnkerSolixApi(USER,PASSWORD,COUNTRY,websession, _LOGGER) if await myapi.async_authenticate(): - print("OK") + CONSOLE.info("OK") else: - print("CACHED") # Login validation will be done during first API call + # Login validation will be done during first API call + CONSOLE.info("CACHED") while True: resp = input(f"\nHow many seconds refresh interval should be used? (10-600, default: {REFRESH}): ") @@ -74,81 +86,81 @@ async def main() -> None: t5 = 6 t6 = 10 while True: - print("\n") + CONSOLE.info("\n") now = datetime.now().astimezone() if next_refr <= now: - print("Running site refresh...") + CONSOLE.info("Running site refresh...") await myapi.update_sites() next_refr = now + timedelta(seconds=REFRESH) if next_dev_refr <= now: - print("Running device details refresh...") + CONSOLE.info("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)}") + CONSOLE.info(f"Solarbank Monitor (refresh {REFRESH} s, details refresh {10*REFRESH} s):") + CONSOLE.info(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()}") + CONSOLE.info(f"{'Device':<{col1}}: {(dev.get('name','NoName')):<{col2}} (Admin: {'YES' if admin else 'NO'})") + CONSOLE.info(f"{'SN':<{col1}}: {sn}") + CONSOLE.info(f"{'PN':<{col1}}: {dev.get('pn','')}") + CONSOLE.info(f"{'Type':<{col1}}: {devtype.capitalize()}") if devtype == "solarbank": siteid = dev.get('site_id','') - print(f"{'Site ID':<{col1}}: {siteid}") + CONSOLE.info(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','')})") + CONSOLE.info(f"{'Wifi state':<{col1}}: {('Unknown' if online is 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'})") + CONSOLE.info(f"{'SW Version':<{col1}}: {dev.get('sw_version','Unknown'):<{col2}} (Auto-Upgrade: {'Unknown' if upgrade is 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','--'))+' %'})") + CONSOLE.info(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}") + CONSOLE.info(f"{'Input Power':<{col1}}: {dev.get('input_power',''):>3} {unit}") + CONSOLE.info(f"{'Charge Power':<{col1}}: {dev.get('charging_power',''):>3} {unit}") + CONSOLE.info(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""" + CONSOLE.info(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}}") + CONSOLE.info(f"{'Schedule':<{col1}}: {now.strftime('%H:%M UTC %z'):<{col2}} (Current Preset: {data.get('current_home_load','')})") + CONSOLE.info(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}}") + CONSOLE.info(f"{str(slot.get('id','')):>{t1}} {slot.get('start_time',''):<{t2}} {slot.get('end_time',''):<{t3}} {('---' if enabled is 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)) + sys.stdoutf("Not a Solarbank device, further details skipped") + CONSOLE.info("") + #CONSOLE.info(json.dumps(myapi.devices, indent=2)) for sec in range(0,REFRESH): now = datetime.now().astimezone() 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) + 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) # noqa: T201 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) + 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) # noqa: T201 time.sleep(1) return False - except Exception as err: - print(f'{type(err)}: {err}') + except Exception as exception: + CONSOLE.info(f'{type(exception)}: {exception}') return False -"""run async main""" +# run async main if __name__ == '__main__': try: if not asyncio.run(main()): - print("\nAborted!") + CONSOLE.info("\nAborted!") except KeyboardInterrupt: - print("\nAborted!") + CONSOLE.info("\nAborted!") except Exception as err: - print(f'{type(err)}: {err}') + CONSOLE.info(f'{type(err)}: {err}') diff --git a/test_api.py b/test_api.py index 569b174..dc5b59c 100644 --- a/test_api.py +++ b/test_api.py @@ -1,103 +1,108 @@ -""" -Example exec module to test the Anker API for various methods or direct endpoint requests with various parameters. -""" +"""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 +import json +import logging +import sys + +from aiohttp import ClientSession +from api import api _LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER.addHandler(logging.StreamHandler(sys.stdout)) -#_LOGGER.setLevel(logging.DEBUG) # enable for detailed API output +# _LOGGER.setLevel(logging.DEBUG) # enable for detailed API output +CONSOLE: logging.Logger = logging.getLogger("console") +CONSOLE.addHandler(logging.StreamHandler(sys.stdout)) +CONSOLE.setLevel(logging.INFO) + async def main() -> None: """Create the aiohttp session and run the example.""" - print("Testing Solix API:") + CONSOLE.info("Testing Solix API:") try: async with ClientSession() as websession: - myapi = api.API("username@domain.com","password","de",websession, _LOGGER) + myapi = api.AnkerSolixApi( + "username@domain.com", "password", "de", websession, _LOGGER + ) - #show login response - ''' + # 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:") + CONSOLE.info("Received Login response:") else: - print("Cached Login response:") - print(json.dumps(myapi._login_response, indent=2)) # show used login response for API reqests - ''' + CONSOLE.info("Cached Login response:") + CONSOLE.info(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)) + CONSOLE.info("System Overview:") + CONSOLE.info(json.dumps(myapi.sites, indent=2)) + CONSOLE.info("Device Overview:") + CONSOLE.info(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)) + """ + CONSOLE.info(json.dumps(await myapi.get_site_list(), indent=2)) + CONSOLE.info(json.dumps(await myapi.get_homepage(), indent=2)) + CONSOLE.info(json.dumps(await myapi.get_bind_devices(), indent=2)) + CONSOLE.info(json.dumps(await myapi.get_user_devices(), indent=2)) + CONSOLE.info(json.dumps(await myapi.get_charging_devices(), indent=2)) + CONSOLE.info(json.dumps(await myapi.get_auto_upgrade(), indent=2)) + CONSOLE.info(json.dumps(await myapi.get_scene_info(siteId='efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'), indent=2)) + CONSOLE.info(json.dumps(await myapi.get_wifi_list(siteId='efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'), indent=2)) + CONSOLE.info(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? + CONSOLE.info(json.dumps(await myapi.get_device_parm(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",paramType="4"), indent=2)) + CONSOLE.info(json.dumps(await myapi.get_power_cutoff(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",deviceSn="9JVB42LJK8J0P5RY"), indent=2)) + CONSOLE.info(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)) - ''' + CONSOLE.info(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)) + CONSOLE.info(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)) + CONSOLE.info(json.dumps(await myapi.energy_daily(siteId="efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c",deviceSn="9JVB42LJK8J0P5RY",startDay=datetime.fromisoformat("2024-01-10"),numDays=10), indent=2)) + CONSOLE.info(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)) - ''' + """ + CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["homepage"],json={})), indent=2)) + CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["site_list"],json={})), indent=2)) + CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["bind_devices"],json={})), indent=2)) + CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["user_devices"],json={})), indent=2)) + CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["charging_devices"],json={})), indent=2)) + CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_auto_upgrade"],json={})), indent=2)) + CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["site_detail"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'})), indent=2)) + CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["wifi_list"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'})), indent=2)) + CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_site_price"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'})), indent=2)) + CONSOLE.info(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? + CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["get_cutoff"],json={"site_id": 'efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c', "device_sn": "9JVB42LJK8J0P5RY"})), indent=2)) + CONSOLE.info(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)) + CONSOLE.info(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)) + CONSOLE.info(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)) + CONSOLE.info(json.dumps((await myapi.request("post", api._API_ENDPOINTS["compatible_process"],json={})), indent=2)) # json parameters unknown + CONSOLE.info(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)) - ''' + CONSOLE.info(json.dumps(myapi.sites,indent=2)) + CONSOLE.info(json.dumps(myapi.devices,indent=2)) + """ - except Exception as err: - print(f'{type(err)}: {err}') + except Exception as exception: + CONSOLE.info(f"{type(exception)}: {exception}") - -"""run async main""" -if __name__ == '__main__': +# run async main +if __name__ == "__main__": try: asyncio.run(main()) except Exception as err: - print(f'{type(err)}: {err}') + CONSOLE.info(f"{type(err)}: {err}")