# onecontrol_client.py # Lippert OneControl BLE Client # Based on reverse engineered protocol from decompiled Xamarin app import asyncio import struct from bleak import BleakClient, BleakScanner from cobs_protocol import CobsEncoder, CobsDecoder from enum import IntEnum # --------------------------------------------------------------------------- # Auth / Key-Seed Exchange (service 0010) # Reverse-engineered from Plugin.BLE.dll :: BleDeviceUnlockManager # --------------------------------------------------------------------------- AUTH_SERVICE = "00000010-0200-a58e-e411-afe28044e62c" SEED_CHAR = "00000012-0200-a58e-e411-afe28044e62c" # read seed from here KEY_CHAR = "00000013-0200-a58e-e411-afe28044e62c" # write response here RV_LINK_CYPHER = 612643285 # MyRvLinkBleGatewayScanResult.RvLinkKeySeedCypher def _tea_encrypt(cypher: int, seed: int) -> int: """Modified TEA used by Lippert OneControl for BLE auth. C# uint arithmetic truncates to 32 bits at each step — every intermediate shift/add must be masked independently, not just the final result.""" M = 0xFFFFFFFF DELTA = 2654435769 # 0x9E3779B9 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: """ Perform key-seed exchange on service 0010 before any commands on 0030. Returns True on success, False on failure. """ for attempt in range(3): try: seed_bytes = await client.read_gatt_char(SEED_CHAR) except Exception as e: print(f"Auth: failed to read seed (attempt {attempt+1}): {e}") await asyncio.sleep(5) continue # If device returns "unlocked" it's already authenticated if len(seed_bytes) == len(b"unlocked") and seed_bytes.lower() == b"unlocked": print("Auth: already unlocked") return True if len(seed_bytes) < 4: print(f"Auth: seed too short ({len(seed_bytes)} bytes), retrying...") await asyncio.sleep(5) continue seed = struct.unpack_from(">I", seed_bytes, 0)[0] # big-endian: Endian.Big=0 if seed == 0: print("Auth: seed is zero, retrying...") await asyncio.sleep(5) continue key = _tea_encrypt(RV_LINK_CYPHER, seed) key_bytes = struct.pack(">I", key) # big-endian: Endian.Big=0 print(f"Auth: seed=0x{seed:08x} → key=0x{key:08x}") try: await client.write_gatt_char(KEY_CHAR, key_bytes, response=False) await asyncio.sleep(0.5) # Verify: 0012 should now return b"unlocked" verify = await client.read_gatt_char(SEED_CHAR) print(f"Auth: verify read = {verify!r} ({verify.hex()})") if verify.lower() == b"unlocked": print("Auth: confirmed unlocked!") else: print("Auth: key written but 'unlocked' not confirmed — may still work") return True except Exception as e: print(f"Auth: failed to write key (attempt {attempt+1}): {e}") return False print("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 class OneControlClient: SERVICE_UUID = "00000030-0200-A58E-E411-AFE28044E62C" WRITE_CHAR = "00000033-0200-A58E-E411-AFE28044E62C" READ_CHAR = "00000034-0200-A58E-E411-AFE28044E62C" def __init__(self, address: str): self.address = address self.client = None self.encoder = CobsEncoder() self.decoder = CobsDecoder() self._seq = 0 self._pending_commands = {} # seq -> cmd_type async def connect(self): """Connect to OneControl device and perform key-seed auth handshake.""" self.client = BleakClient(self.address, disconnected_callback=self._on_disconnect) await self.client.connect(timeout=20.0) if not await perform_auth(self.client): await self.client.disconnect() raise RuntimeError("Auth handshake failed") await asyncio.sleep(1.0) # Give panel a moment after auth before commands await self.client.start_notify(self.READ_CHAR, self._notification_handler) def _on_disconnect(self, _client: BleakClient): print("[!] Device disconnected unexpectedly") async def disconnect(self): """Disconnect from device""" if self.client and self.client.is_connected: await self.client.disconnect() def _notification_handler(self, sender, data): """Handle notifications from device""" try: decoded = self.decoder.decode(data) # Response Structure: [Event(1)] [Seq(2)] [RespType(1)] [Payload...] if len(decoded) < 4: return event_type = decoded[0] seq = struct.unpack(">H", decoded[1:3])[0] resp_type = decoded[3] payload = decoded[4:] # resp_type: 1=SuccessMultiple, 129=SuccessCompleted, 2=FailureMultiple, 130=FailureCompleted completed = resp_type >= 128 # 2 = DeviceCommand response if event_type == 2: cmd_type = self._pending_commands.get(seq) if cmd_type: if cmd_type == CommandType.GET_DEVICES: if completed and len(payload) == 5: # Completion packet: [CRC32 (4 bytes BE)][DeviceCount (1)] crc = struct.unpack_from(">I", payload, 0)[0] count = payload[4] print(f"GetDevices complete: {count} devices, table CRC=0x{crc:08x}") else: self._parse_get_devices_response(payload) elif cmd_type == CommandType.ACTION_SWITCH: print(f"Switch Response (Seq {seq}) resp=0x{resp_type:02x}: {payload.hex()}") else: print(f"Response (Seq {seq}, Cmd {cmd_type}) resp=0x{resp_type:02x}: {payload.hex()}") if completed: self._pending_commands.pop(seq, None) else: print(f"Response for unknown Seq {seq} resp=0x{resp_type:02x}: {payload.hex()}") else: self._handle_status_event(event_type, decoded) except Exception as e: print(f"Error decoding: {e}") DEVICE_NAMES = { 4: "water pump", 5: "gas water heater", 6: "exterior lights", 7: "interior lights", 8: "grey tank 2", 9: "grey tank 1", 10: "black tank", 11: "fresh water tank", } _QUIET_EVENTS = {1, 3, 4, 26, 32} # noisy bookkeeping events suppressed from output _EVENT_NAMES = { 5: "RelayLatchingType1", 8: "DimmableLightStatus", 9: "RgbLightStatus", 11: "HvacStatus", 13: "HBridgeType1", 14: "HBridgeType2", } def _handle_status_event(self, event_type: int, decoded: bytes): """Decode and print unsolicited status broadcasts.""" if event_type == 7 and len(decoded) >= 6: # RvStatus: battery voltage + optional external temp v_raw = struct.unpack_from(">H", decoded, 1)[0] t_raw = struct.unpack_from(">H", decoded, 3)[0] features = decoded[5] if features & 1 and v_raw != 0xFFFF: voltage = v_raw / 256.0 print(f"[Battery] {voltage:.2f}V", end="") else: print(f"[Battery] N/A", end="") if features & 2 and t_raw != 0x7FFF: temp_c = struct.unpack_from(">h", decoded, 3)[0] / 256.0 print(f" [Ext Temp] {temp_c:.1f}°C / {temp_c*9/5+32:.1f}°F", end="") print() elif event_type == 12: # TankSensorStatus: [event][table][devId][pct] repeated i = 2 while i + 1 < len(decoded): dev_id = decoded[i] pct = decoded[i + 1] name = self.DEVICE_NAMES.get(dev_id, f"tank DevID={dev_id}") print(f"[Tank] {name}: {pct}%") i += 2 elif event_type == 6: # RelayBasicLatchingStatusType2: [event][table] then 7 bytes per device # [devId][state_byte][5 more bytes] i = 2 while i + 1 < len(decoded): dev_id = decoded[i] state_byte = decoded[i + 1] on = bool(state_byte & 0x01) name = self.DEVICE_NAMES.get(dev_id, f"DevID={dev_id}") print(f"[Switch] {name}: {'ON' if on else 'OFF'}") i += 7 else: if event_type not in self._QUIET_EVENTS: name = self._EVENT_NAMES.get(event_type, f"Event{event_type}") print(f"[{name}] {decoded.hex()}") def _parse_get_devices_response(self, payload: bytes): """Parse GetDevices intermediate response payload. ExtendedData layout (payload = bytes 4+ of decoded packet): [TableId (1)][StartingDeviceId (1)][DeviceCount (1)][device entries...] Each device entry: [Protocol (1)][PayloadSize (1)][PayloadSize bytes] Protocol 1 (IdsCan): PayloadSize=10 [DeviceType (1)][DeviceInstance (1)][ProductId (2 BE)][MAC (4)] Protocol 2 (Host): varies """ if not payload: print("GetDevices: empty payload") return print(f"GetDevices raw payload: {payload.hex()}") if len(payload) < 3: return table_id = payload[0] start_id = payload[1] dev_count = payload[2] print(f" TableId={table_id} StartId={start_id} DeviceCount={dev_count}") PROTOCOL_NAMES = {0: "None", 1: "IdsCan", 2: "Host"} i = 3 entry_num = start_id while i + 1 < len(payload): protocol = payload[i] payload_size = payload[i + 1] entry_data = payload[i + 2 : i + 2 + payload_size] proto_name = PROTOCOL_NAMES.get(protocol, f"Proto{protocol}") if protocol == 1 and payload_size >= 10: # IdsCan dev_type = entry_data[0] dev_instance = entry_data[1] product_id = struct.unpack_from(">H", entry_data, 2)[0] mac = entry_data[4:8].hex(":") print(f" [{entry_num}] {proto_name} DevType=0x{dev_type:02x} Instance={dev_instance} ProductId=0x{product_id:04x} MAC={mac}") else: print(f" [{entry_num}] {proto_name} data={entry_data.hex()}") i += 2 + payload_size entry_num += 1 def _next_seq(self) -> int: """Get next sequence number""" self._seq = (self._seq + 1) & 0xFFFF return self._seq async def send_command(self, cmd_type: CommandType, table_id: int, payload: bytes): """Send command to device""" if not self.client or not self.client.is_connected: raise RuntimeError("Not connected — call connect() first") 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) # Write-without-response (Command) — panel rejects write-with-response # with ATT 0x0E. See client/onecontrol_client.py for details. await self.client.write_gatt_char(self.WRITE_CHAR, encoded, response=False) print(f"Sent (Seq {seq}): raw={packet.hex()} encoded={encoded.hex()}") async def get_devices(self, start_id: int = 0, max_count: int = 255): """Get list of devices (Command 1)""" payload = bytes([start_id, max_count]) await self.send_command(CommandType.GET_DEVICES, 1, payload) async def turn_on_light(self, device_id: int): """Turn on a light/switch (Command 64)""" payload = bytes([SwitchState.ON, device_id]) await self.send_command(CommandType.ACTION_SWITCH, 1, payload) async def turn_off_light(self, device_id: int): """Turn off a light/switch (Command 64)""" payload = bytes([SwitchState.OFF, device_id]) await self.send_command(CommandType.ACTION_SWITCH, 1, payload) async def toggle_light(self, device_id: int): """Toggle a light/switch (Command 64)""" payload = bytes([SwitchState.TOGGLE, device_id]) await self.send_command(CommandType.ACTION_SWITCH, 1, payload) async def control_awning(self, device_id: int, state: MovementState): """Control awning/slide movement (Command 65)""" payload = bytes([state, device_id]) await self.send_command(CommandType.ACTION_MOVEMENT, 1, payload) async def set_dimmer(self, device_id: int, level: int): """ Set dimmable light level (Command 67) level: 0-100 """ payload = bytes([level, device_id]) await self.send_command(CommandType.ACTION_DIMMABLE, 1, payload) async def scan_for_onecontrol(): """Scan for OneControl devices""" print("Scanning for OneControl devices...") devices = await BleakScanner.discover(timeout=10.0, return_adv=True) found = [] for address, (device, adv) in devices.items(): if device.name and "LCIRemote" in device.name: print(f"Found by name: {device.name} ({address})") found.append(device) elif any("00000030-0200" in uuid.lower() for uuid in adv.service_uuids): print(f"Found by service UUID: {device.name or '(unnamed)'} ({address})") found.append(device) if not found: print("No OneControl devices found.") return found async def main(): address = "30:C6:F7:E0:80:8E" client = OneControlClient(address) try: print("Connecting...") await client.connect() print("Connected! Requesting device list...") # Panel has ~30s idle timeout — send a periodic ping to stay connected while True: await asyncio.sleep(20) await client.get_devices() finally: await client.disconnect() print("Disconnected.") if __name__ == "__main__": asyncio.run(main())