macropad-itprotech/macropad_gui/backends/itprotech.py

"""
ITPROTECH MK series backend (USB vendor id 8808). Confirmed on the MK-09,
the 9 key pad, by reading and writing it; the wire format and the reasons
behind it are in itprotech_proto.py.

Three things about this firmware shape the code:

  * One read command returns everything: every key's record, the lighting
    mode and the custom colour. There are no layers and no knobs, and a
    key is addressed by its number, 1 upwards, left to right, top to bottom.
  * It acts on every other output report. Each report goes out twice, and
    the pad does it exactly once (see REPEAT in the proto module).
  * It can be read back, so a write is checked: after sending, the key is
    read again and has to come back byte for byte as sent. A write that
    doesn't is reported as such, and the control is marked unknown.

Lighting and colour have no place in the GUI yet; they are a subcommand:

    macropad lighting show
    macropad lighting rgb "press: blue"
    macropad lighting colour ff8800
"""

import errno
import time

from . import itprotech_proto as wire
from .base import Backend


class Itprotech(Backend):
    key = "itprotech"
    name = "ITPROTECH MK series"
    vid = wire.VID
    pids = wire.CONFIRMED  # 660a; siblings are offered detection first
    layers = 1
    max_keys = 12
    max_knobs = 0
    sequences = True       # multikey: several keys, one modifier mask

    def __init__(self):
        self._data_len = {}          # device path -> key record length seen

    # ------------------------------------------------------------ protocol

    def action_byte(self, control):
        """Keys are their own number, from 1. There are no knobs."""
        if control.startswith("key"):
            return int(control[3:])
        raise ValueError(f"{self.name} pads have no knobs: {control}")

    def check(self, binding):
        try:
            wire.encode(binding, self._data_len_guess())
        except ValueError as e:
            return str(e)
        return None

    def _data_len_guess(self, device=None):
        if device is not None and device.path in self._data_len:
            return self._data_len[device.path]
        if device is not None and device.pid in wire.DATA_LEN:
            return wire.DATA_LEN[device.pid]
        return wire.DATA_LEN[wire.PID_MK09]

    # ---------------------------------------------------------------- read

    def _dump(self, device):
        """Everything the pad holds, parsed. Raises core.ReadError."""
        from .. import core
        replies = core._exchange(device.path, wire.read_request(),
                                 repeat=wire.REPEAT)
        dump = wire.parse_dump(replies)
        if not dump["keys"]:
            # One more go: a stray report from earlier can upset the pad's
            # every-other-one rhythm, and a second pair puts it right.
            replies = core._exchange(device.path, wire.read_request(),
                                     repeat=wire.REPEAT)
            dump = wire.parse_dump(replies)
        if dump["data_len"]:
            self._data_len[device.path] = dump["data_len"]
        return dump

    def info(self, device):
        """(keys, 0) as the pad reports them, or None if it says nothing."""
        dump = self._dump(device)
        if wire.consistent(dump):
            return None
        return len(dump["keys"]), 0

    def _bindings(self, dump):
        from .. import core
        out = {}
        for index, report in sorted(dump["keys"].items()):
            got = wire.decode(report)
            if got and got[1] is not None:
                out[f"key{index}"] = got[1]
        return out

    def read_layer(self, device, layer=1, layout=None):
        return self._bindings(self._dump(device))

    def lighting(self, device):
        """(mode, (r, g, b)) as stored on the pad."""
        dump = self._dump(device)
        return dump["rgb"], dump["colour"]

    # --------------------------------------------------------------- write

    def _send(self, device, report, _write=None):
        """One report, sent the way this firmware needs it: twice."""
        import macropad as proto
        write = _write or proto.write_report
        for _ in range(wire.REPEAT):
            write(device.path, report)
            time.sleep(wire.WRITE_GAP)

    def write(self, device, control, binding, state, layer=1, _write=None):
        from .. import core
        index = self.action_byte(control)
        data_len = self._data_len_guess(device)
        try:
            kind, body = wire.encode(binding, data_len)
            report = wire.key_frame(data_len, index, kind, body)
        except ValueError as e:
            raise core.WriteError(f"{control}: {e}", pad_touched=False)

        try:
            self._send(device, report, _write)
        except OSError as e:
            why = core._explain(e)
            untouched = e.errno in (errno.ENOENT, errno.ENODEV,
                                    errno.EACCES, errno.EPERM)
            if not untouched:
                state.mark_unknown(control, f"write interrupted: {why}", layer)
                state.save()
            raise core.WriteError(f"{control}: {why}",
                                  pad_touched=not untouched, cause=e)

        # The pad can be read, so check the write landed before believing it.
        try:
            dump = self._dump(device)
        except core.ReadError as e:
            state.mark_unknown(control, f"written, but couldn't read back: {e}",
                               layer)
            state.save()
            raise core.WriteError(f"{control}: sent, but the pad didn't answer "
                                  f"when asked to confirm ({e})",
                                  pad_touched=True, cause=e)
        got = dump["keys"].get(index, b"")
        want = report[1:data_len + 2]              # len byte + record
        if got[:data_len + 1] != want:
            state.mark_unknown(control, "the pad holds something other than "
                               "what was written", layer)
            state.save()
            raise core.WriteError(f"{control}: the pad didn't take the write "
                                  f"(it holds {got[:6].hex(' ') or 'nothing'})",
                                  pad_touched=True)
        state.record(control, binding, layer)
        state.save()

    def set_lighting(self, device, mode, colour=None, _write=None):
        """Lighting mode 0..14, and the custom colour when mode is 14."""
        self._send(device, wire.rgb_frame(mode), _write)
        if mode == wire.RGB_CUSTOM and colour is not None:
            self._send(device, wire.colour_frame(*colour), _write)

    # -------------------------------------------------------------- detect

    def detect(self, device):
        """
        Ask an unknown 8808 pad what it is, reading only.

        The read command is the only query there is, so the test is whether
        its answer looks like one pad: key records 1..n that agree on their
        length, plus a lighting reply. A pad that speaks something else sends
        nothing we recognise and is left alone.
        """
        from .. import core
        try:
            dump = self._dump(device)
        except core.ReadError as e:
            return core.Detection(why=f"it didn't answer: {e}")
        why = wire.consistent(dump)
        if why:
            return core.Detection(why=why)
        n = len(dump["keys"])
        if not self.supports(n, 0):
            return core.Detection(why=f"it reported {n} keys, more than this "
                                      "protocol is known to cover")
        model = wire.MODELS.get(device.pid, (None, None, f"{n} keys"))[2]
        return core.Detection(speaks=True, keys=n, knobs=0, layers=1,
                              bindings=self._bindings(dump),
                              model=f"ITPROTECH {model}")