Fix bleak 3.x compatibility and code review cleanup
- Fix scan_for_onecontrol() to use return_adv=True and adv.service_uuids (device.metadata removed in bleak 3.x) - Add missing _parse_get_devices_response() — was called but never defined, would crash on first real device response - main() now auto-connects when exactly one device is found via scan - Remove premature turn_on/off_light(5) calls from main() — device IDs unknown until get_devices() has been run against real hardware - Guard send_command() and disconnect() against unconnected state - control_awning() now takes MovementState enum instead of raw int - Bump post-get_devices sleep from 2s to 5s for real device latency Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 4.6
+58
-44
@@ -41,25 +41,24 @@ class OneControlClient:
|
||||
self.encoder = CobsEncoder()
|
||||
self.decoder = CobsDecoder()
|
||||
self._seq = 0
|
||||
self._pending_commands = {} # seq -> cmd_type
|
||||
self._pending_commands = {} # seq -> cmd_type
|
||||
|
||||
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:
|
||||
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)
|
||||
# print(f"Received: {decoded.hex()}")
|
||||
# print(f"Received raw: {decoded.hex()}")
|
||||
|
||||
# Response Structure: [Event(1)] [Seq(2)] [RespType(1)] [Payload...]
|
||||
if len(decoded) < 4:
|
||||
@@ -81,7 +80,6 @@ class OneControlClient:
|
||||
else:
|
||||
print(f"Response (Seq {seq}, Cmd {cmd_type}): {payload.hex()}")
|
||||
|
||||
# If completed, remove from pending?
|
||||
# 0x81 (129) = SuccessCompleted, 0x82 (130) = FailureCompleted
|
||||
if resp_type >= 128:
|
||||
self._pending_commands.pop(seq, None)
|
||||
@@ -93,6 +91,23 @@ class OneControlClient:
|
||||
except Exception as e:
|
||||
print(f"Error decoding: {e}")
|
||||
|
||||
def _parse_get_devices_response(self, payload: bytes):
|
||||
"""Parse GetDevices response payload and print each device entry"""
|
||||
if not payload:
|
||||
print("GetDevices: empty payload")
|
||||
return
|
||||
# Raw hex first — useful for cross-referencing against PROTOCOL_FINDINGS.md
|
||||
print(f"GetDevices raw payload: {payload.hex()}")
|
||||
# Each device entry is 4 bytes: [DeviceID, DeviceType, TableID, Status]
|
||||
# Structure inferred from protocol docs — verify against real hardware
|
||||
entry_size = 4
|
||||
for i in range(0, len(payload) - (entry_size - 1), entry_size):
|
||||
device_id = payload[i]
|
||||
device_type = payload[i + 1]
|
||||
table_id = payload[i + 2]
|
||||
status = payload[i + 3]
|
||||
print(f" Device ID={device_id} Type=0x{device_type:02x} Table={table_id} Status=0x{status:02x}")
|
||||
|
||||
def _next_seq(self) -> int:
|
||||
"""Get next sequence number"""
|
||||
self._seq = (self._seq + 1) & 0xFFFF
|
||||
@@ -100,18 +115,15 @@ class OneControlClient:
|
||||
|
||||
async def send_command(self, cmd_type: CommandType, table_id: int, payload: bytes):
|
||||
"""Send command to device"""
|
||||
# Build packet
|
||||
seq = self._next_seq()
|
||||
if not self.client or not self.client.is_connected:
|
||||
raise RuntimeError("Not connected — call connect() first")
|
||||
|
||||
# Track pending command
|
||||
seq = self._next_seq()
|
||||
self._pending_commands[seq] = cmd_type
|
||||
|
||||
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 (Seq {seq}): {packet.hex()}")
|
||||
|
||||
@@ -135,12 +147,9 @@ class OneControlClient:
|
||||
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])
|
||||
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):
|
||||
@@ -152,52 +161,57 @@ class OneControlClient:
|
||||
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)
|
||||
devices = await BleakScanner.discover(timeout=10.0, return_adv=True)
|
||||
|
||||
for device in devices:
|
||||
# Check if device advertises our service
|
||||
found = []
|
||||
for address, (device, adv) in devices.items():
|
||||
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})")
|
||||
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)
|
||||
|
||||
return devices
|
||||
if not found:
|
||||
print("No OneControl devices found.")
|
||||
return found
|
||||
|
||||
|
||||
# Example usage
|
||||
async def main():
|
||||
# Scan for devices
|
||||
await scan_for_onecontrol()
|
||||
found = await scan_for_onecontrol()
|
||||
|
||||
# Connect to specific device
|
||||
ADDRESS = "XX:XX:XX:XX:XX:XX" # Your device address
|
||||
if len(found) == 1:
|
||||
address = found[0].address
|
||||
print(f"Auto-connecting to {found[0].name} ({address})")
|
||||
elif len(found) > 1:
|
||||
print("Multiple OneControl devices found — set ADDRESS manually:")
|
||||
for d in found:
|
||||
print(f" {d.name} ({d.address})")
|
||||
return
|
||||
else:
|
||||
# No device found via scan — set address manually if you already know it
|
||||
ADDRESS = "XX:XX:XX:XX:XX:XX"
|
||||
if ADDRESS == "XX:XX:XX:XX:XX:XX":
|
||||
print("No device found. Set ADDRESS manually if you know the BT address.")
|
||||
return
|
||||
address = ADDRESS
|
||||
|
||||
client = OneControlClient(ADDRESS)
|
||||
client = OneControlClient(address)
|
||||
|
||||
try:
|
||||
print("Connecting...")
|
||||
await client.connect()
|
||||
print("Connected!")
|
||||
print("Connected! Requesting device list...")
|
||||
|
||||
# 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)
|
||||
await asyncio.sleep(5) # Wait for all response packets
|
||||
|
||||
finally:
|
||||
await client.disconnect()
|
||||
print("Disconnected.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user