lda
Verified
038e152bbc
Checks / tests (ubuntu-latest, 3.11) (push) Canceled after 0s
Checks / tests (ubuntu-latest, 3.14) (push) Canceled after 0s
Checks / tests (windows-latest, 3.11) (push) Canceled after 0s
Checks / tests (windows-latest, 3.14) (push) Canceled after 0s
Checks / package (push) Canceled after 0s
96 lines
3.6 KiB
Python
96 lines
3.6 KiB
Python
"""Samsung app packet framing, independent of Bluetooth and the desktop UI."""
|
||
|
||
from dataclasses import dataclass
|
||
|
||
EQ = dict(
|
||
zip(("Party", "Standard", "Hip-hop", "EDM", "Rock", "Latin", "House", "Reggae"), range(1, 9))
|
||
)
|
||
LIGHTS = dict(
|
||
zip(("Off", "Ambient", "Party", "Dance", "Thunder", "Star", "Lover", "Color"), range(8))
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Frame:
|
||
kind: int
|
||
sequence: int
|
||
payload: bytes
|
||
checksum_valid: bool = True
|
||
|
||
|
||
class Encoder:
|
||
"""Match CommandUtil, including its unusual signed-byte sequence rollover."""
|
||
|
||
def __init__(self):
|
||
self.sequence = 0
|
||
|
||
def encode(self, payload):
|
||
payload = bytes(payload)
|
||
if not 1 <= len(payload) <= 65535:
|
||
raise ValueError("Payload must contain 1–65535 bytes")
|
||
self.sequence = self.sequence % 127 + 1
|
||
body = bytes((1, self.sequence)) + len(payload).to_bytes(2, "big") + payload
|
||
return b"\x00\xbb" + body + bytes((sum(body) & 255,))
|
||
|
||
|
||
class Decoder:
|
||
"""Buffer split/combined reads and optionally reject unknown checksums.
|
||
|
||
Hardware captures show both checksum variants: connection/source replies
|
||
exclude the type byte, while an EQ reply includes it. Accept both observed
|
||
sums plus the observed fixed-value lighting formula. The exact firmware
|
||
paths selecting these variants are not yet known. Compatibility mode
|
||
retains mismatches but marks Frame.checksum_valid false.
|
||
A conservative 4096-byte cap prevents bogus lengths from indefinitely
|
||
hiding subsequent control frames.
|
||
"""
|
||
|
||
def __init__(self, strict_checksums=True):
|
||
self.buffer = bytearray()
|
||
self.strict_checksums = strict_checksums
|
||
|
||
def feed(self, chunk):
|
||
self.buffer.extend(chunk)
|
||
frames = []
|
||
while True:
|
||
start = self.buffer.find(b"\x00\xbb")
|
||
if start < 0:
|
||
self.buffer[:] = b"\x00" if self.buffer.endswith(b"\x00") else b""
|
||
break
|
||
del self.buffer[:start]
|
||
if len(self.buffer) < 7:
|
||
break
|
||
size = int.from_bytes(self.buffer[4:6], "big")
|
||
if self.buffer[2] not in (0, 1) or size > 4096:
|
||
del self.buffer[0]
|
||
continue
|
||
total = size + 7
|
||
if len(self.buffer) < total:
|
||
break
|
||
checksum = sum(self.buffer[3 : total - 1]) & 255
|
||
valid = self.buffer[total - 1] in (checksum, (checksum + self.buffer[2]) & 255)
|
||
# All captured 51/03 lighting reports (Ambient, Dance, Star, Lover)
|
||
# match a sum with the mode slot fixed at 1. This is an observed
|
||
# compatibility formula, not a claim about firmware implementation.
|
||
if (
|
||
self.buffer[2] == 1
|
||
and size == 3
|
||
and self.buffer[6:8] == b"\x51\x03"
|
||
and self.buffer[8] <= 7
|
||
):
|
||
lighting_checksum = (checksum - self.buffer[8] + 1) & 255
|
||
valid |= self.buffer[total - 1] == lighting_checksum
|
||
# Captured play/pause echoes sum as if the key slot were 1:
|
||
# sequence CD, payload 70/11 -> checksum 40; CE -> 41.
|
||
# Scope this rule to the observed payload bytes.
|
||
if self.buffer[2] == 1 and size == 2 and self.buffer[6:8] == b"\x70\x11":
|
||
valid |= self.buffer[total - 1] == (checksum - 0x11 + 1) & 255
|
||
if not valid and self.strict_checksums:
|
||
del self.buffer[0]
|
||
continue
|
||
frames.append(
|
||
Frame(self.buffer[2], self.buffer[3], bytes(self.buffer[6 : total - 1]), valid)
|
||
)
|
||
del self.buffer[:total]
|
||
return frames
|