wes
7dd4f55a0c
✅ Protocol fully reversed from decompiled Xamarin app ✅ All 431 .NET assemblies extracted and decompiled ✅ COBS encoder/decoder implemented in Python ✅ CRC8 checksum implementation ✅ Complete BLE client for OneControl devices ✅ Comprehensive documentation Files included: - cobs_protocol.py: COBS encoding/decoding + CRC8 - onecontrol_client.py: Full BLE client implementation - Complete protocol documentation - Home Assistant integration guide - ESPHome Bluetooth Proxy setup - Extraction scripts for reference Ready for testing with RV hardware (April 2025)
122 lines
3.3 KiB
Python
122 lines
3.3 KiB
Python
# 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
|
|
|
|
from typing import List
|
|
|
|
|
|
class Crc8:
|
|
"""CRC8 with init value 0x55 (from Crc8.cs)"""
|
|
POLY = 0x07
|
|
|
|
def __init__(self, init: int = 0x55):
|
|
self.value = init
|
|
|
|
def update(self, byte: int):
|
|
self.value ^= byte
|
|
for _ in range(8):
|
|
if self.value & 0x80:
|
|
self.value = ((self.value << 1) ^ self.POLY) & 0xFF
|
|
else:
|
|
self.value = (self.value << 1) & 0xFF
|
|
|
|
def update_buffer(self, data: bytes):
|
|
for b in data:
|
|
self.update(b)
|
|
|
|
@staticmethod
|
|
def calculate(data: bytes, init: int = 0x55) -> int:
|
|
crc = Crc8(init)
|
|
crc.update_buffer(data)
|
|
return crc.value
|
|
|
|
|
|
class CobsEncoder:
|
|
"""
|
|
COBS Encoder matching DirectConnectionMyRvLinkBle.cs:44
|
|
prependStartFrame=true, useCrc=true, frameByte=0, numDataBits=6
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.frame_byte = 0x00
|
|
self.max_data_bytes = (1 << 6) - 1 # 63 bytes (6-bit)
|
|
|
|
def encode(self, source: bytes) -> bytes:
|
|
"""Encode with CRC8 and COBS"""
|
|
if not source:
|
|
return bytes([self.frame_byte])
|
|
|
|
# 1. Calculate CRC8
|
|
crc = Crc8.calculate(source)
|
|
data_with_crc = source + bytes([crc])
|
|
|
|
# 2. COBS encode
|
|
output = bytearray([self.frame_byte]) # Prepend start frame
|
|
|
|
i = 0
|
|
while i < len(data_with_crc):
|
|
code_index = len(output)
|
|
output.append(0xFF) # Placeholder for code byte
|
|
|
|
count = 0
|
|
while i < len(data_with_crc) and count < self.max_data_bytes:
|
|
byte = data_with_crc[i]
|
|
if byte == self.frame_byte:
|
|
break
|
|
output.append(byte)
|
|
i += 1
|
|
count += 1
|
|
|
|
# Update code byte
|
|
output[code_index] = count + 1
|
|
|
|
# Skip frame bytes
|
|
while i < len(data_with_crc) and data_with_crc[i] == self.frame_byte:
|
|
i += 1
|
|
|
|
return bytes(output)
|
|
|
|
|
|
class CobsDecoder:
|
|
"""COBS Decoder for responses"""
|
|
|
|
def __init__(self):
|
|
self.frame_byte = 0x00
|
|
|
|
def decode(self, encoded: bytes) -> bytes:
|
|
"""Decode COBS packet"""
|
|
if not encoded or encoded[0] != self.frame_byte:
|
|
raise ValueError("Invalid COBS packet")
|
|
|
|
output = bytearray()
|
|
i = 1 # Skip start frame
|
|
|
|
while i < len(encoded):
|
|
code = encoded[i]
|
|
if code == 0:
|
|
break
|
|
|
|
i += 1
|
|
count = code - 1
|
|
|
|
# Copy data bytes
|
|
for _ in range(count):
|
|
if i >= len(encoded):
|
|
break
|
|
output.append(encoded[i])
|
|
i += 1
|
|
|
|
# Add frame byte if not at end
|
|
if code < 0xFF and i < len(encoded):
|
|
output.append(self.frame_byte)
|
|
|
|
# Verify CRC
|
|
if len(output) > 0:
|
|
data = output[:-1]
|
|
received_crc = output[-1]
|
|
calculated_crc = Crc8.calculate(bytes(data))
|
|
if received_crc != calculated_crc:
|
|
raise ValueError(f"CRC mismatch: {received_crc:02x} != {calculated_crc:02x}")
|
|
|
|
return bytes(output[:-1]) # Remove CRC
|