robin
/
anker-solix-api
mirror of https://github.com/thomluther/anker-solix-api.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

style: apply pylint/flake8/etc, add project setup

This commit is contained in:
Jan Stürtz 2024-02-21 21:54:38 +01:00
parent 99e64fa128
commit 0d1ebe470d
No known key found for this signature in database
GPG Key ID: 8183CC01FD23D435
11 changed files with 246 additions and 179 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -16,3 +16,4 @@ coverage.xml
**/credentials* **/credentials*
/node_modules/ /node_modules/
/client.py /client.py
/daily_energy_*.csv

12
README.md
View File

@ -48,13 +48,16 @@ from aiohttp import ClientSession
from api import api, errors from api import api, errors
_LOGGER: logging.Logger = logging.getLogger(__name__) _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: async def main() -> None:
"""Create the aiohttp session and run the example.""" """Create the aiohttp session and run the example."""
async with ClientSession() as websession: async with ClientSession() as websession:
"""put your code here, example""" """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_sites()
await myapi.update_device_details() await myapi.update_device_details()
print("System Overview:") print("System Overview:")
@ -62,12 +65,13 @@ async def main() -> None:
print("Device Overview:") print("Device Overview:")
print(json.dumps(myapi.devices, indent=2)) print(json.dumps(myapi.devices, indent=2))
# run async main # run async main
if __name__ == '__main__': if __name__ == "__main__":
try: try:
asyncio.run(main()) asyncio.run(main())
except Exception as err: except Exception as err:
print(f'{type(err)}: {err}') print(f"{type(err)}: {err}")
``` ```
The AnkerSolixApi class provides 2 main methods: The AnkerSolixApi class provides 2 main methods:

10
api/api.py
View File

@ -5,6 +5,8 @@ pip install cryptography
pip install aiohttp 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 from __future__ import annotations
import contextlib import contextlib
@ -247,7 +249,7 @@ class SolarbankStatus(Enum):
unknown = "unknown" 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.""" """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__( def __init__(
@ -644,7 +646,7 @@ class AnkerSolixApi:
endpoint: str, endpoint: str,
*, *,
headers: dict | None = None, headers: dict | None = None,
json: dict | None = None, # noqa: W0621 json: dict | None = None, # noqa: W0621 # pylint: disable=redefined-outer-name
) -> dict: ) -> dict:
"""Handle all requests to the API. This is also called recursively by login requests if necessary.""" """Handle all requests to the API. This is also called recursively by login requests if necessary."""
if not headers: if not headers:
@ -768,7 +770,9 @@ class AnkerSolixApi:
self._logger.error("Response Text: %s", body_text) self._logger.error("Response Text: %s", body_text)
raise err 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. """Get the latest info for all accessible sites and update class site and device variables.
Example data: Example data:

8
api/credentials.py
View File

@ -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"

53
common.py Normal file
View File

@ -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

65
energy_csv.py
View File

