From 0426cd64e5d6352d74e4d2c93ab3b4cdfa75724a Mon Sep 17 00:00:00 2001 From: Shawn Stricker Date: Tue, 8 Sep 2026 02:44:50 -0400 Subject: [PATCH 01/11] feat(ble): parse the advertisement capability byte before connecting The Anker manufacturer record (company id 0xffff) carries the device MAC, model, and a capability byte declaring which negotiation path the device accepts -- readable at scan time, before any frame is sent. Add a parser for it as the basis for choosing the cleartext vs encrypted handshake per device rather than by product class. Capability is length-relative (last byte when present, absent on the F3800), so it is derived from the sku length rather than read at a fixed offset. Decoded against the app's own field values for five bench records. Co-Authored-By: Claude Opus 4.8 --- SolixBLE/advertisement.py | 116 ++++++++++++++++++++++++++++++++++++ tests/test_advertisement.py | 99 ++++++++++++++++++++++++++++++ 2 files changed, 215 insertions(+) create mode 100644 SolixBLE/advertisement.py create mode 100644 tests/test_advertisement.py diff --git a/SolixBLE/advertisement.py b/SolixBLE/advertisement.py new file mode 100644 index 0000000..89e04d1 --- /dev/null +++ b/SolixBLE/advertisement.py @@ -0,0 +1,116 @@ +"""Parsing of the Anker manufacturer advertisement record. + +Anker devices place a fixed-shape record under BLE company identifier +``0xffff`` in their advertisements. It carries the real MAC, the model +(``product_type`` and ``sku``), and a ``capability`` byte that declares which +negotiation path the device accepts, all readable before a connection is made. + +.. moduleauthor:: kb1ibt + +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from bleak.backends.scanner import AdvertisementData + +#: BLE company identifier the record is published under (a reserved/test id). +ANKER_COMPANY_ID = 0xFFFF + +#: Capability bit meaning the device accepts the encrypted ``4xxx`` negotiation +#: under the static key. Devices that set it also accept the cleartext path; +#: newer firmware may accept only the encrypted one. +CAPABILITY_ENCRYPTED_ECDH = 0x04 + +#: Byte offsets within the record. ``capability`` has no fixed offset -- it is +#: the byte after the sku when present, and absent on some models -- so it is +#: derived from the sku length rather than listed here. +_MAC = slice(1, 7) +_BIND_TYPE = 7 +_PRODUCT_TYPE = slice(8, 10) +_SKU_START = 10 + +#: Length of the ascii sku, keyed by ``version_code``. The record is otherwise +#: fixed up to the sku, and the capability byte (if any) follows the sku. +_SKU_LENGTH = {1: 3, 2: 4} + +#: Shortest valid record: everything up to the sku, plus the shortest sku. +_MIN_LENGTH = _SKU_START + min(_SKU_LENGTH.values()) + + +@dataclass(frozen=True) +class AnkerAdvertisement: + """Decoded Anker manufacturer record. + + :param version_code: Record layout version (selects the sku length). + :param mac: Device MAC as lowercase hex without separators. + :param bind_type: Provisioning-state byte (dynamic per device). + :param product_type: Two-byte model key as lowercase hex. + :param sku: Ascii sku, a substring of the device serial. + :param capability: Capability mask, or None when the record omits it. + """ + + version_code: int + mac: str + bind_type: int + product_type: str + sku: str + capability: int | None + + +def parse_manufacturer_record(data: bytes) -> AnkerAdvertisement | None: + """Decode an Anker ``0xffff`` manufacturer record. + + The layout is ``version_code(1) | mac(6) | bind_type(1) | product_type(2) | + sku(3-4 ascii) | capability(1, optional)``. The sku length follows the + version code, and the capability byte is present only on some models, so it + must be read relative to the sku rather than at a fixed offset. + + :param data: Raw manufacturer-data bytes for company id ``0xffff``. + :returns: The decoded record, or None if it is too short or malformed. + """ + if len(data) < _MIN_LENGTH: + return None + + version_code = data[0] + sku_length = _SKU_LENGTH.get(version_code) + if sku_length is None: + return None + + sku_end = _SKU_START + sku_length + if len(data) < sku_end: + return None + + try: + sku = data[_SKU_START:sku_end].decode("ascii") + except UnicodeDecodeError: + return None + + capability = data[sku_end] if len(data) > sku_end else None + + return AnkerAdvertisement( + version_code=version_code, + mac=data[_MAC].hex(), + bind_type=data[_BIND_TYPE], + product_type=data[_PRODUCT_TYPE].hex(), + sku=sku, + capability=capability, + ) + + +def capability_from_advertisement(advertisement: AdvertisementData) -> int | None: + """Read the capability byte from an advertisement, if it carries one. + + :param advertisement: Advertisement data from a bleak scan callback. + :returns: The capability mask, or None if there is no Anker record or the + record omits the byte. + """ + record = advertisement.manufacturer_data.get(ANKER_COMPANY_ID) + if record is None: + return None + + parsed = parse_manufacturer_record(record) + return parsed.capability if parsed is not None else None diff --git a/tests/test_advertisement.py b/tests/test_advertisement.py new file mode 100644 index 0000000..ca27a8d --- /dev/null +++ b/tests/test_advertisement.py @@ -0,0 +1,99 @@ +"""Tests for the Anker manufacturer advertisement record parser. + +Vectors are the real passive-scan records for the bench devices. + +.. moduleauthor:: kb1ibt +""" + +from unittest import mock + +import pytest + +from SolixBLE.advertisement import ( + ANKER_COMPANY_ID, + CAPABILITY_ENCRYPTED_ECDH, + AnkerAdvertisement, + capability_from_advertisement, + parse_manufacturer_record, +) + + +@pytest.mark.parametrize( + ("record", "expected"), + [ + pytest.param( + "01f49d8a2519b200b4014a544200", + AnkerAdvertisement(1, "f49d8a2519b2", 0x00, "b401", "JTB", 0x00), + id="a91b2_station_capability_0", + ), + pytest.param( + "01f49d8a2f05f000b402514a4204", + AnkerAdvertisement(1, "f49d8a2f05f0", 0x00, "b402", "QJB", 0x04), + id="a2345_charger_capability_4", + ), + pytest.param( + "01007f1d44e79e01b402514a4204", + AnkerAdvertisement(1, "007f1d44e79e", 0x01, "b402", "QJB", 0x04), + id="a2345_sealed_first_boot", + ), + pytest.param( + "027ce91346c50c00b11a444b4b4504", + AnkerAdvertisement(2, "7ce91346c50c", 0x00, "b11a", "DKKE", 0x04), + id="c2000g2_a1783_capability_4", + ), + pytest.param( + "01aabbccddeeff02b106373434", + AnkerAdvertisement(1, "aabbccddeeff", 0x02, "b106", "744", None), + id="f3800_no_capability_byte", + ), + ], +) +def test_parse_manufacturer_record(record: str, expected: AnkerAdvertisement) -> None: + """The record decodes to the app's own field values. + + :param record: Hex of the raw ``0xffff`` manufacturer record. + :param expected: The fields the app reports for that record. + """ + assert parse_manufacturer_record(bytes.fromhex(record)) == expected + + +@pytest.mark.parametrize( + ("record", "reason"), + [ + pytest.param("01f49d8a25", "too_short", id="too_short"), + pytest.param( + "03f49d8a2519b200b4014a544200", + "unknown_version", + id="unknown_version", + ), + pytest.param( + "01f49d8a2519b200b401ffffff00", + "non_ascii_sku", + id="non_ascii_sku", + ), + ], +) +def test_parse_manufacturer_record_rejects(record: str, reason: str) -> None: + """A malformed record decodes to None rather than a wrong guess. + + :param record: Hex of a record that cannot be trusted. + :param reason: Why the record is rejected (documentation only). + """ + assert reason + assert parse_manufacturer_record(bytes.fromhex(record)) is None + + +def test_capability_from_advertisement() -> None: + """The capability byte is read from the ``0xffff`` record on an advert.""" + advertisement = mock.Mock() + advertisement.manufacturer_data = { + ANKER_COMPANY_ID: bytes.fromhex("027ce91346c50c00b11a444b4b4504"), + } + assert capability_from_advertisement(advertisement) == CAPABILITY_ENCRYPTED_ECDH + + +def test_capability_from_advertisement_absent() -> None: + """No Anker record means no capability, not an error.""" + advertisement = mock.Mock() + advertisement.manufacturer_data = {0x004C: b"\x02\x15"} + assert capability_from_advertisement(advertisement) is None From bd6b84643803f1e5f7c523ce79237da000d02364 Mon Sep 17 00:00:00 2001 From: Shawn Stricker Date: Tue, 8 Sep 2026 14:02:20 -0400 Subject: [PATCH 02/11] feat(ble): capability-driven encrypted negotiation and token authorization Reads the advertised capability byte to pick the negotiation path, and adds the encrypted GCM (4xxx) handshake plus client-token authorization to the base class, built to the A1783 comms-module firmware. A device whose advert sets the ECDH capability bit (or a class that defaults to it) negotiates under the static GCM key, echoes the device's own auth mode in 4005, carries a signed int32 UTC offset in the 4022 confer, and authorizes the link with a 4027 registration of a stable, generated client token. On hardened firmware a fresh token is accepted after a physical button press, whose grant arrives unsolicited on pattern 030101; an already-registered token authorizes immediately. `negotiated` gates on that authorization for the encrypted path. Cleartext devices are unchanged (the branch is a no-op when the capability bit is clear), and PrimeDevice keeps its own negotiation, so its captured vectors and telemetry are untouched. The static key, nonce and AAD are the values the module firmware derives from two DROM constants at connect time; the client token replaces the account owner-id binding, since the device stores it as an opaque enrollable handle and needs no cloud account. Co-Authored-By: Claude Opus 4.8 --- SolixBLE/device.py | 379 +++++++++++++++++++++++++++- tests/test_encrypted_negotiation.py | 121 +++++++++ 2 files changed, 494 insertions(+), 6 deletions(-) create mode 100644 tests/test_encrypted_negotiation.py diff --git a/SolixBLE/device.py b/SolixBLE/device.py index 8e1d935..0d1da72 100644 --- a/SolixBLE/device.py +++ b/SolixBLE/device.py @@ -9,6 +9,7 @@ import inspect import logging import time +import uuid from collections.abc import Callable from datetime import datetime from functools import partial @@ -26,6 +27,7 @@ ) from cryptography.hazmat.primitives.padding import PKCS7 +from SolixBLE.advertisement import CAPABILITY_ENCRYPTED_ECDH from SolixBLE.constructs import FragmentedPayload, Packet, ParameterDict, Parameters from SolixBLE.utilities import _to_bytes, get_posix_tz @@ -49,6 +51,21 @@ #: The UUID sent to the device during negotiation UUID_STRING = "b2dc0b17-b75d-4abf-ba6e-ec7c997c23e7" +#: Static AES-GCM key, nonce and AAD for the encrypted negotiation, used before +#: the ECDH secret exists. The device derives the key and nonce from two DROM +#: constants at connect time (A1783 module firmware, confirmed 2026-09-08); the +#: results are fixed, so the values are inlined here. +NEGOTIATION_KEY = "b8ff7422955d4eb6d554a2c470280559" +NEGOTIATION_NONCE = "6ba3e3f2f3a60f2971ce5d1f" +NEGOTIATION_AAD = "3322110077665544bbaa9988ffeeddcc" + +#: The client's ECDH public key (uncompressed P-256 point without the ``04`` +#: prefix) matching ``const.PRIVATE_KEY``, sent in the ``4021`` exchange. +CLIENT_PUBLIC_KEY = ( + "060ea168f232aedb37fb2d120c49180329ac72ab5ec3eb8fd30a2f252dc5e151" + "dabccd9b1dc1e288704ca760a0d8c918e5c94823a1f609a4bf07fb4c33ee2190" +) + class SolixBLEDevice: """Solix BLE device object.""" @@ -61,8 +78,32 @@ class SolixBLEDevice: #: The maximum packet size an Anker device is able to send _mtu = 253 - def __init__(self, ble_device: BLEDevice) -> None: - """Initialise device object. Does not connect automatically.""" + #: Whether to negotiate on the encrypted path when the advertised + #: capability is unknown. A known capability byte overrides this; it is the + #: fallback for a device constructed without one. + _DEFAULT_ENCRYPTED_NEGOTIATION: bool = False + + def __init__( + self, + ble_device: BLEDevice, + capability: int | None = None, + client_token: str | None = None, + ) -> None: + """Initialise device object. Does not connect automatically. + + :param ble_device: The bleak device to wrap. + :param capability: The device's advertised capability byte, if known. + It selects the negotiation path (encrypted when the ECDH bit is + set). ``BLEDevice`` is slotted and cannot carry it, so the caller + reads it from the advertisement (see + :func:`SolixBLE.advertisement.capability_from_advertisement`) and + passes it here; None falls back to the class default. + :param client_token: Stable per-client identifier registered with the + device on the encrypted path. On hardened firmware the first use of + a new token needs a physical button press; the device then accepts + it on every later connection, so the caller should persist it and + pass the same value each time. A random one is generated if omitted. + """ _LOGGER.debug( f"Initializing Solix device '{ble_device.name}' with" @@ -83,6 +124,23 @@ def __init__(self, ble_device: BLEDevice) -> None: self._disconnect_event: asyncio.Event = asyncio.Event() self._connection_attempts: int = 0 self._shared_secret: bytes | None = None + self._capability: int | None = capability + self._authorized: bool = False + self._client_token: str = client_token or str(uuid.uuid4()) + self._auth_mode: bytes | None = None + + @property + def _encrypted_negotiation(self) -> bool: + """Whether to negotiate on the encrypted (GCM / ``4xxx``) path. + + Chosen from the advertised capability's ECDH bit when the byte is + known, else the class default. The device MCU's ``auth_mode`` is the + real policy, but it is only readable once stage 2 arrives, so the + pre-connect advert picks the initial cipher. + """ + if self._capability is not None: + return bool(self._capability & CAPABILITY_ENCRYPTED_ECDH) + return self._DEFAULT_ENCRYPTED_NEGOTIATION def add_callback(self, function: Callable[[], None]) -> None: """Register a callback to be run on state updates. @@ -103,7 +161,25 @@ def remove_callback(self, function: Callable[[], None]) -> None: self._state_changed_callbacks.remove(function) async def _initiate_negotiations(self) -> None: - """Send the negotiation initiation command.""" + """Send the negotiation initiation command. + + The encrypted path opens with ``4001`` under the static GCM key; the + cleartext path opens with ``0001`` carrying the client UUID. + """ + if self._encrypted_negotiation: + await self._send_packet( + pattern=NEGOTIATION_PATTERN, + cmd="4001", + parameters={ + "a1": { + "key": bytes.fromhex("a1"), + "type": None, + "value": lambda self: self._timestamp(), + }, + }, + ) + return + await self._send_packet(pattern=NEGOTIATION_PATTERN, cmd="0001", parameters={ "a1": { @@ -311,7 +387,11 @@ def negotiated(self) -> bool: :returns: True/False if session has been negotiated and connected. """ - return self.connected and self._shared_secret is not None + return ( + self.connected + and self._shared_secret is not None + and (not self._encrypted_negotiation or self._authorized) + ) @property def available(self) -> bool: @@ -378,8 +458,41 @@ def _parse_string(self, key: str, begin: int = None, end: int = None) -> str: else DEFAULT_METADATA_STRING ) + def _gcm_key_nonce(self) -> tuple[bytes, bytes]: + """Return the GCM (key, nonce): the ECDH secret if derived, else static. + + Before the ECDH exchange the encrypted path is keyed on the static + negotiation key and nonce; afterwards on the derived shared secret. + """ + if self._shared_secret is not None: + return self._shared_secret[:16], self._shared_secret[16:28] + return bytes.fromhex(NEGOTIATION_KEY), bytes.fromhex(NEGOTIATION_NONCE) + + def _decrypt_payload_gcm(self, payload: bytes) -> bytes: + """AES-GCM decrypt a payload on the encrypted negotiation path. + + The last 16 bytes are the authentication tag. + """ + key, nonce = self._gcm_key_nonce() + mac = payload[-16:] + body = payload[:-16] + cipher = AES.new(key, AES.MODE_GCM, nonce=nonce) + cipher.update(bytes.fromhex(NEGOTIATION_AAD)) + try: + return cipher.decrypt_and_verify(body, mac) + except ValueError: + _LOGGER.exception("GCM tag verify failed; decrypting without verify") + cipher = AES.new(key, AES.MODE_GCM, nonce=nonce) + return cipher.decrypt(body) + def _decrypt_payload(self, payload: bytes) -> bytes: - """Decrypt payload using negotiated shared secret and IV if available.""" + """Decrypt payload using negotiated shared secret and IV if available. + + The encrypted path uses AES-GCM (static key before the ECDH secret + exists); the cleartext path uses AES-CBC once the secret is derived. + """ + if self._encrypted_negotiation: + return self._decrypt_payload_gcm(payload) if self._shared_secret is None: _LOGGER.debug("Skipping decryption as key not negotiated...") @@ -394,7 +507,17 @@ def _decrypt_payload(self, payload: bytes) -> bytes: return unpadded_data + unpadder.finalize() def _encrypt_payload(self, payload: bytes) -> bytes: - """Encrypt payload using negotiated shared secret if available.""" + """Encrypt payload using negotiated shared secret if available. + + AES-GCM on the encrypted path (static key before the secret exists), + AES-CBC on the cleartext path. + """ + if self._encrypted_negotiation: + key, nonce = self._gcm_key_nonce() + cipher = AES.new(key, AES.MODE_GCM, nonce=nonce) + cipher.update(bytes.fromhex(NEGOTIATION_AAD)) + encrypted, mac = cipher.encrypt_and_digest(payload) + return encrypted + mac if self._shared_secret is None: _LOGGER.debug("Skipping encryption as key not negotiated...") @@ -557,6 +680,12 @@ async def _process_notification( else: _LOGGER.debug(f"Received unknown message of type: {cmd.hex()}") + # The unsolicited authorization grant the device pushes after a + # physical button press on the encrypted path. + case "030101": + _LOGGER.debug("Received authorization grant message!") + return await self._process_arm_grant(cmd, payload) + case _: _LOGGER.warning( f"Unexpected packet type '{pattern}' sent by device! Packet: {data.hex()}" @@ -605,6 +734,13 @@ async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: plain_text_payload = self._decrypt_payload(payload) _LOGGER.debug(f"Plain-text payload: {plain_text_payload.hex()}") + + # The encrypted path has its own stages and a status-9 reply that does + # not parse as parameters, so branch before the generic parse below. + if self._encrypted_negotiation: + await self._process_negotiation_encrypted(cmd, plain_text_payload) + return + parameters = Parameters.parse(plain_text_payload) _LOGGER.debug(f"Parameters: {parameters.to_str(verbose=True, types=False)}") @@ -780,6 +916,235 @@ async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: f"Received unexpected negotiation request response from device! cmd: '{cmd}', parameters: '{parameters}'" ) + async def _process_negotiation_encrypted( + self, + cmd: bytes, + plaintext: bytes, + ) -> None: + """Drive the encrypted (GCM / ``4xxx``) negotiation and authorization. + + Built to the A1783 comms-module firmware: ``4005`` echoes the device's + own auth mode, the ``4022`` confer carries a signed UTC offset, and the + link is not authorized until a ``4027`` registration succeeds. That + succeeds at once for an already-registered client token; otherwise the + device replies status ``9`` and authorizes only after a physical button + press, which arrives as an unsolicited grant on pattern ``030101``. + + :param cmd: The negotiation response command code. + :param plaintext: The GCM-decrypted response payload. + """ + match cmd.hex(): + # Stage 1: propose capabilities. + case "4801": + await self._send_packet( + pattern=NEGOTIATION_PATTERN, + cmd="4003", + parameters={ + "a1": { + "key": bytes.fromhex("a1"), + "type": None, + "value": lambda self: self._timestamp(), + }, + "a3": { + "key": bytes.fromhex("a3"), + "type": None, + "value": bytes.fromhex("20"), + }, + "a4": { + "key": bytes.fromhex("a4"), + "type": None, + "value": bytes.fromhex("00f0"), + }, + }, + ) + + # Stage 2: record the device MTU and auth mode, ask for device info. + case "4803": + parameters = Parameters.parse(plaintext) + self._mtu = int.from_bytes( + parameters["a2"].value_legacy, + byteorder="little", + ) + self._auth_mode = parameters["a5"].value_legacy + await self._send_packet( + pattern=NEGOTIATION_PATTERN, + cmd="4029", + parameters={ + "a1": { + "key": bytes.fromhex("a1"), + "type": None, + "value": lambda self: self._timestamp(), + }, + }, + ) + + # Stage 3: set capabilities, echoing the device's declared auth mode. + case "4829": + await self._send_packet( + pattern=NEGOTIATION_PATTERN, + cmd="4005", + parameters={ + "a1": { + "key": bytes.fromhex("a1"), + "type": None, + "value": lambda self: self._timestamp(), + }, + "a3": { + "key": bytes.fromhex("a3"), + "type": None, + "value": bytes.fromhex("20"), + }, + # The firmware does not read the MTU echo; send the + # declared value for correctness. + "a4": { + "key": bytes.fromhex("a4"), + "type": None, + "value": self._mtu.to_bytes(2, byteorder="little"), + }, + # a5 selects the cipher; the 0x44 bits mean ECDH. + "a5": { + "key": bytes.fromhex("a5"), + "type": None, + "value": bytes.fromhex("44"), + }, + # a6 must equal the auth mode the device gave in 4803. + "a6": { + "key": bytes.fromhex("a6"), + "type": None, + "value": self._auth_mode or bytes.fromhex("02"), + }, + }, + ) + + # Stage 4: send our ECDH public key. + case "4805": + await self._send_packet( + pattern=NEGOTIATION_PATTERN, + cmd="4021", + parameters={ + "a1": { + "key": bytes.fromhex("a1"), + "type": None, + "value": bytes.fromhex(CLIENT_PUBLIC_KEY), + }, + }, + ) + + # Stage 5: derive the shared secret, send the timezone confer. + case "4821": + parameters = Parameters.parse(plaintext) + self._negotiation_timestamp = time.time() + device_public_key_bytes = ( + bytes.fromhex("04") + parameters["a1"].value_legacy + ) + device_public_key = EllipticCurvePublicKey.from_encoded_point( + SECP256R1(), + device_public_key_bytes, + ) + private_value = int.from_bytes( + bytes.fromhex(PRIVATE_KEY), + byteorder="big", + ) + private_key = derive_private_key(private_value, SECP256R1()) + self._shared_secret = private_key.exchange( + ECDH(), + device_public_key, + ) + await self._send_packet( + pattern=NEGOTIATION_PATTERN, + cmd="4022", + parameters={ + "a1": { + "key": bytes.fromhex("a1"), + "type": None, + "value": lambda self: self._timestamp(), + }, + # a3: UTC offset, signed int32 LE, seconds west of UTC. + "a3": { + "key": bytes.fromhex("a3"), + "type": None, + "value": self._offset_seconds_west(), + }, + "a5": { + "key": bytes.fromhex("a5"), + "type": None, + "value": (get_posix_tz() or FALLBACK_TZ).encode(), + }, + }, + ) + + # Stage 6: register the client token to authorize the link. + case "4822": + await self._send_packet( + pattern=NEGOTIATION_PATTERN, + cmd="4027", + parameters={ + "a1": { + "key": bytes.fromhex("a1"), + "type": None, + "value": lambda self: self._timestamp(), + }, + "a2": { + "key": bytes.fromhex("a2"), + "type": None, + "value": self._client_token.encode(), + }, + }, + ) + + # Stage 7: authorization result. + case "4827": + if plaintext[:1] == b"\x00": + _LOGGER.debug("Client registration accepted; link authorized!") + self._authorized = True + elif plaintext[:1] == b"\x09": + _LOGGER.info( + "Device is awaiting a physical button press to authorize " + "this client; press the button on the device.", + ) + else: + _LOGGER.warning( + "Unexpected 4027 registration status: %s", + plaintext[:1].hex(), + ) + + case _: + _LOGGER.warning( + "Received unexpected encrypted negotiation response! cmd: %s", + cmd.hex(), + ) + + async def _process_arm_grant(self, cmd: bytes, payload: bytes) -> None: + """Handle the unsolicited authorization grant on pattern ``030101``. + + The device pushes a ``4827`` with status ``0`` here once the operator + presses the button for a newly registered client token, authorizing the + link. + + :param cmd: The command code of the grant frame. + :param payload: The (still encrypted) frame payload. + """ + plaintext = self._decrypt_payload(payload) + if cmd.hex() == "4827" and plaintext[:1] == b"\x00": + _LOGGER.debug("Client authorized via button press!") + self._authorized = True + else: + _LOGGER.debug( + "Unexpected 030101 frame: cmd %s, payload %s", + cmd.hex(), + plaintext.hex(), + ) + + def _offset_seconds_west(self) -> bytes: + """UTC offset as the firmware reads it: signed int32 LE, seconds west. + + POSIX counts seconds *west* of UTC, so US-Eastern in summer is + ``+14400`` and zones east of UTC are negative. + """ + gmtoff = time.localtime().tm_gmtoff + seconds_west = -gmtoff if gmtoff is not None else 0 + return seconds_west.to_bytes(4, byteorder="little", signed=True) + def _timestamp(self) -> bytes: """Unix timestamp in byte form (4B).""" return int(time.time()).to_bytes(length=4, byteorder="little", signed=False) @@ -1053,6 +1418,8 @@ def _reset_session(self, reset_data: bool = True) -> None: self._fragment_buffers = {} self._fragment_totals = {} self._shared_secret = None + self._authorized = False + self._auth_mode = None self._last_packet_timestamp = None self._negotiation_timestamp = None self._packet_futures: dict[bytes, list[asyncio.Future]] = {} diff --git a/tests/test_encrypted_negotiation.py b/tests/test_encrypted_negotiation.py new file mode 100644 index 0000000..ff9c3e3 --- /dev/null +++ b/tests/test_encrypted_negotiation.py @@ -0,0 +1,121 @@ +"""Tests for the capability-driven encrypted negotiation and authorization. + +The device response plaintexts and frames below were captured from a live +C2000 Gen 2 (A1783) on comms-module firmware v0.3.3.0. + +.. moduleauthor:: kb1ibt +""" + +from unittest import mock + +import pytest + +from SolixBLE.constructs import Packet +from SolixBLE.device import SolixBLEDevice +from tests.const import MOCK_BLE_DEVICE + +#: Static-key GCM negotiation frames as sent by the device, and their plaintexts. +FRAME_4801 = "ff091e000300014801ab273ed3e27270c3f4d676ac7d69a00572793732a6" +FRAME_4803 = ( + "ff092b000300014803ab273ed04438d4b25db54c6d4a6ec3d481f5ad58ff7cc2be8bc8369" + "fd98c0b914e03" +) +PLAIN_4801 = "00a10101" +PLAIN_4803 = "00a10102a202fd00a30144a40101a50102" +PLAIN_4829 = ( + "00a10103a2054553503332a307302e302e302e33a411415043444b4b4530463339363030" + "303131a5067ce91346c50c" +) +PLAIN_4805 = "00" +PLAIN_4821 = ( + "00a1405dff69533d15aae7194ccfce70978889ed3b090f0ea76c9d1b44bfcb145c80f8eb5" + "59e5734fd9a17ea03a903eb6024786c009faa14d837031c9636c42910e490" +) +PLAIN_4822 = "00" +PLAIN_4827_OK = "00" +PLAIN_4827_BUTTON = "09a1021e00" + +EXPECTED_MTU = 253 +AUTH_MODE_ENCRYPTED = b"\x02" + + +def _device(token: str = "test-token-0001") -> SolixBLEDevice: # noqa: S107 + """Build a base device on the encrypted path with a fixed client token.""" + return SolixBLEDevice(MOCK_BLE_DEVICE, capability=4, client_token=token) + + +async def _feed(device: SolixBLEDevice, cmd: str, plaintext: str) -> None: + """Feed one decrypted negotiation response into the state machine.""" + await device._process_negotiation_encrypted( # noqa: SLF001 + bytes.fromhex(cmd), + bytes.fromhex(plaintext), + ) + + +def test_encrypted_path_selected_from_capability() -> None: + """A capability with the ECDH bit set selects the encrypted path.""" + assert _device()._encrypted_negotiation is True # noqa: SLF001 + cleartext = SolixBLEDevice(MOCK_BLE_DEVICE, capability=0) + assert cleartext._encrypted_negotiation is False # noqa: SLF001 + + +@pytest.mark.parametrize( + ("frame", "plaintext"), + [ + pytest.param(FRAME_4801, PLAIN_4801, id="stage1"), + pytest.param(FRAME_4803, PLAIN_4803, id="stage2"), + ], +) +def test_static_key_gcm_decrypt(frame: str, plaintext: str) -> None: + """The base GCM decrypt recovers the real device frames under the static key.""" + payload = Packet.parse(bytes.fromhex(frame)).payload_bytes + assert _device()._decrypt_payload(payload).hex() == plaintext # noqa: SLF001 + + +@pytest.mark.asyncio +async def test_encrypted_negotiation_reaches_authorized() -> None: + """Driving the stages emits the right commands and authorizes at 4827/00.""" + device = _device() + with mock.patch.object(device, "_send_packet", new=mock.AsyncMock()) as send: + await _feed(device, "4801", PLAIN_4801) + await _feed(device, "4803", PLAIN_4803) + assert device._mtu == EXPECTED_MTU # noqa: SLF001 + assert device._auth_mode == AUTH_MODE_ENCRYPTED # noqa: SLF001 + await _feed(device, "4829", PLAIN_4829) + await _feed(device, "4805", PLAIN_4805) + await _feed(device, "4821", PLAIN_4821) + assert device._shared_secret is not None # noqa: SLF001 + await _feed(device, "4822", PLAIN_4822) + + # The 4005 echo must carry the ECDH cipher bits and the device auth mode. + cmd_4005 = next(c for c in send.await_args_list if c.kwargs["cmd"] == "4005") + params = cmd_4005.kwargs["parameters"] + assert params["a5"]["value"] == bytes.fromhex("44") + assert params["a6"]["value"] == AUTH_MODE_ENCRYPTED + # The 4027 registration carries the client token. + cmd_4027 = next(c for c in send.await_args_list if c.kwargs["cmd"] == "4027") + assert cmd_4027.kwargs["parameters"]["a2"]["value"] == b"test-token-0001" + + sent = [c.kwargs["cmd"] for c in send.await_args_list] + assert sent == ["4003", "4029", "4005", "4021", "4022", "4027"] + + assert device._authorized is False # noqa: SLF001 + await _feed(device, "4827", PLAIN_4827_OK) + assert device._authorized is True # noqa: SLF001 + + +@pytest.mark.asyncio +async def test_encrypted_negotiation_awaits_button() -> None: + """A 4827 status-9 reply does not authorize; it waits for the button.""" + device = _device() + await _feed(device, "4827", PLAIN_4827_BUTTON) + assert device._authorized is False # noqa: SLF001 + + +@pytest.mark.asyncio +async def test_button_press_grant_authorizes() -> None: + """The unsolicited 030101 grant authorizes the link.""" + device = _device() + payload = device._encrypt_payload(b"\x00") # noqa: SLF001 -- GCM, static key + await device._process_arm_grant(bytes.fromhex("4827"), payload) # noqa: SLF001 + assert device._authorized is True # noqa: SLF001 From 36f1859284bdf0bf0d66a65444b513d3b759657d Mon Sep 17 00:00:00 2001 From: Shawn Stricker Date: Tue, 8 Sep 2026 14:16:21 -0400 Subject: [PATCH 03/11] docs: document the encrypted negotiation path and client-token pairing Adds a usage page covering how the advertised capability byte selects the cleartext versus encrypted handshake, and how to pair a client with a stable, persisted token -- including the one-time physical button press that firmware enforcing pairing requires on the first connection. Links it into the toctree. Co-Authored-By: Claude Opus 4.8 --- docs/source/encrypted_negotiation.rst | 64 +++++++++++++++++++++++++++ docs/source/index.rst | 1 + 2 files changed, 65 insertions(+) create mode 100644 docs/source/encrypted_negotiation.rst diff --git a/docs/source/encrypted_negotiation.rst b/docs/source/encrypted_negotiation.rst new file mode 100644 index 0000000..38ce55a --- /dev/null +++ b/docs/source/encrypted_negotiation.rst @@ -0,0 +1,64 @@ +================================= +Encrypted negotiation and pairing +================================= + +.. _BLEDevice: https://bleak.readthedocs.io/en/latest/api/index.html#bleak.backends.device.BLEDevice/ + + +Some Anker devices negotiate their session over an encrypted handshake rather +than the cleartext one, and some firmware additionally requires the client to +be *paired* to the device -- confirmed by a physical button press -- before it +will stream telemetry or accept commands. Newer firmware in particular enforces +this. Both are handled automatically once the device is told which path to use +and given a stable client token. + + +Choosing the negotiation path +----------------------------- + +Each device advertises a capability byte that says whether it accepts the +encrypted negotiation, readable at scan time before connecting. Read it from the +manufacturer record in the advertisement and pass it to the device when you +construct it; the correct path is then chosen automatically -- encrypted when +the capability's ECDH bit is set, cleartext otherwise:: + + from SolixBLE.advertisement import capability_from_advertisement + + capability = capability_from_advertisement(advertising_data) + device = C1000G2(ble_device, capability=capability) + +.. note:: + + A `BLEDevice`_ cannot carry the capability itself, so it is passed to the + constructor rather than attached to the device. When it is not supplied the + class default is used. The ``advertising_data`` comes from your own Bleak + scan (for example a Home Assistant Bluetooth callback); the capability is a + property of the device model and does not change once a unit is bound. + + +Pairing with a client token +--------------------------- + +On the encrypted path the device authorizes a specific *client*, identified by a +token you provide. Pass a stable value and **persist it**, so the same client is +recognised on every later connection:: + + device = C1000G2(ble_device, capability=capability, client_token=my_token) + +If you do not pass one a random token is generated, which means a new client on +every run. On firmware that enforces pairing, the **first** connection with a +new token needs a one-time physical confirmation: + +#. Call :py:meth:`.connect`. The device reports that it is awaiting confirmation. +#. Press the button on the device (the same one used to wake its Bluetooth). +#. The connection completes and telemetry begins. + +After that first pairing the token is remembered, and every later connection is +authorized immediately with no button press. The whole process is local -- no +Anker account or cloud service is involved. + +.. note:: + + A device that does not enforce pairing authorizes as soon as the handshake + completes, so no button press is needed; the token is still registered for + later connections. diff --git a/docs/source/index.rst b/docs/source/index.rst index e1ac02f..9227fc7 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -186,6 +186,7 @@ Contents Home examples usage + encrypted_negotiation api limitations new_devices From 7b7269582e7f56bde71a0407dc3c41b30bcb3700 Mon Sep 17 00:00:00 2001 From: Shawn Stricker Date: Wed, 9 Sep 2026 12:27:44 -0400 Subject: [PATCH 04/11] Move the timezone-offset helper out of device.py Harvey noted device.py is already large/complicated. The _offset_seconds_west helper (UTC offset as a signed int32 LE, seconds west, sent as a3 in the 4022 timezone confer) uses no instance state, so move it to SolixBLE/utilities.py next to the existing get_posix_tz timezone helper and import it. Co-Authored-By: Claude Opus 4.8 --- SolixBLE/device.py | 14 ++------------ SolixBLE/utilities.py | 14 ++++++++++++++ 2 files changed, 16 insertions(+), 12 deletions(-) diff --git a/SolixBLE/device.py b/SolixBLE/device.py index 0d1da72..3247de9 100644 --- a/SolixBLE/device.py +++ b/SolixBLE/device.py @@ -29,7 +29,7 @@ from SolixBLE.advertisement import CAPABILITY_ENCRYPTED_ECDH from SolixBLE.constructs import FragmentedPayload, Packet, ParameterDict, Parameters -from SolixBLE.utilities import _to_bytes, get_posix_tz +from SolixBLE.utilities import _offset_seconds_west, _to_bytes, get_posix_tz from .const import ( DEFAULT_METADATA_INT, @@ -1063,7 +1063,7 @@ async def _process_negotiation_encrypted( "a3": { "key": bytes.fromhex("a3"), "type": None, - "value": self._offset_seconds_west(), + "value": _offset_seconds_west(), }, "a5": { "key": bytes.fromhex("a5"), @@ -1135,16 +1135,6 @@ async def _process_arm_grant(self, cmd: bytes, payload: bytes) -> None: plaintext.hex(), ) - def _offset_seconds_west(self) -> bytes: - """UTC offset as the firmware reads it: signed int32 LE, seconds west. - - POSIX counts seconds *west* of UTC, so US-Eastern in summer is - ``+14400`` and zones east of UTC are negative. - """ - gmtoff = time.localtime().tm_gmtoff - seconds_west = -gmtoff if gmtoff is not None else 0 - return seconds_west.to_bytes(4, byteorder="little", signed=True) - def _timestamp(self) -> bytes: """Unix timestamp in byte form (4B).""" return int(time.time()).to_bytes(length=4, byteorder="little", signed=False) diff --git a/SolixBLE/utilities.py b/SolixBLE/utilities.py index 5438e88..0a8593f 100644 --- a/SolixBLE/utilities.py +++ b/SolixBLE/utilities.py @@ -8,6 +8,7 @@ import importlib.resources as resources import inspect import logging +import time from typing import Callable import tzlocal @@ -103,3 +104,16 @@ def get_posix_tz() -> str | None: return lines[-1].decode("ascii").strip() except Exception: _LOGGER.exception("Unable to determine system time zone!") + + +def _offset_seconds_west() -> bytes: + """UTC offset as the firmware reads it: signed int32 LE, seconds west. + + POSIX counts seconds *west* of UTC, so US-Eastern in summer is ``+14400`` + and zones east of UTC are negative. + + :returns: The signed 4-byte little-endian offset in seconds west of UTC. + """ + gmtoff = time.localtime().tm_gmtoff + seconds_west = -gmtoff if gmtoff is not None else 0 + return seconds_west.to_bytes(4, byteorder="little", signed=True) From 3222311d83e3509ec386bc9e5bc8deeda1a12192 Mon Sep 17 00:00:00 2001 From: Shawn Stricker Date: Wed, 9 Sep 2026 14:10:10 -0400 Subject: [PATCH 05/11] Name the capability-negotiation proposal constants Harvey asked for constants over the magic values in the negotiation. Extract the two the firmware decode gives clear meaning: the client's MTU proposal (a4 of 0003/0005 -- u16 LE 0x00f0 = 61440 = "no limit", device streams at min(this, its ceiling)) and the encryptMethod it confirms (a5 of 0005 -- the device selects ECDH on a5 & 0x44). The a3 = 0x20 field stays inline: the comms- module firmware logs and ignores it, so a name would imply meaning it does not have. Co-Authored-By: Claude Opus 4.8 --- SolixBLE/device.py | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/SolixBLE/device.py b/SolixBLE/device.py index 3247de9..1930b71 100644 --- a/SolixBLE/device.py +++ b/SolixBLE/device.py @@ -66,6 +66,15 @@ "dabccd9b1dc1e288704ca760a0d8c918e5c94823a1f609a4bf07fb4c33ee2190" ) +#: Client's proposed MTU in the capability negotiation (``a4`` of ``0003``/``0005``). +#: ``u16`` little-endian ``0x00f0`` = 61440 = "no limit"; the device then streams at +#: ``min(this, its own ceiling)``. +NEGOTIATION_MTU_PROPOSAL = "00f0" + +#: encryptMethod the client confirms in ``0005`` (``a5``). The device selects ECDH when +#: ``a5 & 0x44`` is set; ``0x40`` is the base flag. +NEGOTIATION_ENCRYPT_METHOD = "40" + class SolixBLEDevice: """Solix BLE device object.""" @@ -774,7 +783,7 @@ async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: }, "a4": { "key": bytes.fromhex("a4"), "type": None, - "value": bytes.fromhex("00f0"), + "value": bytes.fromhex(NEGOTIATION_MTU_PROPOSAL), }, }, ) @@ -826,11 +835,11 @@ async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: }, "a4": { "key": bytes.fromhex("a4"), "type": None, - "value": bytes.fromhex("00f0"), + "value": bytes.fromhex(NEGOTIATION_MTU_PROPOSAL), }, "a5": { "key": bytes.fromhex("a5"), "type": None, - "value": bytes.fromhex("40"), + "value": bytes.fromhex(NEGOTIATION_ENCRYPT_METHOD), }, }, ) @@ -953,7 +962,7 @@ async def _process_negotiation_encrypted( "a4": { "key": bytes.fromhex("a4"), "type": None, - "value": bytes.fromhex("00f0"), + "value": bytes.fromhex(NEGOTIATION_MTU_PROPOSAL), }, }, ) From b7da75f6af9cff1ae78079a8d6b5ae764af3b125 Mon Sep 17 00:00:00 2001 From: Shawn Stricker Date: Tue, 8 Sep 2026 14:40:52 -0400 Subject: [PATCH 06/11] feat(ble): decode the c490 protobuf device-summary Reintroduces the protobuf walker (parsing.py) that the packet-layer rewrite removed, and the base-class hooks that route a protobuf device-summary frame (the C2000 G2's c490) into a `.path`-keyed `summary` map rather than the flat TLV the other telemetry frames use. The frame's protobuf blob is sliced out of the outer a2 field before walking. No device enables it yet. Co-Authored-By: Claude Opus 4.8 --- SolixBLE/device.py | 59 ++++++++++++++++++ SolixBLE/parsing.py | 142 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 201 insertions(+) create mode 100644 SolixBLE/parsing.py diff --git a/SolixBLE/device.py b/SolixBLE/device.py index 1930b71..c993461 100644 --- a/SolixBLE/device.py +++ b/SolixBLE/device.py @@ -29,6 +29,7 @@ from SolixBLE.advertisement import CAPABILITY_ENCRYPTED_ECDH from SolixBLE.constructs import FragmentedPayload, Packet, ParameterDict, Parameters +from SolixBLE.parsing import walk_protobuf from SolixBLE.utilities import _offset_seconds_west, _to_bytes, get_posix_tz from .const import ( @@ -84,6 +85,12 @@ class SolixBLEDevice: #: (e.g the C1000 Gen 2 uses ``c421``/``c900`` instead of ``c402``/``c405``). _TELEMETRY_COMMANDS: tuple[str, ...] = ("c402", "4300", "c405") + #: Telemetry command codes whose payload is a protobuf device-summary blob + #: (walked via :func:`SolixBLE.parsing.walk_protobuf`) rather than the flat + #: TLV the other telemetry frames use. Subclasses set this (e.g the C2000 + #: G2's ``c490``). + _PROTOBUF_TELEMETRY_COMMANDS: tuple[str, ...] = () + #: The maximum packet size an Anker device is able to send _mtu = 253 @@ -137,6 +144,7 @@ def __init__( self._authorized: bool = False self._client_token: str = client_token or str(uuid.uuid4()) self._auth_mode: bytes | None = None + self._summary: dict[str, object] = {} @property def _encrypted_negotiation(self) -> bool: @@ -434,6 +442,18 @@ def last_update(self) -> datetime | None: """ return self._last_data_timestamp + @property + def summary(self) -> dict[str, object]: + """Fields from the latest protobuf device-summary frame, if any. + + Populated from a ``_PROTOBUF_TELEMETRY_COMMANDS`` frame (e.g. the C2000 + G2's ``c490``) by :func:`SolixBLE.parsing.walk_protobuf`, keyed by + protobuf ``.path``. Empty until such a frame is received. + + :returns: Mapping of ``.path`` to value. + """ + return self._summary + def _parse_int( self, key: str, begin: int = None, end: int = None, signed: bool = False ) -> int: @@ -467,6 +487,32 @@ def _parse_string(self, key: str, begin: int = None, end: int = None) -> str: else DEFAULT_METADATA_STRING ) + @staticmethod + def _protobuf_body(payload: bytes) -> bytes: + """Return the protobuf blob carried in a device-post's outer ``a2`` field. + + A protobuf device post (e.g. the C2000 G2's ``c490``) is a multi-field + outer TLV: ``a1`` -- a one-byte command echo -- then ``a2``, whose value + *is* the protobuf blob, then a trailing ``a3`` string. ``a2`` is a + ``bin`` field with a 2-byte little-endian length and an ``04`` type byte, + so the header is ``a1 a2 04``. The slice is bounded + to ``a2``'s declared length so the walk sees exactly the protobuf and + nothing else. + + :param payload: The decrypted device-post frame. + :returns: The protobuf blob (``a2``'s value), or the whole payload if it + is too short to carry the wrapper. + """ + if len(payload) <= 6: + return payload + # Skip the a1 TLV (tag + 1-byte length + value) to reach the a2 field. + a2_start = 2 + payload[1] + # a2's 2-byte length counts its 04 type byte + the protobuf value, so the + # blob is that length minus the type byte, after tag+len+type. + a2_length = int.from_bytes(payload[a2_start + 1 : a2_start + 3], "little") + blob_start = a2_start + 4 + return payload[blob_start : blob_start + a2_length - 1] + def _gcm_key_nonce(self) -> tuple[bytes, bytes]: """Return the GCM (key, nonce): the ECDH secret if derived, else static. @@ -682,6 +728,18 @@ async def _process_notification( _LOGGER.debug("Received encrypted telemetry message!") decrypted_payload = self._decrypt_payload(payload) _LOGGER.debug(f"Plain-text payload: {decrypted_payload.hex()}") + + # Protobuf device-summary frames (e.g the c490) are not + # the flat TLV the other telemetry frames use. + if cmd.hex() in self._PROTOBUF_TELEMETRY_COMMANDS: + self._summary = walk_protobuf( + self._protobuf_body(decrypted_payload) + ) + _LOGGER.debug( + f"Protobuf summary ({len(self._summary)} fields)" + ) + return None + parameters = Parameters.parse(decrypted_payload) return await self._process_telemetry(parameters) @@ -1419,6 +1477,7 @@ def _reset_session(self, reset_data: bool = True) -> None: self._shared_secret = None self._authorized = False self._auth_mode = None + self._summary = {} self._last_packet_timestamp = None self._negotiation_timestamp = None self._packet_futures: dict[bytes, list[asyncio.Future]] = {} diff --git a/SolixBLE/parsing.py b/SolixBLE/parsing.py new file mode 100644 index 0000000..0c4c297 --- /dev/null +++ b/SolixBLE/parsing.py @@ -0,0 +1,142 @@ +"""Walkers for nested self-delimiting telemetry payloads. + +.. moduleauthor:: kb1ibt + +Most telemetry fields are flat, fixed-layout TLV that +:meth:`SolixBLE.device.SolixBLEDevice._parse_payload` decodes directly. A few fields +instead carry a *nested* self-delimiting structure that fixed offsets cannot decode, +in one of two encodings (distinguished by the value's leading type byte): + +* **protobuf** (type byte ``0x07``) -- e.g. the C2000 G2's ``c490`` device summary. + Walked by :func:`walk_protobuf`. +* **length-value** (a ```` sequence, type byte ``0x04`` binary) -- e.g. + the ``ce`` combination-battery block in the C2000 G2's ``c421`` telemetry. Walked by + :func:`walk_lv`. + +Both encodings self-delimit, so a value that grows past a byte boundary never shifts +the fields after it -- the whole point over a brittle fixed-offset map. +""" + +from __future__ import annotations + + +def read_varint(buf: bytes, pos: int) -> tuple[int, int]: + """Read one LEB128 varint from ``buf`` at ``pos``. + + :param buf: Buffer to read from. + :param pos: Index to start reading at. + :returns: ``(value, next_pos)``. + """ + result = shift = 0 + while True: + byte = buf[pos] + result |= (byte & 0x7F) << shift + pos += 1 + if not byte & 0x80: + return result, pos + shift += 7 + + +def _is_protobuf_message(sub: bytes) -> bool: + """True if ``sub`` parses cleanly as a protobuf message (so it is recursed into).""" + pos = 0 + try: + while pos < len(sub): + tag, pos = read_varint(sub, pos) + wire = tag & 7 + if wire == 0: + _, pos = read_varint(sub, pos) + elif wire == 2: + length, pos = read_varint(sub, pos) + pos += length + elif wire == 5: + pos += 4 + elif wire == 1: + pos += 8 + else: + return False + return pos == len(sub) + except (IndexError, ValueError): + return False + + +def walk_protobuf( + buf: bytes, + prefix: str = "", + out: dict[str, object] | None = None, +) -> dict[str, object]: + """Flatten a protobuf(-like) blob to a ``.field.subfield`` -> value map. + + Repeated tags keep wire order (occurrence index appended as ``#n``) and every field + is addressed by its ``.path``, so byte offsets never matter -- a leaf value crossing + a varint byte boundary grows in place without shifting anything after it. + Length-delimited fields that themselves parse cleanly as a sub-message are recorded + as their byte length **and** recursed into (so the container and its leaves both + appear); otherwise they are kept as an ASCII string (if fully printable) or hex. A + wire-type 3/4 group marker is recorded as ``None`` and stops the walk. Every field + is recorded -- silently dropping containers under-counts the message, which is a + wrong decode. + + :param buf: The (decrypted, reassembled) protobuf payload. + :param prefix: Path prefix used during recursion. + :param out: Accumulator dict used during recursion. + :returns: Mapping of ``.path`` to value (int, str, hex str or None). + """ + if out is None: + out = {} + pos = 0 + seen: dict[int, int] = {} + while pos < len(buf): + try: + tag, pos = read_varint(buf, pos) + except IndexError: + break + fnum, wire = tag >> 3, tag & 7 + occ = seen.get(fnum, 0) + seen[fnum] = occ + 1 + path = f"{prefix}.{fnum}" + (f"#{occ}" if occ else "") + if wire == 0: + out[path], pos = read_varint(buf, pos) + elif wire == 2: + length, pos = read_varint(buf, pos) + sub = buf[pos : pos + length] + pos += length + if length and _is_protobuf_message(sub) and any(sub): + out[path] = length # container: record its length, then its fields + walk_protobuf(sub, path, out) + elif sub and all(32 <= b < 127 for b in sub): + out[path] = sub.decode("ascii") + else: + out[path] = sub.hex() + elif wire == 5: + out[path] = int.from_bytes(buf[pos : pos + 4], "little") + pos += 4 + elif wire == 1: + out[path] = int.from_bytes(buf[pos : pos + 8], "little") + pos += 8 + else: + out[path] = None # group marker (wire type 3/4): record and stop + break + return out + + +def walk_lv(buf: bytes) -> list[bytes]: + """Walk a length-value blob into its fields. + + Each field is a single length byte followed by that many value bytes, repeated to + the end of ``buf``. Used for nested ``bin`` fields such as the C2000 G2's ``ce`` + combination-battery block, whose first field is a fixed 16-byte device ID (all-zero + when no unit is combined). Trailing zero padding therefore appears as trailing + zero-length fields. Pass the value **without** its leading type byte. + + :param buf: The field value, with its ``0x04`` type byte already stripped. + :returns: The fields in wire order. + """ + fields: list[bytes] = [] + pos = 0 + while pos < len(buf): + length = buf[pos] + pos += 1 + fields.append(buf[pos : pos + length]) + pos += length + return fields From e63f006f17a767f4756e6c087b0569239d332c21 Mon Sep 17 00:00:00 2001 From: Shawn Stricker Date: Wed, 2 Sep 2026 13:25:34 -0400 Subject: [PATCH 07/11] fix(gen2): restore the subscribe command, and add the C2000 G2 (A1783) Two independent things, the first of which the second depends on. fix: C1000G2._post_connect and MagGo3in1._post_connect both still call _send_command(cmd=..., payload=...), but _send_command now takes a required `parameters` argument and no `payload`. Both raise TypeError on every connection. For the Gen 2 that means the subscribe command never goes out and the device streams no telemetry at all; there was no test covering _post_connect on any device, which is why it went unnoticed. Both are switched to the parameters interface and both now have a regression test asserting the exact bytes. feat: adds C2000G2 (A1783), the larger sibling of the C1000 G2. It shares the Gen 2 framing, TLV map and AC/DC control, so ports and power come from C1000G2 unchanged. On top it adds the parts of the frame that were not decoded when C1000G2 was written: - 4103 system group: display switch, brightness and timeout, plus the SoC charge cap and discharge floor. The firmware accepts any percentage, not just the app's menu -- 99 was set over BLE and read back verbatim -- so the setters range-check rather than restrict. - AC and DC output auto-off countdowns, settable and readable. These are a u32 under type 03 -- the same 4-byte form the `fe` timestamp takes, and not the u16 the display timeout uses -- so the value width is keyed off the type code rather than assumed. - 4057 realtime latch. This is the device's actual "start streaming" switch; 4100 is a one-shot poll despite its name. It is only honoured with routing byte 0x21 -- the MQTT-side 0x22 is accepted and silently dropped. The disable carries a warning: it also gates the device's periodic protobuf summary, and re-arming does not bring that back, so it must never be called from teardown. - a3[0] work status and a5[1] charge/discharge status. These are two different fields: the second trips on any flow, the first waits for a threshold, so they disagree at low load. Both test frames are real, and the attached one captures that disagreement at 2 W. - a3[0] also takes a fourth value during a firmware update that is not in the status enum, surfaced as `firmware_updating` rather than silently mapping to UNKNOWN. - a6[6:8] time remaining, bidirectional and in deci-hours. - the rest of the a4 settings block: AC input limit and frequency, AC and 12 V DC output modes, device idle timeout, ultrafast-charge and port-memory switches. - f9 version block: seven little-endian quads, exposed per submodule. The inverter slot reads zero whenever the inverter is not energised, which means idle rather than absent. - c0 expansion block for the BP2000. It is emitted whether or not a pack is attached -- absent, the fields are padded with sentinels (a 239 temperature, a 0 percentage) -- so every expansion property gates on subPackageConnectionStatus rather than on the tag being present. The block is also variable width: the serial is 16 bytes absent and 17 attached, so both length prefixes are walked instead of assuming offsets. Telemetry tests use two real decrypted c421 frames from an A1783, one before the BP2000 was attached and one after, so the absent and attached layouts are both covered. The two frames also differ in their device idle timeout, which pins that field independently of the layout. The support matrix gains a C2000 G2 column. Its cells are realigned because the table padded to visual width, which had drifted the emoji columns out of alignment with the separator row. Not included: Total Power In, which the Gen 2 frame carries only as separate AC and DC inputs with no grand total. Co-Authored-By: Claude Opus 5 --- SolixBLE/__init__.py | 2 + SolixBLE/devices/__init__.py | 2 + SolixBLE/devices/c1000g2.py | 10 +- SolixBLE/devices/c2000g2.py | 685 +++++++++++++++++++++++++++++++++ SolixBLE/devices/maggo_3in1.py | 10 +- docs/source/api.rst | 1 + docs/source/c2000g2.rst | 9 + docs/source/index.rst | 80 ++-- tests/devices/c1000g2.py | 11 + tests/devices/c2000g2.py | 202 ++++++++++ tests/test_commands.py | 2 + tests/test_devices.py | 95 +++++ 12 files changed, 1063 insertions(+), 46 deletions(-) create mode 100644 SolixBLE/devices/c2000g2.py create mode 100644 docs/source/c2000g2.rst create mode 100644 tests/devices/c2000g2.py diff --git a/SolixBLE/__init__.py b/SolixBLE/__init__.py index e19bddd..c8621bb 100644 --- a/SolixBLE/__init__.py +++ b/SolixBLE/__init__.py @@ -11,6 +11,7 @@ C800, C1000, C1000G2, + C2000G2, F2000, F2600, F3800, @@ -42,6 +43,7 @@ "C800", "C1000", "C1000G2", + "C2000G2", "F2000", "F2600", "F3800", diff --git a/SolixBLE/devices/__init__.py b/SolixBLE/devices/__init__.py index b4be66f..4d0497f 100644 --- a/SolixBLE/devices/__init__.py +++ b/SolixBLE/devices/__init__.py @@ -9,6 +9,7 @@ from .c800 import C800 from .c1000 import C1000 from .c1000g2 import C1000G2 +from .c2000g2 import C2000G2 from .f2000 import F2000 from .f2600 import F2600 from .f3800 import F3800 @@ -26,6 +27,7 @@ "C800", "C1000", "C1000G2", + "C2000G2", "F2000", "F2600", "F3800", diff --git a/SolixBLE/devices/c1000g2.py b/SolixBLE/devices/c1000g2.py index 279b316..25aba9f 100644 --- a/SolixBLE/devices/c1000g2.py +++ b/SolixBLE/devices/c1000g2.py @@ -10,7 +10,11 @@ #: Command sent after connecting to start the telemetry stream. Unlike the gen-1 #: models, the Gen 2 streams nothing until it receives this subscribe command. CMD_SUBSCRIBE = "4100" -SUBSCRIBE_PAYLOAD = "a10121" +SUBSCRIBE_PARAMETERS = { + "a1": { + "value": "21", + }, +} CMD_AC_OUTPUT = "4101" CMD_DC_OUTPUT = "4102" @@ -60,8 +64,8 @@ async def _post_connect(self) -> None: it after every (re)connection. """ await self._send_command( - cmd=bytes.fromhex(CMD_SUBSCRIBE), - payload=bytes.fromhex(SUBSCRIBE_PAYLOAD), + cmd=CMD_SUBSCRIBE, + parameters=SUBSCRIBE_PARAMETERS, ) async def turn_ac_on(self) -> None: diff --git a/SolixBLE/devices/c2000g2.py b/SolixBLE/devices/c2000g2.py new file mode 100644 index 0000000..022775f --- /dev/null +++ b/SolixBLE/devices/c2000g2.py @@ -0,0 +1,685 @@ +"""C2000(X) Gen 2 power station model. + +.. moduleauthor:: kb1ibt + +""" + +from ..const import ( + DEFAULT_METADATA_BOOL, + DEFAULT_METADATA_FLOAT, + DEFAULT_METADATA_INT, + DEFAULT_METADATA_STRING, +) +from ..states import ChargingStatus +from .c1000g2 import ( + C1000G2, + CMD_AC_OUTPUT, + CMD_DC_OUTPUT, + CMD_SUBSCRIBE, + SUBSCRIBE_PARAMETERS, +) + +#: System-parameters group. The Gen 2 has no per-setting opcodes -- display +#: switch, brightness, timeout and the SoC limits are all fields of this one +#: command, selected by payload tag. +CMD_SYSTEM = "4103" + +#: Realtime telemetry latch. This is the device's actual "start streaming" +#: switch; ``4100`` is a one-shot poll despite its name. +CMD_REALTIME = "4057" + +#: ``4057`` is only honoured with this routing byte. ``0x22`` -- the form the app +#: uses over MQTT -- and ``0x31`` are accepted onto the wire and silently +#: dropped, so a wrong value here looks like a device that ignores the command. +REALTIME_ROUTING = "21" + +#: ``a3[0]`` value emitted while a firmware update is in progress. It is absent +#: from the published status enum, and a decoder treating the field as tri-state +#: mis-renders every frame sent during an update. +WORK_STATUS_UPDATING = 5 + +BRIGHTNESS_VALUES = (1, 2, 3) +DISPLAY_TIMEOUT_VALUES = (60, 300, 1800) + +#: Seconds between polls for fresh telemetry. +KEEP_ALIVE_INTERVAL = 2 + +#: Highest accepted battery percentage for the charge cap / discharge floor. +MAX_PERCENTAGE = 100 + +#: Bounds on the AC and DC output auto-off countdowns. +MAX_TIMEOUT_SECONDS = 86400 +TIMEOUT_STEP_SECONDS = 300 + +#: ``subPackageConnectionStatus`` value meaning a pack is actually attached. The +#: block is emitted with padded fields even when nothing is connected, so this is +#: the only field that distinguishes the two cases. +SUB_PACKAGE_CONNECTED = 1 + +#: Bytes of the ``c0`` block that follow its variable-length serial. +SUB_PACKAGE_TAIL_LENGTH = 15 + + +#: Value width in bytes for each TLV type code. Note the code is not the width: +#: type 3 carries four bytes, the same form the `fe` timestamp uses. +TYPE_WIDTHS = {1: 1, 2: 2, 3: 4} + + +def _parameters(key: str, value: int, type_: int = 1) -> dict: + """Build a single-field command payload for the group commands. + + Values are encoded here rather than left as ints: the packet builder + converts a bare int with a one-byte big-endian default, which overflows for + anything wider. + + :param key: Payload tag selecting the setting (e.g. "a2", "aa"). + :param value: Value to set. + :param type_: TLV type code -- 1 for a byte, 2 for a u16, 3 for a u32. + :returns: Parameter dictionary for :meth:`_send_command`. + """ + return { + "a1": { + "value": "21", + }, + key: { + "type": type_, + "value": value.to_bytes(TYPE_WIDTHS[type_], byteorder="little"), + }, + } + + +def _validate_timeout(seconds: int) -> None: + """Range-check an output auto-off countdown. + + :param seconds: Countdown in seconds. + :raises ValueError: If out of range or not a whole step. + """ + if not 0 <= seconds <= MAX_TIMEOUT_SECONDS or seconds % TIMEOUT_STEP_SECONDS: + msg = ( + f"Timeout must be 0-{MAX_TIMEOUT_SECONDS} " + f"in steps of {TIMEOUT_STEP_SECONDS}" + ) + raise ValueError(msg) + + +def _version(raw: bytes) -> str: + """Format one of the ``f9`` version quads. + + The four bytes are packed little-endian, so ``02 02 09 01`` is v1.9.2.2. + + :param raw: The four version bytes in wire order. + :returns: Dotted version string. + """ + return ".".join(str(b) for b in reversed(raw)) + + +class C2000G2(C1000G2): + """ + C2000(X) Gen 2 Power Station. + + Use this class to connect, monitor and control a Gen 2 C2000(X) power + station. This model is also known as the A1783. + + The C2000 G2 is the larger sibling of the C1000 G2 (A1763) and shares its + Gen 2 BLE stack: the same ``c421``/``c900`` telemetry framing and TLV field + map, the same ``4100`` poll command, and the same AC (``4101``) and DC + (``4102``) control. Its three USB-C ports, single USB-A port, AC, DC and + solar all decode identically, so the port and power properties come + unchanged from :class:`~SolixBLE.devices.c1000g2.C1000G2`. + + On top of that it adds the parts of the Gen 2 frame that had not been + decoded when the C1000 G2 class was written -- the ``4103`` system group + (display switch, brightness, timeout, and the SoC limits), the ``a3``/``a6`` + status and time-remaining fields, the ``f9`` version block including its + per-submodule slots, and the ``c0`` expansion-battery block for the BP2000. + """ + + async def _keep_alive(self) -> int | None: + """Poll for fresh telemetry. + + Despite its name ``4100`` is a **poll**, not a subscription: each one + returns a single reading and the device then goes quiet again. So a + repeat is what turns it into a steady feed, and each one draws a + ``c900`` alongside the ``c421`` carrying the same values twice. + + :meth:`enable_realtime_telemetry` is the better mechanism where it is + available -- with the latch armed the device reports every change on its + own, and this poll is only needed as a liveness heartbeat. + + :returns: Seconds until the next poll. + """ + await self._send_command( + cmd=CMD_SUBSCRIBE, + parameters=SUBSCRIBE_PARAMETERS, + ) + return KEEP_ALIVE_INTERVAL + + ##################### + # Realtime telemetry# + ##################### + + async def enable_realtime_telemetry(self) -> None: + """Arm the realtime telemetry latch. + + With this armed the device reports every change on its own, rather than + answering one reading per poll. The latch is persistent: it survives + disconnect, reconnect and session change, so it only needs arming once. + """ + await self._send_command( + cmd=CMD_REALTIME, + parameters={ + "a1": {"value": REALTIME_ROUTING}, + "a2": {"type": 1, "value": (1).to_bytes(1)}, + }, + ) + + async def disable_realtime_telemetry(self) -> None: + """Disarm the realtime telemetry latch. + + .. warning:: + This is not symmetric with :meth:`enable_realtime_telemetry`, and it + is not something to do on the way out of a session. The latch also + gates the device's periodic protobuf summary, and re-arming does + **not** bring that back -- nothing sent over BLE does. Recovering it + needs the device to reach the vendor cloud once, which is not + possible for a unit deliberately kept off the network. + + Ordinary telemetry does come back on the next enable. Only call this + if losing the summary until the next cloud session is acceptable, and + never from teardown or error handling. + """ + await self._send_command( + cmd=CMD_REALTIME, + parameters={ + "a1": {"value": REALTIME_ROUTING}, + "a2": {"type": 1, "value": (0).to_bytes(1)}, + }, + ) + + ################### + # System settings # + ################### + + async def turn_display_on(self) -> None: + """Turn the display on.""" + await self._send_command(cmd=CMD_SYSTEM, parameters=_parameters("a2", 1)) + + async def turn_display_off(self) -> None: + """Turn the display off.""" + await self._send_command(cmd=CMD_SYSTEM, parameters=_parameters("a2", 0)) + + async def set_display_brightness(self, brightness: int) -> None: + """Set the display brightness. + + :param brightness: 1 (low), 2 (medium) or 3 (high). + :raises ValueError: If brightness is not one of those values. + """ + if brightness not in BRIGHTNESS_VALUES: + raise ValueError(f"Brightness must be one of {BRIGHTNESS_VALUES}") + + await self._send_command( + cmd=CMD_SYSTEM, parameters=_parameters("a3", brightness), + ) + + async def set_display_timeout(self, seconds: int) -> None: + """Set the display timeout. + + :param seconds: 60, 300 or 1800. + :raises ValueError: If seconds is not one of those values. + """ + if seconds not in DISPLAY_TIMEOUT_VALUES: + raise ValueError(f"Timeout must be one of {DISPLAY_TIMEOUT_VALUES}") + + await self._send_command( + cmd=CMD_SYSTEM, parameters=_parameters("a4", seconds, type_=2), + ) + + async def set_max_battery_percentage(self, percentage: int) -> None: + """Set the charge cap, above which the device stops charging. + + The app only offers 80/85/90/95/100, but that is a UI convention -- the + firmware accepts any percentage, verified by setting 99 over BLE. + + :param percentage: Charge cap, 0-100. + :raises ValueError: If the percentage is out of range. + """ + if not 0 <= percentage <= MAX_PERCENTAGE: + raise ValueError(f"Percentage must be 0-{MAX_PERCENTAGE}") + + await self._send_command( + cmd=CMD_SYSTEM, parameters=_parameters("aa", percentage), + ) + + async def set_min_battery_percentage(self, percentage: int) -> None: + """Set the discharge floor, below which the device stops discharging. + + The app only offers 1/5/10/15/20, but the firmware accepts any + percentage. + + :param percentage: Discharge floor, 0-100. + :raises ValueError: If the percentage is out of range. + """ + if not 0 <= percentage <= MAX_PERCENTAGE: + raise ValueError(f"Percentage must be 0-{MAX_PERCENTAGE}") + + await self._send_command( + cmd=CMD_SYSTEM, parameters=_parameters("ab", percentage), + ) + + async def set_ac_output_timeout(self, seconds: int) -> None: + """Set the AC output auto-off countdown. + + Read it back with :attr:`ac_output_timeout`. + + :param seconds: Countdown in seconds, 0-86400 in steps of 300. 0 + disables the timer. + :raises ValueError: If the value is out of range or not a whole step. + """ + _validate_timeout(seconds) + await self._send_command( + cmd=CMD_AC_OUTPUT, parameters=_parameters("a3", seconds, type_=3), + ) + + async def set_dc_output_timeout(self, seconds: int) -> None: + """Set the DC output auto-off countdown. + + Read it back with :attr:`dc_output_timeout`. + + :param seconds: Countdown in seconds, 0-86400 in steps of 300. 0 + disables the timer. + :raises ValueError: If the value is out of range or not a whole step. + """ + _validate_timeout(seconds) + await self._send_command( + cmd=CMD_DC_OUTPUT, parameters=_parameters("a3", seconds, type_=3), + ) + + ############## + # Status # + ############## + + @property + def charging_status(self) -> ChargingStatus: + """Whether the battery is charging, discharging or idle. + + This is the device's ``workStatus``, which applies a load threshold of + roughly 11 W. The device carries a second, more sensitive flow field + (:attr:`charge_discharge_status`) that trips on any flow at all, so the + two disagree at low load -- always with this one reading idle. + + A firmware update puts the field in a fourth state that is not part of + the status enum; use :attr:`firmware_updating` to detect it. + + :returns: Charging status, or UNKNOWN if there is no data. + """ + if self._data is None: + return ChargingStatus.UNKNOWN + + try: + return ChargingStatus(self._parse_int("a3", begin=1, end=2)) + except ValueError: + return ChargingStatus.UNKNOWN + + @property + def charge_discharge_status(self) -> ChargingStatus: + """Whether current is flowing into or out of the battery. + + The device's ``chargeDischargeStatus``. Unlike :attr:`charging_status` + this trips on any flow rather than waiting for a load threshold, so it + leads that field into and out of both charge and discharge. + + :returns: Charging status, or UNKNOWN if there is no data. + """ + if self._data is None: + return ChargingStatus.UNKNOWN + + try: + return ChargingStatus(self._parse_int("a5", begin=2, end=3)) + except ValueError: + return ChargingStatus.UNKNOWN + + @property + def firmware_updating(self) -> bool: + """Whether a firmware update is in progress. + + The unit reports this while updating either its own firmware or an + attached expansion pack's. + + :returns: True while updating, else False, or default bool value if + there is no data. + """ + if self._data is None: + return DEFAULT_METADATA_BOOL + + return self._parse_int("a3", begin=1, end=2) == WORK_STATUS_UPDATING + + @property + def time_remaining(self) -> float: + """Time remaining to full or empty, in hours. + + The field is bidirectional: it counts down to empty while discharging + and to full while charging. + + :returns: Hours remaining, or default float value if there is no data. + """ + if self._data is None: + return DEFAULT_METADATA_FLOAT + + return self._parse_int("a6", begin=7, end=9) / 10.0 + + @property + def hours_remaining(self) -> float: + """Time remaining to full/empty, with whole days overflowed out. + + Use :attr:`time_remaining` for the total including days. + + :returns: Hours remaining, or default float value if there is no data. + """ + if self._data is None: + return DEFAULT_METADATA_FLOAT + + return round(divmod(self.time_remaining, 24)[1], 1) + + @property + def days_remaining(self) -> int: + """Time remaining to full/empty, whole days only. + + Use :attr:`time_remaining` for the total including hours. + + :returns: Days remaining, or default int value if there is no data. + """ + if self._data is None: + return DEFAULT_METADATA_INT + + return int(divmod(self.time_remaining, 24)[0]) + + @property + def ac_frequency(self) -> int: + """AC mains frequency (Hz). + + :returns: Frequency in Hz, or default int value if there is no data. + """ + return self._parse_int("a4", begin=7, end=8) + + @property + def ac_input_limit(self) -> int: + """Configured ceiling on AC input power (W). + + :returns: Limit in watts, or default int value if there is no data. + """ + return self._parse_int("a4", begin=5, end=7) + + @property + def ac_output_timeout(self) -> int: + """AC output auto-off countdown (s), 0 when no timer is set. + + :returns: Countdown in seconds, or default int value if there is no + data. + """ + return self._parse_int("a4", begin=1, end=5) + + @property + def dc_output_timeout(self) -> int: + """DC output auto-off countdown (s), 0 when no timer is set. + + :returns: Countdown in seconds, or default int value if there is no + data. + """ + return self._parse_int("a4", begin=9, end=13) + + @property + def ac_output_mode(self) -> int: + """AC output mode -- 0 normal, 1 smart (auto-off below 14 W). + + :returns: Mode, or default int value if there is no data. + """ + return self._parse_int("a4", begin=8, end=9) + + @property + def dc_12v_output_mode(self) -> int: + """12 V DC output mode -- 0 normal, 1 smart (auto-off below 3 W). + + :returns: Mode, or default int value if there is no data. + """ + return self._parse_int("a4", begin=13, end=14) + + @property + def device_timeout_minutes(self) -> int: + """Minutes of inactivity before the unit powers itself down. + + :returns: Timeout in minutes, 0 for never, or default int value if + there is no data. + """ + return self._parse_int("a4", begin=14, end=16) + + @property + def ac_fast_charge_enabled(self) -> bool: + """Whether ultrafast AC charging is enabled. + + :returns: True if enabled, else False, or default bool value if there + is no data. + """ + if self._data is None: + return DEFAULT_METADATA_BOOL + + return bool(self._parse_int("a4", begin=21, end=22)) + + @property + def port_memory_enabled(self) -> bool: + """Whether output ports return to their previous state after a restart. + + :returns: True if enabled, else False, or default bool value if there + is no data. + """ + if self._data is None: + return DEFAULT_METADATA_BOOL + + return bool(self._parse_int("a4", begin=23, end=24)) + + @property + def display_on(self) -> bool: + """Whether the display is currently on. + + :returns: True if lit, else False, or default bool value if there is no + data. + """ + if self._data is None: + return DEFAULT_METADATA_BOOL + + return bool(self._parse_int("a4", begin=22, end=23)) + + @property + def display_brightness(self) -> int: + """Display brightness, 1 (low) to 3 (high). + + :returns: Brightness, or default int value if there is no data. + """ + return self._parse_int("a4", begin=18, end=19) + + @property + def display_timeout(self) -> int: + """Seconds of inactivity before the display turns itself off. + + :returns: Timeout in seconds, or default int value if there is no data. + """ + return self._parse_int("a4", begin=16, end=18) + + #################### + # Firmware versions# + #################### + + def _version_slot(self, index: int) -> str: + """Read one of the seven 4-byte version quads out of ``f9``. + + :param index: Slot index, 0-6. + :returns: Dotted version string, or default str value if there is no + data or the block is short. + """ + if self._data is None or "f9" not in self._data: + return DEFAULT_METADATA_STRING + + # value_legacy retains the leading type byte. + block = self._data["f9"].value_legacy[1:] + begin = index * 4 + if len(block) < begin + 4: + return DEFAULT_METADATA_STRING + + return _version(block[begin : begin + 4]) + + @property + def software_version(self) -> str: + """Main software version. + + :returns: Firmware version or default str value. + """ + return self._version_slot(0) + + @property + def software_version_sub_mcu(self) -> str: + """Software version of the sub-MCU. + + :returns: Firmware version or default str value. + """ + return self._version_slot(1) + + @property + def software_version_inverter(self) -> str: + """Software version of the inverter. + + Reads zero whenever the inverter is not energised -- by either AC path, + not only mains input -- so a zero here means idle rather than absent. + + :returns: Firmware version or default str value. + """ + return self._version_slot(3) + + @property + def software_version_bms(self) -> str: + """Software version of the battery management system. + + :returns: Firmware version or default str value. + """ + return self._version_slot(4) + + @property + def software_version_module(self) -> str: + """Software version of the wireless module. + + :returns: Firmware version or default str value. + """ + return self._version_slot(6) + + ###################### + # Expansion battery # + ###################### + + @property + def _sub_package(self) -> bytes | None: + """Fixed-layout tail of the ``c0`` block, past the variable-length serial. + + ``c0`` is **not** fixed width: its first field is a length-prefixed + serial that is 17 bytes with a pack attached and 16 without, so every + following offset shifts by one. Both length prefixes have to be walked + rather than assuming a base offset. + + :returns: The bytes from ``subPackageNumber`` onwards, or None if the + tag is absent or too short to contain them. + """ + if self._data is None or "c0" not in self._data: + return None + + # value_legacy retains the leading type byte (0x04, a `bin` field). + block = self._data["c0"].value_legacy[1:] + if not block: + return None + + fields = block[1 + block[0] :] + return fields if len(fields) >= SUB_PACKAGE_TAIL_LENGTH else None + + @property + def expansion_present(self) -> bool: + """Whether a BP2000 expansion battery is attached. + + The ``c0`` block is emitted whether or not a pack is connected -- with + no pack the fields are padded rather than omitted, and carry sentinel + values (a 239 temperature, a 0 percentage). Callers must therefore gate + on this property rather than on the tag being present, or those + sentinels reach consumers as real readings. + + :returns: True if an expansion battery is attached, else False. + """ + fields = self._sub_package + return bool(fields) and fields[12] == SUB_PACKAGE_CONNECTED + + @property + def num_expansion(self) -> int: + """Number of expansion batteries attached. + + The C2000 G2 has a single expansion slot, so this is 1 or 0. The + underlying ``subPackageNumber`` field is a slot index rather than a + count -- it reads 1 with nothing attached -- so it is not usable here. + + :returns: 1 if an expansion battery is attached, else 0. + """ + return int(self.expansion_present) + + @property + def serial_number_expansion(self) -> str: + """Serial number of the expansion battery. + + :returns: Serial number, or default str value if no pack is attached. + """ + if self._data is None or not self.expansion_present: + return DEFAULT_METADATA_STRING + + block = self._data["c0"].value_legacy[1:] + return block[1 : 1 + block[0]].decode("ascii") + + @property + def temperature_expansion(self) -> int: + """Temperature of the expansion battery (C). + + :returns: Temperature in degrees C, or default int value if no pack is + attached. + """ + fields = self._sub_package + if not self.expansion_present: + return DEFAULT_METADATA_INT + + return int.from_bytes(fields[5:6], byteorder="little", signed=True) + + @property + def battery_percentage_expansion(self) -> int: + """Battery percentage of the expansion battery. + + :returns: Percentage charge, or default int value if no pack is attached. + """ + fields = self._sub_package + if not self.expansion_present: + return DEFAULT_METADATA_INT + + return fields[7] + + @property + def battery_health_expansion(self) -> int: + """Battery health of the expansion battery as a percentage. + + :returns: Percentage health, or default int value if no pack is attached. + """ + fields = self._sub_package + if not self.expansion_present: + return DEFAULT_METADATA_INT + + return fields[8] + + @property + def software_version_expansion(self) -> str: + """Software version of the expansion battery. + + :returns: Firmware version, or default str value if no pack is attached. + """ + fields = self._sub_package + if not self.expansion_present: + return DEFAULT_METADATA_STRING + + return _version(fields[1:5]) diff --git a/SolixBLE/devices/maggo_3in1.py b/SolixBLE/devices/maggo_3in1.py index 1517054..fa4183c 100644 --- a/SolixBLE/devices/maggo_3in1.py +++ b/SolixBLE/devices/maggo_3in1.py @@ -12,7 +12,11 @@ #: C1000G2, this charger streams nothing until it receives this subscribe #: command. CMD_SUBSCRIBE = "4200" -SUBSCRIBE_PAYLOAD = "a10121" +SUBSCRIBE_PARAMETERS = { + "a1": { + "value": "21", + }, +} class MagGo3in1(PrimeDevice): @@ -52,8 +56,8 @@ async def _post_connect(self) -> None: send it after every (re)connection. """ await self._send_command( - cmd=bytes.fromhex(CMD_SUBSCRIBE), - payload=bytes.fromhex(SUBSCRIBE_PAYLOAD), + cmd=CMD_SUBSCRIBE, + parameters=SUBSCRIBE_PARAMETERS, ) @property diff --git a/docs/source/api.rst b/docs/source/api.rst index efb7111..b873b53 100644 --- a/docs/source/api.rst +++ b/docs/source/api.rst @@ -17,6 +17,7 @@ the list of properties for that class. c800 c1000 c1000g2 + c2000g2 f2000 f2600 f3800 diff --git a/docs/source/c2000g2.rst b/docs/source/c2000g2.rst new file mode 100644 index 0000000..363f27f --- /dev/null +++ b/docs/source/c2000g2.rst @@ -0,0 +1,9 @@ +C2000(X) G2 +=========== + +.. autoclass:: SolixBLE.C2000G2 + :members: + :inherited-members: connect, disconnect, add_callback, remove_callback, connected, available, address, name, supports_telemetry, last_update + :special-members: __init__ + :member-order: groupwise + :no-index: diff --git a/docs/source/index.rst b/docs/source/index.rst index 9227fc7..fb98304 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -33,46 +33,46 @@ No pairing is required in order to receive telemetry data or control the device. Power station support --------------------- -======================= ======== ========== ========= ========= ========= ============ ====== ====== -Parameter C300(X) C300(X) DC C800(X) C1000(X) C1000 G2 F2000 (767) F2600 F3800 -======================= ======== ========== ========= ========= ========= ============ ====== ====== -Charging status ✅ ✅ ❌ ❌ ❌ ❌ ✅ ✅ -Time remaining ✅ ✅ ✅ ✅ ❌ ✅ ✅ ✅ -Battery percentage ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ -Battery health ❌ ✅ ✅ ✅ ✅ ✅ ✅ ❌ -Temperature ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ -Total Power In ✅ ✅ ✅ ✅ ❌ ❌ ✅ ✅ -Total Power Out ✅ ✅ ✅ ✅ ✅ ❌ ✅ ✅ -AC on/off control ✅ N/A ✅ ✅ ✅ ❌ ✅ ✅ -AC Power in ✅ N/A ✅ ✅ ✅ ✅ ✅ ✅ -AC Power out ✅ N/A ✅ ✅ ✅ ✅ ✅ ✅ -AC on/off state ✅ N/A ✅ ✅ ✅ ❌ ✅ ✅ -AC Timer ✅ N/A ✅ ✅ ❌ ❌ ✅ ❌ -DC on/off control ✅ ✅ ✅ ✅ ✅ ❌ ✅ ✅ -DC Power in ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ -DC Power out ✅ ✅ ❌ ✅ ✅ ✅ ✅ ✅ -DC Power in status ✅ ✅ ❌ ❌ ✅ ❌ ✅ ❌ -DC Power out status ✅ ❌ ❌ ✅ ✅ ❌ ✅ ✅ -DC Timer ✅ ✅ ❌ ❌ ❌ ❌ ✅ ❌ -USB Power out ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ -USB Port status ✅ ✅ ❌ ❌ ✅ ❌ ✅ ✅ -Light control ✅ ✅ ✅ ✅ ❌ ❌ ✅ ❌ -Light status ✅ ✅ ❌ ❌ N/A ❌ ✅ ❌ -Display on/off control ✅ ✅ ✅ ✅ ❌ ❌ ✅ ❌ -Display on/off status ❌ ✅ ❌ ❌ ❌ ❌ ✅ ❌ -Display brightness ctrl ✅ ✅ ✅ ✅ ❌ ❌ ✅ ❌ -Display brightness stat ❌ ✅ ❌ ❌ ❌ ❌ ✅ ❌ -Display timeout ctrl ✅ ✅ ✅ ✅ ❌ ❌ ✅ ❌ -Display timeout stat ❌ ✅ ❌ ❌ ❌ ❌ ✅ ❌ -Firmware version ✅ ✅ ✅ ✅ ❌ ✅ ✅ ✅ -Serial number ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ -Expansion temperature N/A N/A N/A ✅ N/A ✅ ✅ ❌ -Expansion percentage N/A N/A N/A ✅ N/A ✅ ✅ ✅ -Expansion health N/A N/A N/A ✅ N/A ✅ ✅ ❌ -Expansion firmware N/A N/A N/A ✅ N/A ✅ ✅ ✅ -Expansion num N/A N/A N/A ✅ N/A ✅ ✅ ❌ -Polled status updates ✅ ❌ ✅ ✅ ❌ ❌ ✅ ❌ -======================= ======== ========== ========= ========= ========= ============ ====== ====== +======================= ======= ========== ======= ======== ======== ======== =========== ===== ===== +Parameter C300(X) C300(X) DC C800(X) C1000(X) C1000 G2 C2000 G2 F2000 (767) F2600 F3800 +======================= ======= ========== ======= ======== ======== ======== =========== ===== ===== +Charging status ✅ ✅ ❌ ❌ ❌ ✅ ❌ ✅ ✅ +Time remaining ✅ ✅ ✅ ✅ ❌ ✅ ✅ ✅ ✅ +Battery percentage ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ +Battery health ❌ ✅ ✅ ✅ ✅ ✅ ✅ ✅ ❌ +Temperature ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ +Total Power In ✅ ✅ ✅ ✅ ❌ ❌ ❌ ✅ ✅ +Total Power Out ✅ ✅ ✅ ✅ ✅ ✅ ❌ ✅ ✅ +AC on/off control ✅ N/A ✅ ✅ ✅ ✅ ❌ ✅ ✅ +AC Power in ✅ N/A ✅ ✅ ✅ ✅ ✅ ✅ ✅ +AC Power out ✅ N/A ✅ ✅ ✅ ✅ ✅ ✅ ✅ +AC on/off state ✅ N/A ✅ ✅ ✅ ✅ ❌ ✅ ✅ +AC Timer ✅ N/A ✅ ✅ ❌ ✅ ❌ ✅ ❌ +DC on/off control ✅ ✅ ✅ ✅ ✅ ✅ ❌ ✅ ✅ +DC Power in ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ +DC Power out ✅ ✅ ❌ ✅ ✅ ✅ ✅ ✅ ✅ +DC Power in status ✅ ✅ ❌ ❌ ✅ ✅ ❌ ✅ ❌ +DC Power out status ✅ ❌ ❌ ✅ ✅ ✅ ❌ ✅ ✅ +DC Timer ✅ ✅ ❌ ❌ ❌ ✅ ❌ ✅ ❌ +USB Power out ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ +USB Port status ✅ ✅ ❌ ❌ ✅ ✅ ❌ ✅ ✅ +Light control ✅ ✅ ✅ ✅ ❌ ❌ ❌ ✅ ❌ +Light status ✅ ✅ ❌ ❌ N/A N/A ❌ ✅ ❌ +Display on/off control ✅ ✅ ✅ ✅ ❌ ✅ ❌ ✅ ❌ +Display on/off status ❌ ✅ ❌ ❌ ❌ ✅ ❌ ✅ ❌ +Display brightness ctrl ✅ ✅ ✅ ✅ ❌ ✅ ❌ ✅ ❌ +Display brightness stat ❌ ✅ ❌ ❌ ❌ ✅ ❌ ✅ ❌ +Display timeout ctrl ✅ ✅ ✅ ✅ ❌ ✅ ❌ ✅ ❌ +Display timeout stat ❌ ✅ ❌ ❌ ❌ ✅ ❌ ✅ ❌ +Firmware version ✅ ✅ ✅ ✅ ❌ ✅ ✅ ✅ ✅ +Serial number ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ +Expansion temperature N/A N/A N/A ✅ N/A ✅ ✅ ✅ ❌ +Expansion percentage N/A N/A N/A ✅ N/A ✅ ✅ ✅ ✅ +Expansion health N/A N/A N/A ✅ N/A ✅ ✅ ✅ ❌ +Expansion firmware N/A N/A N/A ✅ N/A ✅ ✅ ✅ ✅ +Expansion num N/A N/A N/A ✅ N/A ✅ ✅ ✅ ❌ +Polled status updates ✅ ❌ ✅ ✅ ❌ ✅ ❌ ✅ ❌ +======================= ======= ========== ======= ======== ======== ======== =========== ===== ===== Solar system support diff --git a/tests/devices/c1000g2.py b/tests/devices/c1000g2.py index b3b2a8c..4552e9b 100644 --- a/tests/devices/c1000g2.py +++ b/tests/devices/c1000g2.py @@ -17,6 +17,17 @@ # raised where appropriate. See test_send_command() in test_commands.py. C1000G2_TEST_COMMANDS = [ + # The Gen 2 streams nothing until it receives this subscribe command, so a + # broken _post_connect costs the device all of its telemetry with no other + # symptom. It had no coverage, and went unnoticed when _send_command moved + # from a payload to a parameters interface. + pytest.param( + C1000G2, + "_post_connect", + [], + [("4100", "a10121")], + id="c1000g2_post_connect_subscribe", + ), pytest.param( C1000G2, "turn_ac_on", diff --git a/tests/devices/c2000g2.py b/tests/devices/c2000g2.py new file mode 100644 index 0000000..031cb2a --- /dev/null +++ b/tests/devices/c2000g2.py @@ -0,0 +1,202 @@ +"""C2000G2 power station device tests. + +.. moduleauthor:: kb1ibt + +""" + +import pytest + +from SolixBLE.devices.c2000g2 import C2000G2 + +######################## +# Test device commands # +######################## + +# These tests are for sending commands to the device and making sure the +# correct calls are made to the command sending functions and errors are +# raised where appropriate. See test_send_command() in test_commands.py. + +C2000G2_TEST_COMMANDS = [ + pytest.param( + C2000G2, + "_post_connect", + [], + [("4100", "a10121")], + id="c2000g2_post_connect_subscribe", + ), + # 4100 is a poll, not a subscription: one request, one reading. Repeating it + # is what turns it into a steady feed. + pytest.param( + C2000G2, + "_keep_alive", + [], + [("4100", "a10121")], + id="c2000g2_keep_alive_poll", + ), + # 4057 is the actual realtime latch. It is only honoured with routing byte + # 0x21 -- the MQTT-side 0x22 is accepted and silently dropped. + pytest.param( + C2000G2, + "enable_realtime_telemetry", + [], + [("4057", "a10121a2020101")], + id="c2000g2_realtime_enable", + ), + pytest.param( + C2000G2, + "disable_realtime_telemetry", + [], + [("4057", "a10121a2020100")], + id="c2000g2_realtime_disable", + ), + # The Gen 2 has no per-setting opcodes: the display switch, brightness, + # timeout and both SoC limits are all fields of the 4103 system group, + # selected by payload tag. + pytest.param( + C2000G2, + "turn_display_on", + [], + [("4103", "a10121a2020101")], + id="c2000g2_display_on", + ), + pytest.param( + C2000G2, + "turn_display_off", + [], + [("4103", "a10121a2020100")], + id="c2000g2_display_off", + ), + pytest.param( + C2000G2, + "set_display_brightness", + [1], + [("4103", "a10121a3020101")], + id="c2000g2_brightness_low", + ), + pytest.param( + C2000G2, + "set_display_brightness", + [3], + [("4103", "a10121a3020103")], + id="c2000g2_brightness_high", + ), + pytest.param( + C2000G2, + "set_display_brightness", + [4], + ValueError, + id="c2000g2_brightness_invalid", + ), + # The timeout is a little-endian u16 under type 02, not a single byte. + pytest.param( + C2000G2, + "set_display_timeout", + [60], + [("4103", "a10121a403023c00")], + id="c2000g2_display_timeout_60", + ), + pytest.param( + C2000G2, + "set_display_timeout", + [1800], + [("4103", "a10121a403020807")], + id="c2000g2_display_timeout_1800", + ), + pytest.param( + C2000G2, + "set_display_timeout", + [90], + ValueError, + id="c2000g2_display_timeout_invalid", + ), + # The app only offers 80/85/90/95/100 and 1/5/10/15/20, but the firmware + # accepts any percentage -- 99 was set over BLE and read back verbatim. + pytest.param( + C2000G2, + "set_max_battery_percentage", + [100], + [("4103", "a10121aa020164")], + id="c2000g2_max_soc_100", + ), + pytest.param( + C2000G2, + "set_max_battery_percentage", + [99], + [("4103", "a10121aa020163")], + id="c2000g2_max_soc_off_menu", + ), + pytest.param( + C2000G2, + "set_max_battery_percentage", + [101], + ValueError, + id="c2000g2_max_soc_invalid", + ), + pytest.param( + C2000G2, + "set_min_battery_percentage", + [5], + [("4103", "a10121ab020105")], + id="c2000g2_min_soc_5", + ), + pytest.param( + C2000G2, + "set_min_battery_percentage", + [-1], + ValueError, + id="c2000g2_min_soc_invalid", + ), + # The output countdowns are a u32 under type 03 -- the same 4-byte form the + # `fe` timestamp uses -- not the u16 the display timeout takes. + pytest.param( + C2000G2, + "set_ac_output_timeout", + [300], + [("4101", "a10121a305032c010000")], + id="c2000g2_ac_timeout", + ), + pytest.param( + C2000G2, + "set_dc_output_timeout", + [600], + [("4102", "a10121a3050358020000")], + id="c2000g2_dc_timeout", + ), + pytest.param( + C2000G2, + "set_ac_output_timeout", + [0], + [("4101", "a10121a3050300000000")], + id="c2000g2_ac_timeout_disable", + ), + pytest.param( + C2000G2, + "set_ac_output_timeout", + [450], + ValueError, + id="c2000g2_ac_timeout_bad_step", + ), + pytest.param( + C2000G2, + "set_dc_output_timeout", + [86700], + ValueError, + id="c2000g2_dc_timeout_out_of_range", + ), + # Inherited from the C1000 G2 unchanged -- the AC and DC output switches use + # the same opcodes and payload on both models. + pytest.param( + C2000G2, + "turn_ac_on", + [], + [("4101", "a10121a2020101")], + id="c2000g2_ac_on", + ), + pytest.param( + C2000G2, + "turn_dc_off", + [], + [("4102", "a10121a2020100")], + id="c2000g2_dc_off", + ), +] diff --git a/tests/test_commands.py b/tests/test_commands.py index 205b6a6..36e3f92 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -31,6 +31,7 @@ C1000_TEST_COMMANDS_RESPONSES, ) from tests.devices.c1000g2 import C1000G2_TEST_COMMANDS, C1000G2_TEST_COMMANDS_E2E +from tests.devices.c2000g2 import C2000G2_TEST_COMMANDS from tests.devices.f2600 import ( F2600_TEST_COMMANDS, F2600_TEST_COMMANDS_E2E, @@ -57,6 +58,7 @@ *C800_TEST_COMMANDS, *C1000_TEST_COMMANDS, *C1000G2_TEST_COMMANDS, + *C2000G2_TEST_COMMANDS, *F2600_TEST_COMMANDS, *F3800_TEST_COMMANDS, *PRIME_CHARGER_160W_TEST_COMMANDS, diff --git a/tests/test_devices.py b/tests/test_devices.py index 437ba26..e4a9645 100644 --- a/tests/test_devices.py +++ b/tests/test_devices.py @@ -17,6 +17,7 @@ C800, C1000, C1000G2, + C2000G2, F2600, ChargingStatus, LightStatus, @@ -336,6 +337,100 @@ }, id="c1000g2_dc_on", ), + # The two cases below are decrypted "c421" telemetry frames captured from + # a real C2000 G2 (A1783), the first on 2026-07-21 with no expansion + # battery attached and the second on 2026-08-28 with a BP2000 attached. + # They lock in the "c0" (subPackageInfo) decode and, importantly, the + # absent case: the block is emitted either way, so with no pack the + # fields are padded and carry sentinels -- a 239 (0xEF) temperature and a + # 0 percentage -- which is why every expansion property gates on + # subPackageConnectionStatus rather than on the tag being present. + # Note "c0" is variable width: the serial is 16 bytes when absent and 17 + # when attached, so the block is 35 vs 44 bytes and every field after the + # serial shifts by one. Both length prefixes have to be walked. + pytest.param( + C2000G2, + "a10131a221062011415043444b4b453046333936303030313100054131373833010102010001a30e040000000008070064cc00580200a41b0400000000e8033c000000000000a0052c010200010000015f0500a506041e005f0000a60a040000000000001a0e5fa70704000000010000a80404000000aa0404010000ab0404000000ac0404000000ae0404000000b20404000000c0230410000000000000000000000000000000000100000000ef0000000100000000022020ce2c0410000000000000000000000000000000000111000000000000000000000000000000000000000000000000d91a040000145f050000000000000000000000000000000000000000da18048100000000000000000001e00138047f0101003804e001dc06040000000000f91d0401010201060201000000000006020100020209010000000006000300fa150401010101001f0700000000000000000000000000fd0e0031373834343830313930333432fe050372d35f6a", + { + "serial_number": "APCDKKE0F39600011", + "part_number": "A1783", + "charging_status": ChargingStatus.IDLE, + "charge_discharge_status": ChargingStatus.IDLE, + "firmware_updating": False, + "time_remaining": 361.0, + "days_remaining": 15, + "hours_remaining": 1.0, + "ac_frequency": 60, + "ac_input_limit": 1000, + "ac_output_timeout": 0, + "dc_output_timeout": 0, + "ac_output_mode": 0, + "dc_12v_output_mode": 0, + # 1440 minutes here against 0 in the later frame -- the setting + # was changed between the two captures. + "device_timeout_minutes": 1440, + "ac_fast_charge_enabled": False, + "port_memory_enabled": True, + "display_on": False, + "display_brightness": 2, + "display_timeout": 300, + "software_version": "1.2.1.1", + "software_version_sub_mcu": "0.1.2.6", + "software_version_inverter": "0.1.2.6", + "software_version_bms": "1.9.2.2", + "software_version_module": "0.3.0.6", + "expansion_present": False, + "num_expansion": 0, + "serial_number_expansion": "Unknown", + "temperature_expansion": -1, + "battery_percentage_expansion": -1, + "battery_health_expansion": -1, + "software_version_expansion": "Unknown", + }, + id="c2000g2_no_expansion", + ), + pytest.param( + C2000G2, + "a10131a221062011415043444b4b453046333936303030313100054131373833010102010001a30e04000001010807003ccc00580200a41b0400000000e8033c00000000000000002c01020001000001551400a506041a01500000a60a040200000000007b0950a70704010000000000a80404000000aa0404000000ab0404010200ac0404000000ae0404000000b20404000000c02c0411415043444b4a4d3046343832303030393701020209011500130001010001000a41313738335f326b5768ce2c0410000000000000000000000000000000000111000000000000000000000000000000000000000000000000d91a040000195514000000009c9d886aacab886a0000000000000000da18048100000000000000000001e00138047f0101013804e001dc06040000000000f91d0401010201060201000000000006020100020209010202090106000300fa150401010101001f0700000000000000000000000000fd0e0031373837383730333336313035fe05031348916a", + { + "serial_number": "APCDKKE0F39600011", + "part_number": "A1783", + # A real instance of the two flow fields disagreeing: the frame + # carries 2 W of output, which trips charge_discharge_status but + # sits under charging_status' ~11 W threshold. + "charging_status": ChargingStatus.IDLE, + "charge_discharge_status": ChargingStatus.DISCHARGING, + "firmware_updating": False, + "time_remaining": 242.7, + "days_remaining": 10, + "hours_remaining": 2.7, + "ac_frequency": 60, + "ac_input_limit": 1000, + "ac_output_timeout": 0, + "dc_output_timeout": 0, + "ac_output_mode": 0, + "dc_12v_output_mode": 0, + "device_timeout_minutes": 0, + "ac_fast_charge_enabled": False, + "port_memory_enabled": True, + "display_on": False, + "display_brightness": 2, + "display_timeout": 300, + "software_version": "1.2.1.1", + "software_version_sub_mcu": "0.1.2.6", + "software_version_inverter": "0.1.2.6", + "software_version_bms": "1.9.2.2", + "software_version_module": "0.3.0.6", + "expansion_present": True, + "num_expansion": 1, + "serial_number_expansion": "APCDKJM0F48200097", + "temperature_expansion": 21, + "battery_percentage_expansion": 19, + "battery_health_expansion": 0, + "software_version_expansion": "1.9.2.2", + }, + id="c2000g2_expansion_attached", + ), pytest.param( C300, "a10131a2050300000000a3050300000000a40302ffffa503020000a603025400a703020000a803020000a903020000aa03020100ab03020000ac03020000ad03020000ae03025500af03020000b003020100b103021b04b20302fc01b30302fc01b403021c00b503027b00b603021b04b7020101b8020100b9020124ba020100bb020164bc020164bd020100be020100bf020100c0020101c1020100c2020100c3020100c4020100c51100415a5653424a30453339323030303438c603024a01c70302a005c803022c01c903023c00ca03020000cb020101cc020100cd020102ce020132cf020100d0020100d1020101", From 78d6b4a319edd2a8b881ae3f40429d14d1f2acf8 Mon Sep 17 00:00:00 2001 From: Shawn Stricker Date: Tue, 8 Sep 2026 15:20:19 -0400 Subject: [PATCH 08/11] feat(gen2): stream the C2000 G2 c490 summary, guarded by schema version Enables the c490 protobuf device-summary on the C2000 G2 -- its telemetry set gains c490 and _PROTOBUF_TELEMETRY_COMMANDS routes it into `summary`. The frame's trailing a3 schema name (charging_pps_series_c_NNNN) versions the protobuf layout, so the field decoding is only correct for one revision. The device now records that schema (exposed as `summary_schema`) and warns when a unit posts an older revision (the 2025 _0002, an old firmware) or one newer than the validated _0005, whose values are then unverified. Adds tests for the a3 extraction, the summary routing, and the schema guard. Co-Authored-By: Claude Opus 4.8 --- SolixBLE/device.py | 97 ++++++++++++++++++++++++++++++- SolixBLE/devices/c2000g2.py | 32 +++++++++-- tests/test_c490.py | 112 ++++++++++++++++++++++++++++++++++++ 3 files changed, 234 insertions(+), 7 deletions(-) create mode 100644 tests/test_c490.py diff --git a/SolixBLE/device.py b/SolixBLE/device.py index c993461..3a147e6 100644 --- a/SolixBLE/device.py +++ b/SolixBLE/device.py @@ -77,6 +77,16 @@ NEGOTIATION_ENCRYPT_METHOD = "40" +def _schema_version(schema: str) -> int | None: + """Extract the trailing ``_NNNN`` revision number from a schema name. + + :param schema: A schema name such as ``charging_pps_series_c_0005``. + :returns: The revision number, or None if the tail is not numeric. + """ + tail = schema.rsplit("_", 1)[-1] + return int(tail) if tail.isdigit() else None + + class SolixBLEDevice: """Solix BLE device object.""" @@ -91,6 +101,13 @@ class SolixBLEDevice: #: G2's ``c490``). _PROTOBUF_TELEMETRY_COMMANDS: tuple[str, ...] = () + #: The protobuf-summary schema this class's field decoding was validated + #: against (the frame's ``a3`` schema name, e.g + #: ``charging_pps_series_c_0005``). The schema versions the protobuf layout, + #: so a device posting a different revision is walked on a best-effort basis + #: and warned about. None disables the check. + _VALIDATED_SUMMARY_SCHEMA: str | None = None + #: The maximum packet size an Anker device is able to send _mtu = 253 @@ -145,6 +162,7 @@ def __init__( self._client_token: str = client_token or str(uuid.uuid4()) self._auth_mode: bytes | None = None self._summary: dict[str, object] = {} + self._summary_schema: str | None = None @property def _encrypted_negotiation(self) -> bool: @@ -454,6 +472,18 @@ def summary(self) -> dict[str, object]: """ return self._summary + @property + def summary_schema(self) -> str | None: + """The ``a3`` schema name of the latest protobuf device-summary frame. + + The schema (e.g ``charging_pps_series_c_0005``) versions the protobuf + layout, so it identifies which revision :attr:`summary` was decoded + against. None until such a frame is received. + + :returns: The schema name, or None. + """ + return self._summary_schema + def _parse_int( self, key: str, begin: int = None, end: int = None, signed: bool = False ) -> int: @@ -513,6 +543,65 @@ def _protobuf_body(payload: bytes) -> bytes: blob_start = a2_start + 4 return payload[blob_start : blob_start + a2_length - 1] + @staticmethod + def _protobuf_schema(payload: bytes) -> str | None: + """Return the c490 frame's trailing ``a3`` schema name, if present. + + The schema (e.g ``charging_pps_series_c_0005``) follows the ``a2`` + protobuf blob and names the revision the protobuf was posted against. + + :param payload: The decrypted device-post frame. + :returns: The schema string, or None if it is absent or not ASCII. + """ + if len(payload) <= 6: + return None + a2_start = 2 + payload[1] + a2_length = int.from_bytes(payload[a2_start + 1 : a2_start + 3], "little") + a3_start = a2_start + a2_length + 3 + if a3_start + 2 > len(payload) or payload[a3_start] != 0xA3: + return None + a3_length = payload[a3_start + 1] + try: + return payload[a3_start + 2 : a3_start + 2 + a3_length].decode("ascii") + except UnicodeDecodeError: + return None + + def _check_summary_schema(self) -> None: + """Warn if the c490 schema differs from the validated revision. + + The field decoding is only correct for + :attr:`_VALIDATED_SUMMARY_SCHEMA`; an older revision (e.g ``_0002``) or + one newer than validated is walked anyway but its values may be wrong, + so it is logged. + """ + validated = self._VALIDATED_SUMMARY_SCHEMA + schema = self._summary_schema + if validated is None or schema is None or schema == validated: + return + got = _schema_version(schema) + want = _schema_version(validated) + if got is None or want is None: + _LOGGER.warning( + "Device-summary schema %r is not the validated %r; " + "values may be wrong.", + schema, + validated, + ) + elif got < want: + _LOGGER.warning( + "Device-summary schema %r predates the validated %r (older " + "firmware); values may be wrong.", + schema, + validated, + ) + else: + _LOGGER.warning( + "Device-summary schema %r is newer than the validated %r; " + "values are unverified.", + schema, + validated, + ) + def _gcm_key_nonce(self) -> tuple[bytes, bytes]: """Return the GCM (key, nonce): the ECDH secret if derived, else static. @@ -735,8 +824,13 @@ async def _process_notification( self._summary = walk_protobuf( self._protobuf_body(decrypted_payload) ) + self._summary_schema = self._protobuf_schema( + decrypted_payload + ) + self._check_summary_schema() _LOGGER.debug( - f"Protobuf summary ({len(self._summary)} fields)" + f"Protobuf summary ({len(self._summary)} fields, " + f"schema {self._summary_schema})" ) return None @@ -1478,6 +1572,7 @@ def _reset_session(self, reset_data: bool = True) -> None: self._authorized = False self._auth_mode = None self._summary = {} + self._summary_schema = None self._last_packet_timestamp = None self._negotiation_timestamp = None self._packet_futures: dict[bytes, list[asyncio.Future]] = {} diff --git a/SolixBLE/devices/c2000g2.py b/SolixBLE/devices/c2000g2.py index 022775f..8eb9399 100644 --- a/SolixBLE/devices/c2000g2.py +++ b/SolixBLE/devices/c2000g2.py @@ -132,8 +132,22 @@ class C2000G2(C1000G2): (display switch, brightness, timeout, and the SoC limits), the ``a3``/``a6`` status and time-remaining fields, the ``f9`` version block including its per-submodule slots, and the ``c0`` expansion-battery block for the BP2000. + + It also receives the ``c490`` protobuf device-summary post (armed by + :meth:`enable_realtime_telemetry`), decoded into the :attr:`summary` map. """ + #: The Gen 2 telemetry set plus the ``c490`` protobuf device-summary post. + _TELEMETRY_COMMANDS: tuple[str, ...] = ("c421", "c900", "c490") + + #: ``c490`` is a protobuf blob, walked into :attr:`summary` rather than the + #: flat TLV the other telemetry frames use. + _PROTOBUF_TELEMETRY_COMMANDS: tuple[str, ...] = ("c490",) + + #: The c490 field decoding here is validated against this schema revision; + #: an older revision (the 2025 ``_0002``) or a newer one is warned about. + _VALIDATED_SUMMARY_SCHEMA: str = "charging_pps_series_c_0005" + async def _keep_alive(self) -> int | None: """Poll for fresh telemetry. @@ -218,7 +232,8 @@ async def set_display_brightness(self, brightness: int) -> None: raise ValueError(f"Brightness must be one of {BRIGHTNESS_VALUES}") await self._send_command( - cmd=CMD_SYSTEM, parameters=_parameters("a3", brightness), + cmd=CMD_SYSTEM, + parameters=_parameters("a3", brightness), ) async def set_display_timeout(self, seconds: int) -> None: @@ -231,7 +246,8 @@ async def set_display_timeout(self, seconds: int) -> None: raise ValueError(f"Timeout must be one of {DISPLAY_TIMEOUT_VALUES}") await self._send_command( - cmd=CMD_SYSTEM, parameters=_parameters("a4", seconds, type_=2), + cmd=CMD_SYSTEM, + parameters=_parameters("a4", seconds, type_=2), ) async def set_max_battery_percentage(self, percentage: int) -> None: @@ -247,7 +263,8 @@ async def set_max_battery_percentage(self, percentage: int) -> None: raise ValueError(f"Percentage must be 0-{MAX_PERCENTAGE}") await self._send_command( - cmd=CMD_SYSTEM, parameters=_parameters("aa", percentage), + cmd=CMD_SYSTEM, + parameters=_parameters("aa", percentage), ) async def set_min_battery_percentage(self, percentage: int) -> None: @@ -263,7 +280,8 @@ async def set_min_battery_percentage(self, percentage: int) -> None: raise ValueError(f"Percentage must be 0-{MAX_PERCENTAGE}") await self._send_command( - cmd=CMD_SYSTEM, parameters=_parameters("ab", percentage), + cmd=CMD_SYSTEM, + parameters=_parameters("ab", percentage), ) async def set_ac_output_timeout(self, seconds: int) -> None: @@ -277,7 +295,8 @@ async def set_ac_output_timeout(self, seconds: int) -> None: """ _validate_timeout(seconds) await self._send_command( - cmd=CMD_AC_OUTPUT, parameters=_parameters("a3", seconds, type_=3), + cmd=CMD_AC_OUTPUT, + parameters=_parameters("a3", seconds, type_=3), ) async def set_dc_output_timeout(self, seconds: int) -> None: @@ -291,7 +310,8 @@ async def set_dc_output_timeout(self, seconds: int) -> None: """ _validate_timeout(seconds) await self._send_command( - cmd=CMD_DC_OUTPUT, parameters=_parameters("a3", seconds, type_=3), + cmd=CMD_DC_OUTPUT, + parameters=_parameters("a3", seconds, type_=3), ) ############## diff --git a/tests/test_c490.py b/tests/test_c490.py new file mode 100644 index 0000000..2a11a55 --- /dev/null +++ b/tests/test_c490.py @@ -0,0 +1,112 @@ +"""Tests for the c490 protobuf device-summary decode on the C2000 Gen 2. + +.. moduleauthor:: kb1ibt +""" + +import logging +from unittest import mock + +import pytest + +from SolixBLE import C2000G2 +from SolixBLE.constructs import Packet +from SolixBLE.device import SolixBLEDevice +from SolixBLE.parsing import walk_protobuf +from tests.const import MOCK_BLE_DEVICE + +#: A minimal protobuf message: field 1 (varint) = 42. +PROTOBUF = "082a" + +#: The c490 outer wrapper: ``a1`` (command echo) then ``a2`` (2-byte length, an +#: ``04`` type byte, and the protobuf blob), so ``a2``'s length counts the type +#: byte plus the two blob bytes. +C490_PLAINTEXT = "a10131" + "a20300" + "04" + PROTOBUF + + +def _c490_plaintext(schema: str) -> bytes: + """Build a c490 device-post plaintext with the given ``a3`` schema name.""" + a3 = b"\xa3" + bytes([len(schema)]) + schema.encode() + return bytes.fromhex(C490_PLAINTEXT) + a3 + + +def _c490_frame(device: C2000G2, schema: str) -> bytes: + """Build an encrypted c490 frame for ``schema`` as the device receives it.""" + ciphertext = device._encrypt_payload(_c490_plaintext(schema)) + return Packet.build( + { + "pattern": bytes.fromhex("03010f"), + "cmd": bytes.fromhex("c490"), + "payload_bytes": ciphertext, + }, + ) + + +def test_c2000g2_declares_c490() -> None: + """The C2000 G2 routes c490 as a protobuf telemetry frame.""" + assert "c490" in C2000G2._TELEMETRY_COMMANDS + assert C2000G2._PROTOBUF_TELEMETRY_COMMANDS == ("c490",) + + +def test_protobuf_body_extracts_a2_blob() -> None: + """The protobuf blob is sliced out of the outer ``a2`` field.""" + body = SolixBLEDevice._protobuf_body(bytes.fromhex(C490_PLAINTEXT)) + assert body.hex() == PROTOBUF + + +@pytest.mark.asyncio +async def test_c490_frame_populates_summary() -> None: + """A c490 frame is walked into the summary map, not TLV-parsed.""" + device = C2000G2(MOCK_BLE_DEVICE, capability=4, client_token="t") # noqa: S106 + client = mock.AsyncMock() + device._client = client + + ciphertext = device._encrypt_payload(bytes.fromhex(C490_PLAINTEXT)) + frame = Packet.build( + { + "pattern": bytes.fromhex("03010f"), + "cmd": bytes.fromhex("c490"), + "payload_bytes": ciphertext, + }, + ) + await device._process_notification(client, 0, frame) + + assert device.summary == walk_protobuf(bytes.fromhex(PROTOBUF)) + assert device.summary + + +def test_protobuf_schema_extracts_a3() -> None: + """The a3 schema name is read from the c490 frame.""" + plaintext = _c490_plaintext("charging_pps_series_c_0005") + assert SolixBLEDevice._protobuf_schema(plaintext) == "charging_pps_series_c_0005" + + +@pytest.mark.asyncio +async def test_older_schema_warns(caplog: pytest.LogCaptureFixture) -> None: + """An older c490 schema is recorded and warned about.""" + device = C2000G2(MOCK_BLE_DEVICE, capability=4, client_token="t") # noqa: S106 + client = mock.AsyncMock() + device._client = client + with caplog.at_level(logging.WARNING): + await device._process_notification( + client, + 0, + _c490_frame(device, "charging_pps_series_c_0002"), + ) + assert device.summary_schema == "charging_pps_series_c_0002" + assert any("predates" in r.message for r in caplog.records) + + +@pytest.mark.asyncio +async def test_validated_schema_no_warn(caplog: pytest.LogCaptureFixture) -> None: + """The validated c490 schema is accepted without a warning.""" + device = C2000G2(MOCK_BLE_DEVICE, capability=4, client_token="t") # noqa: S106 + client = mock.AsyncMock() + device._client = client + with caplog.at_level(logging.WARNING): + await device._process_notification( + client, + 0, + _c490_frame(device, "charging_pps_series_c_0005"), + ) + assert device.summary_schema == "charging_pps_series_c_0005" + assert not any("schema" in r.message for r in caplog.records) From 3b91fbbbec58490edb7cad96a9edc369fb676801 Mon Sep 17 00:00:00 2001 From: Shawn Stricker Date: Tue, 8 Sep 2026 16:05:13 -0400 Subject: [PATCH 09/11] docs(gen2): fold in the c490-derived C2000 G2 support rows Add the four fields the C2000 G2 exposes only through the cloud-armed c490 device-summary -- Max charge power, Pack voltage, Cumulative energy out and Charge presence -- to the power-station table, plus a marks legend and a note explaining they are carried in the raw summary map rather than decoded into named properties. Co-Authored-By: Claude Opus 4.8 --- docs/source/index.rst | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/docs/source/index.rst b/docs/source/index.rst index fb98304..bade421 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -29,6 +29,11 @@ No pairing is required in order to receive telemetry data or control the device. This project is under active development. +The support tables below use these marks: ✅ supported · 🚧 known but not yet +implemented · ❌ not supported · N/A not applicable · ❔ not investigated. A +``read/control`` pair such as ✅/🚧 gives the two states separately (e.g. the max +charge limit is readable but not yet settable). + Power station support --------------------- @@ -54,6 +59,10 @@ DC Power out ✅ ✅ ❌ ✅ ✅ ✅ DC Power in status ✅ ✅ ❌ ❌ ✅ ✅ ❌ ✅ ❌ DC Power out status ✅ ❌ ❌ ✅ ✅ ✅ ❌ ✅ ✅ DC Timer ✅ ✅ ❌ ❌ ❌ ✅ ❌ ✅ ❌ +Max charge power ❔ ❔ ❔ ❔ ✅/🚧 ✅/🚧 ❔ ❔ ❔ +Pack voltage ❔ ❔ ❔ ❔ 🚧 🚧 ❔ ❔ ❔ +Cumulative energy out ❔ ❔ ❔ ❔ 🚧 🚧 ❔ ❔ ❔ +Charge presence ❔ ❔ ❔ ❔ 🚧 🚧 ❔ ❔ ❔ USB Power out ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ USB Port status ✅ ✅ ❌ ❌ ✅ ✅ ❌ ✅ ✅ Light control ✅ ✅ ✅ ✅ ❌ ❌ ❌ ✅ ❌ @@ -74,6 +83,12 @@ Expansion num N/A N/A N/A ✅ N/A ✅ Polled status updates ✅ ❌ ✅ ✅ ❌ ✅ ❌ ✅ ❌ ======================= ======= ========== ======= ======== ======== ======== =========== ===== ===== +The C2000 G2 (A1783) ``Pack voltage``, ``Cumulative energy out`` and ``Charge +presence`` rows come from the ~9-minute, cloud-armed ``c490`` device-summary +rather than the live per-second stream, so the library keeps them in the raw +summary map without yet decoding them into named properties. Every other C2000 G2 +row is decoded from the live stream. + Solar system support -------------------- From e82328fde9b404dd5d6badcb0ff75d66008f7297 Mon Sep 17 00:00:00 2001 From: Shawn Stricker Date: Wed, 9 Sep 2026 08:31:26 -0400 Subject: [PATCH 10/11] Address review feedback on the C2000 G2 / c490 line Move the C2000 G2 class docstring's implementation detail (the shared Gen 2 stack, the added 4103/a3/f9/c0 blocks, the c490 summary) into a collapsible note, keeping the user-facing description at the top. Rename the cached protobuf-summary attributes _summary / _summary_schema to _data_summary / _data_summary_schema, consistent with the existing _data. The public summary / summary_schema properties are unchanged. Co-Authored-By: Claude Opus 4.8 --- SolixBLE/device.py | 22 +++++++++++----------- SolixBLE/devices/c2000g2.py | 33 ++++++++++++++++++--------------- 2 files changed, 29 insertions(+), 26 deletions(-) diff --git a/SolixBLE/device.py b/SolixBLE/device.py index 3a147e6..4530b61 100644 --- a/SolixBLE/device.py +++ b/SolixBLE/device.py @@ -161,8 +161,8 @@ def __init__( self._authorized: bool = False self._client_token: str = client_token or str(uuid.uuid4()) self._auth_mode: bytes | None = None - self._summary: dict[str, object] = {} - self._summary_schema: str | None = None + self._data_summary: dict[str, object] = {} + self._data_summary_schema: str | None = None @property def _encrypted_negotiation(self) -> bool: @@ -470,7 +470,7 @@ def summary(self) -> dict[str, object]: :returns: Mapping of ``.path`` to value. """ - return self._summary + return self._data_summary @property def summary_schema(self) -> str | None: @@ -482,7 +482,7 @@ def summary_schema(self) -> str | None: :returns: The schema name, or None. """ - return self._summary_schema + return self._data_summary_schema def _parse_int( self, key: str, begin: int = None, end: int = None, signed: bool = False @@ -575,7 +575,7 @@ def _check_summary_schema(self) -> None: so it is logged. """ validated = self._VALIDATED_SUMMARY_SCHEMA - schema = self._summary_schema + schema = self._data_summary_schema if validated is None or schema is None or schema == validated: return got = _schema_version(schema) @@ -821,16 +821,16 @@ async def _process_notification( # Protobuf device-summary frames (e.g the c490) are not # the flat TLV the other telemetry frames use. if cmd.hex() in self._PROTOBUF_TELEMETRY_COMMANDS: - self._summary = walk_protobuf( + self._data_summary = walk_protobuf( self._protobuf_body(decrypted_payload) ) - self._summary_schema = self._protobuf_schema( + self._data_summary_schema = self._protobuf_schema( decrypted_payload ) self._check_summary_schema() _LOGGER.debug( - f"Protobuf summary ({len(self._summary)} fields, " - f"schema {self._summary_schema})" + f"Protobuf summary ({len(self._data_summary)} fields, " + f"schema {self._data_summary_schema})" ) return None @@ -1571,8 +1571,8 @@ def _reset_session(self, reset_data: bool = True) -> None: self._shared_secret = None self._authorized = False self._auth_mode = None - self._summary = {} - self._summary_schema = None + self._data_summary = {} + self._data_summary_schema = None self._last_packet_timestamp = None self._negotiation_timestamp = None self._packet_futures: dict[bytes, list[asyncio.Future]] = {} diff --git a/SolixBLE/devices/c2000g2.py b/SolixBLE/devices/c2000g2.py index 8eb9399..bb41bbe 100644 --- a/SolixBLE/devices/c2000g2.py +++ b/SolixBLE/devices/c2000g2.py @@ -120,21 +120,24 @@ class C2000G2(C1000G2): Use this class to connect, monitor and control a Gen 2 C2000(X) power station. This model is also known as the A1783. - The C2000 G2 is the larger sibling of the C1000 G2 (A1763) and shares its - Gen 2 BLE stack: the same ``c421``/``c900`` telemetry framing and TLV field - map, the same ``4100`` poll command, and the same AC (``4101``) and DC - (``4102``) control. Its three USB-C ports, single USB-A port, AC, DC and - solar all decode identically, so the port and power properties come - unchanged from :class:`~SolixBLE.devices.c1000g2.C1000G2`. - - On top of that it adds the parts of the Gen 2 frame that had not been - decoded when the C1000 G2 class was written -- the ``4103`` system group - (display switch, brightness, timeout, and the SoC limits), the ``a3``/``a6`` - status and time-remaining fields, the ``f9`` version block including its - per-submodule slots, and the ``c0`` expansion-battery block for the BP2000. - - It also receives the ``c490`` protobuf device-summary post (armed by - :meth:`enable_realtime_telemetry`), decoded into the :attr:`summary` map. + .. note:: + :collapsible: closed + + The C2000 G2 is the larger sibling of the C1000 G2 (A1763) and shares its + Gen 2 BLE stack: the same ``c421``/``c900`` telemetry framing and TLV field + map, the same ``4100`` poll command, and the same AC (``4101``) and DC + (``4102``) control. Its three USB-C ports, single USB-A port, AC, DC and + solar all decode identically, so the port and power properties come + unchanged from :class:`~SolixBLE.devices.c1000g2.C1000G2`. + + On top of that it adds the parts of the Gen 2 frame that had not been + decoded when the C1000 G2 class was written -- the ``4103`` system group + (display switch, brightness, timeout, and the SoC limits), the ``a3``/``a6`` + status and time-remaining fields, the ``f9`` version block including its + per-submodule slots, and the ``c0`` expansion-battery block for the BP2000. + + It also receives the ``c490`` protobuf device-summary post (armed by + :meth:`enable_realtime_telemetry`), decoded into the :attr:`summary` map. """ #: The Gen 2 telemetry set plus the ``c490`` protobuf device-summary post. From dde09222b189501d5f5d2718815efdad4f30dbf4 Mon Sep 17 00:00:00 2001 From: Shawn Stricker Date: Wed, 9 Sep 2026 16:54:31 -0400 Subject: [PATCH 11/11] Add C2000 G2 input-power reads and AC charging-power control Read the hardware max input power (a3), the a6 mirror of the main battery SoC, and the AC input port status, and add set_ac_charging_power() to drive the a4 charge limit under 4101. Co-Authored-By: Claude Opus 4.8 --- SolixBLE/devices/c2000g2.py | 55 ++++++++++++++++++++++++++++++++++++- docs/source/index.rst | 8 +++--- tests/devices/c2000g2.py | 14 ++++++++++ tests/test_devices.py | 42 ++++++++++++++-------------- 4 files changed, 93 insertions(+), 26 deletions(-) diff --git a/SolixBLE/devices/c2000g2.py b/SolixBLE/devices/c2000g2.py index bb41bbe..30ebd10 100644 --- a/SolixBLE/devices/c2000g2.py +++ b/SolixBLE/devices/c2000g2.py @@ -10,7 +10,7 @@ DEFAULT_METADATA_INT, DEFAULT_METADATA_STRING, ) -from ..states import ChargingStatus +from ..states import ChargingStatus, PortStatus from .c1000g2 import ( C1000G2, CMD_AC_OUTPUT, @@ -317,6 +317,23 @@ async def set_dc_output_timeout(self, seconds: int) -> None: parameters=_parameters("a3", seconds, type_=3), ) + async def set_ac_charging_power(self, watts: int) -> None: + """Set the AC charging-power limit (W). + + Read it back with :attr:`ac_input_limit`. + + :param watts: AC charging power limit, 500-1800 W (the range the app + offers; sent as ``a4`` under the ``4101`` AC command). + :raises ValueError: If the value is out of range. + """ + if not 500 <= watts <= 1800: + raise ValueError("AC charging power must be between 500 and 1800 W") + + await self._send_command( + cmd=CMD_AC_OUTPUT, + parameters=_parameters("a4", watts, type_=2), + ) + ############## # Status # ############## @@ -432,6 +449,42 @@ def ac_input_limit(self) -> int: """ return self._parse_int("a4", begin=5, end=7) + @property + def max_input_power(self) -> int: + """Device maximum charge-input power (W) -- the hardware ceiling. + + Distinct from :attr:`ac_input_limit`, which is the user-configured + limit; this is the fixed maximum the unit can draw. + + :returns: Maximum input power in watts, or default int value if there + is no data. + """ + return self._parse_int("a3", begin=5, end=7) + + @property + def battery_percentage_a6_9(self) -> int: + """Battery percentage reported at ``a6`` offset 9. + + Mirrors :attr:`battery_percentage` (the main-pack SoC) -- the two carry + the same value. Neither is the aggregate SoC when an expansion battery + is attached; both report the main pack alone. + + :returns: Main-pack SoC percent, or default int value if there is no + data. + """ + return self._parse_int("a6", begin=9, end=10) + + @property + def ac_input_port(self) -> PortStatus: + """AC input (mains) status. + + PortStatus.INPUT signifies the mains lead is present, NOT_CONNECTED that + it is absent. Presence only -- it does not imply current is flowing. + + :returns: Status of the AC input. + """ + return PortStatus.from_input_only(self._parse_int("a7", begin=4, end=5)) + @property def ac_output_timeout(self) -> int: """AC output auto-off countdown (s), 0 when no timer is set. diff --git a/docs/source/index.rst b/docs/source/index.rst index bade421..9aeda7a 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -59,10 +59,10 @@ DC Power out ✅ ✅ ❌ ✅ ✅ ✅ DC Power in status ✅ ✅ ❌ ❌ ✅ ✅ ❌ ✅ ❌ DC Power out status ✅ ❌ ❌ ✅ ✅ ✅ ❌ ✅ ✅ DC Timer ✅ ✅ ❌ ❌ ❌ ✅ ❌ ✅ ❌ -Max charge power ❔ ❔ ❔ ❔ ✅/🚧 ✅/🚧 ❔ ❔ ❔ -Pack voltage ❔ ❔ ❔ ❔ 🚧 🚧 ❔ ❔ ❔ -Cumulative energy out ❔ ❔ ❔ ❔ 🚧 🚧 ❔ ❔ ❔ -Charge presence ❔ ❔ ❔ ❔ 🚧 🚧 ❔ ❔ ❔ +Max charge power ❔ ❔ ❔ ❔ ❔ ✅ ❔ ❔ ❔ +Pack voltage ❔ ❔ ❔ ❔ ❌ 🚧 ❔ ❔ ❔ +Cumulative energy out ❔ ❔ ❔ ❔ ❌ 🚧 ❔ ❔ ❔ +Charge presence ❔ ❔ ❔ ❔ ❌ 🚧 ❔ ❔ ❔ USB Power out ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅ USB Port status ✅ ✅ ❌ ❌ ✅ ✅ ❌ ✅ ✅ Light control ✅ ✅ ✅ ✅ ❌ ❌ ❌ ✅ ❌ diff --git a/tests/devices/c2000g2.py b/tests/devices/c2000g2.py index 031cb2a..3d805d4 100644 --- a/tests/devices/c2000g2.py +++ b/tests/devices/c2000g2.py @@ -183,6 +183,20 @@ ValueError, id="c2000g2_dc_timeout_out_of_range", ), + pytest.param( + C2000G2, + "set_ac_charging_power", + [1200], + [("4101", "a10121a40302b004")], + id="c2000g2_ac_charging_power", + ), + pytest.param( + C2000G2, + "set_ac_charging_power", + [400], + ValueError, + id="c2000g2_ac_charging_power_out_of_range", + ), # Inherited from the C1000 G2 unchanged -- the AC and DC output switches use # the same opcodes and payload on both models. pytest.param( diff --git a/tests/test_devices.py b/tests/test_devices.py index e4a9645..a57df1b 100644 --- a/tests/test_devices.py +++ b/tests/test_devices.py @@ -362,6 +362,9 @@ "hours_remaining": 1.0, "ac_frequency": 60, "ac_input_limit": 1000, + "max_input_power": 1800, + "battery_percentage_a6_9": 95, + "ac_input_port": PortStatus.INPUT, "ac_output_timeout": 0, "dc_output_timeout": 0, "ac_output_mode": 0, @@ -406,6 +409,9 @@ "hours_remaining": 2.7, "ac_frequency": 60, "ac_input_limit": 1000, + "max_input_power": 1800, + "battery_percentage_a6_9": 80, + "ac_input_port": PortStatus.NOT_CONNECTED, "ac_output_timeout": 0, "dc_output_timeout": 0, "ac_output_mode": 0, @@ -1119,9 +1125,9 @@ async def test_values( await device._process_telemetry(parameters) for class_property, expected_value in mapping.items(): - assert ( - getattr(device, class_property) == expected_value - ), f"Mismatch for property '{class_property}'!" + assert getattr(device, class_property) == expected_value, ( + f"Mismatch for property '{class_property}'!" + ) @pytest.mark.asyncio @@ -1267,7 +1273,6 @@ async def test_negotiation( # noqa: PLR0913 :param secret: The expected shared secret. """ async with MockDevice() as mock_bluetooth: - device = device_class(MOCK_BLE_DEVICE) for packet in packets: @@ -1280,9 +1285,9 @@ async def test_negotiation( # noqa: PLR0913 assert await device.connect(), "Expected connect to return True" # Assert that the correct shared secret is calculated - assert ( - bytes.fromhex(secret) == device._shared_secret - ), "Shared secret does not match expected" + assert bytes.fromhex(secret) == device._shared_secret, ( + "Shared secret does not match expected" + ) mock_bluetooth.check_assertions() @@ -1529,7 +1534,7 @@ def test_payload_decryption( ), ], ) -async def test_telemetry_packet_processing( # noqa: PLR0913, PLR0917 +async def test_telemetry_packet_processing( # noqa: PLR0913 fake_time, # noqa: ANN001, ARG001 fast_sleep, # noqa: ANN001, ARG001 fast_timeouts, # noqa: ANN001, ARG001 @@ -1559,7 +1564,6 @@ async def test_telemetry_packet_processing( # noqa: PLR0913, PLR0917 ) async with MockDevice() as mock_bluetooth: - # We first expect a negotiation for expected, response in negotiation_responses.items(): mock_bluetooth.expect_ordered( @@ -1579,10 +1583,7 @@ async def test_telemetry_packet_processing( # noqa: PLR0913, PLR0917 for packet in packets: await mock_bluetooth.send_data([bytes.fromhex(packet)]) - device_parameters = ( - device._data.to_str(verbose=False) - if device._data else None - ) + device_parameters = device._data.to_str(verbose=False) if device._data else None assert parameters == device_parameters, "Parameters do not match expected!" @@ -1607,7 +1608,7 @@ async def test_telemetry_packet_processing( # noqa: PLR0913, PLR0917 ), ], ) -async def test_generic_packet_processing( # noqa: PLR0913, PLR0917 +async def test_generic_packet_processing( # noqa: PLR0913 caplog, # noqa: ANN001 fake_time, # noqa: ANN001, ARG001 fast_sleep, # noqa: ANN001, ARG001 @@ -1639,7 +1640,6 @@ async def test_generic_packet_processing( # noqa: PLR0913, PLR0917 async with MockDevice() as mock_bluetooth: with caplog.at_level(logging.DEBUG): - # We first expect a negotiation for expected, response in negotiation_responses.items(): mock_bluetooth.expect_ordered( @@ -1660,9 +1660,9 @@ async def test_generic_packet_processing( # noqa: PLR0913, PLR0917 await mock_bluetooth.send_data([bytes.fromhex(packet)]) for expected_log_entry in expected_logs: - assert ( - expected_log_entry in str(caplog.text) - ), f"Expected to find '{expected_log_entry}' in logs but it was not found!" + assert expected_log_entry in str(caplog.text), ( + f"Expected to find '{expected_log_entry}' in logs but it was not found!" + ) @pytest.mark.asyncio @@ -1753,6 +1753,6 @@ async def test_bad_values( await device._process_telemetry(parameters) for class_property, expected_value in mapping.items(): - assert ( - getattr(device, class_property) == expected_value - ), f"Mismatch for property '{class_property}'!" + assert getattr(device, class_property) == expected_value, ( + f"Mismatch for property '{class_property}'!" + )