#!/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)