285 lines
14 KiB
Python
Executable File
285 lines
14 KiB
Python
Executable File
#!/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.
|
|
|
|
""" # noqa: D205
|
|
|
|
import asyncio
|
|
from datetime import datetime, timedelta
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
from aiohttp import ClientSession
|
|
from aiohttp.client_exceptions import ClientError
|
|
from api import api, errors
|
|
import common
|
|
|
|
_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)
|
|
|
|
REFRESH = 30 # default refresh interval in seconds
|
|
INTERACTIVE = True
|
|
|
|
|
|
def clearscreen():
|
|
"""Clear the terminal screen."""
|
|
if sys.stdin is sys.__stdin__: # check if not in IDLE shell
|
|
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?
|
|
|
|
|
|
def get_subfolders(folder: str) -> list:
|
|
"""Get the full pathnames of all subfolders for given folder as list."""
|
|
if os.path.isdir(folder):
|
|
return [os.path.abspath(f) for f in os.scandir(folder) if f.is_dir()]
|
|
return []
|
|
|
|
|
|
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 REFRESH # pylint: disable=global-statement # noqa: PLW0603
|
|
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 INTERACTIVE:
|
|
if exampleslist:
|
|
CONSOLE.info("\nSelect the input source for the monitor:")
|
|
CONSOLE.info("(0) Real time from Anker cloud")
|
|
for idx, filename in enumerate(exampleslist, start=1):
|
|
CONSOLE.info("(%s) %s", idx, filename)
|
|
selection = input(f"Input Source number (0-{len(exampleslist)}): ")
|
|
if (
|
|
not selection.isdigit()
|
|
or int(selection) < 0
|
|
or int(selection) > len(exampleslist)
|
|
):
|
|
return False
|
|
if (selection := int(selection)) == 0:
|
|
use_file = False
|
|
else:
|
|
use_file = True
|
|
testfolder = exampleslist[selection - 1]
|
|
else:
|
|
use_file = False
|
|
try:
|
|
async with ClientSession() as websession:
|
|
myapi = api.AnkerSolixApi(
|
|
common.user(), common.password(), common.country(), websession, _LOGGER
|
|
)
|
|
if use_file:
|
|
# set the correct test folder for Api
|
|
myapi.testDir(testfolder)
|
|
elif await myapi.async_authenticate():
|
|
CONSOLE.info("Anker Cloud authentication: OK")
|
|
else:
|
|
# Login validation will be done during first API call
|
|
CONSOLE.info("Anker Cloud authentication: CACHED")
|
|
|
|
while True:
|
|
resp = input(
|
|
f"\nHow many seconds refresh interval should be used? (10-600, default: {REFRESH}): "
|
|
)
|
|
if not resp:
|
|
break
|
|
if resp.isdigit() and 10 <= int(resp) <= 600:
|
|
REFRESH = int(resp)
|
|
break
|
|
|
|
# Run loop to update Solarbank parameters
|
|
now = datetime.now().astimezone()
|
|
next_refr = now
|
|
next_dev_refr = now
|
|
col1 = 15
|
|
col2 = 23
|
|
col3 = 14
|
|
t1 = 2
|
|
t2 = 5
|
|
t3 = 5
|
|
t4 = 6
|
|
t5 = 6
|
|
t6 = 10
|
|
t7 = 6
|
|
t8 = 6
|
|
while True:
|
|
CONSOLE.info("\n")
|
|
now = datetime.now().astimezone()
|
|
if next_refr <= now:
|
|
CONSOLE.info("Running site refresh...")
|
|
await myapi.update_sites(fromFile=use_file)
|
|
next_refr = now + timedelta(seconds=REFRESH)
|
|
if next_dev_refr <= now:
|
|
CONSOLE.info("Running device details refresh...")
|
|
await myapi.update_device_details(fromFile=use_file)
|
|
next_dev_refr = next_refr + timedelta(seconds=REFRESH * 9)
|
|
# schedules = {}
|
|
clearscreen()
|
|
CONSOLE.info(
|
|
"Solarbank Monitor (refresh %s s, details refresh %s s):",
|
|
REFRESH,
|
|
10 * REFRESH,
|
|
)
|
|
if use_file:
|
|
CONSOLE.info("Using input source folder: %s", myapi.testDir())
|
|
if len(myapi.sites) > 0:
|
|
update_time = (
|
|
(next(iter(myapi.sites.values()))).get("solarbank_info") or {}
|
|
).get("updated_time") or "Unknown"
|
|
else:
|
|
update_time = "Unknown"
|
|
CONSOLE.info(
|
|
f"Sites: {len(myapi.sites)}, Devices: {len(myapi.sites)}, Data-Updated: {update_time}"
|
|
)
|
|
# pylint: disable=logging-fstring-interpolation
|
|
for sn, dev in myapi.devices.items():
|
|
devtype = dev.get("type", "Unknown")
|
|
admin = dev.get("is_admin", False)
|
|
CONSOLE.info(
|
|
f"{'Device':<{col1}}: {(dev.get('name','NoName')):<{col2}} {'Alias':<{col3}}: {dev.get('alias','Unknown')}"
|
|
)
|
|
CONSOLE.info(
|
|
f"{'Serialnumber':<{col1}}: {sn:<{col2}} {'Admin':<{col3}}: {'YES' if admin else 'NO'}"
|
|
)
|
|
siteid = dev.get("site_id", "")
|
|
CONSOLE.info(f"{'Site ID':<{col1}}: {siteid}")
|
|
for fsn, fitting in (dev.get("fittings") or {}).items():
|
|
CONSOLE.info(
|
|
f"{'Accessory':<{col1}}: {fitting.get('device_name',''):<{col2}} {'Serialnumber':<{col3}}: {fsn}"
|
|
)
|
|
CONSOLE.info(
|
|
f"{'Wifi SSID':<{col1}}: {dev.get('wifi_name',''):<{col2}}"
|
|
)
|
|
online = dev.get("wifi_online")
|
|
CONSOLE.info(
|
|
f"{'Wifi state':<{col1}}: {('Unknown' if online is None else 'Online' if online else 'Offline'):<{col2}} {'Signal':<{col3}}: {dev.get('wifi_signal','---'):>4} %"
|
|
)
|
|
if devtype == "solarbank":
|
|
upgrade = dev.get("auto_upgrade")
|
|
CONSOLE.info(
|
|
f"{'SW Version':<{col1}}: {dev.get('sw_version','Unknown'):<{col2}} {'Auto-Upgrade':<{col3}}: {'Unknown' if upgrade is None else 'Enabled' if upgrade else 'Disabled'}"
|
|
)
|
|
CONSOLE.info(
|
|
f"{'Cloud Status':<{col1}}: {dev.get('status_desc','Unknown'):<{col2}} {'Status code':<{col3}}: {str(dev.get('status','-'))}"
|
|
)
|
|
CONSOLE.info(
|
|
f"{'Charge Status':<{col1}}: {dev.get('charging_status_desc','Unknown'):<{col2}} {'Status code':<{col3}}: {str(dev.get('charging_status','-'))}"
|
|
)
|
|
soc = f"{dev.get('battery_soc','---'):>4} %"
|
|
CONSOLE.info(
|
|
f"{'State Of Charge':<{col1}}: {soc:<{col2}} {'Min SOC':<{col3}}: {str(dev.get('power_cutoff','--')):>4} %"
|
|
)
|
|
energy = f"{dev.get('battery_energy','----'):>4} Wh"
|
|
CONSOLE.info(
|
|
f"{'Battery Energy':<{col1}}: {energy:<{col2}} {'Capacity':<{col3}}: {str(dev.get('battery_capacity','----')):>4} Wh"
|
|
)
|
|
unit = dev.get("power_unit", "W")
|
|
CONSOLE.info(
|
|
f"{'Solar Power':<{col1}}: {dev.get('input_power',''):>4} {unit:<{col2-5}} {'Output Power':<{col3}}: {dev.get('output_power',''):>4} {unit}"
|
|
)
|
|
preset = dev.get("set_output_power") or "---"
|
|
site_preset = dev.get("set_system_output_power") or "---"
|
|
CONSOLE.info(
|
|
f"{'Charge Power':<{col1}}: {dev.get('charging_power',''):>4} {unit:<{col2-5}} {'Device Preset':<{col3}}: {preset:>4} {unit}"
|
|
)
|
|
# update schedule with device details refresh and print it
|
|
if admin:
|
|
data = dev.get("schedule", {})
|
|
CONSOLE.info(
|
|
f"{'Schedule (Now)':<{col1}}: {now.strftime('%H:%M:%S UTC %z'):<{col2}} {'System Preset':<{col3}}: {str(site_preset).replace('W',''):>4} W"
|
|
)
|
|
CONSOLE.info(
|
|
f"{'ID':<{t1}} {'Start':<{t2}} {'End':<{t3}} {'Export':<{t4}} {'Output':<{t5}} {'ChargePrio':<{t6}} {'SB1':>{t7}} {'SB2':>{t8}} Name"
|
|
)
|
|
# for slot in (data.get("home_load_data",{})).get("ranges",[]):
|
|
for slot in data.get("ranges", []):
|
|
enabled = slot.get("turn_on")
|
|
load = slot.get("appliance_loads", [])
|
|
load = load[0] if len(load) > 0 else {}
|
|
solarbanks = slot.get("device_power_loads", [])
|
|
sb1 = str(
|
|
solarbanks[0].get("power")
|
|
if len(solarbanks) > 0
|
|
else "---"
|
|
)
|
|
sb2 = str(
|
|
solarbanks[1].get("power")
|
|
if len(solarbanks) > 1
|
|
else "---"
|
|
)
|
|
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}} {sb1+' W':>{t7}} {sb2+' W':>{t8}} {str(load.get('name',''))}"
|
|
)
|
|
elif devtype == "inverter":
|
|
upgrade = dev.get("auto_upgrade")
|
|
CONSOLE.info(
|
|
f"{'SW Version':<{col1}}: {dev.get('sw_version','Unknown'):<{col2}} {'Auto-Upgrade':<{col3}}: {'Unknown' if upgrade is None else 'Enabled' if upgrade else 'Disabled'}"
|
|
)
|
|
CONSOLE.info(
|
|
f"{'Status':<{col1}}: {dev.get('status_desc','Unknown'):<{col2}} {'Status code':<{col3}}: {str(dev.get('status','-'))}"
|
|
)
|
|
unit = dev.get("power_unit", "W")
|
|
CONSOLE.info(
|
|
f"{'AC Power':<{col1}}: {dev.get('generate_power',''):>3} {unit}"
|
|
)
|
|
else:
|
|
CONSOLE.warning(
|
|
"Neither Solarbank nor Inverter device, further details will be skipped"
|
|
)
|
|
CONSOLE.info("")
|
|
CONSOLE.info("Api Requests: %s", myapi.request_count)
|
|
CONSOLE.debug(json.dumps(myapi.devices, indent=2))
|
|
for sec in range(REFRESH):
|
|
now = datetime.now().astimezone()
|
|
if sys.stdin is sys.__stdin__:
|
|
print( # noqa: T201
|
|
f"Site refresh: {int((next_refr-now).total_seconds()):>3} sec, Device details refresh: {int((next_dev_refr-now).total_seconds()):>3} sec (CTRL-C to abort)",
|
|
end="\r",
|
|
flush=True,
|
|
)
|
|
elif sec == 0:
|
|
# IDLE may be used and does not support cursor placement, skip time progress display
|
|
print( # noqa: T201
|
|
f"Site refresh: {int((next_refr-now).total_seconds()):>3} sec, Device details refresh: {int((next_dev_refr-now).total_seconds()):>3} sec (CTRL-C to abort)",
|
|
end="",
|
|
flush=True,
|
|
)
|
|
time.sleep(1)
|
|
return False
|
|
|
|
except (ClientError, errors.AnkerSolixError) as err:
|
|
CONSOLE.error("%s: %s", type(err), err)
|
|
return False
|
|
|
|
|
|
# run async main
|
|
if __name__ == "__main__":
|
|
try:
|
|
if not asyncio.run(main()):
|
|
CONSOLE.warning("\nAborted!")
|
|
except KeyboardInterrupt:
|
|
CONSOLE.warning("\nAborted!")
|
|
except Exception as exception: # pylint: disable=broad-exception-caught
|
|
CONSOLE.exception("%s: %s", type(exception), exception)
|