110 lines
7.2 KiB
Python
110 lines
7.2 KiB
Python
"""Example exec module to test the Anker API for various methods or direct endpoint requests with various parameters."""
|
|
|
|
import asyncio
|
|
from datetime import datetime
|
|
import json
|
|
import logging
|
|
import sys
|
|
|
|
from aiohttp import ClientSession
|
|
from api import api
|
|
from settings import *
|
|
|
|
_LOGGER: logging.Logger = logging.getLogger(__name__)
|
|
_LOGGER.addHandler(logging.StreamHandler(sys.stdout))
|
|
# _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."""
|
|
CONSOLE.info("Testing Solix API:")
|
|
try:
|
|
async with ClientSession() as websession:
|
|
myapi = api.AnkerSolixApi(
|
|
USER, PASSWORD, 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
|
|
"""
|
|
|
|
# test site api methods
|
|
|
|
await myapi.update_sites()
|
|
await myapi.update_device_details()
|
|
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
|
|
"""
|
|
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))
|
|
"""
|
|
|
|
# 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'})), 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)
|
|
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}")
|
|
|
|
|
|
# run async main
|
|
if __name__ == "__main__":
|
|
try:
|
|
asyncio.run(main())
|
|
except Exception as err:
|
|
CONSOLE.info(f"{type(err)}: {err}")
|