soundtower-controller/controller/messages.py

"""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 "")