diff --git a/.gitignore b/.gitignore index 5d830f1..8b590bf 100644 --- a/.gitignore +++ b/.gitignore @@ -16,3 +16,4 @@ coverage.xml **/credentials* /node_modules/ /client.py +/daily_energy_*.csv diff --git a/README.md b/README.md index 8478bc6..c846d8e 100644 --- a/README.md +++ b/README.md @@ -48,13 +48,16 @@ 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.AnkerSolixApi("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,12 +65,13 @@ async def main() -> None: print("Device Overview:") print(json.dumps(myapi.devices, indent=2)) + # run async main -if __name__ == '__main__': +if __name__ == "__main__": try: asyncio.run(main()) except Exception as err: - print(f'{type(err)}: {err}') + print(f"{type(err)}: {err}") ``` The AnkerSolixApi class provides 2 main methods: diff --git a/api/api.py b/api/api.py index 315fb3f..24975ff 100644 --- a/api/api.py +++ b/api/api.py @@ -5,6 +5,8 @@ pip install cryptography pip install aiohttp """ +# pylint: disable=too-many-lines,too-many-arguments,too-many-branches,too-many-statements,too-many-public-methods + from __future__ import annotations import contextlib @@ -247,7 +249,7 @@ class SolarbankStatus(Enum): unknown = "unknown" -class AnkerSolixApi: +class AnkerSolixApi: # pylint: disable=too-many-instance-attributes """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__( @@ -644,7 +646,7 @@ class AnkerSolixApi: endpoint: str, *, headers: dict | None = None, - json: dict | None = None, # noqa: W0621 + json: dict | None = None, # noqa: W0621 # pylint: disable=redefined-outer-name ) -> dict: """Handle all requests to the API. This is also called recursively by login requests if necessary.""" if not headers: @@ -768,7 +770,9 @@ class AnkerSolixApi: self._logger.error("Response Text: %s", body_text) raise err - async def update_sites(self, fromFile: bool = False) -> dict: + async def update_sites( # pylint: disable=too-many-locals + self, fromFile: bool = False + ) -> dict: """Get the latest info for all accessible sites and update class site and device variables. Example data: diff --git a/api/credentials.py b/api/credentials.py deleted file mode 100644 index 9ae18ef..0000000 --- a/api/credentials.py +++ /dev/null @@ -1,8 +0,0 @@ -"""Define your Anker Cloud credentials to be used by other modules for testing. - -This file is added to .gitignore and local changes will be skipped for commits. -""" - -USERNAME = "username@domain.com" -PASSWORD = "password" -COUNTRYID = "de" diff --git a/common.py b/common.py new file mode 100644 index 0000000..1b713ca --- /dev/null +++ b/common.py @@ -0,0 +1,53 @@ +# -*- mode: python: coding: utf-8 -*- +""" +a collection of helper functions for pyscripts +""" +import getpass +import logging +import os + +CONSOLE: logging.Logger = logging.getLogger("console") + +# Optional default Anker Account credentials to be used +_CREDENTIALS = { + "USER": os.getenv("USER"), + "PASSWORD": os.getenv("PASSWORD"), + "COUNTRY": os.getenv("COUNTRY"), +} + + +def user(): + """ + Get anker account user + """ + if _CREDENTIALS.get("USER"): + return _CREDENTIALS["USER"] + CONSOLE.info("\nEnter Anker Account credentials:") + username = input("Username (email): ") + while not username: + username = input("Username (email): ") + return username + + +def password(): + """ + Get anker account password + """ + if _CREDENTIALS.get("PASSWORD"): + return _CREDENTIALS["PASSWORD"] + pwd = getpass.getpass("Password: ") + while not pwd: + pwd = getpass.getpass("Password: ") + return pwd + + +def country(): + """ + Get anker account country + """ + if _CREDENTIALS.get("COUNTRY"): + return _CREDENTIALS["COUNTRY"] + countrycode = input("Country ID (e.g. DE): ") + while not countrycode: + countrycode = input("Country ID (e.g. DE): ") + return countrycode diff --git a/energy_csv.py b/energy_csv.py index 76c5268..5824bd7 100755 --- a/energy_csv.py +++ b/energy_csv.py @@ -1,13 +1,19 @@ #!/usr/bin/env python -"""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. The received daily values +will be exported into a csv file. -This method will prompt for the Anker account details if not pre-set in the header. -Then you can specify a start day and the number of days for data extraction from the Anker Cloud. -Note: The Solar production and Solarbank discharge can be queried across the full range. The solarbank -charge however can be queried only as total for an interval (e.g. day). Therefore when solarbank charge -data is also selected for export, an additional API query per day is required. -The received daily values will be exported into a csv file. """ +# pylint: disable=duplicate-code import asyncio import csv @@ -15,10 +21,10 @@ import json import logging import sys from datetime import datetime -from getpass import getpass from aiohttp import ClientSession +import common from api import api _LOGGER: logging.Logger = logging.getLogger(__name__) @@ -28,31 +34,16 @@ 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 = "" - async def main() -> None: """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 == "": - CONSOLE.info("\nEnter Anker Account credentials:") - USER = input("Username (email): ") - if USER == "": - return False - PASSWORD = getpass("Password: ") - if PASSWORD == "": - return False - COUNTRY = input("Country ID (e.g. DE): ") - if COUNTRY == "": - return False try: async with ClientSession() as websession: CONSOLE.info("\nTrying authentication...") - myapi = api.AnkerSolixApi(USER, PASSWORD, COUNTRY, websession, _LOGGER) + myapi = api.AnkerSolixApi( + common.user(), common.password(), common.country(), websession, _LOGGER + ) if await myapi.async_authenticate(): CONSOLE.info("OK") else: @@ -65,19 +56,19 @@ async def main() -> None: CONSOLE.info("NO INFO") return False CONSOLE.info("OK") - CONSOLE.info(f"\nDevices: {len(myapi.devices)}") + CONSOLE.info("\nDevices: %s", len(myapi.devices)) _LOGGER.debug(json.dumps(myapi.devices, indent=2)) for sn, device in myapi.devices.items(): if device.get("type") == "solarbank": - CONSOLE.info(f"Found {device.get('name')} SN: {sn}") + CONSOLE.info("Found %s SN: %s", device.get("name"), sn) try: daystr = input( "\nEnter start day for daily energy data (yyyy-mm-dd) or enter to skip: " ) if daystr == "": CONSOLE.info( - f"Skipped SN: {sn}, checking for next Solarbank..." + "Skipped SN: %s, checking for next Solarbank...", sn ) continue startday = datetime.fromisoformat(daystr) @@ -94,7 +85,8 @@ async def main() -> None: except ValueError: return False CONSOLE.info( - f"Queries may take up to {numdays*daytotals + 2} seconds...please wait..." + "Queries may take up to %s seconds...please wait...", + numdays * daytotals + 2, ) data = await myapi.energy_daily( siteId=device.get("site_id"), @@ -114,7 +106,8 @@ async def main() -> None: writer.writeheader() writer.writerows(data.values()) CONSOLE.info( - f"\nCompleted: Successfully exported data to {filename}" + "\nCompleted: Successfully exported data to %s", + filename, ) return True @@ -123,8 +116,8 @@ async def main() -> None: CONSOLE.info("No accepted Solarbank device found.") return False - except Exception as err: - CONSOLE.info(f"{type(err)}: {err}") + except Exception as err: # pylint: disable=broad-exception-caught + CONSOLE.error("%s: %s", type(err), err) return False @@ -132,6 +125,6 @@ async def main() -> None: if __name__ == "__main__": try: if not asyncio.run(main()): - CONSOLE.info("Aborted!") - except Exception as exception: - CONSOLE.info(f"{type(exception)}: {exception}") + CONSOLE.warning("Aborted!") + except Exception as exception: # pylint: disable=broad-exception-caught + CONSOLE.exception("%s: %s", type(exception), exception) diff --git a/export_system.py b/export_system.py index 511faf6..5dc7bd1 100755 --- a/export_system.py +++ b/export_system.py @@ -1,12 +1,22 @@ #!/usr/bin/env python -"""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. -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. + +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. + """ +# pylint: disable=duplicate-code import asyncio import json @@ -16,11 +26,11 @@ import random import string import sys import time -from getpass import getpass from aiohttp import ClientSession from aiohttp.client_exceptions import ClientError +import common from api import api, errors _LOGGER: logging.Logger = logging.getLogger(__name__) @@ -30,11 +40,6 @@ 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 @@ -44,7 +49,6 @@ def randomize(val, key: str = "") -> str: Reuse same randomization if value was already randomized """ - global RANDOMDATA # noqa: PLW0602 if not RANDOMIZE: return str(val) randomstr = RANDOMDATA.get(val, "") @@ -118,7 +122,7 @@ def export(filename: str, d: dict = None, randomkeys: bool = False) -> None: if len(d) == 0: CONSOLE.info("WARNING: File %s not saved because JSON is empty", filename) return - elif RANDOMIZE: + if RANDOMIZE: d = check_keys(d) # Randomize also the keys for the api dictionary export if randomkeys: @@ -137,25 +141,19 @@ def export(filename: str, d: dict = None, randomkeys: bool = False) -> None: return -async def main() -> bool: # noqa: C901 +async def main() -> ( + bool +): # noqa: C901 # pylint: disable=too-many-branches,too-many-statements """Run main function to export config.""" - global USER, PASSWORD, COUNTRY, RANDOMIZE # noqa: PLW0603, W0603 + global RANDOMIZE # noqa: PLW0603, W0603 # pylint: disable=global-statement CONSOLE.info("Exporting found Anker Solix system data for all assigned sites:") - if USER == "": - CONSOLE.info("\nEnter Anker Account credentials:") - USER = input("Username (email): ") - if USER == "": - return False - PASSWORD = getpass("Password: ") - if PASSWORD == "": - return False - COUNTRY = input("Country ID (e.g. DE): ") - if COUNTRY == "": - return False try: + user = common.user() async with ClientSession() as websession: CONSOLE.info("\nTrying authentication...") - myapi = api.AnkerSolixApi(USER, PASSWORD, COUNTRY, websession, _LOGGER) + myapi = api.AnkerSolixApi( + user, common.password(), common.country(), websession, _LOGGER + ) if await myapi.async_authenticate(): CONSOLE.info("OK") else: @@ -187,6 +185,7 @@ async def main() -> bool: # noqa: C901 CONSOLE.info("Sites: %s, Devices: %s", len(myapi.sites), len(myapi.devices)) _LOGGER.debug(json.dumps(myapi.devices, indent=2)) + # pylint: disable=protected-access # Query API using direct endpoints to save full response of each query in json files CONSOLE.info("\nExporting homepage...") export( @@ -397,7 +396,7 @@ async def main() -> bool: # noqa: C901 ) CONSOLE.info( - "\nCompleted export of Anker Solix system data for user %s", USER + "\nCompleted export of Anker Solix system data for user %s", user ) if RANDOMIZE: CONSOLE.info( @@ -415,7 +414,7 @@ async def main() -> bool: # noqa: C901 return True except (ClientError, errors.AnkerSolixError) as err: - CONSOLE.info("%s: %s", type(err), err) + CONSOLE.error("%s: %s", type(err), err) return False @@ -425,6 +424,6 @@ if __name__ == "__main__": if not asyncio.run(main()): CONSOLE.info("Aborted!") except KeyboardInterrupt: - CONSOLE.info("Aborted!") - except Exception as exception: - CONSOLE.info("%s: %s", type(exception), exception) + CONSOLE.warning("Aborted!") + except Exception as exception: # pylint: disable=broad-exception-caught + CONSOLE.exception("%s: %s", type(exception), exception) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..406512d --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,21 @@ +[project] +name = "Anker Solix Api" +version = "0.0.1" +description = "Python library for Anker Solix Power devices (Solarbank, Inverter etc)" +readme = "README.md" +requires-python = ">=3.11" +classifiers = [ + "Programming Language :: Python :: 3", + "Operating System :: OS Independent", +] + +[project.urls] +"Homepage" = "https://github.com/thomluther/anker-solix-api" +"Bug Tracker" = "https://github.com/thomluther/anker-solix-api/issues" + +[tool.isort] +profile="black" + +[tool.pylint] +jobs=0 +disable = ["import-error", "line-too-long", "invalid-name"] diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 0000000..35f4f15 --- /dev/null +++ b/setup.cfg @@ -0,0 +1,2 @@ +[flake8] +ignore = E501,E226,W503,E231 diff --git a/solarbank_monitor.py b/solarbank_monitor.py index a66b2d5..36746f9 100755 --- a/solarbank_monitor.py +++ b/solarbank_monitor.py @@ -1,9 +1,16 @@ #!/usr/bin/env python -"""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. +"""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 @@ -13,11 +20,11 @@ import os import sys import time from datetime import datetime, timedelta -from getpass import getpass from aiohttp import ClientSession from aiohttp.client_exceptions import ClientError +import common from api import api, errors _LOGGER: logging.Logger = logging.getLogger(__name__) @@ -27,11 +34,8 @@ 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 # default refresh interval in seconds +INTERACTIVE = True def clearscreen(): @@ -51,15 +55,17 @@ def get_subfolders(folder: str) -> list: return [] -async def main() -> None: # noqa: C901 +async def main() -> ( # noqa: C901 # pylint: disable=too-many-locals,too-many-branches,too-many-statements + None +): """Run Main routine to start Solarbank monitor in a loop.""" - global USER, PASSWORD, COUNTRY, REFRESH # noqa: W0603, PLW0603 + global REFRESH # pylint: disable=global-statement CONSOLE.info("Solarbank Monitor:") # get list of possible example and export folders to test the monitor against exampleslist = get_subfolders( os.path.join(os.path.dirname(__file__), "examples") ) + get_subfolders(os.path.join(os.path.dirname(__file__), "exports")) - if USER == "": + if INTERACTIVE: if exampleslist: CONSOLE.info("\nSelect the input source for the monitor:") CONSOLE.info("(0) Real time from Anker cloud") @@ -74,16 +80,6 @@ async def main() -> None: # noqa: C901 return False if (selection := int(selection)) == 0: use_file = False - CONSOLE.info("\nEnter Anker Account credentials:") - USER = input("Username (email): ") - if USER == "": - return False - PASSWORD = getpass("Password: ") - if PASSWORD == "": - return False - COUNTRY = input("Country ID (e.g. DE): ") - if COUNTRY == "": - return False else: use_file = True testfolder = exampleslist[selection - 1] @@ -91,7 +87,9 @@ async def main() -> None: # noqa: C901 use_file = False try: async with ClientSession() as websession: - myapi = api.AnkerSolixApi(USER, PASSWORD, COUNTRY, websession, _LOGGER) + myapi = api.AnkerSolixApi( + common.user(), common.password(), common.country(), websession, _LOGGER + ) if use_file: # set the correct test folder for Api myapi.testDir(testfolder) @@ -107,7 +105,7 @@ async def main() -> None: # noqa: C901 ) if not resp: break - elif resp.isdigit() and 10 <= int(resp) <= 600: + if resp.isdigit() and 10 <= int(resp) <= 600: REFRESH = int(resp) break @@ -137,13 +135,16 @@ async def main() -> None: # noqa: C901 # schedules = {} clearscreen() CONSOLE.info( - f"Solarbank Monitor (refresh {REFRESH} s, details refresh {10*REFRESH} s):" + "Solarbank Monitor (refresh %s s, details refresh %s s):", + REFRESH, + 10 * REFRESH, ) if use_file: - CONSOLE.info(f"Using input source folder: {myapi.testDir()}") + CONSOLE.info("Using input source folder: %s", myapi.testDir()) CONSOLE.info( - f"Sites: {len(myapi.sites)}, Devices: {len(myapi.devices)}" + "Sites: %s, Devices: %s", len(myapi.sites), len(myapi.devices) ) + # pylint: disable=logging-fstring-interpolation for sn, dev in myapi.devices.items(): devtype = dev.get("type", "Unknown") admin = dev.get("is_admin", False) @@ -255,7 +256,7 @@ async def main() -> None: # noqa: C901 return False except (ClientError, errors.AnkerSolixError) as err: - CONSOLE.info("%s: %s", type(err), err) + CONSOLE.error("%s: %s", type(err), err) return False @@ -263,8 +264,8 @@ async def main() -> None: # noqa: C901 if __name__ == "__main__": try: if not asyncio.run(main()): - CONSOLE.info("\nAborted!") + CONSOLE.warning("\nAborted!") except KeyboardInterrupt: - CONSOLE.info("\nAborted!") - except Exception as exception: + CONSOLE.warning("\nAborted!") + except Exception as exception: # pylint: disable=broad-exception-caught CONSOLE.exception("%s: %s", type(exception), exception) diff --git a/test_api.py b/test_api.py index 44f5209..d6195cd 100755 --- a/test_api.py +++ b/test_api.py @@ -1,16 +1,19 @@ #!/usr/bin/env python -"""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. +""" +# pylint: disable=duplicate-code import asyncio import json import logging -import os import sys -from datetime import datetime from aiohttp import ClientSession -from api import api, credentials +import common +from api import api _LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER.addHandler(logging.StreamHandler(sys.stdout)) @@ -28,23 +31,21 @@ async def main() -> None: # Update your account credentials in api.credentials.py or directly in this file for testing # Both files are added to .gitignore to avoid local changes being comitted to git myapi = api.AnkerSolixApi( - credentials.USERNAME, - credentials.PASSWORD, - credentials.COUNTRYID, + common.user(), + common.password(), + common.country(), websession, _LOGGER, ) # show login response - """ - #new = await myapi.async_authenticate(restart=True) # enforce new login data from server - new = await myapi.async_authenticate() # receive new or load cached login data - if new: - CONSOLE.info("Received Login response:") - else: - CONSOLE.info("Cached Login response:") - CONSOLE.info(json.dumps(myapi._login_response, indent=2)) # show used login response for API reqests - """ + # 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: + # CONSOLE.info("Received Login response:") + # else: + # 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 @@ -56,62 +57,58 @@ async def main() -> None: CONSOLE.info(json.dumps(myapi.devices, indent=2)) # test api methods - """ - 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)) + # 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)) - 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)) - """ + # 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 - """ - 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', "device_sn": "9JVB42LJK8J0P5RY"})), indent=2)) # json parameters unknown: May need site_id and device_sn of inverter in system? - 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 endpoints directly + # """ + # 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', "device_sn": "9JVB42LJK8J0P5RY"})), indent=2)) # json parameters unknown: May need site_id and device_sn of inverter in system? + # 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(os.path.join(os.path.dirname(__file__), "examples", "example1")) - await myapi.update_sites(fromFile=True) - await myapi.update_device_details(fromFile=True) - CONSOLE.info(json.dumps(myapi.sites,indent=2)) - CONSOLE.info(json.dumps(myapi.devices,indent=2)) - """ + # myapi.testDir(os.path.join(os.path.dirname(__file__), "examples", "example1")) + # await myapi.update_sites(fromFile=True) + # await myapi.update_device_details(fromFile=True) + # CONSOLE.info(json.dumps(myapi.sites,indent=2)) + # CONSOLE.info(json.dumps(myapi.devices,indent=2)) - except Exception as exception: - CONSOLE.info(f"{type(exception)}: {exception}") + except Exception as exception: # pylint: disable=broad-exception-caught + CONSOLE.error("%s: %s", type(exception), exception) # run async main if __name__ == "__main__": try: asyncio.run(main()) - except Exception as err: - CONSOLE.info(f"{type(err)}: {err}") + except Exception as err: # pylint: disable=broad-exception-caught + CONSOLE.exception("%s: %s", type(err), err)