diff --git a/SolixBLE/__init__.py b/SolixBLE/__init__.py index e19bddd..dfa93d6 100644 --- a/SolixBLE/__init__.py +++ b/SolixBLE/__init__.py @@ -11,6 +11,7 @@ C800, C1000, C1000G2, + C2000G2, F2000, F2600, F3800, @@ -35,29 +36,30 @@ from .utilities import discover_devices __all__ = [ - "SolixBLEDevice", - "PrimeDevice", "C300", "C300DC", "C800", "C1000", "C1000G2", + "C2000G2", "F2000", "F2600", "F3800", - "Solarbank2", - "Solarbank3", - "PrimeCharger160w", - "PrimeCharger250w", - "PrimePowerBank20k", - "MagGo3in1", - "Generic", "ChargingStatus", "ChargingStatusF3800", "DisplayTimeout", + "Generic", "LightStatus", + "MagGo3in1", + "PortOverload", "PortStatus", + "PrimeCharger160w", + "PrimeCharger250w", + "PrimeDevice", + "PrimePowerBank20k", + "Solarbank2", + "Solarbank3", + "SolixBLEDevice", "TemperatureUnit", - "PortOverload", "discover_devices", ] diff --git a/SolixBLE/device.py b/SolixBLE/device.py index 2a52206..0bdf80b 100644 --- a/SolixBLE/device.py +++ b/SolixBLE/device.py @@ -45,6 +45,7 @@ UUID_COMMAND, UUID_TELEMETRY, ) +from .parsing import walk_protobuf _LOGGER = logging.getLogger(__name__) @@ -57,12 +58,25 @@ 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 protobuf, not the 1-byte-tag TLV + #: format :meth:`_parse_payload` understands (e.g. the C2000 G2's ``c490`` + #: device-summary post). They are still reassembled and decrypted -- so a + #: consumer that understands the frame can decode the cleartext -- but the base + #: skips TLV-parsing them, which would misread the protobuf varint tags. + _PROTOBUF_TELEMETRY_COMMANDS: tuple[str, ...] = () + + #: Fixed ff09-frame overhead between the on-wire notification value and the + #: ``payload`` that :meth:`_split_packet` returns: ``ff09`` (2) + length (2) + + #: pattern (3) + cmd (2) + checksum (1). Added back so the fragmentation gate can + #: compare the notification length to the live ``ATT_MTU - 3`` cap. + _FRAME_OVERHEAD: int = 10 + def __init__(self, ble_device: BLEDevice) -> None: """Initialise device object. Does not connect automatically.""" _LOGGER.debug( f"Initializing Solix device '{ble_device.name}' with" - f"address '{ble_device.address}' and details '{ble_device.details}'" + f"address '{ble_device.address}' and details '{ble_device.details}'", ) self._ble_device: BLEDevice = ble_device @@ -70,6 +84,7 @@ def __init__(self, ble_device: BLEDevice) -> None: self._fragment_buffers: dict[bytes, dict[int, bytes]] = {} self._fragment_totals: dict[bytes, int] = {} self._data: dict[str, bytes] | None = None + self._summary: dict[str, object] = {} self._last_data_timestamp: datetime | None = None self._last_packet_timestamp: datetime | None = None self._negotiation_timestamp: float | None = None @@ -119,7 +134,6 @@ async def connect(self, max_attempts: int = 3, run_callbacks: bool = True) -> bo self._connection_attempts = self._connection_attempts + 1 try: - # If we have an old client get rid of it if self._client is not None: await self._dispose_of_client() @@ -139,23 +153,24 @@ async def connect(self, max_attempts: int = 3, run_callbacks: bool = True) -> bo except BleakError: _LOGGER.exception( - f"Error establishing initial connection to '{self.name}'!" + f"Error establishing initial connection to '{self.name}'!", ) # If we are still not connected then we have failed if not self.connected: _LOGGER.error( - f"Failed to establish initial connection to '{self.name}' on attempt {self._connection_attempts}!" + f"Failed to establish initial connection to '{self.name}' on attempt {self._connection_attempts}!", ) return False _LOGGER.debug( - f"Established initial connection to '{self.name}' on attempt {self._connection_attempts}!" + f"Established initial connection to '{self.name}' on attempt {self._connection_attempts}!", ) try: _LOGGER.debug(f"Subscribing to notifications from device '{self.name}'!") await self._client.start_notify( - UUID_TELEMETRY, partial(self._process_notification, self._client) + UUID_TELEMETRY, + partial(self._process_notification, self._client), ) except BleakError: _LOGGER.exception(f"Error subscribing/negotiating with '{self.name}'!") @@ -164,10 +179,8 @@ async def connect(self, max_attempts: int = 3, run_callbacks: bool = True) -> bo # Negotiate try: async with asyncio.timeout(NEGOTIATION_TIMEOUT): - # While negotiations have not completed while not self.negotiated: - # If we have not received any packet from the device in # any stage then restart negotiations from the start if ( @@ -175,16 +188,15 @@ async def connect(self, max_attempts: int = 3, run_callbacks: bool = True) -> bo or (time.time() - self._last_packet_timestamp) > NEGOTIATION_RESPONSE_TIMEOUT ): - _LOGGER.debug( - f"Sending negotiation initiation request to '{self.name}'..." + f"Sending negotiation initiation request to '{self.name}'...", ) await self._initiate_negotiations() # Wait at this long to see if we get any response to # our initial request in stage 0. This weird layout # allows us to exit immediately when negotiation occurs - for _ in range(0, NEGOTIATION_RESPONSE_TIMEOUT): + for _ in range(NEGOTIATION_RESPONSE_TIMEOUT): await asyncio.sleep(1) if self.negotiated: break @@ -233,7 +245,6 @@ async def _post_connect(self) -> None: for example, send a subscribe command to start a telemetry stream (see :class:`~SolixBLE.devices.c1000g2.C1000G2`). """ - pass async def _keep_alive(self) -> int | None: """Execute designated keep-alive command periodically after good negotiation. @@ -334,8 +345,24 @@ 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 + self, + key: str, + begin: int = None, + end: int = None, + signed: bool = False, ) -> int: """Parse an integer at the specified key in the telemetry data. @@ -379,23 +406,24 @@ def _split_packet(self, packet: bytes) -> tuple[bytes, bytes, bytes]: # Validate encoded length is correct packet_length = int.from_bytes( - bytes([packet_copy.pop(0), packet_copy.pop(0)]), byteorder="little" + bytes([packet_copy.pop(0), packet_copy.pop(0)]), + byteorder="little", ) if packet_length != len(packet): raise ValueError( - f"Packet length is encoded as {packet_length} but its length was {len(packet)}!" + f"Packet length is encoded as {packet_length} but its length was {len(packet)}!", ) # Validate checksum is correct packet_checksum = packet_copy.pop(-1).to_bytes() if packet_checksum != self._checksum(packet[:-1]): raise ValueError( - f"Packet checksum is encoded as {packet_checksum.hex()} but it is actually {self._checksum(packet[:-1]).hex()}!" + f"Packet checksum is encoded as {packet_checksum.hex()} but it is actually {self._checksum(packet[:-1]).hex()}!", ) # Extract pattern packet_pattern = bytes( - [packet_copy.pop(0), packet_copy.pop(0), packet_copy.pop(0)] + [packet_copy.pop(0), packet_copy.pop(0), packet_copy.pop(0)], ) # Extract command @@ -469,17 +497,19 @@ def _verbose_pop(data: bytearray, length: int, name: str) -> bytes: # Sometimes there is just a param_id with no length or values if len(remaining_data) == 0: - parsed_data[param_id] = bytes() + parsed_data[param_id] = b"" break # Extract encoded length of parameter param_len = int.from_bytes( - _verbose_pop(remaining_data, 1, f"param_len (id={param_id})") + _verbose_pop(remaining_data, 1, f"param_len (id={param_id})"), ) # Extract data/body from parameter param_data = _verbose_pop( - remaining_data, param_len, f"param_data (id={param_id})" + remaining_data, + param_len, + f"param_data (id={param_id})", ) parsed_data[param_id] = param_data @@ -487,13 +517,15 @@ def _verbose_pop(data: bytearray, length: int, name: str) -> bytes: _LOGGER.exception( f"Unexpected end of packet! Data may be missing or invalid!" f" Extracted so far: '{self._parameters_to_str(parsed_data)}'." - f" Payload: '{payload.hex()}'" + f" Payload: '{payload.hex()}'", ) return parsed_data def _parameters_to_str( - self, parameters: dict[str, bytes], types: bool = False + self, + parameters: dict[str, bytes], + types: bool = False, ) -> str: if types: with_types = { @@ -506,8 +538,7 @@ def _parameters_to_str( for k, v in parameters.items() } return json.dumps(with_types, indent=4, sort_keys=True) - else: - return str({k: v.hex() for k, v in parameters.items()}) + return str({k: v.hex() for k, v in parameters.items()}) def _log_diff(self, old: dict[str, bytes], new: dict[str, bytes]) -> None: """Log any differences between parameters.""" @@ -522,13 +553,15 @@ def _log_diff(self, old: dict[str, bytes], new: dict[str, bytes]) -> None: if new[k] != old[k] } _LOGGER.debug( - f"Parameter changes: \n{json.dumps(differences, indent=4, sort_keys=True)}" + f"Parameter changes: \n{json.dumps(differences, indent=4, sort_keys=True)}", ) def _decrypt_payload(self, payload: bytes) -> bytes: """Decrypt telemetry packet using negotiated shared secret and IV.""" cipher = AES.new( - self._shared_secret[:16], AES.MODE_CBC, iv=self._shared_secret[16:] + self._shared_secret[:16], + AES.MODE_CBC, + iv=self._shared_secret[16:], ) decrypted = cipher.decrypt(payload) unpadder = PKCS7(128).unpadder() @@ -543,60 +576,150 @@ def _encrypt_payload(self, payload: bytes) -> bytes: padded_data = padder.update(payload) padded_data += padder.finalize() cipher = AES.new( - self._shared_secret[:16], AES.MODE_CBC, iv=self._shared_secret[16:] + self._shared_secret[:16], + AES.MODE_CBC, + iv=self._shared_secret[16:], ) return cipher.encrypt(padded_data) - async def _process_telemetry_packet( - self, payload: bytes, cmd: bytes = None - ) -> None: - """Process a telemetry packet from the device. - - This performs the default processing of telemetry packets in which - telemetry payloads are spread across multiple packets. This is - overridden for devices which do not use multi-packet payloads for - telemetry. + def _reassemble(self, cmd: bytes, payload: bytes) -> bytes | None: + """Reassemble a possibly-fragmented session frame into one payload. + + Fragmentation is a transport effect that sits *below* the cipher: a frame + larger than a single notification (``ATT_MTU - 3`` bytes on the wire) is + split, and every fragment is prefixed with a ```` byte (high + nibble = 1-based index, low nibble = fragment count). A frame that fits in + one notification arrives whole. Because this operates on the still-encrypted + payload, the same reassembler serves telemetry and unknown session frames + alike, regardless of the AES variant used to decrypt the result (SolixBLE + #42). + + Single frames are told apart from fragments by length, not by trusting the + first byte: only a full-length notification (or the continuation of an open + run) is a fragment. A short frame is a standalone single, and it carries a + ```` byte *only* if that byte is a valid single marker + (``0x11``); some families (e.g. the A91B2 station) put no frag byte on + singles, so their first byte is already ciphertext and must be kept. + + :returns: the complete (still-encrypted) payload ready to decrypt, or + ``None`` while fragments are still outstanding. """ + if not payload: + return payload - # First byte encodes fragment info (high nibble = index, low = total) - fragment_index = (payload[0] >> 4) & 0x0F - fragment_total = payload[0] & 0x0F + cmd_key = bytes(cmd) + index = (payload[0] >> 4) & 0x0F + total = payload[0] & 0x0F - # Multi-part message - if fragment_total > 1: - fragment_data = payload[1:] - cmd_key = bytes(cmd) + # Continuation (or short tail) of a run already in progress for this cmd. + if cmd_key in self._fragment_buffers: _LOGGER.debug( - f"Fragment {fragment_index}/{fragment_total} for cmd {cmd.hex()}, {len(fragment_data)} bytes" + f"Fragment {index}/{self._fragment_totals[cmd_key]} for cmd " + f"{cmd.hex()}, {len(payload) - 1} bytes", ) + self._fragment_buffers[cmd_key][index] = payload[1:] + return self._join_fragments(cmd_key) + + # A run only starts on a full-length first fragment. The length guard stops a + # short frame whose first ciphertext byte happens to look like a ``0x1x`` + # header from opening a run that never completes. + notification_length = len(payload) + self._FRAME_OVERHEAD + if ( + index == 1 + and total > 1 + and notification_length >= self._client.mtu_size - 3 + ): + _LOGGER.debug( + f"Fragment 1/{total} for cmd {cmd.hex()}, {len(payload) - 1} bytes", + ) + self._fragment_buffers[cmd_key] = {1: payload[1:]} + self._fragment_totals[cmd_key] = total + return self._join_fragments(cmd_key) + + # Standalone single notification. Strip the frag byte only when it is a valid + # single marker; otherwise the whole payload is data. + if payload[0] == 0x11: + return payload[1:] + return payload + + def _join_fragments(self, cmd_key: bytes) -> bytes | None: + """Join a completed fragment run, or ``None`` if more fragments are due.""" + if len(self._fragment_buffers[cmd_key]) < self._fragment_totals[cmd_key]: + _LOGGER.debug("Waiting for remaining fragments...") + return None - # Store fragment - if cmd_key not in self._fragment_buffers or fragment_index == 1: - self._fragment_buffers[cmd_key] = {} - self._fragment_totals[cmd_key] = fragment_total - - self._fragment_buffers[cmd_key][fragment_index] = fragment_data + payload = b"".join( + self._fragment_buffers[cmd_key][i] + for i in sorted(self._fragment_buffers[cmd_key]) + ) + del self._fragment_buffers[cmd_key] + del self._fragment_totals[cmd_key] + _LOGGER.debug(f"Reassembled payload: {len(payload)} bytes") + return payload + + @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. Unlike the flat telemetry + frames, ``a2`` is a ``bin`` field with a **2-byte** little-endian length and + an ``04`` type byte (the payload is ~340 bytes, past the 1-byte length + :meth:`_parse_payload` handles), so the header is ``a1 a2 + 04`` -- 7 bytes for the usual ``a1 01 31 a2 04``. Walking the + whole frame instead would misread those header bytes as protobuf tags and + yield almost nothing; running past ``a2`` into the ``a3`` trailer would append + a spurious field. 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, starting 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] - # Wait until all fragments have arrived - if len(self._fragment_buffers[cmd_key]) < fragment_total: - _LOGGER.debug("Waiting for remaining fragments...") - return + async def _process_telemetry_packet( + self, + payload: bytes, + cmd: bytes = None, + ) -> None: + """Decrypt and dispatch a (reassembled) telemetry payload. - # Reassemble in order - payload = b"".join( - self._fragment_buffers[cmd_key][i] - for i in sorted(self._fragment_buffers[cmd_key]) + The payload has already been made whole by :meth:`_reassemble`, so this just + decrypts and parses it. Kept as an override point for devices whose telemetry + needs post-processing (e.g. the A91B2 station remaps two frame layouts onto + one property set). + """ + try: + decrypted_payload = self._decrypt_payload(payload) + except Exception: # noqa: BLE001 + # A fragment that arrived with no matching run (out of order, or its + # first fragment was lost) is returned whole by :meth:`_reassemble` and + # will not decrypt; drop it rather than crash the notification handler. + _LOGGER.debug( + f"Discarding undecryptable telemetry frame for cmd " + f"{cmd.hex() if cmd else '?'}", ) - del self._fragment_buffers[cmd_key] - del self._fragment_totals[cmd_key] - _LOGGER.debug(f"Reassembled payload: {len(payload)} bytes") - - else: - # Strip fragment info - payload = payload[1:] - - decrypted_payload = self._decrypt_payload(payload) + return None _LOGGER.debug(f"Decrypted payload: {decrypted_payload.hex()}") + # Protobuf telemetry (e.g. the C2000 G2's c490 device summary) is not the + # 1-byte-tag TLV format _parse_payload understands. Walk the protobuf blob + # into a `.path` field map (see :mod:`SolixBLE.parsing`), exposed via + # :attr:`summary`, rather than TLV-parsing it -- which would misread the + # varint tags. + if cmd is not None and 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 = self._parse_payload(decrypted_payload) return await self._process_telemetry(parameters) @@ -607,12 +730,11 @@ async def _process_telemetry(self, parameters: dict[str, bytes]) -> None: if _LOGGER.isEnabledFor(logging.DEBUG): _LOGGER.debug( - f"Telemetry parameters: {self._parameters_to_str(parameters)}" + f"Telemetry parameters: {self._parameters_to_str(parameters)}", ) # Print state update if changes if state_changed: - # If we have previous data to compare against log the diff if self._data is not None: _LOGGER.debug("Parameters have changed since previous update!") @@ -621,7 +743,7 @@ async def _process_telemetry(self, parameters: dict[str, bytes]) -> None: # Else log the parameters but with the types else: _LOGGER.debug( - f"Telemetry parameters: {self._parameters_to_str(parameters, types=True)}" + f"Telemetry parameters: {self._parameters_to_str(parameters, types=True)}", ) # Update internal parameters @@ -630,12 +752,14 @@ async def _process_telemetry(self, parameters: dict[str, bytes]) -> None: # Run callbacks if state changed if state_changed: - _LOGGER.debug(self) self._run_state_changed_callbacks() async def _process_notification( - self, client: BleakClient, handle: int, data: bytearray + self, + client: BleakClient, + handle: int, + data: bytearray, ) -> None: """Process a notification from the device.""" @@ -643,11 +767,11 @@ async def _process_notification( if self._client is not client: _LOGGER.debug("Ignoring notification from old client") - return + return None # Split packet into pattern, command, and payload _LOGGER.debug( - f"Received notification from '{self.name}'. length: {len(data)}, packet: '{data.hex()}'" + f"Received notification from '{self.name}'. length: {len(data)}, packet: '{data.hex()}'", ) self._last_packet_timestamp = time.time() pattern, cmd, payload = self._split_packet(data) @@ -660,15 +784,14 @@ async def _process_notification( # future instead of processing it here if pattern + cmd in self._packet_futures: _LOGGER.debug( - "Packet has future(s) registered. Triggering future(s) and ignoring packet..." + "Packet has future(s) registered. Triggering future(s) and ignoring packet...", ) for future in self._packet_futures[pattern + cmd]: future.set_result(payload) - return + return None # Match against common message types match pattern.hex(): - # Negotiation messages case "030001": _LOGGER.debug("Received negotiation message!") @@ -676,59 +799,60 @@ async def _process_notification( # Session messages case "03010f" | "030111": - # Non-encrypted telemetry messages if cmd.hex() == "0300": _LOGGER.debug("Received non-encrypted telemetry message!") parameters = self._parse_payload(payload) return await self._process_telemetry(parameters) + # Reassemble multi-fragment frames before the cipher, so telemetry + # and unknown session frames share one reassembler (SolixBLE #42). + payload = self._reassemble(cmd, payload) + if payload is None: + return None + # Encrypted telemetry messages - elif cmd.hex() in self._TELEMETRY_COMMANDS: + if cmd.hex() in self._TELEMETRY_COMMANDS: _LOGGER.debug("Received encrypted telemetry message!") return await self._process_telemetry_packet(payload, cmd) # Unknown messages - else: - _LOGGER.debug(f"Received unknown message of type: {cmd.hex()}") - try: - - # If the payload is one byte too short and we are - # using the default AES (CBC) then try putting the - # last byte of the cmd in front of it - if ( - len(payload) % 16 == 15 - and self._decrypt_payload - is SolixBLEDevice._decrypt_payload - ): - _LOGGER.debug( - "Using special trick of embedded part of CMD in payload..." - ) - payload = cmd[1].to_bytes() + payload - - decrypted_payload = self._decrypt_payload(payload) - _LOGGER.debug( - f"Decrypted payload: {decrypted_payload.hex()}" - ) - parameters = self._parse_payload(decrypted_payload) + _LOGGER.debug(f"Received unknown message of type: {cmd.hex()}") + try: + # If the payload is one byte too short and we are + # using the default AES (CBC) then try putting the + # last byte of the cmd in front of it + if ( + len(payload) % 16 == 15 + and self._decrypt_payload is SolixBLEDevice._decrypt_payload + ): _LOGGER.debug( - f"Parameters: {self._parameters_to_str(parameters, types=True)}" - ) - except Exception: - _LOGGER.exception( - "Exception decrypting unknown message type" + "Using special trick of embedded part of CMD in payload...", ) + payload = cmd[1].to_bytes() + payload + + decrypted_payload = self._decrypt_payload(payload) + _LOGGER.debug( + f"Decrypted payload: {decrypted_payload.hex()}", + ) + parameters = self._parse_payload(decrypted_payload) + _LOGGER.debug( + f"Parameters: {self._parameters_to_str(parameters, types=True)}", + ) + except Exception: + _LOGGER.exception( + "Exception decrypting unknown message type", + ) case _: _LOGGER.warning( - f"Unexpected packet type '{pattern}' sent by device! Packet: {data.hex()}" + f"Unexpected packet type '{pattern}' sent by device! Packet: {data.hex()}", ) async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: """Negotiate encryption with the device.""" match cmd.hex(): - # There is a "stage 0" in which we automatically send a negotiation # request as soon as we establish the initial connection. That # should lead to the power station sending a response landing us @@ -737,56 +861,60 @@ async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: # Negotiation stage 1 case "0801": _LOGGER.debug( - "Entered negotiation stage 1 due to response from device!" + "Entered negotiation stage 1 due to response from device!", ) parameters = self._parse_payload(payload) _LOGGER.debug(f"Parameters: {self._parameters_to_str(parameters)}") _LOGGER.debug("Sending stage 1 response message...") return await self._client.write_gatt_char( - UUID_COMMAND, bytes.fromhex(NEGOTIATION_COMMAND_1) + UUID_COMMAND, + bytes.fromhex(NEGOTIATION_COMMAND_1), ) # Negotiation stage 2 case "0803": _LOGGER.debug( - "Entered negotiation stage 2 due to response from device!" + "Entered negotiation stage 2 due to response from device!", ) parameters = self._parse_payload(payload) _LOGGER.debug(f"Parameters: {self._parameters_to_str(parameters)}") _LOGGER.debug("Sending stage 2 response message...") return await self._client.write_gatt_char( - UUID_COMMAND, bytes.fromhex(NEGOTIATION_COMMAND_2) + UUID_COMMAND, + bytes.fromhex(NEGOTIATION_COMMAND_2), ) # Negotiation stage 3 case "0829": _LOGGER.debug( - "Entered negotiation stage 3 due to response from device!" + "Entered negotiation stage 3 due to response from device!", ) parameters = self._parse_payload(payload) _LOGGER.debug(f"Parameters: {self._parameters_to_str(parameters)}") self._negotiation_timestamp = time.time() _LOGGER.debug("Sending stage 3 response message...") return await self._client.write_gatt_char( - UUID_COMMAND, bytes.fromhex(NEGOTIATION_COMMAND_3) + UUID_COMMAND, + bytes.fromhex(NEGOTIATION_COMMAND_3), ) # Negotiation stage 4 case "0805": _LOGGER.debug( - "Entered negotiation stage 4 due to response from device!" + "Entered negotiation stage 4 due to response from device!", ) parameters = self._parse_payload(payload) _LOGGER.debug(f"Parameters: {self._parameters_to_str(parameters)}") _LOGGER.debug("Sending stage 4 response message...") return await self._client.write_gatt_char( - UUID_COMMAND, bytes.fromhex(NEGOTIATION_COMMAND_4) + UUID_COMMAND, + bytes.fromhex(NEGOTIATION_COMMAND_4), ) # Negotiation stage 5 case "0821": _LOGGER.debug( - "Entered negotiation stage 5 due to response from device!" + "Entered negotiation stage 5 due to response from device!", ) parameters = self._parse_payload(payload) _LOGGER.debug(f"Parameters: {self._parameters_to_str(parameters)}") @@ -795,14 +923,16 @@ async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: device_public_key_bytes = bytes.fromhex("04") + parameters["a1"] _LOGGER.debug(f"Public key of device: {device_public_key_bytes.hex()}") device_public_key = EllipticCurvePublicKey.from_encoded_point( - SECP256R1(), device_public_key_bytes + SECP256R1(), + device_public_key_bytes, ) # Calculate the shared secret # The first half of the shared secret is the encryption key # and the second half is the IV private_value = int.from_bytes( - bytes.fromhex(PRIVATE_KEY), byteorder="big" + bytes.fromhex(PRIVATE_KEY), + byteorder="big", ) private_key = derive_private_key(private_value, SECP256R1()) self._shared_secret = private_key.exchange(ECDH(), device_public_key) @@ -810,7 +940,8 @@ async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: _LOGGER.debug("Sending stage 5 response message...") return await self._client.write_gatt_char( - UUID_COMMAND, bytes.fromhex(NEGOTIATION_COMMAND_5) + UUID_COMMAND, + bytes.fromhex(NEGOTIATION_COMMAND_5), ) # Negotiation stage 6 (Optional) @@ -819,7 +950,7 @@ async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: # but it does not hurt to decrypt it anyway. case "4822": _LOGGER.debug( - "Entered negotiation stage 6 (optional) due to response from device!" + "Entered negotiation stage 6 (optional) due to response from device!", ) decrypted_payload = self._decrypt_payload(payload) parameters = self._parse_payload(decrypted_payload) @@ -827,7 +958,7 @@ async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: case _: _LOGGER.warning( - f"Received unexpected negotiation request response from device! cmd: '{cmd}', parameters: '{self._parameters_to_str(parameters)}'" + f"Received unexpected negotiation request response from device! cmd: '{cmd}', parameters: '{self._parameters_to_str(parameters)}'", ) def _checksum(self, packet: bytes) -> bytes: @@ -851,10 +982,12 @@ async def _send_command(self, cmd: bytes, payload: bytes) -> None: # and that timestamp is set during negotiations time_passed = int(time.time() - self._negotiation_timestamp) base_timestamp = int.from_bytes( - bytes.fromhex(BASE_TIMESTAMP), byteorder="little" + bytes.fromhex(BASE_TIMESTAMP), + byteorder="little", ) new_timestamp = (base_timestamp + time_passed).to_bytes( - length=4, byteorder="little" + length=4, + byteorder="little", ) new_payload = payload + bytes.fromhex("fe0503") + new_timestamp await self._send_encrypted_packet(cmd, new_payload) @@ -882,7 +1015,7 @@ def _build_packet(self, pattern: bytes, cmd: bytes, payload: bytes) -> bytes: async def _send_encrypted_packet(self, cmd: bytes, payload: bytes) -> None: """Send an encrypted packet using negotiated shared secret and IV.""" _LOGGER.debug( - f"Building packet with cmd: {cmd.hex()} and payload: {payload.hex()}" + f"Building packet with cmd: {cmd.hex()} and payload: {payload.hex()}", ) encrypted_payload = self._encrypt_payload(payload) @@ -893,7 +1026,10 @@ async def _send_encrypted_packet(self, cmd: bytes, payload: bytes) -> None: await self._client.write_gatt_char(UUID_COMMAND, packet) def _register_future( - self, future: asyncio.Future, pattern: bytes, cmd: bytes + self, + future: asyncio.Future, + pattern: bytes, + cmd: bytes, ) -> None: """Register a future to be triggered when the pattern and cmd bytes are received.""" @@ -907,7 +1043,10 @@ def _register_future( self._packet_futures[pattern + cmd].append(future) def _deregister_future( - self, future: asyncio.Future, pattern: bytes, cmd: bytes + self, + future: asyncio.Future, + pattern: bytes, + cmd: bytes, ) -> None: """Deregister a future to be triggered when the pattern and cmd bytes are received.""" @@ -927,7 +1066,10 @@ def _deregister_future( self._packet_futures.pop(pattern + cmd) async def _listen_for_packet( - self, pattern: bytes, cmd: bytes, timeout: int = 10 + self, + pattern: bytes, + cmd: bytes, + timeout: int = 10, ) -> bytes | None: """Wait for a response and return its payload bytes. @@ -960,7 +1102,7 @@ def _run_state_changed_callbacks(self) -> None: function() except Exception: _LOGGER.exception( - f"Exception raised by a registered state change callback '{function}'!" + f"Exception raised by a registered state change callback '{function}'!", ) async def _auto_reconnect(self) -> None: @@ -980,26 +1122,24 @@ def _can_retry() -> bool: ) try: - # If callbacks need to be run on reconnection, we silently # reconnect if the timeout has not been exceeded, else we # run callbacks to let subscribers know we were disconnected run_callbacks_on_reconnect = False while _can_retry(): - # If we are already connected and negotiated then wait for disconnection if self.negotiated: _LOGGER.debug( - f"Automatic reconnect task ready and waiting for disconnect event from '{self.name}'!" + f"Automatic reconnect task ready and waiting for disconnect event from '{self.name}'!", ) await self._disconnect_event.wait() _LOGGER.debug( - f"Disconnection event signalled by '{self.name}', starting reconnection..." + f"Disconnection event signalled by '{self.name}', starting reconnection...", ) else: _LOGGER.debug( - f"We are still not connected to '{self.name}', starting reconnection..." + f"We are still not connected to '{self.name}', starting reconnection...", ) # If we have reached this stage we are not connected @@ -1009,18 +1149,16 @@ def _can_retry() -> bool: # we have to trigger callbacks to let subscribers know we # are disconnected async with asyncio.timeout(DISCONNECT_TIMEOUT): - while _can_retry(): - await asyncio.sleep(RECONNECT_DELAY) try: attempt_number = self._connection_attempts if await self.connect( - run_callbacks=run_callbacks_on_reconnect + run_callbacks=run_callbacks_on_reconnect, ): _LOGGER.debug( - f"""Successfully reconnected to '{self.name}' {"silently" if not run_callbacks_on_reconnect else ""} on attempt {attempt_number}!""" + f"""Successfully reconnected to '{self.name}' {"silently" if not run_callbacks_on_reconnect else ""} on attempt {attempt_number}!""", ) # Reset back to false on successful connection @@ -1030,13 +1168,13 @@ def _can_retry() -> bool: break except Exception: _LOGGER.exception( - f"""Exception raised attempting to {"silently" if not run_callbacks_on_reconnect else ""} reconnect to '{self.name}'!""" + f"""Exception raised attempting to {"silently" if not run_callbacks_on_reconnect else ""} reconnect to '{self.name}'!""", ) # If timeout exceeded - except asyncio.TimeoutError: + except TimeoutError: _LOGGER.warning( - f"Timed out attempting to silently reconnect to '{self.name}', callbacks will be triggered due to disconnect!" + f"Timed out attempting to silently reconnect to '{self.name}', callbacks will be triggered due to disconnect!", ) self._reset_session(reset_data=True) self._run_state_changed_callbacks() @@ -1045,8 +1183,7 @@ def _can_retry() -> bool: # need to run them again on reconnect run_callbacks_on_reconnect = True - else: - _LOGGER.warning("Maximum reconnect limit exceeded!") + _LOGGER.warning("Maximum reconnect limit exceeded!") except asyncio.CancelledError: _LOGGER.debug("Automatic reconnect task has been canceled/stopped") @@ -1103,7 +1240,7 @@ def _disconnect_callback(self, client: BaseBleakClient) -> None: # Ignore disconnect callbacks from old clients if client is not self._client: _LOGGER.debug( - f"Disconnect of '{self.name}' came from other client. Ignoring..." + f"Disconnect of '{self.name}' came from other client. Ignoring...", ) return @@ -1123,7 +1260,7 @@ async def _dispose_of_client(self) -> None: await client.disconnect() except Exception: _LOGGER.exception( - f"Exception raised when disposing of bleak client '{client}'!" + f"Exception raised when disposing of bleak client '{client}'!", ) def _reset_session(self, reset_data: bool = True) -> None: @@ -1131,6 +1268,7 @@ def _reset_session(self, reset_data: bool = True) -> None: if reset_data: self._data = None + self._summary = {} self._last_data_timestamp = None self._fragment_buffers = {} @@ -1159,7 +1297,7 @@ def _safe_get(name: str, prop: property) -> str: return prop.fget(self) except Exception as e: _LOGGER.exception( - f"Failed to parse property '{name}' when stringifying class! Is there an undocumented state?" + f"Failed to parse property '{name}' when stringifying class! Is there an undocumented state?", ) return f"{type(e).__name__}: {e}" diff --git a/SolixBLE/devices/__init__.py b/SolixBLE/devices/__init__.py index b4be66f..47cbb91 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,14 +27,15 @@ "C800", "C1000", "C1000G2", + "C2000G2", "F2000", "F2600", "F3800", - "Solarbank2", - "Solarbank3", + "Generic", + "MagGo3in1", "PrimeCharger160w", "PrimeCharger250w", "PrimePowerBank20k", - "MagGo3in1", - "Generic", + "Solarbank2", + "Solarbank3", ] diff --git a/SolixBLE/devices/c1000g2.py b/SolixBLE/devices/c1000g2.py index 424700d..66ad704 100644 --- a/SolixBLE/devices/c1000g2.py +++ b/SolixBLE/devices/c1000g2.py @@ -4,6 +4,7 @@ """ +from ..const import DEFAULT_METADATA_FLOAT, DEFAULT_METADATA_INT from ..device import SolixBLEDevice from ..states import PortStatus @@ -37,7 +38,14 @@ class C1000G2(SolixBLEDevice): _EXPECTED_TELEMETRY_LENGTH: int = 253 #: The Gen 2 pushes telemetry on different command codes to the gen-1 models. - _TELEMETRY_COMMANDS: tuple[str, ...] = ("c421", "c900") + #: ``c490`` is the device-summary the C2000 G2 (A1783) posts unsolicited every + #: ~9 min: routed here (not the dropped unknown-frame branch) so it's reassembled + #: and decrypted, but its payload is protobuf, so it's flagged below to skip the + #: TLV parse -- a protobuf-aware consumer decodes the delivered cleartext. + _TELEMETRY_COMMANDS: tuple[str, ...] = ("c421", "c900", "c490") + + #: ``c490`` carries a protobuf device summary, not the TLV format the base parses. + _PROTOBUF_TELEMETRY_COMMANDS: tuple[str, ...] = ("c490",) async def _post_connect(self) -> None: """Subscribe to telemetry once connected. @@ -57,7 +65,8 @@ async def turn_ac_on(self) -> None: :raises BleakError: If command transmission fails. """ await self._send_command( - cmd=bytes.fromhex(CMD_AC_OUTPUT), payload=bytes.fromhex(PAYLOAD_ON) + cmd=bytes.fromhex(CMD_AC_OUTPUT), + payload=bytes.fromhex(PAYLOAD_ON), ) async def turn_ac_off(self) -> None: @@ -67,7 +76,8 @@ async def turn_ac_off(self) -> None: :raises BleakError: If command transmission fails. """ await self._send_command( - cmd=bytes.fromhex(CMD_AC_OUTPUT), payload=bytes.fromhex(PAYLOAD_OFF) + cmd=bytes.fromhex(CMD_AC_OUTPUT), + payload=bytes.fromhex(PAYLOAD_OFF), ) async def turn_dc_on(self) -> None: @@ -81,7 +91,8 @@ async def turn_dc_on(self) -> None: :raises BleakError: If command transmission fails. """ await self._send_command( - cmd=bytes.fromhex(CMD_DC_OUTPUT), payload=bytes.fromhex(PAYLOAD_ON) + cmd=bytes.fromhex(CMD_DC_OUTPUT), + payload=bytes.fromhex(PAYLOAD_ON), ) async def turn_dc_off(self) -> None: @@ -91,7 +102,8 @@ async def turn_dc_off(self) -> None: :raises BleakError: If command transmission fails. """ await self._send_command( - cmd=bytes.fromhex(CMD_DC_OUTPUT), payload=bytes.fromhex(PAYLOAD_OFF) + cmd=bytes.fromhex(CMD_DC_OUTPUT), + payload=bytes.fromhex(PAYLOAD_OFF), ) @property @@ -299,3 +311,131 @@ def min_battery_percentage(self) -> int: :returns: Battery charge percentage lower limit or default int value. """ return self._parse_int("d9", begin=5, end=6) + + @property + def max_input_power(self) -> int: + """Maximum charge-input limit in watts (tag ``a3``). + + :returns: Maximum input limit in watts or default int value. + """ + return self._parse_int("a3", begin=5, end=7) + + @property + def dc_input_power(self) -> int: + """DC/solar input power in watts (tag ``a6``). + + :returns: DC input power in watts or default int value. + """ + return self._parse_int("a6", begin=5, end=7) + + @property + def remaining_time_hours(self) -> float: + """Remaining time in hours (tag ``a6``, deci-hours ``x0.1``). + + Direction-aware: time until full when charging, time until empty when + discharging. + + :returns: Remaining time in hours or default float value. + """ + raw = self._parse_int("a6", begin=7, end=9) + return raw / 10 if raw != DEFAULT_METADATA_INT else DEFAULT_METADATA_FLOAT + + @property + def main_battery_soc(self) -> int: + """Main-battery state of charge in percent, excluding expansion (tag ``a6``). + + :returns: Main battery SOC percent or default int value. + """ + return self._parse_int("a6", begin=9, end=10) + + # -- c490 device-summary fields (walker output; see :attr:`summary`) ---------- + # ``c490`` is a ~9-minute rollup, armed by a cloud session, so these read their + # defaults until a frame arrives. Fields that also exist in the per-second + # ``0421`` stream are exposed namespaced ``c490_*`` so a stale summary value + # never masks the live one. + + def _summary_int(self, path: str) -> int: + """Read an int leaf from the latest c490 summary, or the default.""" + value = self._summary.get(path) + return value if isinstance(value, int) else DEFAULT_METADATA_INT + + @property + def battery_voltage(self) -> float: + """Battery pack voltage in volts, from the c490 summary (``.15.3``, ``x0.1``). + + :returns: Pack voltage in volts or default float value. + """ + raw = self._summary_int(".15.3") + return raw / 10 if raw != DEFAULT_METADATA_INT else DEFAULT_METADATA_FLOAT + + @property + def flow_state(self) -> int: + """Power-flow state from the c490 summary (``.23.7``). + + ``0`` = idle/balanced, ``1`` = net discharge, ``2`` = charge. + + :returns: Flow state or default int value. + """ + return self._summary_int(".23.7") + + @property + def charge_presence(self) -> int: + """Charge-presence indicator from the c490 summary (``.23.6``); units uncalibrated. + + :returns: Charge presence value or default int value. + """ + return self._summary_int(".23.6") + + @property + def cumulative_discharge_energy(self) -> int: + """Cumulative discharge energy in watt-hours, from the c490 summary (``.19.8``). + + :returns: Cumulative discharge energy in Wh or default int value. + """ + return self._summary_int(".19.8") + + @property + def c490_battery_soc(self) -> int: + """Battery SOC percent from the ~9-min c490 summary (``.23.1``). + + See :attr:`battery_percentage` for the live per-second value. + + :returns: SOC percent or default int value. + """ + return self._summary_int(".23.1") + + @property + def c490_output_power_total(self) -> int: + """Total output power in watts from the c490 summary (``.23.2``). + + :returns: Total output power in watts or default int value. + """ + return self._summary_int(".23.2") + + @property + def c490_ac_output_power(self) -> int: + """AC output power in watts from the c490 summary (``.23.3``). + + DC/USB output = :attr:`c490_output_power_total` minus this. + + :returns: AC output power in watts or default int value. + """ + return self._summary_int(".23.3") + + @property + def c490_input_power_total(self) -> int: + """Total input power in watts from the c490 summary (``.23.4``). + + :returns: Total input power in watts or default int value. + """ + return self._summary_int(".23.4") + + @property + def c490_dc_input_power(self) -> int: + """DC/solar input power in watts from the c490 summary (``.23.5``). + + AC input = :attr:`c490_input_power_total` minus this. + + :returns: DC input power in watts or default int value. + """ + return self._summary_int(".23.5") diff --git a/SolixBLE/devices/c2000g2.py b/SolixBLE/devices/c2000g2.py new file mode 100644 index 0000000..88e1542 --- /dev/null +++ b/SolixBLE/devices/c2000g2.py @@ -0,0 +1,53 @@ +"""C2000(X) Gen 2 power station model. + +.. moduleauthor:: kb1ibt + +""" + +from ..parsing import walk_lv +from .c1000g2 import C1000G2 + + +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`` subscribe 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 -- confirmed against a live A1783 frame (serial, + part number ``A1783``, temperature, battery percentage and min/max SOC all + read correctly through the inherited + :class:`~SolixBLE.devices.c1000g2.C1000G2` properties) -- so it is driven + almost entirely by that inherited behaviour. + + The C2000 additionally reports a parallel/expansion battery (a BP2000) in tag + ``ce`` of the ``c421`` telemetry. Unlike the flat fields, ``ce`` is a nested + length-value block (the app's "并机"/``DeviceCombinationInfo``) whose first + field is a 16-byte combination-device ID -- all-zero when no unit is combined. + :attr:`expansion_present` decodes that ID via :func:`SolixBLE.parsing.walk_lv`; + the remaining sub-fields (the combined pack's identity/mode/power) are left + undecoded pending a capture from a device that actually has one combined. The + ``c490`` protobuf device summary (exposed via :attr:`summary`) carries the + cumulative energy ledgers. + """ + + @property + def expansion_present(self) -> bool: + """Whether a parallel/expansion battery (e.g. a BP2000) is combined. + + The ``ce`` tag's first length-value field is a 16-byte combination-device + ID: all-zero means no unit is combined (the app's + ``DeviceCombinationInfo.status == single``), non-zero means one is. + + :returns: True if a parallel/expansion unit is combined, else False. + """ + if not self._data or "ce" not in self._data: + return False + # ce value is a `bin` field (0x04 type byte) wrapping a length-value block. + fields = walk_lv(self._data["ce"][1:]) + return bool(fields and any(fields[0])) 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 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 e1ac02f..1d8bd5d 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -33,46 +33,63 @@ 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 ✅ ❌ ✅ ✅ ❌ ❌ ✅ ❌ -======================= ======== ========== ========= ========= ========= ============ ====== ====== +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 power is readable but not yet settable). + +======================= ======= ========== ======= ======== ======== ======== =========== ===== ===== +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 ✅ ✅ ❌ ❌ ✅ ✅ ❌ ✅ ✅ +Max charge power ❔ ❔ ❔ ❔ ✅/🚧 ✅/🚧 ❔ ❔ ❔ +Pack voltage ❔ ❔ ❔ ❔ ✅ ✅ ❔ ❔ ❔ +Cumulative energy out ❔ ❔ ❔ ❔ ✅ ✅ ❔ ❔ ❔ +Charge presence ❔ ❔ ❔ ❔ ✅ ✅ ❔ ❔ ❔ +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 ✅ ❌ ✅ ✅ 🚧 🚧 ❌ ✅ ❌ +======================= ======= ========== ======= ======== ======== ======== =========== ===== ===== + +The C2000 G2 (A1783) shares the C1000 G2 (A1763) telemetry stack, so the two +columns track together. The ``Max charge power``, ``Pack voltage``, +``Cumulative energy out`` and ``Charge presence`` rows are read from the +``c490`` protobuf device summary rather than the live per-second stream, and +are exposed through :attr:`summary` alongside dedicated properties. Only the +A1783 was verified against hardware; the A1763 column reflects the shared code +path. Solar system support diff --git a/tests/helpers.py b/tests/helpers.py index 0bd60b3..1f0270f 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -8,7 +8,7 @@ import logging from collections.abc import Callable from dataclasses import dataclass, field -from typing import Any, Union +from typing import Any from unittest import mock from bleak import BleakClient @@ -29,7 +29,7 @@ class RequestResponse: Name of request to produce more useful error messages. """ - expected: Union[bytes, None] + expected: bytes | None """ The bytes expected by this request. Use none to accept any bytes. """ @@ -105,19 +105,22 @@ def custom_init(*args, **kwargs): # We give it a name so we can tell the difference between them in logs mock_bleak_client = mock.AsyncMock( - name=f"bleak_client_{len(self._mock_bleak_clients)}" + name=f"bleak_client_{len(self._mock_bleak_clients)}", ) # Set functions/properties mock_bleak_client.write_gatt_char.side_effect = self.write_gatt_char mock_bleak_client.start_notify.side_effect = self.start_notify + # Emulate an Anker 256-byte ATT MTU so fragment reassembly (which gates on + # the live ``mtu_size - 3`` notification cap) behaves as it does on device. + mock_bleak_client.mtu_size = 256 type(mock_bleak_client).is_connected = mock.PropertyMock( - side_effect=lambda: self._is_connected + side_effect=lambda: self._is_connected, ) # Add it to the list of all bleak clients self._mock_bleak_clients.append( - (mock_bleak_client, [kwargs["disconnected_callback"]], []) + (mock_bleak_client, [kwargs["disconnected_callback"]], []), ) # Set this as the most current bleak client and return it @@ -161,7 +164,9 @@ def disconnect(self): callback(bleak_client) def expect_ordered( - self, value: Union[bytes, None] = None, response: list[bytes] = [] + self, + value: bytes | None = None, + response: list[bytes] = [], ): """ Expect an ordered request to be made to the mock device with @@ -175,8 +180,10 @@ def expect_ordered( """ self._assertions.append( RequestResponse( - name=f"num {len(self._assertions)}", expected=value, response=response - ) + name=f"num {len(self._assertions)}", + expected=value, + response=response, + ), ) def expect_ordered_all(self, requests: list[RequestResponse]): @@ -215,7 +222,7 @@ async def send_data(self, data: list[bytes]) -> None: for callback in n_callbacks: for packet in data: _LOGGER.debug( - f"Mock device sending '{packet.hex()}' to client '{client}' for callback '{callback}'..." + f"Mock device sending '{packet.hex()}' to client '{client}' for callback '{callback}'...", ) # Handle is not used await callback(None, packet) @@ -224,7 +231,10 @@ async def send_data(self, data: list[bytes]) -> None: await asyncio.sleep(0.1) async def write_gatt_char( - self, char_specifier: str, data: bytes, response: bool = False + self, + char_specifier: str, + data: bytes, + response: bool = False, ): """ Patched version of the bleak clients write_gatt_char function @@ -244,15 +254,15 @@ async def write_gatt_char( request_response = self._assertions[self._position] except IndexError: print(self._assertions) - assert ( - False - ), f"Received an unexpected request '{data.hex()}'. Number: {self._position+1}, Num expected: {len(self._assertions)}" + assert False, ( + f"Received an unexpected request '{data.hex()}'. Number: {self._position + 1}, Num expected: {len(self._assertions)}" + ) if request_response.expected is not None: # Assert it matches - assert ( - request_response.expected == data - ), f"Expected bytes {request_response.expected.hex()}' for request '{request_response.name}' ({self._position+1}) but got '{data.hex()}'!" + assert request_response.expected == data, ( + f"Expected bytes {request_response.expected.hex()}' for request '{request_response.name}' ({self._position + 1}) but got '{data.hex()}'!" + ) # Increment position self._position = self._position + 1 @@ -273,9 +283,9 @@ def check_assertions(self): Check that all specified requests have been made by the module. """ for i, item in enumerate(self._assertions): - assert ( - item.called - ), f"Request '{item.name}' ({i}) with expected bytes '{item.expected.hex()}' was not called!" + assert item.called, ( + f"Request '{item.name}' ({i}) with expected bytes '{item.expected.hex()}' was not called!" + ) async def __aexit__(self, *exc): """ diff --git a/tests/test_parsing.py b/tests/test_parsing.py new file mode 100644 index 0000000..d725748 --- /dev/null +++ b/tests/test_parsing.py @@ -0,0 +1,102 @@ +"""Tests for the nested-payload walkers (protobuf + length-value).""" + +import pytest + +from SolixBLE.parsing import read_varint, walk_lv, walk_protobuf + + +@pytest.mark.parametrize( + "hexstr,pos,expected", + [ + # 0x96 0x01 -> 150 (the canonical protobuf varint example) + ("9601", 0, (150, 2)), + ("00", 0, (0, 1)), + ("7f", 0, (127, 1)), # largest single-byte varint + ("8001", 0, (128, 2)), # smallest two-byte varint + ("ff01", 0, (255, 2)), + ("ffff03", 0, (65535, 3)), + # read starting partway through a buffer + ("aa9601", 1, (150, 3)), + ], +) +def test_read_varint(hexstr: str, pos: int, expected: tuple[int, int]) -> None: + assert read_varint(bytes.fromhex(hexstr), pos) == expected + + +@pytest.mark.parametrize( + "hexstr,expected,note", + [ + # field 1 = varint 150 (08 96 01); field 2 = "ABC" (12 03 414243) + ("0896011203414243", {".1": 150, ".2": "ABC"}, "varint + printable string"), + # field 1 = submessage{ field 1 = varint 150 }: the container records its byte + # length AND its leaves are recursed -- dropping the container under-counts. + ("0a03089601", {".1": 3, ".1.1": 150}, "submessage records container + leaf"), + # repeated field 1 keeps wire order (second occurrence suffixed #1) + ("08010802", {".1": 1, ".1#1": 2}, "repeated tag keeps order"), + # field 1, wire type 5 (32-bit fixed), little-endian 0x12345678 + ("0d78563412", {".1": 0x12345678}, "wire 5 (32-bit fixed)"), + # field 1, wire type 1 (64-bit fixed), little-endian 01..08 + ( + "090102030405060708", + {".1": int.from_bytes(bytes(range(1, 9)), "little")}, + "wire 1 (64-bit fixed)", + ), + # field 2, non-printable length-delimited bytes -> kept as hex + ("120200ff", {".2": "00ff"}, "non-printable bytes leaf -> hex"), + # field 2, zero-length length-delimited -> empty (falls through to hex "") + ("1200", {".2": ""}, "empty length-delimited leaf"), + # field 2, all-zero sub is NOT recursed (any(sub) guard) -> kept as hex + ("12020000", {".2": "0000"}, "all-zero sub kept as hex, not recursed"), + # field 1 varint, then a wire-type-3 group marker for field 1 (0x0b): the group + # is recorded as None and stops the walk -- the trailing 0802 is not parsed. + ("08010b0802", {".1": 1, ".1#1": None}, "wire 3 group -> record None and stop"), + ], +) +def test_walk_protobuf(hexstr: str, expected: dict, note: str) -> None: + assert walk_protobuf(bytes.fromhex(hexstr)) == expected, note + + +def test_walk_protobuf_empty_buffer() -> None: + assert walk_protobuf(b"") == {} + + +@pytest.mark.parametrize( + "value_hex,first_field,first_nonzero,note", + [ + # Real C2000 G2 `ce` value with no BP2000: a 16-byte zero combination-device ID, + # then a 1-byte status (0x11), then zero padding. Passed without the 04 type byte. + ( + "10000000000000000000000000000000000111" + "000000000000000000000000000000000000000000000000", + b"\x00" * 16, + False, + "empty combination id (no expansion unit)", + ), + # A non-zero 16-byte ID -> first field carries it verbatim. + ( + (bytes([0x10]) + bytes(range(1, 17)) + bytes.fromhex("0111")).hex(), + bytes(range(1, 17)), + True, + "populated combination id (expansion unit present)", + ), + ], +) +def test_walk_lv_first_field( + value_hex: str, + first_field: bytes, + first_nonzero: bool, + note: str, +) -> None: + fields = walk_lv(bytes.fromhex(value_hex)) + assert fields[0] == first_field, note + assert any(fields[0]) is first_nonzero + + +def test_walk_lv_reads_all_fields_and_trailing_padding() -> None: + # pairs to the end; trailing zero padding appears as zero-length fields. + fields = walk_lv(bytes.fromhex("02aabb01cc0000")) + assert fields == [b"\xaa\xbb", b"\xcc", b"", b""] + + +def test_walk_lv_empty_buffer() -> None: + assert walk_lv(b"") == [] diff --git a/tests/test_reassembly.py b/tests/test_reassembly.py new file mode 100644 index 0000000..b37b14b --- /dev/null +++ b/tests/test_reassembly.py @@ -0,0 +1,70 @@ +"""Tests for multi-fragment session-frame reassembly (SolixBLE #42). + +``_reassemble`` runs below the cipher and serves telemetry and unknown session +frames alike, so these tests exercise it directly on raw (still-encrypted) +payloads for a link with a 256-byte ATT MTU (``ATT_MTU - 3 == 253`` on the wire). +""" + +from unittest import mock + +from SolixBLE import SolixBLEDevice +from tests.const import MOCK_BLE_DEVICE + +#: ``ATT_MTU - 3`` for a 256-byte link -- a full (fragmenting) notification value. +CAP = 253 +CMD = b"\xc4\x90" + + +def _device() -> SolixBLEDevice: + dev = SolixBLEDevice(MOCK_BLE_DEVICE) + dev._client = mock.Mock(mtu_size=256) + return dev + + +def test_short_single_without_frag_byte_kept_whole() -> None: + # A91B2-style single: first byte 0xd8 is ciphertext (index 13 / total 8 is an + # impossible header), so the whole payload is data and must be kept. + device = _device() + payload = bytes([0xD8]) + b"\x11" * 200 + assert device._reassemble(CMD, payload) == payload + assert device._fragment_buffers == {} + + +def test_short_single_with_frag_byte_stripped() -> None: + # MagGo/Prime-style single: 0x11 is a valid single marker (index 1 / total 1), + # so the frag byte is stripped before it reaches the (GCM/CBC) decrypt. + device = _device() + payload = bytes([0x11]) + b"\xcd" * 60 + assert device._reassemble(CMD, payload) == b"\xcd" * 60 + assert device._fragment_buffers == {} + + +def test_two_fragment_reassembly_with_short_tail() -> None: + device = _device() + frag1 = bytes([0x12]) + b"\xaa" * (CAP - 1) # full-length first fragment + frag2 = bytes([0x22]) + b"\xbb" * 40 # short tail closes the run + assert device._reassemble(CMD, frag1) is None # still buffering + assert device._reassemble(CMD, frag2) == b"\xaa" * (CAP - 1) + b"\xbb" * 40 + assert device._fragment_buffers == {} + + +def test_exact_multiple_no_short_tail() -> None: + # Two full 253-byte notifications, no short tail (the 506-on-the-wire case). + # Termination must come from the count, not from a shorter + # packet closing the run -- otherwise this hangs forever. + device = _device() + frag1 = bytes([0x12]) + b"\xaa" * (CAP - 1) + frag2 = bytes([0x22]) + b"\xbb" * (CAP - 1) + assert device._reassemble(CMD, frag1) is None + assert device._reassemble(CMD, frag2) == b"\xaa" * (CAP - 1) + b"\xbb" * (CAP - 1) + assert device._fragment_buffers == {} + + +def test_cold_non_first_fragment_decoded_whole() -> None: + # A full-length frame whose header reads index != 1 with no open run cannot be a + # joinable fragment (a run always opens on index 1). Treat it as a single and + # keep the whole payload rather than opening a run that never completes. + device = _device() + payload = bytes([0x35]) + b"\x00" * (CAP - 1) # index 3 / total 5, no buffer + assert device._reassemble(CMD, payload) == payload + assert device._fragment_buffers == {} diff --git a/tests/test_summary.py b/tests/test_summary.py new file mode 100644 index 0000000..a784962 --- /dev/null +++ b/tests/test_summary.py @@ -0,0 +1,134 @@ +"""Tests for the c490 protobuf device-summary path (C1000 G2 / C2000 G2). + +The C2000 G2 (A1783) posts a protobuf device summary on command ``c490`` every +~9 min. It is delivered whole by the reassembler, decrypted, then walked into a +``.path`` field map exposed via :attr:`~SolixBLE.device.SolixBLEDevice.summary` +(see :mod:`SolixBLE.parsing`). The protobuf is wrapped in an outer ``a1``/``a2`` +TLV, so it must be un-wrapped before walking. The frames below are real captures +(decrypted cleartext from the collector's journald log), re-encrypted with the +test secret to exercise the whole decrypt -> unwrap -> walk -> property path. +""" + +import pytest + +from SolixBLE import C1000G2, C2000G2 +from SolixBLE.const import DEFAULT_METADATA_FLOAT, DEFAULT_METADATA_INT +from SolixBLE.device import SolixBLEDevice +from tests.const import MOCK_BLE_DEVICE + +#: AES key/IV used to re-encrypt the captured cleartext (same secret the other +#: telemetry tests use); the decrypt round-trips it back to the cleartext below. +SECRET = "5609bc39f79166da75139feb7c335fb7524b3bf0d730db96bf6ebf450d3e165b" + +C490_CMD = b"\xc4\x90" + +#: Real A1763 c490 frame, idle regime (flow-state 0, 95 % SoC), decrypted cleartext. +FRAME_IDLE = "a10131a25001040a0541313736331200320538b4094007722e0a200000040075411d61894900005201000000013c00000000000100000080000000100018cb1e20cf1628003000722c0a200000000000000000000000000000000000000000000000000000000000000000100018002000280030007a1b08820f1002189504208f04280330093800400048005000580060007a18080010001800200028003000380040004800500058006000920110080010001800200028003000380040009a011a08cefd0110900118d5ea0320aad00228c63a302438df0a409c42a2011608ee01100018e40220c101281330003800400048ee01aa01130800100018002000280030003800400048ee01b001da9702ba0119085f08001000180020002800302a3800409a1c48bb0950b202c20127080410011800200028003000380040004800500058006000680070007800800100880100900100a31c046368617267696e675f7070735f7365726965735f635f3030303500" # noqa: E501 + +#: Real A1763 c490 frame, discharge regime (flow-state 1, 44 W out, 52.9 V). +FRAME_DISCHARGE = "a10131a25001040a0541313736331200320538b4094007722e0a200000000075001d61894800005201000000003c00000000000100000080000000100018cb1e20d01628003000722c0a200000000000000000000000000000000000000000000000000000000000000000100018002000280030007a1b08820f1002189104209104280830093800400048005000580060007a18080010001800200028003000380040004800500058006000920110080010001800200028003000380040009a011a08cefd0110900118d5ea0320aad00228d73a302438df0a40a442a2011608ee01100018e40220c101281330003800400048ee01aa01130800100018002000280030003800400048ee01b001eb9702ba0119085f0800102c180020002800302a380140910348bb0950b202c20127080410011800200028003000380040004800500058006000680070007800800100880100900100a31c046368617267696e675f7070735f7365726965735f635f3030303500" # noqa: E501 + + +async def _feed_c490(device: SolixBLEDevice, frame_hex: str) -> None: + """Re-encrypt a captured cleartext frame and run it through the telemetry path.""" + device._shared_secret = bytes.fromhex(SECRET) + encrypted = device._encrypt_payload(bytes.fromhex(frame_hex)) + await device._process_telemetry_packet(encrypted, cmd=C490_CMD) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "device_class,frame_hex,mapping", + [ + pytest.param( + C1000G2, + FRAME_IDLE, + { + "c490_battery_soc": 95, + "c490_output_power_total": 0, + "c490_ac_output_power": 0, + "c490_input_power_total": 0, + "c490_dc_input_power": 0, + "charge_presence": 42, + "flow_state": 0, + "battery_voltage": 53.3, + "cumulative_discharge_energy": 8476, + }, + id="c1000g2_c490_idle", + ), + pytest.param( + C2000G2, + FRAME_DISCHARGE, + { + "c490_battery_soc": 95, + "c490_output_power_total": 44, + "flow_state": 1, + "battery_voltage": 52.9, + "cumulative_discharge_energy": 8484, + }, + id="c2000g2_c490_discharge", + ), + ], +) +async def test_c490_summary_properties( + device_class: type[SolixBLEDevice], + frame_hex: str, + mapping: dict, +) -> None: + """A real c490 frame decrypts, unwraps and walks into the summary properties.""" + device = device_class(MOCK_BLE_DEVICE) + await _feed_c490(device, frame_hex) + + for prop, expected in mapping.items(): + assert getattr(device, prop) == expected, f"Mismatch for '{prop}'" + + +@pytest.mark.asyncio +async def test_c490_summary_is_faithful_field_map() -> None: + """The walk records every field (containers included) and stops at the trailer. + + Recording message containers (not just their leaves) keeps the map faithful -- + an under-count is a wrong decode. The one wire-type-3 marker is the trailing + ``a3`` string field past the protobuf, recorded as ``None`` and terminating the + walk (see :func:`SolixBLE.parsing.walk_protobuf`). + """ + device = C2000G2(MOCK_BLE_DEVICE) + await _feed_c490(device, FRAME_IDLE) + + summary = device.summary + assert summary[".1"] == "A1763" # part number leaf + assert isinstance(summary[".23"], int) # the .23 rollup container is recorded + assert summary[".23.1"] == 95 # ... and its leaf + # the walk is bounded to a2's declared length, so it never runs into the trailing + # a3 string field: no spurious .452 group marker, and every value is a real field. + assert ".452" not in summary + assert None not in summary.values() + + +@pytest.mark.parametrize( + "payload_hex,expected_hex", + [ + # a1 01 31 | a2 type(04)+3-byte blob> 04 | 089601 -> just 089601 + ("a10131a2040004089601", "089601"), + # an 8-byte blob: a2 length is 9 (04 type + 8) -> returns the 8 blob bytes + ("a10131a2090004" + "08" * 8, "08" * 8), + # a trailing a3 string field past a2 is bounded out, not appended to the blob + ("a10131a2040004089601a31c0463686172", "089601"), + # too short to carry the wrapper -> returned whole rather than mis-sliced + ("a101", "a101"), + ], +) +def test_protobuf_body_strips_wrapper(payload_hex: str, expected_hex: str) -> None: + body = SolixBLEDevice._protobuf_body(bytes.fromhex(payload_hex)) + assert body == bytes.fromhex(expected_hex) + + +def test_summary_defaults_before_any_c490() -> None: + """With no c490 frame yet, the summary is empty and its properties read defaults.""" + device = C2000G2(MOCK_BLE_DEVICE) + + assert device.summary == {} + assert device.battery_voltage == DEFAULT_METADATA_FLOAT + assert device.flow_state == DEFAULT_METADATA_INT + assert device.c490_battery_soc == DEFAULT_METADATA_INT + assert device.cumulative_discharge_energy == DEFAULT_METADATA_INT