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