Bjorn Blomberg
625c95f09c
A backend that lists lighting_modes gets a Lighting button in the footer: one entry per mode and a colour picker for the custom one. The tick follows what the pad reported in its last read, since every read of an ITPROTECH pad carries the lighting along with the keys. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
416 lines
17 KiB
Python
416 lines
17 KiB
Python
"""
|
|
The ITPROTECH backend, against what a real MK-09 said.
|
|
|
|
REAL_DUMP is the pad's answer to the read command on 2026-10-23, fresh out
|
|
of the box: keys 1 to 9 bound to a s d f g h j k l, lighting mode 12,
|
|
custom colour white. WRITES are reports that were sent to the same pad and
|
|
read back byte for byte. A fake pad answers out of those, so the backend
|
|
is held to what the hardware actually did, not to what the vendor's code
|
|
appears to do.
|
|
"""
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import macropad as proto
|
|
from macropad_gui import backends, core
|
|
from macropad_gui.backends import itprotech_proto as wire
|
|
from macropad_gui.backends.itprotech import Itprotech
|
|
|
|
# One 64 byte input report per line, as /dev/hidraw delivered them. The
|
|
# bytes after the record are whatever the firmware had lying around.
|
|
REAL_DUMP = [bytes.fromhex(h) + bytes(64 - 16) for h in (
|
|
"0d 01 01 00 04 00 00 00 00 00 00 00 00 00 00 fd",
|
|
"0d 02 01 00 16 00 00 00 00 00 00 00 00 00 00 fd",
|
|
"0d 03 01 00 07 00 00 00 00 00 00 00 00 00 00 fd",
|
|
"0d 04 01 00 09 00 00 00 00 00 00 00 00 00 00 fd",
|
|
"0d 05 01 00 0a 00 00 00 00 00 00 00 00 00 00 fd",
|
|
"0d 06 01 00 0b 00 00 00 00 00 00 00 00 00 00 fd",
|
|
"0d 07 01 00 0d 00 00 00 00 00 00 00 00 00 00 fd",
|
|
"0d 08 01 00 0e 00 00 00 00 00 00 00 00 00 00 fd",
|
|
"0d 09 01 00 0f 00 00 00 00 00 00 00 00 00 00 fd",
|
|
"02 63 0c 00 0f 00 00 00 00 00 00 00 00 00 00 fd",
|
|
"04 64 ff ff ff 00 00 00 00 00 00 00 00 00 00 fd",
|
|
)]
|
|
|
|
# (what was written, what key 9 then read back as)
|
|
WRITES = [
|
|
("00 0d 09 01 00 10", "0d 09 01 00 10 00 00 00 00 00 00 00 00 00"), # m
|
|
("00 0d 09 03 04 e9 00 00", "0d 09 03 04 e9 00 00 00 00 00 00 00 00 00"), # vol+
|
|
("00 0d 09 04 00 0b 08 0f 0f 12", "0d 09 04 00 0b 08 0f 0f 12 00 00 00 00 00"), # hello
|
|
("00 0d 09 01 00 0f", "0d 09 01 00 0f 00 00 00 00 00 00 00 00 00"), # l again
|
|
]
|
|
|
|
|
|
def fake_device(pid="660a", path="/dev/hidraw-mk09"):
|
|
return core.Device(path=path, report_id=0, writable=True, vid="8808", pid=pid)
|
|
|
|
|
|
class FakePad:
|
|
"""
|
|
Behaves like the MK-09 on the desk: answers the read command with the
|
|
dump, acts on every OTHER output report, and stores key records.
|
|
"""
|
|
|
|
def __init__(self, replies=None):
|
|
self.records = {}
|
|
for r in (replies or REAL_DUMP):
|
|
if wire.is_key_report(r):
|
|
self.records[r[1]] = bytes(r)
|
|
self.other = [r for r in (replies or REAL_DUMP) if not wire.is_key_report(r)]
|
|
self.seen = 0 # output reports received
|
|
self.acted = [] # the ones the firmware actually processed
|
|
self.written = [] # every report, as the host sent it
|
|
|
|
def _receive(self, report):
|
|
self.seen += 1
|
|
self.written.append(bytes(report))
|
|
if self.seen % 2: # the every-other-one rule
|
|
return []
|
|
self.acted.append(bytes(report))
|
|
length, payload = report[1], report[2:]
|
|
if length == 1 and payload[0] == wire.CMD_READ:
|
|
return [self.records[i] for i in sorted(self.records)] + self.other
|
|
if length >= 4 and payload[0] not in (wire.CMD_RGB, wire.CMD_COLOUR):
|
|
record = bytes([length]) + payload[:length]
|
|
self.records[record[1]] = record + bytes(64 - len(record))
|
|
return []
|
|
|
|
def exchange(self, path, request, want=None, repeat=1):
|
|
out = []
|
|
for _ in range(repeat):
|
|
out += self._receive(request)
|
|
return out
|
|
|
|
def write_report(self, path, report):
|
|
self._receive(report)
|
|
|
|
|
|
class Frames(unittest.TestCase):
|
|
"""The bytes, before any hardware is involved."""
|
|
|
|
def test_read_request_is_what_the_vendor_sends(self):
|
|
self.assertEqual(wire.read_request()[:3], bytes.fromhex("00 01 55"))
|
|
self.assertEqual(len(wire.read_request()), 65)
|
|
|
|
def test_key_writes_match_the_pad(self):
|
|
for sent, _ in WRITES:
|
|
want = bytes.fromhex(sent)
|
|
record = want[1:]
|
|
kind, body = record[2], record[3:]
|
|
got = wire.key_frame(0x0d, record[1], kind, body)
|
|
self.assertEqual(got[:len(want)], want)
|
|
self.assertEqual(got[len(want):], bytes(65 - len(want)))
|
|
|
|
def test_lighting_frames(self):
|
|
self.assertEqual(wire.rgb_frame(12)[:4], bytes.fromhex("00 02 57 0c"))
|
|
# green first, as the vendor software sends it
|
|
self.assertEqual(wire.colour_frame(0x11, 0x22, 0x33)[:6],
|
|
bytes.fromhex("00 04 70 22 11 33"))
|
|
|
|
def test_the_bootloader_command_cannot_be_framed(self):
|
|
with self.assertRaises(ValueError):
|
|
wire.frame([wire.CMD_BOOTLOADER])
|
|
|
|
def test_a_record_that_does_not_fit_is_refused(self):
|
|
with self.assertRaises(ValueError):
|
|
wire.key_frame(0x0d, 1, wire.TYPE_MULTI, bytes(12))
|
|
|
|
|
|
class Encoding(unittest.TestCase):
|
|
|
|
def enc(self, **kw):
|
|
return wire.encode(core.Binding(**kw).validate(), 0x0d)
|
|
|
|
def test_shortcut(self):
|
|
self.assertEqual(self.enc(keys="ctrl+c"), (wire.TYPE_KEY, bytes([0x01, 0x06])))
|
|
|
|
def test_sequence_shares_one_modifier_byte(self):
|
|
self.assertEqual(self.enc(keys="h,e,l,l,o"),
|
|
(wire.TYPE_MULTI, bytes([0, 0x0b, 0x08, 0x0f, 0x0f, 0x12])))
|
|
self.assertEqual(self.enc(keys="ctrl+a,ctrl+c"),
|
|
(wire.TYPE_MULTI, bytes([0x01, 0x04, 0x06])))
|
|
|
|
def test_sequence_with_mixed_modifiers_is_refused(self):
|
|
with self.assertRaises(ValueError):
|
|
self.enc(keys="ctrl+a,c")
|
|
|
|
def test_sequence_longer_than_the_record_is_refused(self):
|
|
with self.assertRaises(ValueError):
|
|
self.enc(keys=",".join("abcdefghijk")) # 11 keys, room for 10
|
|
self.assertEqual(self.enc(keys=",".join("abcdefghij"))[0], wire.TYPE_MULTI)
|
|
|
|
def test_delay_is_refused(self):
|
|
with self.assertRaises(ValueError):
|
|
self.enc(keys="a,b", delay=50)
|
|
|
|
def test_media(self):
|
|
self.assertEqual(self.enc(media="volumeup"),
|
|
(wire.TYPE_MEDIA, bytes([0x04, 0xe9, 0x00, 0x00])))
|
|
self.assertEqual(self.enc(media="wwwhome"),
|
|
(wire.TYPE_MEDIA, bytes([0x04, 0x23, 0x02, 0x00])))
|
|
|
|
def test_nothing_is_a_keyboard_binding_with_no_key(self):
|
|
self.assertEqual(self.enc(none=True), (wire.TYPE_KEY, bytes([0, 0])))
|
|
|
|
def test_every_vendor_media_key_has_a_name_the_cli_accepts(self):
|
|
for usage, name in wire.VENDOR_MEDIA.items():
|
|
self.assertIn(name, proto.MEDIA_KEYS, name)
|
|
self.assertEqual(proto.MEDIA_KEYS[name], usage, name)
|
|
|
|
|
|
class Decoding(unittest.TestCase):
|
|
|
|
def test_the_factory_bindings(self):
|
|
got = {wire.decode(r)[0]: wire.decode(r)[1] for r in REAL_DUMP[:9]}
|
|
self.assertEqual([got[i].keys for i in range(1, 10)], list("asdfghjkl"))
|
|
|
|
def test_round_trips_of_what_was_written(self):
|
|
for sent, back in WRITES:
|
|
index, binding = wire.decode(bytes.fromhex(back))
|
|
self.assertEqual(index, 9)
|
|
kind, body = wire.encode(binding, 0x0d)
|
|
self.assertEqual(wire.key_frame(0x0d, 9, kind, body)[:len(bytes.fromhex(sent))],
|
|
bytes.fromhex(sent), binding)
|
|
|
|
def test_labels(self):
|
|
self.assertEqual(wire.decode(bytes.fromhex(WRITES[1][1]))[1].label(), "Volume up")
|
|
self.assertEqual(wire.decode(bytes.fromhex(WRITES[2][1]))[1].keys, "h,e,l,l,o")
|
|
|
|
def test_a_mouse_key_is_seen_but_not_named(self):
|
|
index, binding = wire.decode(bytes.fromhex("0d 05 02 01 00 00 00") + bytes(7))
|
|
self.assertEqual((index, binding), (5, None))
|
|
|
|
def test_an_empty_slot_is_seen_but_not_named(self):
|
|
self.assertEqual(wire.decode(bytes.fromhex("0d 05 ff") + bytes(11)), (5, None))
|
|
|
|
def test_dump_parsing(self):
|
|
dump = wire.parse_dump(REAL_DUMP)
|
|
self.assertEqual(sorted(dump["keys"]), list(range(1, 10)))
|
|
self.assertEqual(dump["data_len"], 0x0d)
|
|
self.assertEqual(dump["rgb"], 12)
|
|
self.assertEqual(dump["colour"], (255, 255, 255))
|
|
self.assertEqual(wire.consistent(dump), "")
|
|
|
|
|
|
class OnThePad(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.pad = FakePad()
|
|
self.backend = Itprotech()
|
|
patch = mock.patch.object(core, "_exchange", self.pad.exchange)
|
|
patch.start()
|
|
self.addCleanup(patch.stop)
|
|
self.dir = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.dir.cleanup)
|
|
self.state = core.State(Path(self.dir.name) / "s.json",
|
|
layout=core.KNOWN_LAYOUTS[("8808", "660a")],
|
|
device_id="8808:660a")
|
|
|
|
def test_info_counts_the_keys(self):
|
|
self.assertEqual(self.backend.info(fake_device()), (9, 0))
|
|
|
|
def test_a_silent_pad_gives_none(self):
|
|
self.pad.records.clear()
|
|
self.pad.other.clear()
|
|
self.assertIsNone(self.backend.info(fake_device()))
|
|
|
|
def test_read_layer(self):
|
|
got = self.backend.read_layer(fake_device())
|
|
self.assertEqual(got["key1"].keys, "a")
|
|
self.assertEqual(got["key9"].keys, "l")
|
|
self.assertEqual(len(got), 9)
|
|
|
|
def test_lighting(self):
|
|
self.assertEqual(self.backend.lighting(fake_device()), (12, (255, 255, 255)))
|
|
|
|
def test_the_read_command_goes_out_twice(self):
|
|
self.backend.info(fake_device())
|
|
self.assertEqual(self.pad.seen, 2)
|
|
self.assertEqual(len(self.pad.acted), 1)
|
|
|
|
def test_write_lands_once_and_is_read_back(self):
|
|
dev = fake_device()
|
|
self.backend.write(dev, "key9", core.Binding(keys="h,e,l,l,o"), self.state,
|
|
_write=self.pad.write_report)
|
|
sent = [w for w in self.pad.written if w[1] == 0x0d]
|
|
self.assertEqual(len(sent), 2, "the record goes out twice")
|
|
self.assertEqual(sent[0], sent[1])
|
|
self.assertEqual(sent[0][:10], bytes.fromhex(WRITES[2][0]))
|
|
self.assertEqual(self.pad.records[9][:14], bytes.fromhex(WRITES[2][1]))
|
|
self.assertEqual(self.state.get("key9").binding.keys, "h,e,l,l,o")
|
|
self.assertFalse(self.state.get("key9").unknown)
|
|
|
|
def test_a_write_the_pad_did_not_take_is_reported(self):
|
|
"""A pad that stores something else must not be believed."""
|
|
dev = fake_device()
|
|
real = self.pad._receive
|
|
|
|
def stubborn(report):
|
|
out = real(report)
|
|
if report[1] == 0x0d:
|
|
self.pad.records[9] = REAL_DUMP[8] # it kept 'l'
|
|
return out
|
|
self.pad._receive = stubborn
|
|
with self.assertRaises(core.WriteError) as cm:
|
|
self.backend.write(dev, "key9", core.Binding(keys="m"), self.state,
|
|
_write=self.pad.write_report)
|
|
self.assertTrue(cm.exception.pad_touched)
|
|
self.assertTrue(self.state.get("key9").unknown)
|
|
|
|
def test_an_impossible_binding_touches_nothing(self):
|
|
with self.assertRaises(core.WriteError) as cm:
|
|
self.backend.write(fake_device(), "key1", core.Binding(keys="ctrl+a,b"),
|
|
self.state, _write=self.pad.write_report)
|
|
self.assertFalse(cm.exception.pad_touched)
|
|
self.assertEqual(self.pad.seen, 0)
|
|
|
|
def test_check_explains_before_writing(self):
|
|
self.assertIsNone(self.backend.check(core.Binding(keys="ctrl+c")))
|
|
self.assertIn("modifiers", self.backend.check(core.Binding(keys="ctrl+a,b")))
|
|
|
|
def test_set_lighting(self):
|
|
self.backend.set_lighting(fake_device(), wire.RGB_CUSTOM, (0xff, 0x88, 0x00),
|
|
_write=self.pad.write_report)
|
|
self.assertEqual([a[:6] for a in self.pad.acted],
|
|
[bytes.fromhex("00 02 57 0e 00 00"), bytes.fromhex("00 04 70 88 ff 00")])
|
|
|
|
|
|
class Detecting(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.pad = FakePad()
|
|
self.backend = Itprotech()
|
|
patch = mock.patch.object(core, "_exchange", self.pad.exchange)
|
|
patch.start()
|
|
self.addCleanup(patch.stop)
|
|
|
|
def test_the_mk09_is_recognised_as_three_by_three(self):
|
|
found = self.backend.detect(fake_device())
|
|
self.assertTrue(found.speaks)
|
|
self.assertEqual((found.keys, found.knobs), (9, 0))
|
|
self.assertEqual((found.layout.rows, found.layout.cols), (3, 3))
|
|
self.assertEqual(found.bindings["key5"].keys, "g")
|
|
self.assertIn("MK-09", found.model)
|
|
|
|
def test_a_sibling_is_named_by_its_count(self):
|
|
found = self.backend.detect(fake_device(pid="66ff"))
|
|
self.assertTrue(found.speaks)
|
|
self.assertEqual(found.model, "ITPROTECH 9 keys")
|
|
|
|
def test_silence_is_not_taken_as_yes(self):
|
|
self.pad.records.clear()
|
|
self.pad.other.clear()
|
|
self.assertFalse(self.backend.detect(fake_device()).speaks)
|
|
|
|
def test_a_gap_in_the_keys_is_not_a_pad_we_know(self):
|
|
del self.pad.records[5]
|
|
found = self.backend.detect(fake_device())
|
|
self.assertFalse(found.speaks)
|
|
self.assertIn("not 1 to 8", found.why)
|
|
|
|
|
|
class Registry(unittest.TestCase):
|
|
|
|
def test_the_mk09_is_claimed_and_siblings_are_not(self):
|
|
self.assertIsInstance(backends.for_vid_pid("8808", "660a"), Itprotech)
|
|
self.assertIsNone(backends.for_vid_pid("8808", "6604"))
|
|
self.assertTrue(Itprotech().could_claim("8808", "6604"))
|
|
|
|
def test_the_udev_rule_covers_the_vendor(self):
|
|
self.assertIn('ATTRS{idVendor}=="8808"', Itprotech().udev_rule())
|
|
self.assertTrue(any('"8808"' in l for l in core.rule_lines()))
|
|
|
|
def test_sequences_are_allowed_here_and_not_on_the_ch57x(self):
|
|
self.assertTrue(Itprotech().sequences)
|
|
self.assertEqual(backends.CH57x().sequences, core.SEQUENCES_VERIFIED)
|
|
|
|
def test_known_layout(self):
|
|
lay = core.KNOWN_LAYOUTS[("8808", "660a")]
|
|
self.assertEqual((lay.keys, lay.knobs, lay.rows, lay.cols), (9, 0, 3, 3))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
class Lighting(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.pad = FakePad()
|
|
self.backend = Itprotech()
|
|
patch = mock.patch.object(core, "_exchange", self.pad.exchange)
|
|
patch.start()
|
|
self.addCleanup(patch.stop)
|
|
|
|
def test_the_backend_offers_its_modes_to_the_gui(self):
|
|
self.assertEqual(len(self.backend.lighting_modes), 15)
|
|
self.assertEqual(self.backend.lighting_modes[self.backend.lighting_custom],
|
|
"custom colour")
|
|
|
|
def test_a_read_remembers_the_lighting(self):
|
|
dev = fake_device()
|
|
self.assertIsNone(self.backend.last_lighting(dev))
|
|
self.backend.read_layer(dev)
|
|
self.assertEqual(self.backend.last_lighting(dev), (12, (255, 255, 255)))
|
|
|
|
def test_setting_updates_what_is_remembered(self):
|
|
dev = fake_device()
|
|
self.backend.read_layer(dev)
|
|
self.backend.set_lighting(dev, 3, _write=self.pad.write_report)
|
|
self.assertEqual(self.backend.last_lighting(dev), (3, (255, 255, 255)))
|
|
self.backend.set_lighting(dev, wire.RGB_CUSTOM, (1, 2, 3),
|
|
_write=self.pad.write_report)
|
|
self.assertEqual(self.backend.last_lighting(dev), (14, (1, 2, 3)))
|
|
|
|
|
|
class LightingMenu(unittest.TestCase):
|
|
"""The GUI's Lighting menu, offscreen, against the fake pad."""
|
|
|
|
def test_the_menu_appears_for_this_pad_and_writes_to_it(self):
|
|
import os
|
|
try:
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
from PySide6.QtWidgets import QApplication
|
|
from macropad_gui import app
|
|
except ImportError:
|
|
self.skipTest("PySide6 not installed")
|
|
import macropad as proto
|
|
pad = FakePad()
|
|
dev = fake_device()
|
|
ready = core.Diagnosis("ready", f"Ready on {dev.path}.", device=dev)
|
|
with tempfile.TemporaryDirectory() as d, \
|
|
mock.patch.dict(os.environ, {"XDG_CONFIG_HOME": d}), \
|
|
mock.patch.object(core, "diagnose", return_value=ready), \
|
|
mock.patch.object(core, "_exchange", pad.exchange), \
|
|
mock.patch.object(proto, "write_report", pad.write_report):
|
|
QApplication.instance() or QApplication([])
|
|
w = app.Window()
|
|
w.timer.stop()
|
|
w._read(quiet=True)
|
|
self.assertFalse(w.light_btn.isHidden())
|
|
self.assertEqual(w.state.device_id, "8808:660a")
|
|
self.assertEqual(w.state.layout.describe(), "9 keys in 3x3")
|
|
# the pad said mode 12, so that one is ticked
|
|
self.assertTrue(w.light_actions[12].isChecked())
|
|
self.assertIn("#ffffff", w.light_custom.text())
|
|
# choosing a mode sends it, twice, and ticks it
|
|
w.light_actions[3].trigger()
|
|
sent = [r for r in pad.written if r[1:3] == bytes([2, wire.CMD_RGB])]
|
|
self.assertEqual([r[3] for r in sent], [3, 3])
|
|
self.assertEqual([a[3] for a in pad.acted if a[2] == wire.CMD_RGB], [3])
|
|
self.assertTrue(w.light_actions[3].isChecked())
|
|
self.assertIn("press: green", w.message.text())
|
|
# a custom colour chosen without the dialog goes out as mode 14 + colour
|
|
w._set_lighting(wire.RGB_CUSTOM, (0x10, 0x20, 0x30))
|
|
colours = [a for a in pad.acted if a[2] == wire.CMD_COLOUR]
|
|
self.assertEqual(colours[-1][3:6], bytes([0x20, 0x10, 0x30]))
|
|
self.assertTrue(w.light_custom.isChecked())
|
|
self.assertIn("#102030", w.light_custom.text())
|
|
# no pad with lighting: the button goes away
|
|
w.diag = core.Diagnosis("unplugged", "No pad connected.")
|
|
w._follow_device()
|
|
self.assertTrue(w.light_btn.isHidden())
|