Add MX-T40 BLE desktop controller and protocol documentation

controller/messages.py

@@ -0,0 +1,136 @@
+"""Translate Samsung payloads into activity text without claiming ACK = applied.
+
+Outgoing settings and incoming reports have different field layouts. Unknown
+commands stay explicitly unmapped; raw packet bytes remain available separately.
+"""
+
+from collections import deque
+
+from controller.protocol import EQ, LIGHTS
+
+SOURCES = {1: "Bluetooth", 2: "USB 1", 3: "USB 2", 4: "AUX 1", 5: "AUX 2"}
+SOUND_TYPES = {
+    1: "volume",
+    2: "subwoofer volume",
+    3: "EQ preset",
+    4: "bass boost",
+    5: "DJ effect",
+    6: "tempo",
+    7: "mute",
+}
+REMOTE_KEYS = {
+    1: "Power",
+    2: "Cycle lighting",
+    7: "Cycle EQ preset",
+    15: "Volume up",
+    16: "Volume down",
+    17: "Play / pause",
+    18: "Next track",
+    19: "Previous track",
+    20: "Toggle mute",
+    21: "Toggle bass boost",
+}
+
+
+def name_for(mapping, value):
+    return next((name for name, code in mapping.items() if code == value), f"Unknown ({value})")
+
+
+def describe_reply(data):
+    """Read TYPE/VALUE/VALUE2 from reports, not the outgoing OPTION layout."""
+    if len(data) >= 4 and data[0] == 0x41:
+        kind, value = data[1:3]
+        if kind == 3:
+            return "EQ: " + name_for(EQ, value)
+        if kind == 1:
+            return f"Volume: {value}"
+        if kind == 2:
+            return f"Subwoofer volume: {value}"
+        if kind == 4:
+            return "Bass boost: " + {0: "On", 1: "Off"}.get(value, f"Unknown ({value})")
+        if kind == 7:
+            return "Mute: " + {1: "On", 2: "Off"}.get(value, f"Unknown ({value})")
+    if len(data) >= 2 and data[0] == 0x31:
+        return "Source: " + SOURCES.get(data[1], f"Unknown ({data[1]})")
+    if len(data) >= 3 and data[:2] == b"\x51\x03":
+        return "Lighting: " + name_for(LIGHTS, data[2])
+    if len(data) >= 2 and data[0] == 0x70:
+        # A remote-key echo is an action event, not an absolute playback state.
+        return "Button event: " + REMOTE_KEYS.get(data[1], f"Unknown key ({data[1]})")
+    if len(data) >= 7 and data[0] == 2 and len(data) >= 7 + data[5]:
+        sources = ", ".join(
+            SOURCES.get(value, f"Unknown ({value})") for value in data[6 : 6 + data[5]]
+        )
+        return "Speaker identified. Available sources: " + sources
+    return "Unmapped speaker reply: " + data.hex(" ")
+
+
+def describe_command(data):
+    """Describe the requested action; success must come from a later report."""
+    if data == b"\x01":
+        return "Request speaker connection"
+    if data == b"\x04":
+        return "Complete speaker handshake"
+    if data == b"\x32":
+        return "Read audio source"
+    if len(data) >= 2 and data[0] == 0x42:
+        return "Read " + SOUND_TYPES.get(data[1], f"sound setting {data[1]}")
+    if data == b"\x52\x03":
+        return "Read lighting mode"
+    if len(data) >= 2 and data[0] == 0x70:
+        return REMOTE_KEYS.get(data[1], f"Remote key {data[1]}")
+    if len(data) >= 2 and data[0] == 0x30:
+        return "Set source to " + SOURCES.get(data[1], f"Unknown ({data[1]})")
+    if len(data) >= 5 and data[0] == 0x40:
+        if data[1] == 3:
+            return "Set EQ to " + name_for(EQ, data[3])
+        if data[1] == 4:
+            return "Set bass boost to " + {0: "On", 1: "Off"}.get(data[3], f"Unknown ({data[3]})")
+    if len(data) >= 3 and data[:2] == b"\x50\x03":
+        return "Set lighting to " + name_for(LIGHTS, data[2])
+    if len(data) >= 5 and data[:2] == b"\x60\x02":
+        return f"Set lighting color to #{data[2]:02X}{data[3]:02X}{data[4]:02X}"
+    return "Unmapped command: " + data.hex(" ")
+
+
+class ActivityLog:
+    """Bounded readable/debug history with ACKs matched to request sequences."""
+
+    def __init__(self, limit=300):
+        self.entries = deque(maxlen=limit)
+        self.pending = {}
+
+    def record(self, kind, value):
+        if kind == "frame":
+            if value.kind == 0:
+                request = self.pending.pop(value.sequence, f"request {value.sequence}")
+                self.entries.append(("Speaker acknowledged: " + request, None))
+            elif value.payload:
+                self.entries.append(("Speaker reports: " + describe_reply(value.payload), None))
+            return
+        if kind != "log":
+            return
+        if value.startswith(("TX BLE ", "TX ")):
+            raw = value.removeprefix("TX BLE ").removeprefix("TX ")
+            try:
+                packet = bytes.fromhex(raw)
+                if len(packet) >= 7:
+                    description = describe_command(packet[6:-1])
+                    self.pending[packet[3]] = description
+                    self.entries.append(("Sending: " + description, value))
+                    return
+            except ValueError:
+                pass
+        if value.startswith(("RX BLE ", "RX ")) and not value.startswith("RX checksum"):
+            self.entries.append((None, value))
+        else:
+            self.entries.append((value, None))
+
+    def text(self, raw=False):
+        lines = []
+        for readable, packet in self.entries:
+            if readable:
+                lines.append(readable)
+            if raw and packet:
+                lines.append("  " + packet)
+        return "\n".join(lines) + ("\n" if lines else "")