wes/lippert-onecontrol
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add HA custom integration; fix bleak 3.x write-without-response regression

Browse Source 
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>
This commit is contained in:
wes
2026-06-09 06:41:14 -04:00
co-authored by Claude Opus 4.8
parent b6d498c829
commit 34155fd7f9
14 changed files with 1207 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/lippert_onecontrol/client/__init__.py
View File
custom_components/lippert_onecontrol/client/cobs_protocol.py
+149
View File
@@ -0,0 +1,149 @@
# 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")
custom_components/lippert_onecontrol/client/onecontrol_client.py
+304
View File
@@ -0,0 +1,304 @@
# onecontrol_client.py
# Lippert OneControl BLE Client
# Based on reverse engineered protocol from decompiled Xamarin app
import asyncio
import logging
import struct
from collections.abc import Callable
from bleak import BleakClient
from bleak_retry_connector import establish_connection, BleakClientWithServiceCache
from .cobs_protocol import CobsEncoder, CobsDecoder
from enum import IntEnum, StrEnum
class EventType(StrEnum):
SWITCH = "switch"
TANK = "tank"
BATTERY = "battery"
COVER_RAW = "cover_raw"
_LOGGER = logging.getLogger(__name__)
AUTH_SERVICE = "00000010-0200-a58e-e411-afe28044e62c"
SEED_CHAR = "00000012-0200-a58e-e411-afe28044e62c"
KEY_CHAR = "00000013-0200-a58e-e411-afe28044e62c"
RV_LINK_CYPHER = 612643285
def _tea_encrypt(cypher: int, seed: int) -> int:
M = 0xFFFFFFFF
DELTA = 2654435769
num = DELTA
for _ in range(32):
t = (((cypher << 4) & M) + 1131376761) & M
t ^= (cypher + num) & M
t ^= ((cypher >> 5) + 1919510376) & M
seed = (seed + t) & M
t = (((seed << 4) & M) + 1948272964) & M
t ^= (seed + num) & M
t ^= ((seed >> 5) + 1400073827) & M
cypher = (cypher + t) & M
num = (num + DELTA) & M
return seed
async def _perform_auth(client: BleakClient) -> bool:
for attempt in range(3):
if not client.is_connected:
_LOGGER.warning("Auth: connection lost before attempt %d", attempt + 1)
return False
try:
seed_bytes = await client.read_gatt_char(SEED_CHAR)
except Exception as e:
_LOGGER.warning("Auth: failed to read seed (attempt %d): %s", attempt + 1, e)
if not client.is_connected:
return False
await asyncio.sleep(2)
continue
if seed_bytes.lower() == b"unlocked":
_LOGGER.debug("Auth: already unlocked")
return True
if len(seed_bytes) < 4:
_LOGGER.warning("Auth: seed too short (%d bytes), retrying", len(seed_bytes))
await asyncio.sleep(2)
continue
seed = struct.unpack_from(">I", seed_bytes, 0)[0]
if seed == 0:
_LOGGER.warning("Auth: seed is zero, retrying")
await asyncio.sleep(2)
continue
key = _tea_encrypt(RV_LINK_CYPHER, seed)
_LOGGER.debug("Auth: seed=0x%08x → key=0x%08x", seed, key)
try:
await client.write_gatt_char(KEY_CHAR, struct.pack(">I", key), response=False)
await asyncio.sleep(0.5)
verify = await client.read_gatt_char(SEED_CHAR)
if verify.lower() == b"unlocked":
_LOGGER.debug("Auth: confirmed unlocked")
return True
_LOGGER.warning("Auth: unexpected verify response: %r", verify)
return True # key was accepted even if verify string differs
except Exception as e:
_LOGGER.error("Auth: failed to write key (attempt %d): %s", attempt + 1, e)
return False
_LOGGER.error("Auth: all attempts exhausted")
return False
class CommandType(IntEnum):
GET_DEVICES = 1
ACTION_SWITCH = 64
ACTION_MOVEMENT = 65
ACTION_DIMMABLE = 67
ACTION_RGB = 68
ACTION_HVAC = 69
class SwitchState(IntEnum):
OFF = 0
ON = 1
TOGGLE = 2
class MovementState(IntEnum):
RETRACT = 0
EXTEND = 1
STOP = 2
# State callback type: (event_type: str, data: dict) -> None
# event_type values: "switch", "tank", "battery", "cover"
StateCallback = Callable[[str, dict], None]
class OneControlClient:
SERVICE_UUID = "00000030-0200-A58E-E411-AFE28044E62C"
WRITE_CHAR = "00000033-0200-A58E-E411-AFE28044E62C"
READ_CHAR = "00000034-0200-A58E-E411-AFE28044E62C"
_QUIET_EVENTS = {1, 3, 4, 26, 32}
def __init__(self, address: str, state_callback: StateCallback | None = None,
disconnect_callback: Callable[[], None] | None = None,
ble_device=None):
self.address = address
self._ble_device = ble_device
self._state_cb = state_callback
self._disconnect_cb = disconnect_callback
self.client: BleakClient | None = None
self.encoder = CobsEncoder()
self.decoder = CobsDecoder()
self._seq = 0
self._pending_commands: dict[int, CommandType] = {}
def set_ble_device(self, ble_device) -> None:
"""Update the BLE device reference (from HA scanner)."""
self._ble_device = ble_device
async def connect(self) -> None:
# Panel often drops the first BLE connection immediately.
# Retry the full connect→auth sequence with a fresh client each time.
# Do NOT call pair() — panel uses proprietary TEA auth, not BLE pairing,
# and the pair request may trigger the disconnect.
last_err: Exception | None = None
for attempt in range(5):
try:
if self._ble_device:
self.client = await establish_connection(
BleakClientWithServiceCache,
self._ble_device,
self.address,
disconnected_callback=self._on_disconnect,
max_attempts=2,
)
else:
self.client = BleakClient(self.address, disconnected_callback=self._on_disconnect)
await self.client.connect(timeout=30.0)
_LOGGER.info("BLE connected (attempt %d), starting auth...", attempt + 1)
# Brief settle time — just enough for BlueZ service discovery
await asyncio.sleep(0.5)
if not self.client.is_connected:
_LOGGER.info("Panel dropped connection on attempt %d — retrying", attempt + 1)
await asyncio.sleep(2.0)
continue
if not await _perform_auth(self.client):
if self.client.is_connected:
# Auth truly failed (not just disconnect) — still retry,
# the panel may need a warmup connection first
await self.client.disconnect()
_LOGGER.warning("Auth failed on attempt %d — retrying full sequence", attempt + 1)
await asyncio.sleep(2.0)
continue
await asyncio.sleep(0.5)
await self.client.start_notify(self.READ_CHAR, self._notification_handler)
_LOGGER.info("OneControl connected and authenticated (attempt %d)", attempt + 1)
return
except Exception as e:
last_err = e
_LOGGER.warning("Connect attempt %d failed: %s", attempt + 1, e)
if self.client and self.client.is_connected:
try:
await self.client.disconnect()
except Exception:
pass
await asyncio.sleep(2.0)
raise RuntimeError(f"OneControl: failed to connect after 4 attempts: {last_err}")
def _on_disconnect(self, _client: BleakClient) -> None:
_LOGGER.warning("OneControl: disconnected unexpectedly")
if self._disconnect_cb:
self._disconnect_cb()
async def disconnect(self) -> None:
if self.client and self.client.is_connected:
await self.client.disconnect()
@property
def is_connected(self) -> bool:
return self.client is not None and self.client.is_connected
def _notification_handler(self, sender, data: bytearray) -> None:
try:
decoded = self.decoder.decode(data)
if len(decoded) < 4:
return
event_type = decoded[0]
seq = struct.unpack(">H", decoded[1:3])[0]
resp_type = decoded[3]
payload = decoded[4:]
completed = resp_type >= 128
if event_type == 2:
cmd_type = self._pending_commands.get(seq)
if cmd_type == CommandType.GET_DEVICES and completed:
self._pending_commands.pop(seq, None)
elif cmd_type is not None and completed:
self._pending_commands.pop(seq, None)
else:
self._handle_status_event(event_type, decoded)
except Exception as e:
_LOGGER.debug("OneControl: error decoding notification: %s", e)
def _handle_status_event(self, event_type: int, decoded: bytes) -> None:
if event_type == 7 and len(decoded) >= 6:
v_raw = struct.unpack_from(">H", decoded, 1)[0]
features = decoded[5]
if features & 1 and v_raw != 0xFFFF:
voltage = v_raw / 256.0
self._emit(EventType.BATTERY, {"voltage": voltage})
elif event_type == 12:
i = 2
while i + 1 < len(decoded):
dev_id = decoded[i]
pct = decoded[i + 1]
self._emit(EventType.TANK, {"device_id": dev_id, "pct": pct})
i += 2
elif event_type == 6:
i = 2
while i + 1 < len(decoded):
dev_id = decoded[i]
state_byte = decoded[i + 1]
on = bool(state_byte & 0x01)
self._emit(EventType.SWITCH, {"device_id": dev_id, "on": on})
i += 7
elif event_type == 14:
# RelayHBridgeMomentaryType2 — covers/slides/awnings
# payload structure not fully decoded; pass raw for now
self._emit(EventType.COVER_RAW, {"data": decoded})
elif event_type not in self._QUIET_EVENTS:
_LOGGER.debug("OneControl: unhandled event %d: %s", event_type, decoded.hex())
def _emit(self, event_type: str, data: dict) -> None:
if self._state_cb:
self._state_cb(event_type, data)
def _next_seq(self) -> int:
self._seq = (self._seq + 1) & 0xFFFF
return self._seq
async def send_command(self, cmd_type: CommandType, table_id: int, payload: bytes) -> None:
if not self.is_connected:
raise RuntimeError("Not connected")
seq = self._next_seq()
self._pending_commands[seq] = cmd_type
packet = struct.pack(">HBB", seq, cmd_type, table_id) + payload
encoded = self.encoder.encode(packet)
# Panel's command char is a streaming (Write Without Response) endpoint,
# like the OEM app. Bleak 3.x defaults to write-with-response when the
# char advertises the "write" property, which the panel rejects with
# ATT 0x0E (Unlikely Error). Force write-without-response.
await self.client.write_gatt_char(self.WRITE_CHAR, encoded, response=False)
async def get_devices(self) -> None:
await self.send_command(CommandType.GET_DEVICES, 1, bytes([0, 255]))
async def set_switch(self, device_id: int, on: bool) -> None:
state = SwitchState.ON if on else SwitchState.OFF
await self.send_command(CommandType.ACTION_SWITCH, 1, bytes([state, device_id]))
async def control_movement(self, device_id: int, state: MovementState) -> None:
await self.send_command(CommandType.ACTION_MOVEMENT, 1, bytes([state, device_id]))
async def set_dimmer(self, device_id: int, level: int) -> None:
await self.send_command(CommandType.ACTION_DIMMABLE, 1, bytes([level, device_id]))