deck.py

import io

import hid
from PIL import Image

import devices


def encode(img, profile, quality=90):
    """Resize + rotate for this device's profile, then JPEG-encode."""
    if img.size != profile.image_size:
        img = img.resize(profile.image_size)
    if profile.rotation:
        # PIL rotates counter-clockwise; profile.rotation is degrees clockwise.
        img = img.rotate(-profile.rotation, expand=True)
    buf = io.BytesIO()
    img.save(buf, format="JPEG", quality=quality)
    return buf.getvalue()


def open(profile=None):
    """Resolve a device profile (auto-detect/config/prompt) and open interface 0."""
    profile = profile or devices.resolve()
    if profile is None:
        raise SystemExit("No supported device found.")
    for d in hid.enumerate(profile.vid, profile.pid):
        if d["interface_number"] == 0:
            dev = hid.device()
            dev.open_path(d["path"])
            return Deck(dev, profile)
    raise SystemExit(f"{profile.name}: interface 0 not found")


def _pad(b, length):
    return b + bytes(length - len(b))


# Input report event codes (byte 9 of a 64-byte read), verified against
# mirajazz's src/inputs.rs for this device family.
INPUT_CODES = {
    0x00: "all_released",
    0x01: "key1", 0x02: "key2", 0x03: "key3",
    0x04: "key4", 0x05: "key5", 0x06: "key6",
    0x25: "scene_a", 0x30: "scene_b", 0x31: "scene_c",
    0x90: "knob_a_ccw", 0x91: "knob_a_cw",
    0x50: "knob_b_ccw", 0x51: "knob_b_cw",
    0x60: "knob_c_ccw", 0x61: "knob_c_cw",
    0x33: "knob_a_press", 0x35: "knob_b_press", 0x34: "knob_c_press",
}


def decode_input(report):
    """Maps a raw input report to an event name, or None if not a known event.

    This device reports protocol_version 2, which mirajazz's DeviceStateReader
    treats as "single state" - every report is a discrete click, there's no
    separate press/release pair to track (matches what NOTES.md observed).
    """
    if len(report) < 10:
        return None
    return INPUT_CODES.get(report[9])


class Deck:
    """A handle to one open device. Key numbers are 1-based and match the
    wire values directly (unlike upstream, which stores 0-based indices and
    adds +1 on the wire) - do not add another +1 here."""

    def __init__(self, dev, profile):
        self.dev = dev
        self.profile = profile

    def _send_command(self, magic, extra=b""):
        payload = _pad(b"CRT\x00\x00" + magic + extra, self.profile.packet_size)
        self.dev.write(b"\x00" + payload)

    def clear_all(self):
        # verified against mirajazz (github.com/4ndv/mirajazz): 3 zero bytes sit
        # between "CLE" and the key byte, not zero. Our old layout put 0xFF right
        # after "CLE" with no gap, so the real key-byte slot was always 0x00 from
        # padding - explains why clear always hit every key regardless of the byte.
        self._send_command(b"CLE", bytes([0x00, 0x00, 0x00, 0xFF]))
        self._send_command(b"STP")

    def clear_key(self, key):
        # No STP here - confirmed against mirajazz's clear_button_image, which
        # sends only CLE for a single key. STP is only for clear_all (0xFF);
        # sending it after a per-key CLE was committing a broader flush that
        # blanked every key instead of just this one (found on hardware while
        # generalizing to the device family - see docs/NOTES.md).
        self._send_command(b"CLE", bytes([0x00, 0x00, 0x00, key]))

    def set_brightness(self, percent):
        """Screen brightness, 0-100. Layout per mirajazz: LIG + 2 zero bytes + pct."""
        self._send_command(b"LIG", bytes([0x00, 0x00, max(0, min(100, percent))]))

    def read_input(self, timeout_ms=5000):
        data = self.dev.read(64, timeout_ms=timeout_ms)
        return decode_input(data) if data else None

    def send_image(self, key, jpeg_bytes, flush_total=8192):
        # ponytail: pad with trailing zero bytes (uncounted in the declared size,
        # so they don't affect JPEG decoding) to scrub stale bytes left in a shared
        # staging buffer by earlier oversized test writes, which otherwise leak
        # onto a neighboring key on commit. Confirmed still required on hardware
        # during the device-family generalization: removing it made clear_key()
        # blank every key instead of just the target one (test_clear_key.py).
        # Don't drop this again without re-running that test.
        size = len(jpeg_bytes)
        header_extra = bytes([(size >> 8) & 0xFF, size & 0xFF, key])
        self._send_command(b"BAT\x00\x00", header_extra)

        payload = jpeg_bytes
        if flush_total and flush_total > len(payload):
            payload = payload + bytes(flush_total - len(payload))

        offset = 0
        while offset < len(payload):
            chunk = _pad(payload[offset: offset + self.profile.packet_size], self.profile.packet_size)
            self.dev.write(b"\x00" + chunk)
            offset += self.profile.packet_size

        self._send_command(b"STP")

    def close(self):
        self.dev.close()