Reorganize repository structure into logical folders

Structure:
- src/ - Python implementation (cobs_protocol.py, onecontrol_client.py)
- docs/ - All documentation markdown files
- scripts/ - Extraction scripts (for reference only)

Changes:
- Moved Python files to src/
- Moved all .md docs to docs/
- Moved extraction scripts to scripts/
- Updated README.md with new structure
- Updated import paths in README examples
- Added placeholder for future Quartz documentation URL

Benefits:
- Cleaner repository organization
- Easier to navigate
- Separates code from documentation
- Follows standard project conventions
This commit is contained in:
wes
2025-12-29 09:33:20 -05:00
parent c62fbf1bd4
commit 718a13c02f
11 changed files with 29 additions and 19 deletions
+170
View File
@@ -0,0 +1,170 @@
# 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())