protocol.py (5082B)
1 """Protocol constants and command builders for in-lite mesh.""" 2 3 from __future__ import annotations 4 5 from collections.abc import Iterable 6 7 # Packet types (block streaming layer) 8 PKT_BLOCK_FLUSH = 0x70 # 112 9 PKT_BLOCK_DATA = 0x71 # 113 10 PKT_BLOCK_ACK = 0x72 # 114 11 PKT_BLOCK_DATA_BLK = 0x73 # 115 12 PKT_BLOCK_STREAM = 0x74 # 116 — hub response to acknowledged commands 13 14 # Opcodes 15 OPCODE_DISCOVER = 0x000C 16 OPCODE_GET_INFO_DEVICES = 0x0005 17 OPCODE_GET_OUTPUT_NAMES = 0x0019 18 OPCODE_IDENTIFY = 0x0014 19 OPCODE_SET_CLOCK = 0x0064 20 OPCODE_SET_MODULE_DATE_TIME = 0x100C 21 OPCODE_SET_OUTLET_MODE = 0x1007 # 4103 22 OPCODE_START_GARDEN_TEACH_IN = 0x002C 23 OPCODE_OOB_ALL_OUTLETS = 0x0021 # broadcast after state changes 24 25 # TLV notifications carried in a PKT_BLOCK_DATA_BLK payload. 26 CMD_TYPE_TLV = 0x54 27 TLV_TYPE_GARDEN_TEACH_IN = 0x04 28 29 # Output states 30 OUTPUT_OFF = 0x00 31 OUTPUT_ON_AUTO = 0x01 # dusk-to-dawn mode 32 OUTPUT_ON_MANUAL = 0x03 # forced on 33 34 # GATT UUIDs 35 SERVICE_UUID = "0000fef1-0000-1000-8000-00805f9b34fb" 36 # Public factory-reset network constant used by the official app. 37 FACTORY_NETWORK_PASSPHRASE = "6DeNmnsD5XUsf4UD" 38 # The "MTL Complete CP" characteristic — bidirectional (write + notify) 39 CHAR_WRITE_UUID = "c4edc000-9daf-11e3-8004-00025b000b00" 40 # The "MTL Continuation CP" characteristic — bidirectional (write + notify) 41 CHAR_CONTINUATION_UUID = "c4edc000-9daf-11e3-8003-00025b000b00" 42 43 44 def build_outlet_mode_data(output_id: int, on: bool) -> bytes: 45 """Build SET_OUTLET_MODE payload: [outputId, modeByte, modeMaskByte]. 46 47 When on=True: modeByte=0x01 (bit 0 = on), modeMask=0x01 48 When on=False: modeByte=0x00, modeMask=0x01 49 """ 50 mode_byte = 0x01 if on else 0x00 51 mode_mask = 0x01 # only the 'on' bit is being changed 52 return bytes([output_id, mode_byte, mode_mask]) 53 54 55 def build_block_data_payload(opcode: int, command_data: bytes) -> bytes: 56 """Build BLK_DATA inner payload: [offset_lo, offset_hi, cmd_type=1, opcode_lo, opcode_hi, data...].""" 57 return bytes([ 58 0x00, 0x00, # offset = 0 (start of stream) 59 0x01, # cmd_type = 1 (standard command) 60 opcode & 0xFF, (opcode >> 8) & 0xFF, 61 ]) + command_data 62 63 64 def build_flush_payload(byte_count: int = 0) -> bytes: 65 """Build BLK_FLUSH payload.""" 66 return bytes([byte_count & 0xFF, (byte_count >> 8) & 0xFF]) 67 68 69 def build_ack_payload(byte_count: int, end: bool = False) -> bytes: 70 """Build BLK_ACK payload.""" 71 result = bytes([byte_count & 0xFF, (byte_count >> 8) & 0xFF]) 72 if end: 73 result += b'\xef' 74 return result 75 76 77 def build_discovery_payload() -> bytes: 78 """Build BLK_DATA_BLK discovery packet.""" 79 return build_unacknowledged_command_payload(OPCODE_DISCOVER) 80 81 82 def build_unacknowledged_command_payload(opcode: int, command_data: bytes = b"") -> bytes: 83 """Build a fire-and-forget BLK_DATA_BLK command payload.""" 84 return bytes([0x01, opcode & 0xFF, (opcode >> 8) & 0xFF]) + command_data 85 86 87 def build_association_payload(network_key: bytes) -> bytes: 88 """Build the local in-lite factory-network association request. 89 90 This is a vendor MCP stream rather than a standard command stream. The 91 network key is the 16-byte MCP key derived from the new passphrase. 92 """ 93 if len(network_key) != 16: 94 raise ValueError("network_key must be 16 bytes") 95 return bytes([0x01, 0x24, 0x00]) + network_key 96 97 98 def build_clock_payload( 99 timestamp: int, 100 utc_offset_seconds: int, 101 transitions: Iterable[tuple[int, int]] = (), 102 ) -> bytes: 103 """Build SET_CLOCK's current and future timezone-offset entries. 104 105 The hub stores UTC transition instants followed by the corresponding UTC 106 offset in quarter-hours. Sending the current entry every day keeps both 107 the time and offset current without requiring a location or cloud service. 108 """ 109 def _entry(entry_timestamp: int, entry_offset_seconds: int) -> bytes: 110 if entry_timestamp < 0 or entry_timestamp > 0xFFFFFFFF: 111 raise ValueError("timestamp must fit in an unsigned 32-bit integer") 112 offset_quarters = round(entry_offset_seconds / (15 * 60)) 113 if not -128 <= offset_quarters <= 127: 114 raise ValueError("UTC offset must fit in a signed byte of quarter-hours") 115 return entry_timestamp.to_bytes(4, "little") + offset_quarters.to_bytes( 116 1, "little", signed=True 117 ) 118 119 return _entry(timestamp, utc_offset_seconds) + b"".join( 120 _entry(transition_timestamp, transition_offset_seconds) 121 for transition_timestamp, transition_offset_seconds in transitions 122 ) 123 124 125 def build_legacy_datetime_payload(timestamp: int) -> bytes: 126 """Build SET_MODULE_DATE_TIME's UTC calendar payload.""" 127 from datetime import UTC, datetime 128 129 value = datetime.fromtimestamp(timestamp, UTC) 130 return bytes( 131 [ 132 value.second, 133 value.minute, 134 value.hour, 135 value.day, 136 value.month, 137 value.year & 0xFF, 138 value.year >> 8, 139 ] 140 )