hub.py (34013B)
1 """InliteHub — BLE controller for in-lite SMART HUB transformers.""" 2 3 from __future__ import annotations 4 5 import asyncio 6 import logging 7 from collections import deque 8 from collections.abc import Callable, Iterable 9 from datetime import datetime 10 from typing import Any 11 12 from bleak import BleakClient, BleakScanner 13 from bleak.backends.device import BLEDevice 14 15 from inlite_ble.crypto import CsrMeshCrypto 16 from inlite_ble.protocol import ( 17 SERVICE_UUID, 18 CHAR_WRITE_UUID, 19 CHAR_CONTINUATION_UUID, 20 PKT_BLOCK_FLUSH, 21 PKT_BLOCK_DATA, 22 PKT_BLOCK_ACK, 23 PKT_BLOCK_DATA_BLK, 24 PKT_BLOCK_STREAM, 25 OPCODE_SET_OUTLET_MODE, 26 OPCODE_IDENTIFY, 27 OPCODE_SET_CLOCK, 28 OPCODE_SET_MODULE_DATE_TIME, 29 OPCODE_START_GARDEN_TEACH_IN, 30 OPCODE_GET_INFO_DEVICES, 31 OPCODE_OOB_ALL_OUTLETS, 32 build_outlet_mode_data, 33 build_block_data_payload, 34 build_flush_payload, 35 build_ack_payload, 36 build_discovery_payload, 37 build_unacknowledged_command_payload, 38 build_association_payload, 39 build_clock_payload, 40 build_legacy_datetime_payload, 41 CMD_TYPE_TLV, 42 TLV_TYPE_GARDEN_TEACH_IN, 43 ) 44 45 _LOGGER = logging.getLogger(__name__) 46 47 WRITE_DELAY = 0.06 # 60ms between BLE writes (matches app timing) 48 ACK_TIMEOUT = 0.6 # Official client retries a segment after 600 ms. 49 TX_STREAM_MAX_RETRIES = 17 # Matches the official client's send budget. 50 STREAM_TIMEOUT = 15.0 # The official app retains an RX stream for 15 seconds. 51 STREAM_REACK_INTERVAL = 5.0 # Re-ACK an incomplete response after this delay. 52 BLE_PACKET_PART_SIZE = 78 # Matches the official app's GATT packet splitter. 53 STREAM_DATA_SIZE = 62 # Maximum official-app payload per PKT_BLOCK_DATA. 54 SMART_HUB_75_PRODUCT_ID = 0x1D 55 56 57 class ZoneState: 58 """Current state of a single light zone.""" 59 60 def __init__( 61 self, 62 output_id: int, 63 output_type: int = 0, 64 cap_mask: int = 0, 65 output_mode: int = 0, 66 dtd1: int = 0, 67 dtd2: int = 0, 68 output_state: int = 0, 69 ) -> None: 70 self.output_id = output_id 71 self.output_type = output_type 72 self.cap_mask = cap_mask 73 self.output_mode = output_mode 74 self.dtd1 = dtd1 75 self.dtd2 = dtd2 76 self.output_state = output_state 77 78 @property 79 def is_on(self) -> bool: 80 """Whether the physical outlet is currently energized. 81 82 ``output_mode`` configures permanent/motion/timer/dusk behaviour. 83 The hub reports the source(s) currently energizing the outlet in the 84 low four bits of ``output_state``. 85 """ 86 return (self.output_state & 0x0F) != 0 87 88 def __repr__(self) -> str: 89 return "ZoneState(id=%d, %s, mode=0x%02X, state=0x%02X)" % ( 90 self.output_id, 91 "ON" if self.is_on else "OFF", 92 self.output_mode, 93 self.output_state, 94 ) 95 96 97 class InliteHub: 98 """Controls an in-lite SMART HUB via BLE mesh. 99 100 The BLE device acts as a gateway to the CSRmesh network; the device_id 101 is the mesh destination address for a specific transformer. 102 103 Args: 104 device_id: The hub's mesh device ID (from cloud API transformers[].deviceId) 105 passphrase: The mesh passphrase, from which the network key is derived. 106 network_key: The already-derived 16-byte CSRmesh key. Mutually exclusive 107 with ``passphrase``; useful when joining a mesh without its passphrase. 108 ble_address: BLE device address or name (e.g., 'inlitebt' or a MAC/UUID) 109 on_state_update: Optional callback invoked when OOB broadcast updates zone states. 110 """ 111 112 def __init__( 113 self, 114 device_id: int, 115 passphrase: str | None = None, 116 ble_address: str = "inlitebt", 117 on_state_update: Callable[[], None] | None = None, 118 on_teach_in_progress: Callable[[int, int], None] | None = None, 119 network_key: bytes | None = None, 120 ) -> None: 121 self._device_id = device_id 122 self._crypto = CsrMeshCrypto(passphrase, network_key) 123 self._ble_address = ble_address 124 self._client: BleakClient | None = None 125 self._loop: asyncio.AbstractEventLoop | None = None 126 # ACK, FLUSH, and STREAM packets share one queue. Their ordering is 127 # significant: splitting them into separate queues lets a later STREAM 128 # overtake a FLUSH/ACK and desynchronise the block-stream state machine. 129 self._packet_queue: asyncio.Queue[tuple[int, bytes]] = asyncio.Queue() 130 self._deferred_packets: deque[tuple[int, bytes]] = deque() 131 self._last_stream_data = b"" 132 self._stream_buffer = bytearray() 133 self._stream_invalid = False 134 self._incoming_ble_buffer = bytearray() 135 self._zone_states: dict[int, ZoneState] = {} 136 self._firmware_version: int | None = None 137 self._product_id: int | None = None 138 self._hidden_output_ids: set[int] = set() 139 self._discovered_device_ids: set[int] = set() 140 self._discovery_event = asyncio.Event() 141 self._notification_callback: Callable[[dict[str, Any]], None] | None = None 142 self._on_state_update = on_state_update 143 self._on_teach_in_progress = on_teach_in_progress 144 145 @property 146 def device_id(self) -> int: 147 return self._device_id 148 149 @property 150 def controller_address(self) -> int: 151 return self._crypto.controller_address 152 153 @property 154 def is_connected(self) -> bool: 155 return self._client is not None and self._client.is_connected 156 157 @property 158 def zone_states(self) -> dict[int, ZoneState]: 159 return self._zone_states 160 161 @property 162 def firmware_version(self) -> int | None: 163 """Firmware byte reported by GET_INFO_DEVICES.""" 164 return self._firmware_version 165 166 def is_output_visible(self, output_id: int) -> bool: 167 """Return whether an output should be exposed by the integration.""" 168 return output_id not in self._hidden_output_ids 169 170 async def scan(self, timeout: float = 10.0) -> BLEDevice | None: 171 """Scan for the in-lite hub by name or address.""" 172 _LOGGER.info("Scanning for %s...", self._ble_address) 173 devices = await BleakScanner.discover(timeout=timeout) 174 for d in devices: 175 if (d.name and d.name.lower() == self._ble_address.lower()) or \ 176 str(d.address).lower() == self._ble_address.lower(): 177 _LOGGER.info("Found hub: %s (%s)", d.name, d.address) 178 return d 179 return None 180 181 async def connect( 182 self, 183 device: BLEDevice | None = None, 184 client: BleakClient | None = None, 185 subscribe: bool = True, 186 notification_handler: Callable[[Any, bytearray], None] | None = None, 187 ) -> bool: 188 """Connect to the hub and enable notifications. 189 190 Args: 191 device: BLEDevice to connect to (scans if None and no client given). 192 client: Pre-established BleakClient (for HA integration). If given, 193 we use it directly and subscribe notifications. 194 """ 195 # Store reference to the running event loop for thread-safe callbacks 196 self._loop = asyncio.get_running_loop() 197 198 if client is not None: 199 self._client = client 200 else: 201 if device is None: 202 device = await self.scan() 203 if device is None: 204 _LOGGER.error("Hub not found") 205 return False 206 self._client = BleakClient(device) 207 await self._client.connect() 208 209 if not self._client.is_connected: 210 _LOGGER.error("Client not connected after setup") 211 self._client = None 212 return False 213 214 _LOGGER.info("Connected") 215 216 # Subscribe to notifications on both bidirectional characteristics 217 if subscribe: 218 handler = notification_handler or self._on_notification 219 # A CSRmesh packet can be split over the continuation and complete 220 # characteristics. Pass the characteristic identity on explicitly: 221 # Bleak otherwise supplies a backend-specific sender (usually an 222 # integer handle), which cannot be compared portably. 223 await self._client.start_notify( 224 CHAR_WRITE_UUID, 225 lambda _sender, data: handler(CHAR_WRITE_UUID, data), 226 ) 227 await self._client.start_notify( 228 CHAR_CONTINUATION_UUID, 229 lambda _sender, data: handler(CHAR_CONTINUATION_UUID, data), 230 ) 231 _LOGGER.info("Notifications enabled") 232 233 return True 234 235 async def disconnect(self) -> None: 236 """Disconnect from the hub.""" 237 if self._client: 238 try: 239 if self._client.is_connected: 240 await self._client.disconnect() 241 except Exception as err: 242 _LOGGER.debug("Disconnect error (ignoring): %s", err) 243 finally: 244 self._client = None 245 self._loop = None 246 _LOGGER.info("Disconnected") 247 248 def detach_client(self) -> None: 249 """Forget a shared client without disconnecting it.""" 250 self._client = None 251 self._loop = None 252 253 def _on_notification(self, sender: Any, data: bytearray) -> None: 254 """Receive a BLE notification from Bleak on an arbitrary thread.""" 255 self.queue_notification(sender, data) 256 257 def queue_notification(self, sender: Any, data: bytearray) -> None: 258 """Schedule notification processing on the hub's event loop. 259 260 This must be the entry point for BLE callbacks. In particular, the 261 continuation buffer is mutable and must never be assembled from the 262 callback thread. 263 """ 264 loop = self._loop 265 if loop is None or loop.is_closed(): 266 return 267 loop.call_soon_threadsafe(self.handle_notification, sender, bytearray(data)) 268 269 def handle_notification(self, sender: Any, data: bytearray) -> None: 270 """Process one notification on the hub's event loop.""" 271 if sender == CHAR_CONTINUATION_UUID: 272 self._incoming_ble_buffer.extend(data) 273 return 274 275 # The complete characteristic terminates one encrypted packet. A 276 # one-piece packet simply has an empty continuation buffer. 277 raw = bytes(self._incoming_ble_buffer) + bytes(data) 278 self._incoming_ble_buffer.clear() 279 decrypted = self._crypto.decrypt_packet(raw) 280 if decrypted is None: 281 return 282 283 pkt_type = decrypted["pkt_type"] 284 payload = decrypted["data"] 285 286 loop = self._loop 287 if loop is None or loop.is_closed(): 288 return 289 290 if pkt_type in (PKT_BLOCK_ACK, PKT_BLOCK_FLUSH, PKT_BLOCK_STREAM): 291 # Control packets for a command are unicast. The hub uses ACKs 292 # while receiving our request, then FLUSH packets for its response. 293 # Filter before queuing so traffic from another mesh node cannot 294 # satisfy this hub's active command. 295 if ( 296 decrypted["src_id"] == self._device_id 297 and decrypted["dest_id"] == self._crypto.controller_address 298 ): 299 loop.call_soon_threadsafe( 300 self._packet_queue.put_nowait, (pkt_type, payload) 301 ) 302 elif pkt_type == PKT_BLOCK_DATA_BLK: 303 if len(payload) >= 3 and payload[1:3] == b"\x0c\x00": 304 self._discovered_device_ids.add(decrypted["src_id"]) 305 loop.call_soon_threadsafe(self._discovery_event.set) 306 if decrypted["src_id"] == self._device_id: 307 loop.call_soon_threadsafe(self._parse_oob_broadcast, payload) 308 self._parse_tlv_notification(decrypted["src_id"], payload) 309 310 if self._notification_callback: 311 loop.call_soon_threadsafe(self._notification_callback, decrypted) 312 313 def _accept_stream_segment(self, payload: bytes) -> bool: 314 """Append one contiguous STREAM segment, rejecting malformed offsets.""" 315 if len(payload) < 2: 316 self._stream_invalid = True 317 return False 318 offset = payload[0] | (payload[1] << 8) 319 if offset != len(self._stream_buffer): 320 _LOGGER.warning("Discarding STREAM segment at offset %d (expected %d)", offset, len(self._stream_buffer)) 321 self._stream_invalid = True 322 return False 323 self._stream_buffer.extend(payload[2:]) 324 self._last_stream_data = bytes(self._stream_buffer) 325 return True 326 327 def _parse_oob_broadcast(self, payload: bytes) -> None: 328 """Parse OOB_ALL_OUTLETS_MODE_UPDATE broadcast to update zone states.""" 329 if len(payload) < 3: 330 return 331 cmd_type = payload[0] 332 opcode = payload[1] | (payload[2] << 8) 333 if cmd_type != 0x03 or opcode != OPCODE_OOB_ALL_OUTLETS: 334 return 335 336 data = payload[3:] 337 i = 0 338 changed = False 339 while i + 3 < len(data): 340 outlet_id = data[i] 341 output_mode = data[i + 1] 342 output_state = data[i + 2] 343 # data[i + 3] = rtcTimer 344 if outlet_id in self._hidden_output_ids: 345 i += 4 346 continue 347 if outlet_id in self._zone_states: 348 old = self._zone_states[outlet_id] 349 if old.output_mode != output_mode or old.output_state != output_state: 350 changed = True 351 old.output_mode = output_mode 352 old.output_state = output_state 353 else: 354 self._zone_states[outlet_id] = ZoneState( 355 output_id=outlet_id, 356 output_mode=output_mode, 357 output_state=output_state, 358 ) 359 changed = True 360 _LOGGER.debug("OOB update: zone %d mode=0x%02X state=0x%02X", 361 outlet_id, output_mode, output_state) 362 i += 4 363 364 if changed and self._on_state_update is not None: 365 self._on_state_update() 366 367 def _parse_tlv_notification(self, source_id: int, payload: bytes) -> None: 368 """Forward documented garden teach-in progress TLVs to the caller.""" 369 # TLV block format: [0x54, type, length, value, ...]. A block can 370 # contain more than one TLV, so validate and walk it defensively. 371 if not payload or payload[0] != CMD_TYPE_TLV: 372 return 373 pos = 1 374 while pos + 2 <= len(payload): 375 tlv_type = payload[pos] 376 length = payload[pos + 1] 377 pos += 2 378 if pos + length > len(payload): 379 _LOGGER.warning("Discarding truncated TLV notification") 380 return 381 value = payload[pos : pos + length] 382 pos += length 383 if ( 384 tlv_type == TLV_TYPE_GARDEN_TEACH_IN 385 and value 386 and self._on_teach_in_progress is not None 387 ): 388 loop = self._loop 389 if loop is not None and not loop.is_closed(): 390 loop.call_soon_threadsafe( 391 self._on_teach_in_progress, source_id, value[0] 392 ) 393 394 async def _write(self, packet: bytes) -> None: 395 """Write an encrypted packet, splitting BLE continuation fragments.""" 396 if not self._client or not self._client.is_connected: 397 raise ConnectionError("Not connected to hub") 398 for offset in range(0, len(packet), BLE_PACKET_PART_SIZE): 399 part = packet[offset : offset + BLE_PACKET_PART_SIZE] 400 is_final = offset + BLE_PACKET_PART_SIZE >= len(packet) 401 await self._client.write_gatt_char( 402 CHAR_WRITE_UUID if is_final else CHAR_CONTINUATION_UUID, 403 part, 404 response=True, 405 ) 406 407 async def _write_mesh(self, dest_id: int, pkt_type: int, data: bytes) -> None: 408 """Encrypt and write a mesh packet.""" 409 packet = self._crypto.encrypt_packet(dest_id, pkt_type, data) 410 await self._write(packet) 411 412 async def _wait_packet( 413 self, packet_types: tuple[int, ...], timeout: float 414 ) -> tuple[int, bytes]: 415 """Return the next packet of a requested type without losing others.""" 416 for packet in self._deferred_packets: 417 if packet[0] in packet_types: 418 self._deferred_packets.remove(packet) 419 return packet 420 loop = asyncio.get_running_loop() 421 deadline = loop.time() + timeout 422 while True: 423 remaining = deadline - loop.time() 424 if remaining <= 0: 425 raise asyncio.TimeoutError 426 packet = await asyncio.wait_for(self._packet_queue.get(), remaining) 427 if packet[0] in packet_types: 428 return packet 429 self._deferred_packets.append(packet) 430 431 async def _wait_ack(self, timeout: float = ACK_TIMEOUT) -> bytes: 432 """Return the next ACK, preserving an early response packet.""" 433 try: 434 _, payload = await self._wait_packet((PKT_BLOCK_ACK,), timeout) 435 return payload 436 except asyncio.TimeoutError: 437 _LOGGER.warning("ACK timeout after %.1fs", timeout) 438 return b"" 439 440 async def _send_command( 441 self, opcode: int, command_data: bytes, acknowledged: bool = True 442 ) -> bool: 443 """Send a command using the block streaming protocol. 444 445 Flow: 446 1. BLK_FLUSH(0x0000) → wait ACK 447 2. BLK_DATA(offset, opcode, data) → wait ACK with byte count 448 3. BLK_FLUSH(byte_count) → wait ACK with 'ef' suffix (done) 449 4. ACK the hub's response 450 """ 451 if not await self._send_raw_stream(build_block_data_payload(opcode, command_data)): 452 return False 453 # Ordinary commands have a response stream too. Drain it even when 454 # the contents are only a status acknowledgement. 455 return await self._receive_response_stream() is not None 456 457 async def _send_raw_stream(self, block_data: bytes) -> bool: 458 """Send a request block stream and wait for its completion ACK. 459 460 Association is the observed exception that changes encryption key and 461 has no response stream; callers of normal commands must subsequently 462 invoke ``_receive_response_stream``. 463 """ 464 dest = self._device_id 465 self._stream_buffer.clear() 466 self._stream_invalid = False 467 while not self._packet_queue.empty(): 468 self._packet_queue.get_nowait() 469 self._deferred_packets.clear() 470 471 # ``build_block_data_payload`` retains its offset prefix for callers 472 # and tests. The stream transport owns that prefix, however, so strip 473 # its initial zero offset before transmitting numbered segments. 474 # Association is already an offset-free vendor payload. 475 stream_payload = block_data[2:] if block_data.startswith(b"\x00\x00") else block_data 476 async def send_and_wait_for_ack( 477 packet_type: int, payload: bytes, expected_min: int, expected_max: int 478 ) -> int | None: 479 """Send one stream phase until its cumulative ACK advances.""" 480 for attempt in range(TX_STREAM_MAX_RETRIES): 481 await self._write_mesh(dest, packet_type, payload) 482 ack = await self._wait_ack() 483 if len(ack) >= 2: 484 acked_bytes = ack[0] | (ack[1] << 8) 485 if expected_min <= acked_bytes <= expected_max: 486 return acked_bytes 487 _LOGGER.debug( 488 "Stream packet type 0x%02X attempt %d/%d was not acknowledged " 489 "in range %d..%d", 490 packet_type, attempt + 1, TX_STREAM_MAX_RETRIES, 491 expected_min, expected_max, 492 ) 493 return None 494 495 if await send_and_wait_for_ack( 496 PKT_BLOCK_FLUSH, build_flush_payload(0), 0, 0 497 ) is None: 498 _LOGGER.warning("Block stream: start FLUSH was not acknowledged") 499 return False 500 501 offset = 0 502 while offset < len(stream_payload): 503 segment = stream_payload[offset : offset + STREAM_DATA_SIZE] 504 acked_bytes = await send_and_wait_for_ack( 505 PKT_BLOCK_DATA, 506 offset.to_bytes(2, "little") + segment, 507 offset + 1, 508 offset + len(segment), 509 ) 510 if acked_bytes is None: 511 _LOGGER.warning("Block stream: DATA was not acknowledged") 512 return False 513 offset = acked_bytes 514 515 # Step 3: Flush (end) 516 completed = await send_and_wait_for_ack( 517 PKT_BLOCK_FLUSH, build_flush_payload(offset), offset, offset 518 ) 519 if completed is None: 520 _LOGGER.warning("Block stream: no completion ACK") 521 return False 522 _LOGGER.info("Block stream sent successfully") 523 return True 524 525 async def _receive_response_stream(self) -> bytes | None: 526 """Drain the hub's FLUSH → STREAM* → FLUSH response exchange.""" 527 dest = self._device_id 528 # A response begins with FLUSH(0). Keep an incomplete response alive, 529 # like the official client, rather than treating one missed interval as 530 # a failed command. 531 try: 532 _, start_flush = await self._wait_packet( 533 (PKT_BLOCK_FLUSH,), STREAM_TIMEOUT 534 ) 535 if len(start_flush) >= 2 and (start_flush[0] | (start_flush[1] << 8)) != 0: 536 _LOGGER.warning("Response stream started at non-zero offset") 537 return None 538 await asyncio.sleep(WRITE_DELAY) 539 await self._write_mesh(dest, PKT_BLOCK_ACK, build_ack_payload(0)) 540 541 started = asyncio.get_running_loop().time() 542 last_activity = started 543 while True: 544 now = asyncio.get_running_loop().time() 545 remaining = STREAM_TIMEOUT - (now - started) 546 if remaining <= 0: 547 _LOGGER.warning("Timed out waiting for hub response stream") 548 return None 549 try: 550 response_type, payload = await self._wait_response_part( 551 min(STREAM_REACK_INTERVAL, remaining) 552 ) 553 except asyncio.TimeoutError: 554 # The target may have missed our last ACK. Re-ACK the 555 # contiguous prefix and retain the receive state. 556 if self._stream_buffer and ( 557 asyncio.get_running_loop().time() - last_activity 558 >= STREAM_REACK_INTERVAL 559 ): 560 _LOGGER.debug( 561 "Re-ACKing incomplete STREAM response at offset %d", 562 len(self._stream_buffer), 563 ) 564 await self._write_mesh( 565 dest, PKT_BLOCK_ACK, 566 build_ack_payload(len(self._stream_buffer)), 567 ) 568 last_activity = asyncio.get_running_loop().time() 569 continue 570 continue 571 572 last_activity = asyncio.get_running_loop().time() 573 if response_type == PKT_BLOCK_STREAM: 574 offset = payload[0] | (payload[1] << 8) if len(payload) >= 2 else None 575 if offset != len(self._stream_buffer): 576 # Re-ACK the contiguous prefix; the hub will retransmit 577 # a duplicate or the missing segment. 578 _LOGGER.debug( 579 "Out-of-order STREAM segment at offset %s (expected %d); re-ACKing", 580 offset, 581 len(self._stream_buffer), 582 ) 583 await self._write_mesh( 584 dest, PKT_BLOCK_ACK, build_ack_payload(len(self._stream_buffer)) 585 ) 586 continue 587 if not self._accept_stream_segment(payload): 588 _LOGGER.warning("Discarding malformed STREAM response") 589 return None 590 # The official app acknowledges every contiguous segment. 591 await self._write_mesh( 592 dest, PKT_BLOCK_ACK, build_ack_payload(len(self._stream_buffer)) 593 ) 594 continue 595 596 flush_count = ( 597 payload[0] | (payload[1] << 8) 598 if len(payload) >= 2 599 else len(self._stream_buffer) 600 ) 601 if flush_count != len(self._stream_buffer): 602 _LOGGER.debug( 603 "STREAM end FLUSH at offset %d (expected %d); re-ACKing", 604 flush_count, 605 len(self._stream_buffer), 606 ) 607 await self._write_mesh( 608 dest, PKT_BLOCK_ACK, build_ack_payload(len(self._stream_buffer)) 609 ) 610 continue 611 await asyncio.sleep(WRITE_DELAY) 612 # The 0xef terminator is a third byte, not an arithmetic adjustment 613 # of the 16-bit byte count. 614 await self._write_mesh( 615 dest, PKT_BLOCK_ACK, build_ack_payload(flush_count, end=True) 616 ) 617 return bytes(self._stream_buffer) 618 except asyncio.TimeoutError: 619 _LOGGER.warning("Timed out waiting for hub response stream") 620 return None 621 622 async def _wait_response_part(self, timeout: float = STREAM_TIMEOUT) -> tuple[int, bytes]: 623 """Return whichever arrives next: a stream segment or final FLUSH.""" 624 return await self._wait_packet((PKT_BLOCK_STREAM, PKT_BLOCK_FLUSH), timeout) 625 626 async def _send_acknowledged_command( 627 self, opcode: int, command_data: bytes 628 ) -> bytes | None: 629 """Send a command that expects a STREAM response (e.g., GET_INFO_DEVICES). 630 631 Flow: 632 1. Send command via block streaming (same as _send_command) 633 2. Hub responds: FLUSH(0) → STREAM(data) → FLUSH(n) 634 3. We ACK each step and return the STREAM data 635 636 Returns: 637 The STREAM payload bytes, or None on failure. 638 """ 639 block_data = build_block_data_payload(opcode, command_data) 640 if not await self._send_raw_stream(block_data): 641 _LOGGER.warning("Acknowledged command 0x%04X: no completion ACK", opcode) 642 return None 643 stream_data = await self._receive_response_stream() 644 if stream_data is None: 645 return None 646 _LOGGER.info("Acknowledged command 0x%04X: got %d bytes", opcode, len(stream_data)) 647 # Restore the stream's offset prefix expected by the response parser. 648 return b"\x00\x00" + bytes(stream_data) 649 650 async def query_zone_states(self) -> dict[int, ZoneState]: 651 """Query all zone states from the hub using GET_INFO_DEVICES. 652 653 Returns: 654 Dict mapping output_id → ZoneState for each zone. 655 """ 656 # Parse STREAM response: 657 # [offset(2), cmd_type=0x02, opcode(2), vendorId, productId, firmware, status, 658 # numZones, then per zone(7 bytes): outputId, outputType, capMask, outputMode, 659 # dtd1, dtd2, outputState] 660 stream = await self._send_acknowledged_command( 661 OPCODE_GET_INFO_DEVICES, b"" 662 ) 663 if stream is None: 664 _LOGGER.warning("Failed to query zone states") 665 return self._zone_states 666 667 if len(stream) < 10: 668 _LOGGER.warning("STREAM too short: %d bytes", len(stream)) 669 return self._zone_states 670 671 offset = stream[0] | (stream[1] << 8) 672 cmd_type = stream[2] 673 opcode = stream[3] | (stream[4] << 8) 674 if offset != 0 or cmd_type != 0x02 or opcode != OPCODE_GET_INFO_DEVICES: 675 _LOGGER.warning("Unexpected opcode in STREAM: 0x%04X", opcode) 676 return self._zone_states 677 678 # vendor_id = stream[5] 679 self._product_id = stream[6] 680 self._firmware_version = stream[7] 681 # status = stream[8] 682 num_zones = stream[9] 683 expected_length = 10 + num_zones * 7 684 if len(stream) != expected_length: 685 _LOGGER.warning( 686 "Malformed GET_INFO_DEVICES response: %d bytes, expected %d", 687 len(stream), expected_length, 688 ) 689 return self._zone_states 690 691 692 zone_records: list[ZoneState] = [] 693 zone_start = 10 694 for i in range(num_zones): 695 pos = zone_start + i * 7 696 zs = ZoneState( 697 output_id=stream[pos], 698 output_type=stream[pos + 1], 699 cap_mask=stream[pos + 2], 700 output_mode=stream[pos + 3], 701 dtd1=stream[pos + 4], 702 dtd2=stream[pos + 5], 703 output_state=stream[pos + 6], 704 ) 705 zone_records.append(zs) 706 _LOGGER.debug("Zone %d: %s", zs.output_id, zs) 707 708 self._hidden_output_ids.clear() 709 if self._product_id == SMART_HUB_75_PRODUCT_ID and len(zone_records) > 2: 710 # The official app suppresses the final raw zone on a Smart Hub 75. 711 hidden_zone = zone_records.pop() 712 self._hidden_output_ids.add(hidden_zone.output_id) 713 _LOGGER.debug("Ignoring unavailable Smart Hub 75 output %d", hidden_zone.output_id) 714 715 zones = {zone.output_id: zone for zone in zone_records} 716 717 self._zone_states = zones 718 return zones 719 720 async def set_outlet_mode(self, output_id: int, on: bool) -> bool: 721 """Turn a light zone on or off. 722 723 Args: 724 output_id: Zone number (0=zone 1, 1=zone 2, 2=zone 3) 725 on: True to turn on, False to turn off 726 """ 727 data = build_outlet_mode_data(output_id, on) 728 _LOGGER.info( 729 "Setting zone %d %s (device 0x%04X)", 730 output_id, "ON" if on else "OFF", self._device_id, 731 ) 732 success = await self._send_command(OPCODE_SET_OUTLET_MODE, data) 733 734 # Update local state immediately on success 735 if success and output_id in self._zone_states: 736 zs = self._zone_states[output_id] 737 if on: 738 zs.output_mode = zs.output_mode | 0x01 739 zs.output_state = zs.output_state | 0x01 740 else: 741 zs.output_mode = zs.output_mode & ~0x01 742 # Match the app: manual off clears every active-output source 743 # (permanent, motion, dusk-to-dawn, and timer). 744 zs.output_state = zs.output_state & ~0x0F 745 746 return success 747 748 async def turn_on(self, output_id: int = 0) -> bool: 749 """Turn on a light zone.""" 750 return await self.set_outlet_mode(output_id, True) 751 752 async def turn_off(self, output_id: int = 0) -> bool: 753 """Turn off a light zone.""" 754 return await self.set_outlet_mode(output_id, False) 755 756 async def send_discovery(self) -> None: 757 """Send a discovery broadcast (also serves as keepalive).""" 758 data = build_discovery_payload() 759 await self._write_mesh(0x0000, PKT_BLOCK_DATA_BLK, data) 760 761 async def discover_hubs(self, timeout: float = 5.0) -> set[int]: 762 """Discover factory-network in-lite devices and return mesh IDs.""" 763 self._discovered_device_ids.clear() 764 self._discovery_event.clear() 765 await self.send_discovery() 766 # Discovery responses arrive independently; keep collecting for the 767 # complete window rather than stopping after the first hub responds. 768 await asyncio.sleep(timeout) 769 return self._discovered_device_ids.copy() 770 771 async def associate(self, new_passphrase: str) -> bool: 772 """Move this factory-reset hub onto ``new_passphrase``'s local mesh.""" 773 return await self._send_raw_stream( 774 build_association_payload(CsrMeshCrypto.derive_key(new_passphrase)) 775 ) 776 777 async def identify(self) -> bool: 778 """Ask a hub to briefly blink its identify light.""" 779 # The official app sends IDENTIFY as a fire-and-forget BLK_DATA_BLK 780 # packet. It has neither block-stream ACKs nor a response stream. 781 await self._write_mesh( 782 self._device_id, 783 PKT_BLOCK_DATA_BLK, 784 build_unacknowledged_command_payload(OPCODE_IDENTIFY), 785 ) 786 return True 787 788 async def sync_clock( 789 self, now: datetime, transitions: Iterable[tuple[int, int]] = () 790 ) -> bool: 791 """Synchronize the hub clock and current local UTC offset. 792 793 Firmware supporting the RTC protocol receives SET_CLOCK (0x0064). 794 Older firmware receives its documented UTC calendar command instead. 795 """ 796 if now.tzinfo is None: 797 raise ValueError("clock synchronization requires an aware datetime") 798 timestamp = int(now.timestamp()) 799 firmware = self._firmware_version or 0 800 if (32 <= firmware < 100) or firmware > 132: 801 offset = now.utcoffset() 802 offset_seconds = int(offset.total_seconds()) if offset else 0 803 return await self._send_command( 804 OPCODE_SET_CLOCK, 805 build_clock_payload(timestamp, offset_seconds, transitions), 806 ) 807 return await self._send_command( 808 OPCODE_SET_MODULE_DATE_TIME, build_legacy_datetime_payload(timestamp) 809 ) 810 811 async def start_teach_in(self) -> bool: 812 """Start the official app's garden teach-in operation. 813 814 Its 0x0a options reapply the mesh password and remove group assignments 815 below ten, matching the app's smart-lamp onboarding flow. 816 """ 817 return await self._send_command(OPCODE_START_GARDEN_TEACH_IN, b"\x0a")