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

1.7.0

This commit is contained in:
Thomas Luther 2024-03-26 01:15:38 +01:00
parent fdf4a1d7e7
commit abd35546eb
8 changed files with 568 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

615
api/api.py
View File

@ -9,13 +9,15 @@ from __future__ import annotations
from base64 import b64encode
import contextlib
import copy
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
import json
import logging
import os
import sys
import time
import time as systime
from aiohttp import ClientSession
from aiohttp.client_exceptions import ClientError
@ -244,6 +246,7 @@ class SolixDeviceType(Enum):
INVERTER = "inverter"
PPS = "pps"
POWERPANEL = "powerpanel"
POWERCOOLER = "powercooler"
class SolixParmType(Enum):
@ -252,40 +255,86 @@ class SolixParmType(Enum):
SOLARBANK_SCHEDULE = "4"
class SolixDeviceCapacity(Enum):
"""Enumuration for Anker Solix device capacities in Wh by Part Number."""
@dataclass(frozen=True)
class SolixDeviceCapacity:
"""Dataclass for Anker Solix device capacities in Wh by Part Number."""
A17C0 = 1600 # SOLIX E1600 Solarbank
A1720 = 256 # Anker PowerHouse 521 Portable Power Station
A1751 = 512 # Anker PowerHouse 535 Portable Power Station
A1760 = 1024 # Anker PowerHouse 555 Portable Power Station
A1770 = 1229 # Anker PowerHouse 757 Portable Power Station
A1771 = 1229 # SOLIX F1200 Portable Power Station
A1780 = 2048 # SOLIX F2000 Portable Power Station (PowerHouse 767)
A1780_1 = 2048 # Expansion Battery for F2000
A1790 = 3840 # SOLIX F3800 Portable Power Station
A1790_1 = 3840 # SOLIX BP3800 Expansion Battery for F3800
A1753 = 768 # SOLIX C800 Portable Power Station
A1761 = 1056 # SOLIX C1000 Portable Power Station
A17C1 = 1056 # SOLIX C1000 Expansion Battery
A17C0: int = 1600 # SOLIX E1600 Solarbank
A1720: int = 256 # Anker PowerHouse 521 Portable Power Station
A1751: int = 512 # Anker PowerHouse 535 Portable Power Station
A1753: int = 768 # SOLIX C800 Portable Power Station
A1754: int = 768 # SOLIX C800 Plus Portable Power Station
A1755: int = 768 # SOLIX C800X Portable Power Station
A1760: int = 1024 # Anker PowerHouse 555 Portable Power Station
A1761: int = 1056 # SOLIX C1000(X) Portable Power Station
A17C1: int = 1056 # SOLIX C1000 Expansion Battery
A1770: int = 1229 # Anker PowerHouse 757 Portable Power Station
A1771: int = 1229 # SOLIX F1200 Portable Power Station
A1772: int = 1536 # SOLIX F1500 Portable Power Station
A1780: int = 2048 # SOLIX F2000 Portable Power Station (PowerHouse 767)
A1780_1: int = 2048 # Expansion Battery for F2000
A1781: int = 2560 # SOLIX F2600 Portable Power Station
A1790: int = 3840 # SOLIX F3800 Portable Power Station
A1790_1: int = 3840 # SOLIX BP3800 Expansion Battery for F3800
class SolixDefaults(Enum):
"""Enumuration for Anker Solix defaults to be used."""
@dataclass(frozen=True)
class SolixDeviceCategory:
"""Dataclass for Anker Solix device types by Part Number to be used for standalone/unbound device categorization."""
PRESET_MIN = 50
PRESET_MAX = 800
PRESET_DEF = 100
ALLOW_DISCHARGE = True
CHARGE_PRIORITY_MIN = 0
CHARGE_PRIORITY_MAX = 100
CHARGE_PRIORITY_DEF = 80
A17C0: str = SolixDeviceType.SOLARBANK.value # SOLIX E1600 Solarbank
A5140: str = SolixDeviceType.INVERTER.value # MI60 Inverter
A5143: str = SolixDeviceType.INVERTER.value # MI80 Inverter
A1720: str = (
SolixDeviceType.PPS.value
) # Anker PowerHouse 521 Portable Power Station
A1751: str = (
SolixDeviceType.PPS.value
) # Anker PowerHouse 535 Portable Power Station
A1753: str = SolixDeviceType.PPS.value # SOLIX C800 Portable Power Station
A1754: str = SolixDeviceType.PPS.value # SOLIX C800 Plus Portable Power Station
A1755: str = SolixDeviceType.PPS.value # SOLIX C800X Portable Power Station
A1760: str = (
SolixDeviceType.PPS.value
) # Anker PowerHouse 555 Portable Power Station
A1761: str = SolixDeviceType.PPS.value # SOLIX C1000(X) Portable Power Station
A1770: str = (
SolixDeviceType.PPS.value
) # Anker PowerHouse 757 Portable Power Station
A1771: str = SolixDeviceType.PPS.value # SOLIX F1200 Portable Power Station
A1772: str = SolixDeviceType.PPS.value # SOLIX F1500 Portable Power Station
A1780: str = (
SolixDeviceType.PPS.value
) # SOLIX F2000 Portable Power Station (PowerHouse 767)
A1781: str = SolixDeviceType.PPS.value # SOLIX F2600 Portable Power Station
A1790: str = SolixDeviceType.PPS.value # SOLIX F3800 Portable Power Station
A17B1: str = SolixDeviceType.POWERPANEL.value # SOLIX Home Power Panel
A17A0: str = SolixDeviceType.POWERCOOLER.value # SOLIX Power Cooler 30
A17A1: str = SolixDeviceType.POWERCOOLER.value # SOLIX Power Cooler 40
A17A2: str = SolixDeviceType.POWERCOOLER.value # SOLIX Power Cooler 50
@dataclass(frozen=True)
class SolixDefaults:
"""Dataclass for Anker Solix defaults to be used."""
PRESET_MIN: int = 0
PRESET_MAX: int = 800
PRESET_DEF: int = 100
ALLOW_DISCHARGE: bool = True
CHARGE_PRIORITY_MIN: int = 0
CHARGE_PRIORITY_MAX: int = 100
CHARGE_PRIORITY_DEF: int = 80
class SolixDeviceStatus(Enum):
"""Enumuration for Anker Solix Device status."""
# TODO(3): Add descriptions once status code usage is observed/known
# The device status code seems to be used for cloud connection status.
offline = "0"
online = "1"
unknown = "unknown"
@ -308,6 +357,17 @@ class SolarbankStatus(Enum):
unknown = "unknown"
@dataclass
class SolarbankTimeslot:
"""Dataclass to define customizable attributes of an Anker Solix Solarbank time slot as used for the schedule definition or update."""
start_time: datetime
end_time: datetime
appliance_load: int | None = None # mapped to appliance_loads setting using a default 50% share for dual solarbank setups
allow_discharge: bool | None = None # mapped to the turn_on boolean
charge_priority_limit: int | None = None # mapped to charge_priority setting
class AnkerSolixApi:
"""Define the API class to handle Anker server authentication and API requests, along with the last state of queried site details and Device information."""
@ -379,6 +439,9 @@ class AnkerSolixApi:
),
) # returns bytes of shared secret
# track active devices bound to any site
self._site_devices: set = set()
# Define class variables saving the most recent site and device data
self.nickname: str = ""
self.sites: dict = {}
@ -526,6 +589,13 @@ class AnkerSolixApi:
try:
if key in ["product_code", "device_pn"] and value:
device.update({"device_pn": str(value)})
# try to get type for standalone device from category definitions if not defined yet
if "type" not in device and hasattr(
SolixDeviceCategory, str(value)
):
device.update(
{"type": getattr(SolixDeviceCategory, str(value))}
)
elif key in ["device_name"] and value:
if value != device.get("name", ""):
calc_capacity = True
@ -640,16 +710,16 @@ class AnkerSolixApi:
for x in ("brand_id", "model_img", "version", "ota_status")
if x in keylist
]:
value.pop(key,None)
value.pop(key, None)
device.update({"solar_info": dict(value)})
# schedule is currently a site wide setting. However, we save this with device details to retain info across site updates
# When individual device schedules are support in future, this info is needed per device anyway
elif key in ["schedule"] and isinstance(value, dict) and value:
device.update({"schedule": dict(value)})
# default active presets to None
device.pop("preset_system_output_power",None)
device.pop("preset_allow_discharge",None)
device.pop("preset_charge_priority",None)
device.pop("preset_system_output_power", None)
device.pop("preset_allow_discharge", None)
device.pop("preset_charge_priority", None)
# get actual presets from current slot
now = datetime.now().time().replace(microsecond=0)
# set now to new daytime if close to end of day
@ -693,13 +763,10 @@ class AnkerSolixApi:
if key in ["battery_power"] or calc_capacity:
# generate battery values when soc updated or device name changed or PN is known
if not (cap := device.get("battery_capacity")):
if hasattr(
SolixDeviceCapacity, device.get("device_pn", "")
):
pn = device.get("device_pn") or ""
if hasattr(SolixDeviceCapacity, pn):
# get battery capacity from known PNs
cap = SolixDeviceCapacity[
device.get("device_pn", "")
].value
cap = getattr(SolixDeviceCapacity, pn)
elif device.get("type") == SolixDeviceType.SOLARBANK.value:
# Derive battery capacity in Wh from latest solarbank name or alias if available
cap = (
@ -804,7 +871,7 @@ class AnkerSolixApi:
datetime.utcoffset(now).total_seconds() * 1000
), # timezone offset in ms, e.g. 'GMT+01:00' => 3600000
"transaction": str(
int(time.mktime(now.timetuple()) * 1000)
int(systime.mktime(now.timetuple()) * 1000)
), # Unix Timestamp in ms as string
},
)
@ -999,7 +1066,7 @@ class AnkerSolixApi:
new_sites = {}
self._logger.debug("Getting site list")
sites = await self.get_site_list(fromFile=fromFile)
act_devices = []
self._site_devices = set()
for site in sites.get("site_list", []):
if site.get("site_id"):
# Update site info
@ -1076,7 +1143,7 @@ class AnkerSolixApi:
isAdmin=admin,
)
if sn:
act_devices.append(sn)
self._site_devices.add(sn)
sb_charges[sn] = charge_calc
# adjust calculated SB charge to match total
if len(sb_charges) == len(sb_list) and str(sb_total_charge).isdigit():
@ -1121,7 +1188,7 @@ class AnkerSolixApi:
isAdmin=admin,
)
if sn:
act_devices.append(sn)
self._site_devices.add(sn)
for solar in mysite.get("solar_list", []):
# work around for device_name which is actually the device_alias in scene info
if "device_name" in solar:
@ -1135,7 +1202,7 @@ class AnkerSolixApi:
isAdmin=admin,
)
if sn:
act_devices.append(sn)
self._site_devices.add(sn)
for powerpanel in mysite.get("powerpanel_list", []):
# work around for device_name which is actually the device_alias in scene info
if "device_name" in powerpanel:
@ -1149,13 +1216,9 @@ class AnkerSolixApi:
isAdmin=admin,
)
if sn:
act_devices.append(sn)
self._site_devices.add(sn)
# Write back the updated sites
self.sites = new_sites
# recycle device list and remove devices no longer used in sites
rem_devices = [dev for dev in self.devices if dev not in act_devices]
for dev in rem_devices:
self.devices.pop(dev,None)
return self.sites
async def update_site_details(self, fromFile: bool = False) -> dict:
@ -1191,6 +1254,7 @@ class AnkerSolixApi:
devtypes = {d.value for d in SolixDeviceType}
self._logger.debug("Updating Device Details")
# Fetch firmware version of device
# This response will also contain unbound / standalone devices not added to a site
self._logger.debug("Getting bind devices")
await self.get_bind_devices(fromFile=fromFile)
# Get the setting for effective automated FW upgrades
@ -1251,6 +1315,7 @@ class AnkerSolixApi:
self.devices.update({sn: device})
# TODO(#0): Fetch other details of specific device types as known and relevant
return self.devices
async def update_device_energy(self, devtypes: set = None) -> dict:
@ -1388,8 +1453,18 @@ class AnkerSolixApi:
else:
resp = await self.request("post", _API_ENDPOINTS["bind_devices"])
data = resp.get("data", {})
active_devices = set()
for device in data.get("data", []):
self._update_dev(device)
if sn := self._update_dev(device):
active_devices.add(sn)
# recycle api device list and remove devices no longer used in sites or bind devices
rem_devices = [
dev
for dev in self.devices
if dev not in (self._site_devices | active_devices)
]
for dev in rem_devices:
self.devices.pop(dev, None)
return data
async def get_user_devices(self, fromFile: bool = False) -> dict:
@ -1853,7 +1928,7 @@ class AnkerSolixApi:
await self.get_device_parm(siteId=siteId, deviceSn=deviceSn)
return True
async def set_home_load(
async def set_home_load( # noqa: C901
self,
siteId: str,
deviceSn: str,
@ -1861,123 +1936,425 @@ class AnkerSolixApi:
preset: int = None,
discharge: bool = None,
charge_prio: int = None,
set_slot: SolarbankTimeslot = None,
insert_slot: SolarbankTimeslot = None,
) -> bool:
"""Set the home load parameters for a given site id and device for actual or all slots in the existing schedule.
Example schedule:
If no time slot is defined for current time, a new slot will be inserted for the gap. This will result in full day definition when no slot is defined.
Optionally when set_slot SolarbankTimeslot is provided, the given slot will replace the existing schedule completely.
When insert_slot SolarbankTimeslot is provided, the given slot will be incoorporated into existing schedule. Adjacent overlapping slot times will be updated and overlayed slots will be replaced.
Example schedule as provided via Api:
{{"ranges":[
{"id":0,"start_time":"00:00","end_time":"08:30","turn_on":true,"appliance_loads":[{"id":0,"name":"Benutzerdefiniert","power":300,"number":1}],"charge_priority":80},
{"id":0,"start_time":"08:30","end_time":"17:00","turn_on":false,"appliance_loads":[{"id":0,"name":"Benutzerdefiniert","power":100,"number":1}],"charge_priority":80},
{"id":0,"start_time":"17:00","end_time":"24:00","turn_on":true,"appliance_loads":[{"id":0,"name":"Benutzerdefiniert","power":300,"number":1}],"charge_priority":0}],
"min_load":100,"max_load":800,"step":0,"is_charge_priority":0,default_charge_priority":0}}
"""
# obtain actual device schedule from internal dict or fetch via api
if not (schedule := (self.devices.get(deviceSn) or {}).get("schedule") or {}):
schedule = (
await self.get_device_load(siteId=siteId, deviceSn=deviceSn)
).get("home_load_data") or {}
# fast quit if nothing to change
if not str(charge_prio).isdigit():
charge_prio = None
if not str(preset).isdigit():
preset = None
if preset is None and discharge is None and charge_prio is None:
return True
now = datetime.now().time().replace(microsecond=0)
# set now to new daytime if close to end of day
if now >= datetime.strptime("23:59:58", "%H:%M:%S").time():
now = datetime.strptime("00:00", "%H:%M").time()
new_ranges = []
dev_serials = []
if (min_load := str(schedule.get("min_load"))).isdigit():
min_load = int(min_load)
if (
preset is None
and discharge is None
and charge_prio is None
and set_slot is None
and insert_slot is None
):
return False
# set flag for required current parameter update
if set_slot is None and insert_slot is None:
pending_now_update = True
else:
min_load = SolixDefaults.PRESET_MIN.value
pending_now_update = False
# obtain actual device schedule from internal dict or fetch via api
if not (schedule := (self.devices.get(deviceSn) or {}).get("schedule") or {}):
schedule = (
await self.get_device_load(siteId=siteId, deviceSn=deviceSn)
).get("home_load_data") or {}
if (min_load := str(schedule.get("min_load"))).isdigit():
# min_load = int(min_load)
# Allow lower min setting as defined by API minimum. This however may be ignored if outsite of appliance defined slot boundaries.
min_load = SolixDefaults.PRESET_MIN
else:
min_load = SolixDefaults.PRESET_MIN
if (max_load := str(schedule.get("max_load"))).isdigit():
max_load = int(max_load)
else:
max_load = SolixDefaults.PRESET_MAX.value
for slot in schedule.get("ranges") or []:
with contextlib.suppress(ValueError):
start_time = datetime.strptime(
slot.get("start_time") or "00:00", "%H:%M"
).time()
end_time = slot.get("end_time") or "00:00"
# "24:00" format not supported in strptime
if end_time == "24:00":
end_time = datetime.strptime("23:59", "%H:%M").time()
else:
end_time = datetime.strptime(end_time, "%H:%M").time()
if all_day or start_time <= now < end_time:
if preset is not None:
(slot.get("appliance_loads") or [{}])[0].update(
max_load = SolixDefaults.PRESET_MAX
ranges = schedule.get("ranges") or []
# get appliance load name from first existing slot to avoid mixture
# NOTE: The solarbank may behave weird if a mixture is found or the name does not match with some internal settings
# The name cannot be queried, but seems to be 'custom' by default. However, the mobile app translates it to whather language is defined in the App
appliance_name = None
pending_insert = False
if len(ranges) > 0:
appliance_name = (ranges[0].get("appliance_loads") or [{}])[0].get("name")
if insert_slot:
# set flag for pending insert slot
pending_insert = True
elif insert_slot:
# use insert_slot for set_slot to define a single new slot when no slots exist
set_slot = insert_slot
new_ranges = []
# update individual values in current slot or insert SolarbankTimeslot and adjust adjacent slots
if not set_slot:
now = datetime.now().time().replace(microsecond=0)
last_time = datetime.strptime("00:00", "%H:%M").time()
# set now to new daytime if close to end of day to determine which slot to modify
if now >= datetime.strptime("23:59:58", "%H:%M:%S").time():
now = datetime.strptime("00:00", "%H:%M").time()
next_start = None
split_slot = {}
for idx, slot in enumerate(ranges, start=1):
with contextlib.suppress(ValueError):
start_time = datetime.strptime(
slot.get("start_time") or "00:00", "%H:%M"
).time()
# "24:00" format not supported in strptime
end_time = datetime.strptime(
(
str(slot.get("end_time") or "00:00").replace(
"24:00", "23:59"
)
),
"%H:%M",
).time()
# check slot timings to update current, or insert new and modify adjacent slots
insert = {}
# Check if parameter update required for current time but it falls into gap of no defined slot.
# Create insert slot for the gap and add before or after current slot at the end of the current slot checks/modifications required for allday usage
if (
not insert_slot
and pending_now_update
and (
last_time <= now < start_time
or (idx == len(ranges) and now >= end_time)
)
):
# Use daily end time if now after last slot
insert = copy.deepcopy(slot)
insert.update(
{"start_time": next_start.isoformat(timespec="minutes")}
)
insert.update(
{
"power": min(
max(int(preset), min_load),
max_load,
)
"end_time": (
start_time.isoformat(timespec="minutes")
).replace("23:59", "24:00")
if now < start_time
else "24:00"
}
)
if discharge is not None:
slot.update({"turn_on": discharge})
if charge_prio is not None:
slot.update(
(insert.get("appliance_loads") or [{}])[0].update(
{
"power": min(
max(
int(
SolixDefaults.PRESET_DEF
if preset is None
else preset
),
min_load,
),
max_load,
),
}
)
insert.update(
{
"turn_on": SolixDefaults.ALLOW_DISCHARGE
if discharge is None
else discharge
}
)
insert.update(
{
"charge_priority": min(
max(
int(charge_prio),
SolixDefaults.CHARGE_PRIORITY_MIN.value,
int(
SolixDefaults.CHARGE_PRIORITY_DEF
if charge_prio is None
else charge_prio
),
SolixDefaults.CHARGE_PRIORITY_MIN,
),
SolixDefaults.CHARGE_PRIORITY_MAX.value,
SolixDefaults.CHARGE_PRIORITY_MAX,
)
}
)
# loookup additional device SNs in schedule for device schedule update later in device details dict
if len(new_ranges) == 0:
for dev in slot.get("device_power_loads") or []:
if (
(sn := dev.get("device_sn"))
and sn != deviceSn
and sn not in dev_serials
# if gap is before current slot, insert now
if now < start_time:
new_ranges.append(insert)
last_time = start_time
insert = {}
if pending_insert and (
insert_slot.start_time.time() <= start_time
or idx == len(ranges)
):
dev_serials.append(sn)
new_ranges.append(slot)
# If no slot defined, set defaults for new all day slot
# copy slot, update and insert the new slot
insert = copy.deepcopy(slot)
insert.update(
{
"start_time": datetime.strftime(
insert_slot.start_time, "%H:%M"
)
}
)
insert.update(
{
"end_time": datetime.strftime(
insert_slot.end_time, "%H:%M"
).replace("23:59", "24:00")
}
)
(insert.get("appliance_loads") or [{}])[0].update(
{
"power": min(
max(
int(
SolixDefaults.PRESET_DEF
if insert_slot.appliance_load is None
else insert_slot.appliance_load
),
min_load,
),
max_load,
),
}
)
insert.update(
{
"turn_on": SolixDefaults.ALLOW_DISCHARGE
if insert_slot.allow_discharge is None
else insert_slot.allow_discharge
}
)
insert.update(
{
"charge_priority": min(
max(
int(
SolixDefaults.CHARGE_PRIORITY_DEF
if insert_slot.charge_priority_limit is None
else insert_slot.charge_priority_limit
),
SolixDefaults.CHARGE_PRIORITY_MIN,
),
SolixDefaults.CHARGE_PRIORITY_MAX,
)
}
)
# insert slot before current slot if not last
if insert_slot.start_time.time() <= start_time:
new_ranges.append(insert)
insert = {}
pending_insert = False
if insert_slot.end_time.time() >= end_time:
# set start of next slot if not end of day
if (
end_time
< datetime.strptime("23:59", "%H:%M").time()
):
next_start = insert_slot.end_time.time()
last_time = insert_slot.end_time.time()
# skip current slot since overlapped by insert slot
continue
if split_slot:
# insert second part of a preceeding slot that was split
new_ranges.append(split_slot)
split_slot = {}
# delay start time of current slot not needed if previous slot was split
else:
# delay start time of current slot
slot.update(
{
"start_time": datetime.strftime(
insert_slot.end_time, "%H:%M"
).replace("23:59", "24:00")
}
)
else:
# create copy of slot when insert slot will split last slot to add it later as well
if insert_slot.end_time.time() < end_time:
split_slot = copy.deepcopy(slot)
split_slot.update(
{
"start_time": datetime.strftime(
insert_slot.end_time, "%H:%M"
).replace("23:59", "24:00")
}
)
if insert_slot.start_time.time() < end_time:
# shorten end time of current slot when appended at the end
slot.update(
{
"end_time": datetime.strftime(
insert_slot.start_time, "%H:%M"
).replace("23:59", "24:00")
}
)
elif pending_insert and insert_slot.start_time.time() <= end_time:
# create copy of slot when insert slot will split current slot to add it later
if insert_slot.end_time.time() < end_time:
split_slot = copy.deepcopy(slot)
split_slot.update(
{
"start_time": datetime.strftime(
insert_slot.end_time, "%H:%M"
).replace("23:59", "24:00")
}
)
# shorten end of preceeding slot
slot.update(
{
"end_time": datetime.strftime(
insert_slot.start_time, "%H:%M"
)
}
)
elif next_start and next_start < end_time:
# delay start of slot following an insert
slot.update(
{
"start_time": (
next_start.isoformat(timespec="minutes")
).replace("23:59", "24:00")
}
)
next_start = None
elif not insert_slot and (all_day or start_time <= now < end_time):
# update parameters in current slot or all slots
if preset is not None:
(slot.get("appliance_loads") or [{}])[0].update(
{
"power": min(
max(int(preset), min_load),
max_load,
)
}
)
if discharge is not None:
slot.update({"turn_on": discharge})
if charge_prio is not None:
slot.update(
{
"charge_priority": min(
max(
int(charge_prio),
SolixDefaults.CHARGE_PRIORITY_MIN,
),
SolixDefaults.CHARGE_PRIORITY_MAX,
)
}
)
# clear flag for pending parameter update for actual time
if start_time <= now < end_time:
pending_now_update = False
if (
last_time
<= datetime.strptime(
(slot.get("start_time") or "00:00").replace("24:00", "23:59"),
"%H:%M",
).time()
):
new_ranges.append(slot)
# fill gap after last slot for current time parameter changes or insert slots
if insert:
slot = insert
new_ranges.append(slot)
if split_slot:
# insert second part of a preceeding slot that was split
new_ranges.append(split_slot)
split_slot = {}
# Track end time of last appended slot in list
last_time = datetime.strptime(
(
str(new_ranges[-1].get("end_time") or "00:00").replace(
"24:00", "23:59"
)
),
"%H:%M",
).time()
# If no slot exists or new slot to be set, set defaults or given set_slot parameters
if len(new_ranges) == 0:
if preset is None:
preset = SolixDefaults.PRESET_DEF.value
if discharge is None:
discharge = SolixDefaults.ALLOW_DISCHARGE.value
if charge_prio is None:
charge_prio = SolixDefaults.CHARGE_PRIORITY_DEF.value
if not set_slot:
# define parameters to be used for a new slot
set_slot = SolarbankTimeslot(
start_time=datetime.strptime("00:00", "%H:%M"),
end_time=datetime.strptime("23:59", "%H:%M"),
appliance_load=SolixDefaults.PRESET_DEF
if preset is None
else preset,
allow_discharge=SolixDefaults.ALLOW_DISCHARGE
if discharge is None
else discharge,
charge_priority_limit=SolixDefaults.CHARGE_PRIORITY_DEF
if charge_prio is None
else charge_prio,
)
# generate the new slot
slot = {
"start_time": "00:00",
"end_time": "24:00",
"turn_on": discharge,
"start_time": datetime.strftime(set_slot.start_time, "%H:%M"),
"end_time": datetime.strftime(set_slot.end_time, "%H:%M").replace(
"23:59", "24:00"
),
"turn_on": SolixDefaults.ALLOW_DISCHARGE
if set_slot.allow_discharge is None
else set_slot.allow_discharge,
"appliance_loads": [
{
"power": min(
max(int(preset), min_load),
max(
int(
SolixDefaults.PRESET_DEF
if set_slot.appliance_load is None
else set_slot.appliance_load
),
min_load,
),
max_load,
),
}
],
"charge_priority": min(
max(
int(charge_prio),
SolixDefaults.CHARGE_PRIORITY_MIN.value,
int(
SolixDefaults.CHARGE_PRIORITY_DEF
if set_slot.charge_priority_limit is None
else set_slot.charge_priority_limit
),
SolixDefaults.CHARGE_PRIORITY_MIN,
),
SolixDefaults.CHARGE_PRIORITY_MAX.value,
SolixDefaults.CHARGE_PRIORITY_MAX,
),
}
# use previous appliance name if a slot was defined originally
if appliance_name:
(slot.get("appliance_loads") or [{}])[0].update(
{"name": appliance_name}
)
new_ranges.append(slot)
self._logger.debug(
"Ranges to apply: %s",
new_ranges,
)
# Make the Api call and check for return code, the set call will also update api dict
# NOTE: set_device_load does not seem to be usable yet for changing the home load
# Make the Api call with final schedule and check for return code, the set call will also update api dict
# NOTE: set_device_load does not seem to be usable yet for changing the home load, or is only usable in dual bank setups for changing the appliance load share as well?
schedule.update({"ranges": new_ranges})
return await self.set_device_parm(
siteId=siteId,
@ -2019,7 +2396,7 @@ class AnkerSolixApi:
for key in [
x for x in ("img_url", "bt_ble_id", "link_time") if x in keylist
]:
fitting.pop(key,None)
fitting.pop(key, None)
fittings[fitting.get("device_sn")] = fitting
self._update_dev({"device_sn": deviceSn, "fittings": fittings})
return data
@ -2201,7 +2578,7 @@ class AnkerSolixApi:
for day in daylist:
daystr = day.strftime("%Y-%m-%d")
if day != daylist[0]:
time.sleep(1) # delay to avoid hammering API
systime.sleep(1) # delay to avoid hammering API
resp = await self.energy_analysis(
siteId=siteId,
deviceSn=deviceSn,

17
common.py
View File

@ -1,7 +1,4 @@
# -*- mode: python: coding: utf-8 -*-
"""
a collection of helper functions for pyscripts
"""
"""A collection of helper functions for pyscripts."""
import getpass
import logging
import os
@ -17,9 +14,7 @@ _CREDENTIALS = {
def user():
"""
Get anker account user
"""
"""Get anker account user."""
if _CREDENTIALS.get("USER"):
return _CREDENTIALS["USER"]
CONSOLE.info("\nEnter Anker Account credentials:")
@ -30,9 +25,7 @@ def user():
def password():
"""
Get anker account password
"""
"""Get anker account password."""
if _CREDENTIALS.get("PASSWORD"):
return _CREDENTIALS["PASSWORD"]
pwd = getpass.getpass("Password: ")
@ -42,9 +35,7 @@ def password():
def country():
"""
Get anker account country
"""
"""Get anker account country."""
if _CREDENTIALS.get("COUNTRY"):
return _CREDENTIALS["COUNTRY"]
countrycode = input("Country ID (e.g. DE): ")

8
energy_csv.py
View File

@ -1,6 +1,5 @@
#!/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
@ -17,15 +16,14 @@ will be exported into a csv file.
import asyncio
import csv
from datetime import datetime
import json
import logging
import sys
from datetime import datetime
from aiohttp import ClientSession
import common
from api import api
import common
_LOGGER: logging.Logger = logging.getLogger(__name__)
_LOGGER.addHandler(logging.StreamHandler(sys.stdout))

49
examples/example3/api_devices.json
View File

@ -6,20 +6,23 @@
"is_admin": true,
"device_pn": "A17C0",
"battery_soc": "38",
"charging_power": 82,
"battery_capacity": "1600",
"battery_energy": "608",
"charging_power": "82",
"power_unit": "W",
"charging_status": "3",
"charging_status_desc": "charge_priority",
"status": "1",
"status_desc": "on",
"status_desc": "online",
"wireless_type": "1",
"input_power": "82",
"output_power": "0",
"set_output_power": "0",
"power_cutoff": 10,
"alias": "Solarbank E1600",
"bt_ble_mac": "E1FCF9AFB36D",
"set_system_output_power": "0",
"bt_ble_mac": "EA7AAD9B60BD",
"name": "Solarbank E1600",
"battery_capacity": "1600",
"battery_energy": "608",
"wifi_online": true,
"charge": false,
"bws_surplus": "0",
@ -27,7 +30,22 @@
"auto_upgrade": true,
"wifi_name": "wifi-network-1",
"wifi_signal": "100",
"power_cutoff": 10,
"power_cutoff_data": [
{
"id": 1,
"is_selected": 1,
"output_cutoff_data": 10,
"lowpower_input_data": 5,
"input_cutoff_data": 10
},
{
"id": 2,
"is_selected": 0,
"output_cutoff_data": 5,
"lowpower_input_data": 4,
"input_cutoff_data": 5
}
],
"solar_info": {
"solar_brand": "Deye",
"solar_model": "SUN600G3-EU-230",
@ -112,8 +130,9 @@
"display_advanced_mode": 0,
"advanced_mode_min_load": 0
},
"set_system_output_power": "0",
"set_output_power": "0",
"preset_system_output_power": 100,
"preset_allow_discharge": true,
"preset_charge_priority": 80,
"fittings": {
"HAIUG6WVVY2VJDRX": {
"device_sn": "HAIUG6WVVY2VJDRX",
@ -123,5 +142,17 @@
"bt_ble_mac": "CCAD45CC0A0C"
}
}
},
"1OVQCE1LTA8QITV5": {
"device_sn": "1OVQCE1LTA8QITV5",
"device_pn": "A1761",
"type": "pps",
"bt_ble_mac": "95F70F1FCEE2",
"name": "SOLIX C1000(X)",
"alias": "SOLIX C1000(X)",
"wifi_online": true,
"charge": false,
"bws_surplus": "0",
"sw_version": "v1.3.2"
}
}
}

2
examples/example3/bind_devices.json
View File

@ -4,7 +4,7 @@
"data": {
"data": [
{
"device_sn": "SE0QPSV3WXDDXN8G",
"device_sn": "Y2T19B29T9HQT209",
"product_code": "A17C0",
"bt_ble_id": "EA:7A:AD:9B:60:BD",
"bt_ble_mac": "EA7AAD9B60BD",

6
export_system.py
View File

@ -415,7 +415,6 @@ async def main() -> bool: # noqa: C901 # pylint: disable=too-many-branches,too-
if not admin:
CONSOLE.warning("Query requires account of site owner!")
CONSOLE.info("\nExporting site rules...")
export(
os.path.join(folder, "site_rules.json"),
@ -424,10 +423,11 @@ async def main() -> bool: # noqa: C901 # pylint: disable=too-many-branches,too-
CONSOLE.info("Exporting message unread status...")
export(
os.path.join(folder, "message_unread.json"),
await myapi.request("get", api._API_ENDPOINTS["get_message_unread"], json={}),
await myapi.request(
"get", api._API_ENDPOINTS["get_message_unread"], json={}
),
)
# update the api dictionaries from exported files to use randomized input data
# this is more efficient and allows validation of randomized data in export files
myapi.testDir(folder)

25
solarbank_monitor.py
View File

@ -121,6 +121,8 @@ async def main() -> ( # noqa: C901 # pylint: disable=too-many-locals,too-many-b
t4 = 9
t5 = 6
t6 = 10
t7 = 6
t8 = 6
while True:
CONSOLE.info("\n")
now = datetime.now().astimezone()
@ -142,7 +144,9 @@ async def main() -> ( # noqa: C901 # pylint: disable=too-many-locals,too-many-b
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"
update_time = (
(next(iter(myapi.sites.values()))).get("solarbank_info") or {}
).get("updated_time") or "Unknown"
else:
update_time = "Unknown"
CONSOLE.info(
@ -160,7 +164,7 @@ async def main() -> ( # noqa: C901 # pylint: disable=too-many-locals,too-many-b
)
siteid = dev.get("site_id", "")
CONSOLE.info(f"{'Site ID':<{col1}}: {siteid}")
for fsn, fitting in (dev.get('fittings') or {}).items():
for fsn, fitting in (dev.get("fittings") or {}).items():
CONSOLE.info(
f"{'Accessory':<{col1}}: {fitting.get('device_name',''):<{col2}} {'Serialnumber':<{col3}}: {fsn}"
)
@ -206,15 +210,26 @@ async def main() -> ( # noqa: C901 # pylint: disable=too-many-locals,too-many-b
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}} {'Discharge':<{t4}} {'Output':<{t5}} {'ChargePrio':<{t6}}"
f"{'ID':<{t1}} {'Start':<{t2}} {'End':<{t3}} {'Discharge':<{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}}"
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")
@ -234,7 +249,7 @@ async def main() -> ( # noqa: C901 # pylint: disable=too-many-locals,too-many-b
)
CONSOLE.info("")
CONSOLE.debug(json.dumps(myapi.devices, indent=2))
for sec in range(0, REFRESH):
for sec in range(REFRESH):
now = datetime.now().astimezone()
if sys.stdin is sys.__stdin__:
print( # noqa: T201

2
test_api.py
View File

@ -46,7 +46,7 @@ async def test_api_methods(myapi: api.AnkerSolixApi) -> None: # noqa: D103
_out(await myapi.get_scene_info(siteId=siteid))
_out(await myapi.get_wifi_list(siteId=siteid))
_out(await myapi.get_solar_info(solarbankSn=devicesn))
_out(await myapi.get_device_parm(siteId=siteid, paramType="4"))
_out(await myapi.get_device_parm(siteId=siteid))
_out(
await myapi.get_power_cutoff(