diff --git a/SolixBLE/device.py b/SolixBLE/device.py index 226afc3..bb45f2b 100644 --- a/SolixBLE/device.py +++ b/SolixBLE/device.py @@ -646,7 +646,7 @@ async def _process_notification( return await self._process_negotiation(cmd, payload) # Session messages - case "03010f" | "030111": + case "03000f" | "03010f" | "030111": # Non-encrypted telemetry messages if cmd.hex() == "0300": diff --git a/SolixBLE/devices/solarbank3.py b/SolixBLE/devices/solarbank3.py index 122cc35..a4f9277 100644 --- a/SolixBLE/devices/solarbank3.py +++ b/SolixBLE/devices/solarbank3.py @@ -4,20 +4,50 @@ """ -from ..const import DEFAULT_METADATA_FLOAT, DEFAULT_METADATA_STRING -from ..device import SolixBLEDevice - - -class Solarbank3(SolixBLEDevice): +import logging +import struct +import time + +from bleak.backends.device import BLEDevice +from cryptography.hazmat.primitives.asymmetric import ec + +from ..const import DEFAULT_METADATA_FLOAT, DEFAULT_METADATA_STRING, UUID_COMMAND +from ..prime_device import PrimeDevice +from ..sb3_protocol import ( + SB3_4001, + SB3_4003, + SB3_4005, + SB3_4029, + SB3_4822_SUCCESS_PLAINTEXT, + SB3_SCHEDULE_MODE_CHARGE, + SB3_SCHEDULE_MODE_DISCHARGE, + SB3_SET_MAX_LOAD_COMMAND, + SB3_SET_SCHEDULE_COMMAND, + aes_gcm_decrypt, + build_account_auth_packet, + build_firmware_request_packet, + build_max_load_plaintext, + build_public_key_packet, + build_schedule_plaintext, + build_security_auth_packet, + build_telemetry_request_packet, + decode_public_key, + validate_account_id, +) + +_LOGGER = logging.getLogger(__name__) + + +class Solarbank3(PrimeDevice): """ SolarBank 3 Power Station. Use this class to connect and monitor a Solarbank 3 power station. This model is also known as the A17C5. - .. note:: - This model was added using data from anker-solix-api. It has not been - tested! + The implementation has been tested against an Anker Solarbank 3 E2700 Pro + (A17C5) running firmware 1.0.7.1. The account identifier is passed to the + constructor because the device validates it during local BLE authentication. .. note:: It should be possible to add more sensors. I think devices with lots of @@ -28,8 +58,271 @@ class Solarbank3(SolixBLEDevice): """ + _TELEMETRY_COMMANDS: tuple[str, ...] = ( + "c405", + "c840", + "4409", + "4830", + "485e", + ) _EXPECTED_TELEMETRY_LENGTH: int = 253 + def __init__(self, ble_device: BLEDevice, anker_user_id: str) -> None: + """Initialise an A17C5 device with its Anker account identifier.""" + super().__init__(ble_device) + self._anker_user_id = validate_account_id(anker_user_id) + self._sb3_session_ready = False + self._sb3_private_key = ec.generate_private_key(ec.SECP256R1()) + self._sb3_raw_fragments: dict[str, dict[int, bytes]] = {} + self._sb3_battery_metadata: bytes | None = None + self._sb3_firmware_metadata: dict[str, str] = {} + self._schedule_mode = SB3_SCHEDULE_MODE_DISCHARGE + + @property + def negotiated(self) -> bool: + """Return True only after the SB3 authentication sequence is complete.""" + return self.connected and self._sb3_session_ready + + def _reset_session(self, reset_data: bool = True) -> None: + """Reset the inherited session and the SB3-specific authentication state.""" + super()._reset_session(reset_data) + self._sb3_session_ready = False + self._sb3_raw_fragments = {} + + @property + def schedule_mode(self) -> str: + """Return the direction used by the next all-day schedule write.""" + return self._schedule_mode + + def set_schedule_mode(self, mode: str) -> None: + """Choose whether a custom schedule charges or discharges the bank.""" + if mode not in (SB3_SCHEDULE_MODE_DISCHARGE, SB3_SCHEDULE_MODE_CHARGE): + raise ValueError("mode must be 'discharge' or 'charge'") + self._schedule_mode = mode + + def _decrypt_payload(self, payload: bytes) -> bytes: + """Authenticate and decrypt an A17C5 session payload without fallback.""" + if self._shared_secret is None: + raise ConnectionError("Solarbank 3 session key is not available") + return aes_gcm_decrypt( + self._shared_secret[:16], + self._shared_secret[16:28], + payload, + ) + + @staticmethod + def _is_complete_tlv_payload(payload: bytes) -> bool: + """Return whether an A17C5 payload is a complete telemetry TLV list.""" + if not payload: + return False + index = 1 if payload[0] == 0 else 0 + if index == len(payload): + return False + while index < len(payload): + if len(payload) - index < 2: + return False + value_length = payload[index + 1] + index += 2 + if len(payload) - index < value_length: + return False + index += value_length + return True + + @staticmethod + def _parse_firmware_metadata(payload: bytes) -> dict[str, str]: + """Decode the compact ASCII TLV body returned by authenticated 4830.""" + index = 1 if payload[:1] in (b"\x00", b"\x04") else 0 + metadata: dict[str, str] = {} + while index + 2 <= len(payload): + parameter_id, value_length = payload[index : index + 2] + index += 2 + value = payload[index : index + value_length] + if len(value) != value_length: + return {} + try: + metadata[f"{parameter_id:02x}"] = value.decode("ascii") + except UnicodeDecodeError: + pass + index += value_length + return metadata if index == len(payload) else {} + + async def _process_telemetry_packet( + self, payload: bytes, cmd: bytes = None + ) -> None: + """Handle A17C5 GCM frames without losing a valid first byte. + + A17C5 uses ``0x12``/``0x22`` only for actual two-part packets. A + single encrypted packet can legitimately begin with ``0x11``; treating + that byte as a fragment marker drops data and makes AES-GCM validation + fail after reconnects. + """ + if cmd is None: + return + command = cmd.hex() + complete_payload = bytes(payload) + if payload: + fragment_index = (payload[0] >> 4) & 0x0F + fragment_total = payload[0] & 0x0F + if 1 <= fragment_index <= fragment_total and 2 <= fragment_total <= 4: + fragments = self._sb3_raw_fragments.setdefault(command, {}) + if fragment_index == 1: + fragments.clear() + fragments[fragment_index] = bytes(payload[1:]) + if len(fragments) < fragment_total: + return + complete_payload = b"".join( + fragments[index] for index in range(1, fragment_total + 1) + ) + self._sb3_raw_fragments.pop(command, None) + + plaintext = self._decrypt_payload(complete_payload) + if len(plaintext) == 4 and plaintext[:3] == b"\x01\xa1\x01": + _LOGGER.debug("Solarbank 3 command acknowledgement: %s", plaintext[-1:]) + return + if command == "4409": + self._sb3_battery_metadata = plaintext + return + if command == "4830": + self._sb3_firmware_metadata = self._parse_firmware_metadata(plaintext) + return + if not self._is_complete_tlv_payload(plaintext): + _LOGGER.debug( + "Ignoring authenticated non-telemetry A17C5 packet %s", command + ) + return + await self._process_telemetry(self._parse_payload(plaintext)) + + async def _process_telemetry(self, parameters: dict[str, bytes]) -> None: + """Merge partial A17C5 updates instead of clearing known fields.""" + if self._data is not None: + parameters = {**self._data, **parameters} + await super()._process_telemetry(parameters) + + async def _initiate_negotiations(self) -> None: + """Start the A17C5 secure-conference handshake.""" + self._sb3_private_key = ec.generate_private_key(ec.SECP256R1()) + await self._client.write_gatt_char(UUID_COMMAND, SB3_4001, response=False) + + async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: + """Advance the A17C5 handshake and start encrypted telemetry.""" + command = cmd.hex() + if command == "4801": + reply = SB3_4003 + elif command == "4803": + reply = SB3_4029 + elif command == "4829": + reply = SB3_4005 + elif command == "4805": + reply = build_public_key_packet(self._sb3_private_key.public_key()) + elif command == "4821": + encrypted = aes_gcm_decrypt( + bytes.fromhex("b8ff7422955d4eb6d554a2c470280559"), + bytes.fromhex("6ba3e3f2f3a60f2971ce5d1f"), + payload, + ) + if not encrypted.startswith(b"\x00\xa1\x40"): + raise ValueError("unexpected Solarbank 3 4821 public-key payload") + device_public_key = decode_public_key(encrypted[3:]) + self._shared_secret = self._sb3_private_key.exchange( + ec.ECDH(), device_public_key + ) + self._negotiation_timestamp = time.time() + reply = build_account_auth_packet( + self._anker_user_id, + self._shared_secret[:16], + self._shared_secret[16:28], + ) + elif command == "4822": + plaintext = self._decrypt_payload(payload) + if plaintext != SB3_4822_SUCCESS_PLAINTEXT: + raise ValueError( + f"Solarbank 3 identity authentication failed: {plaintext.hex()}" + ) + reply = build_security_auth_packet( + self._anker_user_id, + self._shared_secret[:16], + self._shared_secret[16:28], + ) + elif command == "4827": + plaintext = self._decrypt_payload(payload) + if plaintext != b"\x00": + raise ValueError( + "Solarbank 3 client-security authentication failed: " + f"{plaintext.hex()}" + ) + self._sb3_session_ready = True + reply = build_telemetry_request_packet( + self._shared_secret[:16], + self._shared_secret[16:28], + int.from_bytes(bytes.fromhex("ef79b569"), "little") + + int(time.time() - self._negotiation_timestamp), + ) + else: + _LOGGER.warning("Unexpected Solarbank 3 negotiation command: %s", command) + return + + await self._client.write_gatt_char(UUID_COMMAND, reply, response=False) + + async def _post_connect(self) -> None: + """Re-arm status telemetry and request read-only firmware metadata.""" + if self._sb3_session_ready: + timestamp = int.from_bytes(bytes.fromhex("ef79b569"), "little") + int( + time.time() - self._negotiation_timestamp + ) + await self._client.write_gatt_char( + UUID_COMMAND, + build_telemetry_request_packet( + self._shared_secret[:16], + self._shared_secret[16:28], + timestamp, + ), + response=False, + ) + await self._client.write_gatt_char( + UUID_COMMAND, + build_firmware_request_packet( + self._shared_secret[:16], + self._shared_secret[16:28], + timestamp + 1, + ), + response=False, + ) + + async def set_schedule( + self, + power_w: int, + *, + start_minutes: int = 0, + end_minutes: int = 1440, + mode: str | None = None, + ) -> None: + """Set a uniform seven-day output schedule using command 405e.""" + await self._send_command( + SB3_SET_SCHEDULE_COMMAND, + build_schedule_plaintext( + power_w, + start_minutes=start_minutes, + end_minutes=end_minutes, + mode=self._schedule_mode if mode is None else mode, + ), + ) + + async def set_max_load(self, max_load_w: int) -> None: + """Set the device maximum output/load limit using command 4080.""" + await self._send_command( + SB3_SET_MAX_LOAD_COMMAND, + build_max_load_plaintext(max_load_w), + ) + + def _parse_sb3_float(self, key: str) -> float: + """Parse an A17C5 typed float, retaining compatibility with integer data.""" + if self._data is None or key not in self._data: + return DEFAULT_METADATA_FLOAT + value = self._data[key] + if value and value[0] == 0x05 and len(value) >= 5: + return struct.unpack(" str: """Device serial number. @@ -38,6 +331,27 @@ def serial_number(self) -> str: """ return self._parse_string("a2", begin=1) + @property + def software_version(self) -> str: + """Return the primary Solarbank firmware reported by ``4830``.""" + return self._sb3_firmware_metadata.get("a2", DEFAULT_METADATA_STRING) + + @property + def firmware_versions(self) -> str: + """Return all verified bank-side firmware and component strings.""" + labels = ( + ("Solarbank", "a2"), + ("Internal MCU", "a1"), + ("MCU component", "a4"), + ("ESP32 component", "a5"), + ) + values = [ + f"{label}: {value}" + for label, key in labels + if (value := self._sb3_firmware_metadata.get(key)) + ] + return " | ".join(values) if values else DEFAULT_METADATA_STRING + @property def battery_percentage_aggregate(self) -> float: """Battery Percentage average across all batteries. @@ -47,7 +361,12 @@ def battery_percentage_aggregate(self) -> float: if self._data is None: return DEFAULT_METADATA_FLOAT - return self._parse_int("a5", begin=1) / 10.0 + percentages = [self._parse_int("a3", begin=1)] + for slot in range(1, 6): + percentage = self._expansion_battery(slot)[1] + if percentage is not None: + percentages.append(percentage) + return float(sum(percentages) // len(percentages)) @property def battery_health(self) -> float: @@ -58,7 +377,7 @@ def battery_health(self) -> float: if self._data is None: return DEFAULT_METADATA_FLOAT - return self._parse_int("a6", begin=1) / 10.0 + return float(self._parse_int("a6", begin=1)) @property def battery_percentage(self) -> int: @@ -66,7 +385,7 @@ def battery_percentage(self) -> int: :returns: Percentage charge of battery or default int value. """ - return self._parse_int("a7", begin=1) + return self._parse_int("a3", begin=1) @property def solar_power_in(self) -> int: @@ -74,7 +393,7 @@ def solar_power_in(self) -> int: :returns: Total solar power in or default int value. """ - return self._parse_int("ab", begin=1) + return round(self._parse_sb3_float("ab")) @property def pv_yield(self) -> int: @@ -82,7 +401,7 @@ def pv_yield(self) -> int: :returns: Total solar power generated or default int value. """ - return self._parse_int("ac", begin=1) + return max(0.0, self._parse_sb3_float("ac")) @property def house_demand(self) -> int: @@ -90,7 +409,7 @@ def house_demand(self) -> int: :returns: Power used by house or default int value. """ - return self._parse_int("b1", begin=1) + return round(self._parse_sb3_float("b1")) @property def house_consumption(self) -> int: @@ -100,7 +419,7 @@ def house_consumption(self) -> int: :returns: Power used by house or default int value. """ - return self._parse_int("b2", begin=1) + return round(self._parse_sb3_float("b2")) @property def battery_power(self) -> int: @@ -112,6 +431,11 @@ def battery_power(self) -> int: """ return self._parse_int("b6", begin=1, signed=True) + @property + def schedule_power(self) -> int: + """Return the active schedule output target reported in field b9.""" + return self._parse_int("b9", begin=1) + @property def charged_energy(self) -> int: """Energy into battery? @@ -136,7 +460,7 @@ def grid_power(self) -> int: :returns: Power in/out of grid or default int value. """ - return self._parse_int("bd", begin=1, signed=True) + return round(self._parse_sb3_float("bd")) @property def grid_import_energy(self) -> int: @@ -152,7 +476,7 @@ def grid_export_energy(self) -> int: :returns: Total energy exported to grid or default int value. """ - return self._parse_int("bf", begin=1) + return round(self._parse_sb3_float("bf")) @property def solar_pv_1_power_in(self) -> int: @@ -160,7 +484,14 @@ def solar_pv_1_power_in(self) -> int: :returns: Solar power in or default int value. """ - return self._parse_int("c7", begin=1) + return self._solar_pv_port_power_in("c6") + + def _solar_pv_port_power_in(self, key: str) -> int: + """Return a non-stale individual MPPT value from c6 through c9.""" + value = max(0.0, self._parse_sb3_float(key)) + if self.solar_power_in <= 0 and value > 0: + return 0 + return round(value) @property def solar_pv_2_power_in(self) -> int: @@ -168,7 +499,7 @@ def solar_pv_2_power_in(self) -> int: :returns: Solar power in or default int value. """ - return self._parse_int("c8", begin=1) + return self._solar_pv_port_power_in("c7") @property def solar_pv_3_power_in(self) -> int: @@ -176,7 +507,7 @@ def solar_pv_3_power_in(self) -> int: :returns: Solar power in or default int value. """ - return self._parse_int("c9", begin=1) + return self._solar_pv_port_power_in("c8") @property def solar_pv_4_power_in(self) -> int: @@ -184,7 +515,7 @@ def solar_pv_4_power_in(self) -> int: :returns: Solar power in or default int value. """ - return self._parse_int("ca", begin=1) + return self._solar_pv_port_power_in("c9") @property def temperature(self) -> int: @@ -192,7 +523,86 @@ def temperature(self) -> int: :returns: Temperature of the unit in degrees C. """ - return self._parse_int("cc", begin=1, signed=True) + return self._parse_int("a5", begin=1, signed=True) + + def _expansion_battery( + self, slot: int + ) -> tuple[str | None, int | None, int | None]: + """Decode one inserted BP1600/BP2700 record from ``4409`` metadata. + + Firmware through 1.0.7.3 uses either ``63 01`` or ``6a 01`` as the + record marker. The 16 ASCII bytes directly ahead of that marker are + the battery serial; SoC and temperature use the verified positions in + the compact record. Unknown layouts remain unavailable rather than + being guessed. + """ + payload = getattr(self, "_sb3_battery_metadata", None) + if payload is None: + return None, None, None + for marker_start in (0x63, 0x6A): + marker = bytes((marker_start, 0x01, slot)) + start = 0 + while (index := payload.find(marker, start)) >= 16: + if index + 7 > len(payload): + start = index + 1 + continue + try: + serial = payload[index - 16 : index].decode("ascii") + except UnicodeDecodeError: + start = index + 1 + continue + return serial, payload[index + 5], payload[index + 3] + return None, None, None + + @property + def num_expansion(self) -> int: + """Return the number of detected expansion batteries.""" + return sum(self._expansion_battery(slot)[0] is not None for slot in range(1, 6)) + + @property + def expansion_battery_1_serial_number(self) -> str | None: + """Return the first expansion serial, when present.""" + return self._expansion_battery(2)[0] + + @property + def expansion_battery_1_percentage(self) -> int | None: + """Return the first expansion state of charge, when present.""" + return self._expansion_battery(2)[1] + + @property + def expansion_battery_1_temperature(self) -> int | None: + """Return the first expansion temperature, when present.""" + return self._expansion_battery(2)[2] + + @property + def expansion_battery_2_serial_number(self) -> str | None: + """Return the second expansion serial, when present.""" + return self._expansion_battery(3)[0] + + @property + def expansion_battery_2_percentage(self) -> int | None: + """Return the second expansion state of charge, when present.""" + return self._expansion_battery(3)[1] + + @property + def expansion_battery_2_temperature(self) -> int | None: + """Return the second expansion temperature, when present.""" + return self._expansion_battery(3)[2] + + @property + def expansion_battery_3_serial_number(self) -> str | None: + """Return the third expansion serial, when present.""" + return self._expansion_battery(4)[0] + + @property + def expansion_battery_3_percentage(self) -> int | None: + """Return the third expansion state of charge, when present.""" + return self._expansion_battery(4)[1] + + @property + def expansion_battery_3_temperature(self) -> int | None: + """Return the third expansion temperature, when present.""" + return self._expansion_battery(4)[2] @property def power_out(self) -> int: @@ -200,7 +610,12 @@ def power_out(self) -> int: :returns: Total power out or default int value. """ - return self._parse_int("d3", begin=1) + return round(self._parse_sb3_float("ad")) + + @property + def power_in(self) -> int: + """Return live battery charging power from verified field ``bc``.""" + return round(self._parse_sb3_float("bc")) @property def grid_to_home_power(self) -> int: diff --git a/SolixBLE/sb3_protocol.py b/SolixBLE/sb3_protocol.py new file mode 100644 index 0000000..c8ce456 --- /dev/null +++ b/SolixBLE/sb3_protocol.py @@ -0,0 +1,272 @@ +"""Protocol helpers for the Solarbank 3 E2700 Pro (A17C5). + +The Solarbank 3 uses the Prime packet framing, but its authentication flow is +different from the older Prime devices. This module intentionally contains +only protocol primitives; device lifecycle and BLE I/O remain in the model +class. +""" + +from __future__ import annotations + +import secrets +import time +from uuid import UUID + +from cryptography.exceptions import InvalidTag +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + +from .const import BASE_TIMESTAMP + +SB3_ACCOUNT_ID_LENGTH = 40 +SB3_DEFAULT_CLIENT_ID = "79ebed35-dc9c-4904-b40c-72c4e8363a10" +SB3_4822_SUCCESS_PLAINTEXT = b"\x04" +SB3_INITIAL_AES_KEY = bytes.fromhex("b8ff7422955d4eb6d554a2c470280559") +SB3_INITIAL_NONCE = bytes.fromhex("6ba3e3f2f3a60f2971ce5d1f") +SB3_AES_GCM_AAD = bytes.fromhex("3322110077665544bbaa9988ffeeddcc") +SB3_MAX_LOAD_VALUES = (350, 600, 800, 1200) +SB3_SCHEDULE_MODE_DISCHARGE = "discharge" +SB3_SCHEDULE_MODE_CHARGE = "charge" +SB3_SCHEDULE_MODES = (SB3_SCHEDULE_MODE_DISCHARGE, SB3_SCHEDULE_MODE_CHARGE) + +SB3_SET_SCHEDULE_COMMAND = bytes.fromhex("405e") +SB3_SET_MAX_LOAD_COMMAND = bytes.fromhex("4080") + +# These packets are stable across the captured A17C5 connections. The +# account-bound authentication starts only after the dynamic ECDH exchange. +SB3_4001 = bytes.fromhex( + "ff09220003000140010a824f0bbd508bb2178c3054ae2df691dab7ce7dd037c5e38b" +) +SB3_4003 = bytes.fromhex( + "ff09290003000140030a824e0bbd508bb25db5286d496f964ade328b233f57fcf51eb1f2639d69c6f9" +) +SB3_4029 = bytes.fromhex( + "ff094a0003000140290a824e0bbd508b9acc816cf1285604b0b741b6b202d4f3b4c28ad6630662ca07b3fef57148a0835a890e253dcdeaf36c2a4ca1d6229283bc963af531b711fd239a" +) +SB3_4005 = bytes.fromhex( + "ff092f0003000140050a824e0bbd508bb25db5286d496f9670823925d138f20cc16133c3ead23c3a1da7e14615bdb8" +) + + +def xor_checksum(data: bytes) -> bytes: + """Return the checksum used by the FF09 packet framing.""" + value = 0 + for byte in data: + value ^= byte + return bytes((value,)) + + +def build_packet(pattern: bytes, command: bytes, payload: bytes) -> bytes: + """Build an FF09 packet and append its checksum.""" + if len(pattern) != 3 or len(command) != 2: + raise ValueError("pattern must be 3 bytes and command must be 2 bytes") + length = 2 + 2 + len(pattern) + len(command) + len(payload) + 1 + packet = b"\xff\x09" + length.to_bytes(2, "little") + pattern + command + payload + return packet + xor_checksum(packet) + + +def parse_packet(packet: bytes) -> tuple[bytes, bytes, bytes]: + """Validate an FF09 packet and return pattern, command and payload.""" + if len(packet) < 10 or packet[:2] != b"\xff\x09": + raise ValueError("invalid FF09 packet") + if int.from_bytes(packet[2:4], "little") != len(packet): + raise ValueError("packet length mismatch") + if packet[-1:] != xor_checksum(packet[:-1]): + raise ValueError("packet checksum mismatch") + return packet[4:7], packet[7:9], packet[9:-1] + + +def aes_gcm_encrypt(key: bytes, nonce: bytes, plaintext: bytes) -> bytes: + """Encrypt a Solarbank 3 payload with AES-GCM and the protocol AAD.""" + return AESGCM(key).encrypt(nonce, plaintext, SB3_AES_GCM_AAD) + + +def aes_gcm_decrypt(key: bytes, nonce: bytes, payload: bytes) -> bytes: + """Authenticate and decrypt a Solarbank 3 payload.""" + try: + return AESGCM(key).decrypt(nonce, payload, SB3_AES_GCM_AAD) + except InvalidTag as error: + raise ValueError("Solarbank 3 AES-GCM authentication failed") from error + + +def encode_public_key(public_key: ec.EllipticCurvePublicKey) -> bytes: + """Encode a P-256 public key as the captured X||Y representation.""" + numbers = public_key.public_numbers() + return numbers.x.to_bytes(32, "big") + numbers.y.to_bytes(32, "big") + + +def decode_public_key(raw_key: bytes) -> ec.EllipticCurvePublicKey: + """Decode and validate a 64-byte P-256 X||Y public key.""" + if len(raw_key) != 64: + raise ValueError("Solarbank 3 public key must contain 64 bytes") + return ec.EllipticCurvePublicKey.from_encoded_point( + ec.SECP256R1(), b"\x04" + raw_key + ) + + +def build_public_key_packet(public_key: ec.EllipticCurvePublicKey) -> bytes: + """Build the dynamic ECDH public-key request (4021).""" + plaintext = b"\xa1\x40" + encode_public_key(public_key) + encrypted = aes_gcm_encrypt(SB3_INITIAL_AES_KEY, SB3_INITIAL_NONCE, plaintext) + return build_packet(b"\x03\x00\x01", b"\x40\x21", encrypted) + + +def validate_account_id(account_id: str) -> str: + """Validate the 40-character hexadecimal Anker account identifier.""" + value = account_id.strip().lower() + if len(value) != SB3_ACCOUNT_ID_LENGTH or any( + char not in "0123456789abcdef" for char in value + ): + raise ValueError( + "Solarbank 3 account ID must be exactly 40 hexadecimal characters" + ) + return value + + +def validate_client_id(client_id: str) -> str: + """Validate the identifier carried by the 4027 security request.""" + value = client_id.strip().lower() + if len(value) == 40 and all(char in "0123456789abcdef" for char in value): + return value + try: + return str(UUID(value)) + except ValueError as error: + raise ValueError( + "Solarbank 3 client ID must be a UUID or 40-char hex ID" + ) from error + + +def _timestamp(timestamp: int | None) -> bytes: + value = int(time.time()) if timestamp is None else timestamp + if not 0 <= value <= 0xFFFFFFFF: + raise ValueError("timestamp does not fit in four bytes") + return value.to_bytes(4, "little") + + +def build_account_auth_packet( + account_id: str, + session_key: bytes, + session_nonce: bytes, + timestamp: int | None = None, +) -> bytes: + """Build the session-encrypted 4022 account-authentication request.""" + account = validate_account_id(account_id).encode("ascii") + plaintext = b"\xa1\x04" + _timestamp(timestamp) + b"\xa2\x28" + account + return build_packet( + b"\x03\x00\x01", + b"\x40\x22", + aes_gcm_encrypt(session_key, session_nonce, plaintext), + ) + + +def build_security_auth_packet( + client_id: str, + session_key: bytes, + session_nonce: bytes, + timestamp: int | None = None, +) -> bytes: + """Build the session-encrypted 4027 security-authentication request.""" + client = validate_client_id(client_id).encode("ascii") + plaintext = ( + b"\xa1\x04" + _timestamp(timestamp) + bytes((0xA2, len(client))) + client + ) + return build_packet( + b"\x03\x00\x01", + b"\x40\x27", + aes_gcm_encrypt(session_key, session_nonce, plaintext), + ) + + +def build_telemetry_request_plaintext(timestamp: int | None = None) -> bytes: + """Build the replay-protected 4040 request body.""" + if timestamp is None: + timestamp = int.from_bytes(bytes.fromhex(BASE_TIMESTAMP), "little") + return b"\xa1\x01\x21\xfe\x05\x03" + _timestamp(timestamp) + + +def build_telemetry_request_packet( + session_key: bytes, session_nonce: bytes, timestamp: int | None = None +) -> bytes: + """Build an encrypted 4040 telemetry request.""" + return build_packet( + b"\x03\x00\x0f", + b"\x40\x40", + aes_gcm_encrypt( + session_key, + session_nonce, + build_telemetry_request_plaintext(timestamp), + ), + ) + + +def build_firmware_request_packet( + session_key: bytes, session_nonce: bytes, timestamp: int | None = None +) -> bytes: + """Build the authenticated ``4030`` firmware-information request. + + A17C5 firmware pages use the same replay-protected request body as + ``4040``. The response is ``4830`` and contains a compact ASCII TLV list. + """ + return build_packet( + b"\x03\x00\x0f", + b"\x40\x30", + aes_gcm_encrypt( + session_key, + session_nonce, + build_telemetry_request_plaintext(timestamp), + ), + ) + + +def build_schedule_plaintext( + power_w: int, + *, + start_minutes: int = 0, + end_minutes: int = 1440, + mode: str = SB3_SCHEDULE_MODE_DISCHARGE, + fd_token: bytes | None = None, +) -> bytes: + """Build the seven-day 405e schedule payload observed on firmware 1.0.7.1.""" + if not isinstance(power_w, int) or isinstance(power_w, bool): + raise TypeError("power_w must be an integer") + if not 0 <= power_w <= 1200 or power_w % 50: + raise ValueError("power_w must be between 0 and 1200 W in 50 W steps") + if not 0 <= start_minutes <= end_minutes <= 1440: + raise ValueError("schedule times must be between 0 and 1440 minutes") + if mode not in SB3_SCHEDULE_MODES: + raise ValueError("mode must be 'discharge' or 'charge'") + if fd_token is None: + fd_token = secrets.token_bytes(4) + if len(fd_token) != 4: + raise ValueError("fd_token must contain exactly four bytes") + + slot = ( + start_minutes.to_bytes(2, "little") + + end_minutes.to_bytes(2, "little") + + power_w.to_bytes(2, "little") + + bytes((0x50, 0x01 if mode == SB3_SCHEDULE_MODE_CHARGE else 0x00)) + ) + schedule = bytearray(b"\xa1\x01\x21\xa2\x02\x01\x01") + for day in range(7): + base = 0xA3 + 4 * day + schedule.extend(bytes((base,)) + b"\x02\x01\x01") + schedule.extend(bytes((base + 1,)) + b"\x09\x04" + slot) + schedule.extend(bytes((base + 2,)) + b"\x02\x01\x00") + schedule.extend(bytes((base + 3,)) + b"\x01\x04") + schedule.extend(b"\xfd\x05\x03" + fd_token) + if len(schedule) != 168: + raise AssertionError("unexpected Solarbank 3 schedule length") + return bytes(schedule) + + +def build_max_load_plaintext(max_load_w: int) -> bytes: + """Build the 4080 maximum-load payload observed on firmware 1.0.7.1.""" + if not isinstance(max_load_w, int) or isinstance(max_load_w, bool): + raise TypeError("max_load_w must be an integer") + if max_load_w not in SB3_MAX_LOAD_VALUES: + raise ValueError("max_load_w must be one of 350, 600, 800 or 1200 W") + return ( + b"\xa1\x01\x21\xa2\x03\x02" + + max_load_w.to_bytes(2, "little") + + b"\xa3\x03\x02\x00\x00" + ) diff --git a/docs/source/solarbank3.rst b/docs/source/solarbank3.rst index 1f3d338..8a8c913 100644 --- a/docs/source/solarbank3.rst +++ b/docs/source/solarbank3.rst @@ -1,6 +1,36 @@ Solarbank 3 =========== +The Solarbank 3 E2700 Pro (A17C5) implementation has been tested with +firmware 1.0.7.1 and 1.0.7.3. It requires the 40-character hexadecimal +Anker account ID used by the device during local authentication:: + + device = Solarbank3(ble_device, anker_user_id="...") + +The device supports the following local controls: + +* ``set_schedule(power_w)`` writes the seven-day ``405e`` schedule in 50 W + steps from 0 to 1200 W. +* ``set_max_load(max_load_w)`` writes the ``4080`` limit for 350, 600, 800 or + 1200 W. + +Stability notes +--------------- + +The implementation uses a fresh P-256 ECDH key pair per connection, validates +the authenticated ``4827`` session result, and re-arms the session with a +``4040`` request after reconnecting. It also accepts both observed ``4409`` +expansion-battery record markers (``63 01`` and ``6a 01``), so expansion-battery +metadata remains available across normal reconnects. The Solarbank firmware +information is requested with ``4030`` and exposed through ``firmware_versions``. + +The schedule supports both discharge and charge slots. Use +``set_schedule_mode("charge")`` before ``set_schedule`` to create charge slots; +the default remains discharge for compatibility. + +These controls change the Solarbank itself over BLE. They do not update the +cloud-side plan metadata shown by the Anker app. + .. autoclass:: SolixBLE.Solarbank3 :members: :inherited-members: connect, disconnect, add_callback, remove_callback, connected, available, address, name, supports_telemetry, last_update diff --git a/tests/test_solarbank3.py b/tests/test_solarbank3.py new file mode 100644 index 0000000..459b028 --- /dev/null +++ b/tests/test_solarbank3.py @@ -0,0 +1,184 @@ +"""Tests for the Solarbank 3 A17C5 protocol and model mappings.""" + +import struct + +import pytest +from bleak.backends.device import BLEDevice +from cryptography.hazmat.primitives.asymmetric import ec + +from SolixBLE.devices.solarbank3 import Solarbank3 +from SolixBLE.sb3_protocol import ( + SB3_DEFAULT_CLIENT_ID, + SB3_INITIAL_AES_KEY, + SB3_INITIAL_NONCE, + SB3_SCHEDULE_MODE_CHARGE, + aes_gcm_decrypt, + aes_gcm_encrypt, + build_max_load_plaintext, + build_schedule_plaintext, + build_security_auth_packet, + encode_public_key, + parse_packet, +) + + +def _float_tlv(value: float) -> bytes: + """Build the typed float representation used by A17C5 telemetry.""" + return b"\x05" + struct.pack(" None: + """The 405e payload contains seven identical weekday slots.""" + payload = build_schedule_plaintext(350, fd_token=b"\x01\x02\x03\x04") + + assert len(payload) == 168 + assert payload[:7] == bytes.fromhex("a10121a2020101") + assert payload[161:] == bytes.fromhex("fd050301020304") + assert [payload[18 + 22 * day] for day in range(7)] == [0x5E] * 7 + + +def test_sb3_charge_schedule_changes_only_the_slot_direction() -> None: + """The app's charge tab uses the second byte after the 0x50 trailer.""" + payload = build_schedule_plaintext( + 300, + mode=SB3_SCHEDULE_MODE_CHARGE, + fd_token=b"\x01\x02\x03\x04", + ) + + assert payload[14:22] == bytes.fromhex("0000a0052c015001") + + +def test_sb3_max_load_uses_little_endian_watts() -> None: + """The 4080 payload encodes the selected maximum load as a LE integer.""" + assert build_max_load_plaintext(350) == bytes.fromhex( + "a10121a203025e01a303020000" + ) + assert build_max_load_plaintext(1200) == bytes.fromhex( + "a10121a20302b004a303020000" + ) + + +def test_sb3_4027_uses_session_gcm_and_client_identifier() -> None: + """The security-authentication request must use the negotiated session.""" + key = bytes(range(16)) + nonce = bytes(range(12)) + packet = build_security_auth_packet( + SB3_DEFAULT_CLIENT_ID, key, nonce, timestamp=1_700_000_000 + ) + _, command, encrypted = parse_packet(packet) + + assert command == bytes.fromhex("4027") + assert aes_gcm_decrypt(key, nonce, encrypted).startswith( + bytes.fromhex("a10400f15365a224") + ) + + +def test_sb3_telemetry_mappings_match_firmware_1071() -> None: + """Firmware 1.0.7.1 uses typed floats and the corrected A17C5 fields.""" + device = Solarbank3.__new__(Solarbank3) + device._data = { + "a2": b"\x02SN", + "a3": b"\x01\x5a", + "a5": b"\x01\x19", + "a6": b"\x01\x64", + "ab": _float_tlv(689), + "ac": _float_tlv(12.5), + "ad": _float_tlv(327), + "b1": _float_tlv(400), + "b2": _float_tlv(390), + "b9": b"\x02\x90\x01", + "c6": _float_tlv(309), + "c7": _float_tlv(23), + "c8": _float_tlv(13), + "c9": _float_tlv(40), + } + + assert device.battery_percentage == 90 + assert device.battery_percentage_aggregate == 90 + assert device.temperature == 25 + assert device.power_out == 327 + assert device.schedule_power == 400 + assert device.solar_power_in == 689 + assert device.solar_pv_1_power_in == 309 + assert device.solar_pv_2_power_in == 23 + assert device.solar_pv_3_power_in == 13 + assert device.solar_pv_4_power_in == 40 + + +class _FakeClient: + """Capture writes made by the model during the protocol test.""" + + def __init__(self) -> None: + self.writes: list[bytes] = [] + + async def write_gatt_char( + self, _characteristic: str, data: bytes, response: bool = False + ) -> None: + del response + self.writes.append(data) + + +@pytest.mark.asyncio +async def test_sb3_authentication_sequence_reaches_session_ready() -> None: + """The complete 4021/4022/4027 flow reaches the authenticated session.""" + device = Solarbank3( + BLEDevice("00:11:22:33:44:55", "A17C5", None), + anker_user_id="1" * 40, + ) + device._client = _FakeClient() + + device_private_key = ec.generate_private_key(ec.SECP256R1()) + device_4821 = aes_gcm_encrypt( + SB3_INITIAL_AES_KEY, + SB3_INITIAL_NONCE, + b"\x00\xa1\x40" + encode_public_key(device_private_key.public_key()), + ) + await device._process_negotiation(bytes.fromhex("4821"), device_4821) + + assert parse_packet(device._client.writes[-1])[1] == bytes.fromhex("4022") + session_key = device._shared_secret[:16] + session_nonce = device._shared_secret[16:28] + + await device._process_negotiation( + bytes.fromhex("4822"), + aes_gcm_encrypt(session_key, session_nonce, b"\x04"), + ) + assert parse_packet(device._client.writes[-1])[1] == bytes.fromhex("4027") + + await device._process_negotiation( + bytes.fromhex("4827"), + aes_gcm_encrypt(session_key, session_nonce, b"\x00"), + ) + assert device._sb3_session_ready + assert parse_packet(device._client.writes[-1])[1] == bytes.fromhex("4040") + + +@pytest.mark.asyncio +async def test_sb3_single_payload_starting_with_11_is_not_a_fragment() -> None: + """A GCM packet beginning with 0x11 must retain that byte verbatim.""" + device = Solarbank3.__new__(Solarbank3) + device._sb3_raw_fragments = {} + observed: list[bytes] = [] + + def decrypt(payload: bytes) -> bytes: + observed.append(payload) + return b"\x01\xa1\x01\x31" + + device._decrypt_payload = decrypt + await device._process_telemetry_packet( + b"\x11" + b"ciphertext", bytes.fromhex("485e") + ) + + assert observed == [b"\x11ciphertext"] + + +def test_sb3_battery_metadata_accepts_new_firmware_marker() -> None: + """Firmware 1.0.7.3 uses 6a01 instead of the older 6301 marker.""" + device = Solarbank3.__new__(Solarbank3) + device._sb3_battery_metadata = ( + b"APCDJF4G72230095" + bytes((0x6A, 0x01, 0x02, 25, 0x02, 80, 0x64)) + ) + + assert device.expansion_battery_1_serial_number == "APCDJF4G72230095" + assert device.expansion_battery_1_percentage == 80 + assert device.expansion_battery_1_temperature == 25