@ -1,13 +1,19 @@
#!/usr/bin/env python #!/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 asyncio
import csv import csv
@ -15,10 +21,10 @@ import json
import logging import logging
import sys import sys
from datetime import datetime from datetime import datetime
from getpass import getpass
from aiohttp import ClientSession from aiohttp import ClientSession
import common
from api import api from api import api
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
@ -28,31 +34,16 @@ CONSOLE: logging.Logger = logging.getLogger("console")
CONSOLE.addHandler(logging.StreamHandler(sys.stdout)) CONSOLE.addHandler(logging.StreamHandler(sys.stdout))
CONSOLE.setLevel(logging.INFO) CONSOLE.setLevel(logging.INFO)
# Optional default Anker Account credentials to be used
USER = ""
PASSWORD = ""
COUNTRY = ""
async def main() -> None: async def main() -> None:
"""Run main to export energy history from cloud.""" """Run main to export energy history from cloud."""
global USER, PASSWORD, COUNTRY # noqa: PLW0603
CONSOLE.info("Exporting daily Energy data for Anker Solarbank:") 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: try:
async with ClientSession() as websession: async with ClientSession() as websession:
CONSOLE.info("\nTrying authentication...") 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(): if await myapi.async_authenticate():
CONSOLE.info("OK") CONSOLE.info("OK")
else: else:
@ -65,19 +56,19 @@ async def main() -> None:
CONSOLE.info("NO INFO") CONSOLE.info("NO INFO")
return False return False
CONSOLE.info("OK") 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)) _LOGGER.debug(json.dumps(myapi.devices, indent=2))
for sn, device in myapi.devices.items(): for sn, device in myapi.devices.items():
if device.get("type") == "solarbank": 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: try:
daystr = input( daystr = input(
"\nEnter start day for daily energy data (yyyy-mm-dd) or enter to skip: " "\nEnter start day for daily energy data (yyyy-mm-dd) or enter to skip: "
) )
if daystr == "": if daystr == "":
CONSOLE.info( CONSOLE.info(
f"Skipped SN: {sn}, checking for next Solarbank..." "Skipped SN: %s, checking for next Solarbank...", sn
) )
continue continue
startday = datetime.fromisoformat(daystr) startday = datetime.fromisoformat(daystr)
@ -94,7 +85,8 @@ async def main() -> None:
except ValueError: except ValueError:
return False return False
CONSOLE.info( 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( data = await myapi.energy_daily(
siteId=device.get("site_id"), siteId=device.get("site_id"),
@ -114,7 +106,8 @@ async def main() -> None:
writer.writeheader() writer.writeheader()
writer.writerows(data.values()) writer.writerows(data.values())
CONSOLE.info( CONSOLE.info(
f"\nCompleted: Successfully exported data to {filename}" "\nCompleted: Successfully exported data to %s",
filename,
) )
return True return True
@ -123,8 +116,8 @@ async def main() -> None:
CONSOLE.info("No accepted Solarbank device found.") CONSOLE.info("No accepted Solarbank device found.")
return False return False
except Exception as err: except Exception as err: # pylint: disable=broad-exception-caught
CONSOLE.info(f"{type(err)}: {err}") CONSOLE.error("%s: %s", type(err), err)
return False return False
@ -132,6 +125,6 @@ async def main() -> None:
if __name__ == "__main__": if __name__ == "__main__":
try: try:
if not asyncio.run(main()): if not asyncio.run(main()):
CONSOLE.info("Aborted!") CONSOLE.warning("Aborted!")
except Exception as exception: except Exception as exception: # pylint: disable=broad-exception-caught
CONSOLE.info(f"{type(exception)}: {exception}") CONSOLE.exception("%s: %s", type(exception), exception)

63
export_system.py
View File

@ -1,12 +1,22 @@
#!/usr/bin/env python #!/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. 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. Upon successfull authentication, you can specify a subfolder for the exported
You can review the response files afterwards. They can be used as examples for dedicated data extraction from the devices. JSON files received as API query response, defaulting to your nick name.
Optionally the API class can use the json files for debugging and testing on various system outputs.
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 asyncio
import json import json
@ -16,11 +26,11 @@ import random
import string import string
import sys import sys
import time import time
from getpass import getpass
from aiohttp import ClientSession from aiohttp import ClientSession
from aiohttp.client_exceptions import ClientError from aiohttp.client_exceptions import ClientError
import common
from api import api, errors from api import api, errors
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
@ -30,11 +40,6 @@ CONSOLE: logging.Logger = logging.getLogger("console")
CONSOLE.addHandler(logging.StreamHandler(sys.stdout)) CONSOLE.addHandler(logging.StreamHandler(sys.stdout))
CONSOLE.setLevel(logging.INFO) CONSOLE.setLevel(logging.INFO)
# Optional default Anker Account credentials to be used
USER = ""
PASSWORD = ""
COUNTRY = ""
RANDOMIZE = True # Global flag to save randomize decission RANDOMIZE = True # Global flag to save randomize decission
RANDOMDATA = {} # Global dict for randomized data, printed at the end 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 Reuse same randomization if value was already randomized
""" """
global RANDOMDATA # noqa: PLW0602
if not RANDOMIZE: if not RANDOMIZE:
return str(val) return str(val)
randomstr = RANDOMDATA.get(val, "") randomstr = RANDOMDATA.get(val, "")
@ -118,7 +122,7 @@ def export(filename: str, d: dict = None, randomkeys: bool = False) -> None:
if len(d) == 0: if len(d) == 0:
CONSOLE.info("WARNING: File %s not saved because JSON is empty", filename) CONSOLE.info("WARNING: File %s not saved because JSON is empty", filename)
return return
elif RANDOMIZE: if RANDOMIZE:
d = check_keys(d) d = check_keys(d)
# Randomize also the keys for the api dictionary export # Randomize also the keys for the api dictionary export
if randomkeys: if randomkeys:
@ -137,25 +141,19 @@ def export(filename: str, d: dict = None, randomkeys: bool = False) -> None:
return 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.""" """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:") 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: try:
user = common.user()
async with ClientSession() as websession: async with ClientSession() as websession:
CONSOLE.info("\nTrying authentication...") 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(): if await myapi.async_authenticate():
CONSOLE.info("OK") CONSOLE.info("OK")
else: else:
@ -187,6 +185,7 @@ async def main() -> bool: # noqa: C901
CONSOLE.info("Sites: %s, Devices: %s", len(myapi.sites), len(myapi.devices)) CONSOLE.info("Sites: %s, Devices: %s", len(myapi.sites), len(myapi.devices))
_LOGGER.debug(json.dumps(myapi.devices, indent=2)) _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 # Query API using direct endpoints to save full response of each query in json files
CONSOLE.info("\nExporting homepage...") CONSOLE.info("\nExporting homepage...")
export( export(
@ -397,7 +396,7 @@ async def main() -> bool: # noqa: C901
) )
CONSOLE.info( 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: if RANDOMIZE:
CONSOLE.info( CONSOLE.info(
@ -415,7 +414,7 @@ async def main() -> bool: # noqa: C901
return True return True
except (ClientError, errors.AnkerSolixError) as err: except (ClientError, errors.AnkerSolixError) as err:
CONSOLE.info("%s: %s", type(err), err) CONSOLE.error("%s: %s", type(err), err)
return False return False
@ -425,6 +424,6 @@ if __name__ == "__main__":
if not asyncio.run(main()): if not asyncio.run(main()):
CONSOLE.info("Aborted!") CONSOLE.info("Aborted!")
except KeyboardInterrupt: except KeyboardInterrupt:
CONSOLE.info("Aborted!") CONSOLE.warning("Aborted!")
except Exception as exception: except Exception as exception: # pylint: disable=broad-exception-caught
CONSOLE.info("%s: %s", type(exception), exception) CONSOLE.exception("%s: %s", type(exception), exception)

21
pyproject.toml Normal file
View File

@ -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"]

2
setup.cfg Normal file
View File

@ -0,0 +1,2 @@
[flake8]
ignore = E501,E226,W503,E231

65
solarbank_monitor.py
View File

@ -1,9 +1,16 @@
#!/usr/bin/env python #!/usr/bin/env python
"""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
This module will prompt for the Anker account details if not pre-set in the header. displaying important solarbank parameters This module will prompt for the Anker
Upon successfull authentication, you will see the solarbank parameters displayed and refreshed at reqular interval. account details if not pre-set in the header. Upon successfull authentication,
Note: When the system owning account is used, more details for the solarbank can be queried and displayed. you will see the solarbank parameters displayed and refreshed at reqular
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. 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 """ # noqa: D205
import asyncio import asyncio
@ -13,11 +20,11 @@ import os
import sys import sys
import time import time
from datetime import datetime, timedelta from datetime import datetime, timedelta
from getpass import getpass
from aiohttp import ClientSession from aiohttp import ClientSession
from aiohttp.client_exceptions import ClientError from aiohttp.client_exceptions import ClientError
import common
from api import api, errors from api import api, errors
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
@ -27,11 +34,8 @@ CONSOLE: logging.Logger = logging.getLogger("console")
CONSOLE.addHandler(logging.StreamHandler(sys.stdout)) CONSOLE.addHandler(logging.StreamHandler(sys.stdout))
CONSOLE.setLevel(logging.INFO) CONSOLE.setLevel(logging.INFO)
# Optional default Anker Account credentials to be used
USER = ""
PASSWORD = ""
COUNTRY = ""
REFRESH = 30 # default refresh interval in seconds REFRESH = 30 # default refresh interval in seconds
INTERACTIVE = True
def clearscreen(): def clearscreen():
@ -51,15 +55,17 @@ def get_subfolders(folder: str) -> list:
return [] 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.""" """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:") CONSOLE.info("Solarbank Monitor:")
# get list of possible example and export folders to test the monitor against # get list of possible example and export folders to test the monitor against
exampleslist = get_subfolders( exampleslist = get_subfolders(
os.path.join(os.path.dirname(__file__), "examples") os.path.join(os.path.dirname(__file__), "examples")
) + get_subfolders(os.path.join(os.path.dirname(__file__), "exports")) ) + get_subfolders(os.path.join(os.path.dirname(__file__), "exports"))
if USER == "": if INTERACTIVE:
if exampleslist: if exampleslist:
CONSOLE.info("\nSelect the input source for the monitor:") CONSOLE.info("\nSelect the input source for the monitor:")
CONSOLE.info("(0) Real time from Anker cloud") CONSOLE.info("(0) Real time from Anker cloud")
@ -74,16 +80,6 @@ async def main() -> None: # noqa: C901
return False return False
if (selection := int(selection)) == 0: if (selection := int(selection)) == 0:
use_file = False 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: else:
use_file = True use_file = True
testfolder = exampleslist[selection - 1] testfolder = exampleslist[selection - 1]
@ -91,7 +87,9 @@ async def main() -> None: # noqa: C901
use_file = False use_file = False
try: try:
async with ClientSession() as websession: 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: if use_file:
# set the correct test folder for Api # set the correct test folder for Api
myapi.testDir(testfolder) myapi.testDir(testfolder)
@ -107,7 +105,7 @@ async def main() -> None: # noqa: C901
) )
if not resp: if not resp:
break break
elif resp.isdigit() and 10 <= int(resp) <= 600: if resp.isdigit() and 10 <= int(resp) <= 600:
REFRESH = int(resp) REFRESH = int(resp)
break break
@ -137,13 +135,16 @@ async def main() -> None: # noqa: C901
# schedules = {} # schedules = {}
clearscreen() clearscreen()
CONSOLE.info( 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: if use_file:
CONSOLE.info(f"Using input source folder: {myapi.testDir()}") CONSOLE.info("Using input source folder: %s", myapi.testDir())
CONSOLE.info( 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(): for sn, dev in myapi.devices.items():
devtype = dev.get("type", "Unknown") devtype = dev.get("type", "Unknown")
admin = dev.get("is_admin", False) admin = dev.get("is_admin", False)
@ -255,7 +256,7 @@ async def main() -> None: # noqa: C901
return False return False
except (ClientError, errors.AnkerSolixError) as err: except (ClientError, errors.AnkerSolixError) as err:
CONSOLE.info("%s: %s", type(err), err) CONSOLE.error("%s: %s", type(err), err)
return False return False
@ -263,8 +264,8 @@ async def main() -> None: # noqa: C901
if __name__ == "__main__": if __name__ == "__main__":
try: try:
if not asyncio.run(main()): if not asyncio.run(main()):
CONSOLE.info("\nAborted!") CONSOLE.warning("\nAborted!")
except KeyboardInterrupt: except KeyboardInterrupt:
CONSOLE.info("\nAborted!") CONSOLE.warning("\nAborted!")
except Exception as exception: except Exception as exception: # pylint: disable=broad-exception-caught
CONSOLE.exception("%s: %s", type(exception), exception) CONSOLE.exception("%s: %s", type(exception), exception)

125
test_api.py
View File

@ -1,16 +1,19 @@
#!/usr/bin/env python #!/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 asyncio
import json import json
import logging import logging
import os
import sys import sys
from datetime import datetime
from aiohttp import ClientSession from aiohttp import ClientSession
from api import api, credentials import common
from api import api
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
_LOGGER.addHandler(logging.StreamHandler(sys.stdout)) _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 # 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 # Both files are added to .gitignore to avoid local changes being comitted to git
myapi = api.AnkerSolixApi( myapi = api.AnkerSolixApi(
credentials.USERNAME, common.user(),
credentials.PASSWORD, common.password(),
credentials.COUNTRYID, common.country(),
websession, websession,
_LOGGER, _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(restart=True) # enforce new login data from server # new = await myapi.async_authenticate() # receive new or load cached login data
new = await myapi.async_authenticate() # receive new or load cached login data # if new:
if new: # CONSOLE.info("Received Login response:")
CONSOLE.info("Received Login response:") # else:
else: # CONSOLE.info("Cached Login response:")
CONSOLE.info("Cached Login response:") # CONSOLE.info(json.dumps(myapi._login_response, indent=2)) # show used login response for API reqests
CONSOLE.info(json.dumps(myapi._login_response, indent=2)) # show used login response for API reqests
"""
# test site api methods # test site api methods
@ -56,62 +57,58 @@ async def main() -> None:
CONSOLE.info(json.dumps(myapi.devices, indent=2)) CONSOLE.info(json.dumps(myapi.devices, indent=2))
# test api methods # test api methods
""" # CONSOLE.info(json.dumps(await myapi.get_site_list(), 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_homepage(), indent=2)) # CONSOLE.info(json.dumps(await myapi.get_bind_devices(), 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_user_devices(), indent=2)) # CONSOLE.info(json.dumps(await myapi.get_charging_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_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_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_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_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_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_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_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="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_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.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.home_load_chart(siteId='efaca6b5-f4a0-e82e-3b2e-6b9cf90ded8c'), indent=2))
""" # """
# test api endpoints directly # # 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["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["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["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["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["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["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["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["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["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["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_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_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_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["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["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)) # 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 # test api from json files
""" # myapi.testDir(os.path.join(os.path.dirname(__file__), "examples", "example1"))
myapi.testDir(os.path.join(os.path.dirname(__file__), "examples", "example1")) # await myapi.update_sites(fromFile=True)
await myapi.update_sites(fromFile=True) # await myapi.update_device_details(fromFile=True)
await myapi.update_device_details(fromFile=True) # CONSOLE.info(json.dumps(myapi.sites,indent=2))
CONSOLE.info(json.dumps(myapi.sites,indent=2)) # CONSOLE.info(json.dumps(myapi.devices,indent=2))
CONSOLE.info(json.dumps(myapi.devices,indent=2))
"""
except Exception as exception: except Exception as exception: # pylint: disable=broad-exception-caught
CONSOLE.info(f"{type(exception)}: {exception}") CONSOLE.error("%s: %s", type(exception), exception)
# run async main # run async main
if __name__ == "__main__": if __name__ == "__main__":
try: try:
asyncio.run(main()) asyncio.run(main())
except Exception as err: except Exception as err: # pylint: disable=broad-exception-caught
CONSOLE.info(f"{type(err)}: {err}") CONSOLE.exception("%s: %s", type(err), err)