diff --git a/README.md b/README.md index 8b82398..96851e3 100644 --- a/README.md +++ b/README.md @@ -49,6 +49,7 @@ See the [support table](https://solixble.readthedocs.io/en/latest) in the docume - C1000 Gen 2 - F2000 (767 PowerHouse) - F3800 +- Solarbank 1 - Solarbank 2 - Solarbank 3 - Prime Charger 160w diff --git a/SolixBLE/__init__.py b/SolixBLE/__init__.py index d6c614f..20a9e2c 100644 --- a/SolixBLE/__init__.py +++ b/SolixBLE/__init__.py @@ -18,6 +18,7 @@ PrimeCharger160w, PrimeCharger250w, PrimePowerBank20k, + Solarbank1, Solarbank2, Solarbank3, ) @@ -43,6 +44,7 @@ "C1000G2", "F2000", "F3800", + "Solarbank1", "Solarbank2", "Solarbank3", "PrimeCharger160w", diff --git a/SolixBLE/devices/__init__.py b/SolixBLE/devices/__init__.py index dbae16a..db597b6 100644 --- a/SolixBLE/devices/__init__.py +++ b/SolixBLE/devices/__init__.py @@ -4,6 +4,7 @@ """ + from .c300 import C300 from .c300dc import C300DC from .c800 import C800 @@ -16,6 +17,7 @@ from .prime_charger_160w import PrimeCharger160w from .prime_charger_250w import PrimeCharger250w from .prime_power_bank_20k import PrimePowerBank20k +from .solarbank1 import Solarbank1 from .solarbank2 import Solarbank2 from .solarbank3 import Solarbank3 @@ -27,6 +29,7 @@ "C1000G2", "F2000", "F3800", + "Solarbank1", "Solarbank2", "Solarbank3", "PrimeCharger160w", diff --git a/SolixBLE/devices/solarbank1.py b/SolixBLE/devices/solarbank1.py new file mode 100644 index 0000000..6ca2737 --- /dev/null +++ b/SolixBLE/devices/solarbank1.py @@ -0,0 +1,419 @@ +"""Solarbank 1 power station model. + +.. moduleauthor:: Simon Mariacher https://github.com/smariacher + +""" + +import logging +import struct +from dataclasses import dataclass + +from ..const import ( + DEFAULT_METADATA_FLOAT, + DEFAULT_METADATA_INT, + DEFAULT_METADATA_STRING, +) +from ..device import SolixBLEDevice +from ..states import ChargingStatus + +CMD_SB_SET_SCHEDULE = "405e" + +MIN_WATTAGE, MAX_WATTAGE = 0, 800 +MIN_SOC, MAX_SOC = 10, 100 + +_LOGGER = logging.getLogger(__name__) + + +@dataclass +class FamilyLoadSchedule: + start_time: int + """ + Start of schedule in minutes since midnight. + """ + end_time: int + """ + End of schedule in minutes since midnight. + """ + output_wattage: int + + max_soc: int + """ + Maximum SOC before Solarbank (presumably) goes into passthrough mode. + """ + + def __str__(self) -> str: + """Convert the integer minutes back to HH:MM format for a nice display""" + start_time_str = f"{self.start_time // 60:02d}:{self.start_time % 60:02d}" + end_time_str = f"{self.end_time // 60:02d}:{self.end_time % 60:02d}" + + return ( + f"Charging Schedule:\n" + f" Time: {start_time_str} - {end_time_str}\n" + f" Wattage: {self.output_wattage}W\n" + f" Max SOC: {self.max_soc}%" + ) + + def __post_init__(self): + if not (MIN_WATTAGE <= self.output_wattage <= MAX_WATTAGE): + raise ValueError( + f"Invalid output_wattage: {self.output_wattage}. " + f"Must be between {MIN_WATTAGE} and {MAX_WATTAGE}." + ) + + if not (MIN_SOC <= self.max_soc <= MAX_SOC): + raise ValueError( + f"Invalid max_soc: {self.max_soc}. " + f"Must be between {MIN_SOC} and {MAX_SOC}." + ) + + if not (self.end_time - self.start_time > 0): + raise ValueError( + f"Invalid time frame: Start: {self.start_time}, End: {self.end_time}. " + f"Start time must be smaller than end time." + ) + + if not (self.start_time >= 0 and self.start_time <= 1440): + raise ValueError( + f"Invalid start time: {self.start_time}. " + f"Start time cannot be less than 0 minutes or greater than 1440 minutes (24 hours)" + ) + + if not (self.end_time >= 0 and self.end_time <= 1440): + raise ValueError( + f"Invalid start time: {self.end_time}. " + f"End time cannot be less than 0 minutes or greater than 1440 minutes (24 hours)" + ) + + @classmethod + def from_time_strings( + cls, start: str, end: str, output_wattage: int, max_soc: int + ) -> "FamilyLoadSchedule": + """Alternative constructor to create a schedule using HH:MM string formats.""" + return cls( + start_time=cls.time_from_string(start), + end_time=cls.time_from_string(end), + output_wattage=output_wattage, + max_soc=max_soc, + ) + + @staticmethod + def time_from_string(time: str) -> int: + """ + Converts a string time in 24-hour HH:MM format to minutes since midnight. + + :param time: Time string in 24-hour HH:MM format. + :returns: Minutes since midnight. + """ + + hours_str, minutes_str = time.split(":") + hours = int(hours_str) + minutes = int(minutes_str) + + if hours > 24 or hours < 0: + raise ValueError( + f"Invalid hour value: {hours}. Hour must be between 0 and 24." + ) + + if minutes > 59 or minutes < 0: + raise ValueError( + f"Invalid minute value: {minutes}. Minute must be between 0 and 59." + ) + + if hours == 24 and minutes != 0: + raise ValueError( + f"Invalid time string: {time}. If hour is set to 24 then minutes may only be 0." + ) + + return hours * 60 + minutes + + +class Solarbank1(SolixBLEDevice): + """ + SolarBank 1 Power Station. + + Use this class to connect and monitor a Solarbank 1 power station. + This model is also known as the A17C0. + + .. note:: + This model was added using data from anker-solix-api as well as logging the actual anker app as described in the SolixBLE docs. + It seems to be working so far, altough not everything has been reverse engineered so far. + + + """ + + _EXPECTED_TELEMETRY_LENGTH: int = 253 + + FamilyLoadSchedule = ( + FamilyLoadSchedule # Added so the user only has to do one import + ) + + @property + def serial_number(self) -> str: + """Device serial number. + + :returns: Device serial number or default str value. + """ + return self._parse_string("a2", begin=1) + + @property + def battery_percentage(self) -> int: + """Battery Percentage. + + :returns: Percentage charge of battery or default int value. + """ + return self._parse_int("a3", begin=1) + + @property + def software_version(self) -> str: + """Main software version. + + :returns: Firmware version or default str value. + """ + if self._data is None: + return DEFAULT_METADATA_STRING + + return ".".join([digit for digit in str(self._parse_int("a6", begin=1))]) + + @property + def software_version_controller(self) -> str: + """Software version of the controller. + + :returns: Firmware version or default str value. + """ + if self._data is None: + return DEFAULT_METADATA_STRING + + return ".".join([digit for digit in str(self._parse_int("a7", begin=1))]) + + @property + def hardware_version(self) -> str: + """Hardware version. + + :returns: Hardware version or default str value. + """ + if self._data is None: + return DEFAULT_METADATA_STRING + + return ".".join([digit for digit in str(self._parse_int("a8", begin=1))]) + + @property + def temperature(self) -> int: + """Temperature of the unit (C). + + :returns: Temperature of the unit in degrees C. + """ + return self._parse_int("aa", begin=1, signed=True) + + @property + def solar_power_in(self) -> float: + """Total Solar Power In. + + :returns: Total solar power in or default float value. + """ + if self._data is None: + return DEFAULT_METADATA_FLOAT + + return self._parse_int("ab", begin=1) + + @property + def output_power(self) -> int: + """Output power. + + :returns: Total power out in watts or default float value. + """ + if self._data is None: + return DEFAULT_METADATA_INT + + return self._parse_int("ac", begin=1) + + @property + def charging_status(self) -> ChargingStatus: + """Retrieve the current charging status of the device. + Parses the charging status from the device data. If device data is unavailable + or does not contain charging status information, returns UNKNOWN. + + :returns: ChargingStatus enum member representing the current charging state + (e.g., CHARGING, DISCHARGING, IDLE, or UNKNOWN if status cannot be determined). + """ + + if self._data is None or "ad" not in self._data: + return ChargingStatus.UNKNOWN + + value = self._parse_int("ad", begin=1) + + try: + return ChargingStatus(value) + except ValueError: + _LOGGER.exception( + f"Invalid ChargingStatus value {value} received from device; returning ChargingStatus.UNKNOWN." + ) + return ChargingStatus.UNKNOWN + + @property + def family_load_schedule(self) -> list[FamilyLoadSchedule]: + """Parse the active daily family load schedule block(s). + + :returns: A list of FamilyLoadSchedule objects representing the current schedule, + or an empty list if no schedule is set. + """ + if self._data is None or "ae" not in self._data: + return [] + + data = self._data["ae"] + + # Safely extract the raw bytes + if isinstance(data, bytes): + raw_bytes = data + else: + return [] + + # A valid payload has a 1-byte header, plus N * 8-byte blocks + if len(raw_bytes) < 9 or (len(raw_bytes) - 1) % 8 != 0: + return [] + + schedules = [] + for i in range(1, len(raw_bytes), 8): + chunk = raw_bytes[i : i + 8] + + start_min = int.from_bytes(chunk[0:2], byteorder="little") + end_min = int.from_bytes(chunk[2:4], byteorder="little") + watts = int.from_bytes(chunk[4:6], byteorder="little") + limit = int.from_bytes(chunk[6:8], byteorder="little") + + schedules.append( + FamilyLoadSchedule( + start_time=start_min, + end_time=end_min, + output_wattage=watts, + max_soc=limit, + ) + ) + + return schedules + + @property + def battery_charge_power(self) -> float: + """Battery charging power. + + :returns: Total battery power in or default float value. + """ + if self._data is None: + return DEFAULT_METADATA_FLOAT + + return self._parse_int("b0", begin=1) / 100.0 + + @property + def pv_yield(self) -> float: + """Solar power generated. + + :returns: Total solar power generated or default float value. + """ + if self._data is None: + return DEFAULT_METADATA_FLOAT + + return self._parse_int("b1", begin=1) / 10000.0 + + @property + def charged_energy(self) -> float: + """Probably aggregated energy charged in Wh? + + :returns: Charged energy or default float value. + """ + if self._data is None: + return DEFAULT_METADATA_FLOAT + + return self._parse_int("b2", begin=1) / 10000.0 + + @property + def output_energy(self) -> float: + """Output energy. + + :returns: Total energy output or default float value. + """ + if self._data is None: + return DEFAULT_METADATA_FLOAT + + return self._parse_int("b3", begin=1) / 10000.0 + + @property + def inverter_brand(self) -> str: + """Brand of the connected inverter. + + :returns: Inverter brand or default str value. + """ + if self._data is None: + return DEFAULT_METADATA_STRING + + return self._parse_string("b7", begin=1) + + @property + def inverter_model(self) -> str: + """Model of the connected inverter. + + :returns: Inverter model or default str value. + """ + if self._data is None: + return DEFAULT_METADATA_STRING + + return self._parse_string("b8", begin=1) + + async def set_schedule(self, schedules: list[FamilyLoadSchedule]) -> None: + """Set the daily charge/discharge schedule on the Solarbank 1. + + .. note:: + :collapsible: closed + + Sends a schedule write command (CMD 0x405e) to the device. + The base class ``_send_command`` automatically appends the current + session timestamp and handles AES-CBC encryption and framing. + + Pass an empty list to clear/delete all schedules. + + Examples:: + + # Single schedule: charge-only midnight-06:00, cap at 80 % SOC + await sb1.set_schedule([ + FamilyLoadSchedule(start=0, end=360, output_wattage=0, max_soc=80) + ]) + + # Two back-to-back schedules + await sb1.set_schedule([ + FamilyLoadSchedule(start=0, end=360, output_wattage=0, max_soc=80), + FamilyLoadSchedule(start=360, end=870, output_wattage=240, max_soc=80), + ]) + + # Clear all schedules + await sb1.set_schedule([]) + + :param schedules: List of FamilyLoadSchedule objects. The device-side upper limit + is unknown but confirmed to be at least 10. + :raises ConnectionError: If not connected/negotiated to the device. + """ + + # ── Build plaintext TLV payload ──────────────────────────────────── + # + # Format per field: + # LENGTH counts the TYPE byte plus data bytes (i.e. len(DATA) + 1). + # + # 0xa1 — command marker: no data, type byte 0x21 only + payload = bytes([0xA1, 0x01, 0x21]) + + # 0xa2 — schedule count: type 0x01, 1-byte unsigned int + payload += bytes([0xA2, 0x02, 0x01, len(schedules)]) + + # 0xa3 — schedule blocks: type 0x04, then N × 8-byte entries + # Each entry: [start_min u16le][end_min u16le][power_W u16le][soc_% u16le] + schedule_bytes = b"" + for schedule in schedules: + schedule_bytes += struct.pack( + " None: + """set_schedule builds the correct TLV payload byte-wise. + .. note:: + :collapsible: closed + + Payload format: + 0xa1: command marker -> a1 01 21 + 0xa2: schedule count -> a2 02 01 + 0xa3: schedule blocks -> a3 <1+8*N> 04 + N*(start end power soc, all u16le) + """ + device = Solarbank1(MOCK_BLE_DEVICE) + device._send_command = mock.AsyncMock() + + # ── Empty list: count 0, block length = 1 (type byte only) ── + await device.set_schedule([]) + device._send_command.assert_awaited_once_with( + bytes.fromhex("405e"), + bytes.fromhex("a10121" "a2020100" "a30104"), + ) + device._send_command.reset_mock() + + # ── Single schedule: start=0, end=360, wattage=0, soc=80 ── + # 0000 6801 0000 5000 (u16le: 0, 360=0x0168, 0, 80=0x50) + # block length = 1 + 8 = 9 = 0x09 + await device.set_schedule( + [device.FamilyLoadSchedule(start_time=0, end_time=360, output_wattage=0, max_soc=80)] + ) + device._send_command.assert_awaited_once_with( + bytes.fromhex("405e"), + bytes.fromhex( + "a10121" + "a2020101" + "a30904" "0000" "6801" "0000" "5000" + ), + ) + device._send_command.reset_mock() + + # ── Two schedules ── + # #1: 0000 6801 0000 5000 + # #2: start=360(6801) end=870(0366) wattage=240(00f0) soc=80(0050) + # block length = 1 + 16 = 17 = 0x11 + await device.set_schedule( + [ + device.FamilyLoadSchedule(start_time=0, end_time=360, output_wattage=0, max_soc=80), + device.FamilyLoadSchedule( + start_time=360, end_time=870, output_wattage=240, max_soc=80 + ), + ] + ) + device._send_command.assert_awaited_once_with( + bytes.fromhex("405e"), + bytes.fromhex( + "a10121" + "a2020102" + "a31104" + "0000" "6801" "0000" "5000" + "6801" "6603" "f000" "5000" + ), + ) + +def test_sb1_family_load_schedule_bytes() -> None: + """family_load_schedule parses the 'ae' TLV payload into FamilyLoadSchedule objects. + + Read-side counterpart to test_sb1_set_schedule_bytes above + """ + device = Solarbank1(MOCK_BLE_DEVICE) + + # ── No schedule set: 'ae' key absent -> empty list ── + assert device.family_load_schedule == [] + + # ── Single schedule: start=0, end=360, wattage=0, soc=80 ── + device._data = device._parse_payload( + bytes.fromhex("ae09" "04" "0000" "6801" "0000" "5000") + ) + assert device.family_load_schedule == [ + device.FamilyLoadSchedule(start_time=0, end_time=360, output_wattage=0, max_soc=80), + ] + + # ── Two schedules ── + device._data = device._parse_payload( + bytes.fromhex( + "ae11" "04" + "0000" "6801" "0000" "5000" + "6801" "6603" "f000" "5000" + ) + ) + assert device.family_load_schedule == [ + device.FamilyLoadSchedule(start_time=0, end_time=360, output_wattage=0, max_soc=80), + device.FamilyLoadSchedule( + start_time=360, end_time=870, output_wattage=240, max_soc=80 + ), + ]