wesandClaude Opus 4.8
34155fd7f9
The OneControl panel's command characteristic is a streaming (Write Without Response) endpoint. Bleak 3.x changed write_gatt_char to default to write-with-response when the char advertises the "write" property, so every command (incl. switch turn_on/off) got rejected by the panel with ATT 0x0E (Unlikely Error), surfaced as BleakGATTProtocolError. Force response=False on the command write (matching the auth key write) to restore control. Also commits the productionized custom_components integration (config flow, coordinator, switch/sensor/cover entities, key-seed TEA auth, COBS codec) and the matching src/ RE client/COBS fixes (big-endian framing, table-driven CRC8, status-event decoding) that were developed but never tracked. Verified live on the campsite HAOS Pi: switch.exterior_lights / interior_lights toggle the physical panel with no GATT error. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
150 lines
5.6 KiB
Python
150 lines
5.6 KiB
Python
# cobs_protocol.py
|
|
# COBS encoding/decoding and CRC8 implementation for Lippert OneControl
|
|
# Based on decompiled source from IDS.Portable.Common.COBS/CobsEncoder.cs and Crc8.cs
|
|
|
|
|
|
class Crc8:
|
|
"""CRC8 using the exact lookup table from IDS.Portable.Common/Crc8.cs (init=0x55)."""
|
|
|
|
_TABLE = [
|
|
0, 94, 188, 226, 97, 63, 221, 131, 194, 156,
|
|
126, 32, 163, 253, 31, 65, 157, 195, 33, 127,
|
|
252, 162, 64, 30, 95, 1, 227, 189, 62, 96,
|
|
130, 220, 35, 125, 159, 193, 66, 28, 254, 160,
|
|
225, 191, 93, 3, 128, 222, 60, 98, 190, 224,
|
|
2, 92, 223, 129, 99, 61, 124, 34, 192, 158,
|
|
29, 67, 161, 255, 70, 24, 250, 164, 39, 121,
|
|
155, 197, 132, 218, 56, 102, 229, 187, 89, 7,
|
|
219, 133, 103, 57, 186, 228, 6, 88, 25, 71,
|
|
165, 251, 120, 38, 196, 154, 101, 59, 217, 135,
|
|
4, 90, 184, 230, 167, 249, 27, 69, 198, 152,
|
|
122, 36, 248, 166, 68, 26, 153, 199, 37, 123,
|
|
58, 100, 134, 216, 91, 5, 231, 185, 140, 210,
|
|
48, 110, 237, 179, 81, 15, 78, 16, 242, 172,
|
|
47, 113, 147, 205, 17, 79, 173, 243, 112, 46,
|
|
204, 146, 211, 141, 111, 49, 178, 236, 14, 80,
|
|
175, 241, 19, 77, 206, 144, 114, 44, 109, 51,
|
|
209, 143, 12, 82, 176, 238, 50, 108, 142, 208,
|
|
83, 13, 239, 177, 240, 174, 76, 18, 145, 207,
|
|
45, 115, 202, 148, 118, 40, 171, 245, 23, 73,
|
|
8, 86, 180, 234, 105, 55, 213, 139, 87, 9,
|
|
235, 181, 54, 104, 138, 212, 149, 203, 41, 119,
|
|
244, 170, 72, 22, 233, 183, 85, 11, 136, 214,
|
|
52, 106, 43, 117, 151, 201, 74, 20, 246, 168,
|
|
116, 42, 200, 150, 21, 75, 169, 247, 182, 232,
|
|
10, 84, 215, 137, 107, 53,
|
|
]
|
|
|
|
RESET_VALUE = 0x55
|
|
|
|
@staticmethod
|
|
def calculate(data: bytes) -> int:
|
|
crc = Crc8.RESET_VALUE
|
|
for b in data:
|
|
crc = Crc8._TABLE[(crc ^ b) & 0xFF]
|
|
return crc
|
|
|
|
|
|
class CobsEncoder:
|
|
"""COBS Encoder matching CobsEncoder.cs (numDataBits=6, prependStartFrame=true, useCrc=true).
|
|
|
|
Code byte format: num_data_bytes + (num_consecutive_zeros * 64)
|
|
This differs from standard COBS — zeros are packed into the code byte, not output inline.
|
|
"""
|
|
|
|
LSB = 1 << 6 # 64 (FrameByteCountLsb)
|
|
MAX_DATA = LSB - 1 # 63 (MaxDataBytes)
|
|
MAX_ZEROS = 255 - MAX_DATA # 192 (MaxCompressedFrameBytes)
|
|
|
|
def __init__(self):
|
|
self.frame_byte = 0x00
|
|
|
|
def encode(self, source: bytes) -> bytes:
|
|
"""Encode source bytes with CRC8 appended, using Lippert COBS variant."""
|
|
crc = Crc8.calculate(source)
|
|
data = source + bytes([crc])
|
|
|
|
output = bytearray([self.frame_byte]) # Start frame
|
|
i = 0
|
|
while i < len(data):
|
|
code_index = len(output)
|
|
output.append(0) # Placeholder for code byte
|
|
num6 = 0
|
|
|
|
# Collect non-zero data bytes (up to MAX_DATA)
|
|
while i < len(data) and num6 < self.MAX_DATA:
|
|
b = data[i]
|
|
if b == self.frame_byte:
|
|
break
|
|
output.append(b)
|
|
num6 += 1
|
|
i += 1
|
|
|
|
# Collect consecutive zero bytes, packing each as +LSB in the code byte
|
|
while i < len(data):
|
|
if data[i] != self.frame_byte:
|
|
break
|
|
num6 += self.LSB
|
|
i += 1
|
|
if num6 >= self.MAX_ZEROS:
|
|
break
|
|
|
|
output[code_index] = num6
|
|
|
|
output.append(self.frame_byte) # End frame
|
|
return bytes(output)
|
|
|
|
|
|
class CobsDecoder:
|
|
"""COBS Decoder — stateful byte-at-a-time algorithm matching CobsDecoder.cs.
|
|
|
|
Code byte encoding (numDataBits=6):
|
|
code_byte = (num_zeros_to_insert << 6) | num_data_bytes_before_trigger
|
|
When the lower 6 bits of the running code_byte reach 0, the upper bits
|
|
tell how many zero bytes to insert. This differs from standard COBS.
|
|
"""
|
|
|
|
NUM_DATA_BITS = 6
|
|
LSB = 1 << NUM_DATA_BITS # 64
|
|
MAX_DATA_MASK = LSB - 1 # 63 (0x3F)
|
|
|
|
def __init__(self):
|
|
self.frame_byte = 0x00
|
|
|
|
def decode(self, encoded: bytes) -> bytes:
|
|
"""Decode one complete COBS packet (start frame … end frame)."""
|
|
output = bytearray()
|
|
code_byte = 0
|
|
|
|
for b in encoded:
|
|
if b == self.frame_byte:
|
|
# Frame byte — either start (no data yet) or end (complete packet)
|
|
if code_byte != 0 or len(output) == 0:
|
|
# Start frame or error: reset state
|
|
output.clear()
|
|
code_byte = 0
|
|
continue
|
|
# End frame: strip CRC and return
|
|
received_crc = output[-1]
|
|
data = output[:-1]
|
|
calc_crc = Crc8.calculate(data)
|
|
if received_crc != calc_crc:
|
|
raise ValueError(f"CRC mismatch: received 0x{received_crc:02x} != calculated 0x{calc_crc:02x}")
|
|
return bytes(data)
|
|
|
|
if code_byte <= 0:
|
|
# Code byte: encodes data-count (low 6 bits) + zero-count (high bits)
|
|
code_byte = b
|
|
else:
|
|
# Data byte
|
|
code_byte -= 1
|
|
output.append(b)
|
|
|
|
# When lower NUM_DATA_BITS of code_byte hit zero, insert implied zeros
|
|
if (code_byte & self.MAX_DATA_MASK) == 0:
|
|
while code_byte > 0:
|
|
output.append(self.frame_byte)
|
|
code_byte -= self.LSB
|
|
|
|
raise ValueError("Incomplete COBS packet — no end frame received")
|