lippert-onecontrol/onecontrol_client.py

# 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


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

    async def connect(self):
        """Connect to OneControl device"""
        self.client = BleakClient(self.address)
        await self.client.connect()
        # Enable notifications
        await self.client.start_notify(self.READ_CHAR, self._notification_handler)

    async def disconnect(self):
        """Disconnect from device"""
        if self.client:
            await self.client.disconnect()

    def _notification_handler(self, sender, data):
        """Handle notifications from device"""
        try:
            decoded = self.decoder.decode(data)
            print(f"Received: {decoded.hex()}")
            # Parse response here
        except Exception as e:
            print(f"Error decoding: {e}")

    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"""
        # Build packet
        seq = self._next_seq()
        packet = struct.pack("<HBB", seq, cmd_type, table_id) + payload

        # Encode
        encoded = self.encoder.encode(packet)

        # Send
        await self.client.write_gatt_char(self.WRITE_CHAR, encoded)
        print(f"Sent: {packet.hex()} -> {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, position: int):
        """
        Control awning/slide movement (Command 65)
        position: 0=Retract, 1=Extend, 2=Stop
        """
        payload = bytes([position, 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)


# Scan for devices
async def scan_for_onecontrol():
    """Scan for OneControl devices"""
    print("Scanning for OneControl devices...")
    devices = await BleakScanner.discover(timeout=10.0)

    for device in devices:
        # Check if device advertises our service
        if device.name and "OneControl" in device.name:
            print(f"Found: {device.name} ({device.address})")
        # Or check UUIDs in advertisement data
        if device.metadata.get("uuids"):
            for uuid in device.metadata["uuids"]:
                if "00000030-0200" in uuid:
                    print(f"Found OneControl: {device.name} ({device.address})")

    return devices


# Example usage
async def main():
    # Scan for devices
    await scan_for_onecontrol()

    # Connect to specific device
    ADDRESS = "XX:XX:XX:XX:XX:XX"  # Your device address

    client = OneControlClient(ADDRESS)

    try:
        await client.connect()
        print("Connected!")

        # Get device list
        await client.get_devices()
        await asyncio.sleep(2)

        # Turn on light (device ID 5 as example)
        await client.turn_on_light(5)
        await asyncio.sleep(1)

        # Turn off light
        await client.turn_off_light(5)

    finally:
        await client.disconnect()


if __name__ == "__main__":
    asyncio.run(main())