Bjorn Blomberg
80c6a5da1c
Confirmed on the MK-09 (8808:660a, 9 keys): the vendor's .NET configurator was decompiled for the wire format and every frame was then written to a real pad and read back. Keys, media keys and multikey sequences are supported; lighting from the command line. The firmware acts on every other output report, so each report is sent twice (as the vendor software does) and every write is verified by reading the key back. The app now keeps one state file per pad and starts with a known layout for a known USB id, and the editor asks the backend whether a binding fits before offering to write it. The actions listener and the udev rule cover every supported vendor. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1013 lines
38 KiB
Python
1013 lines
38 KiB
Python
"""
|
|
Core of the macro pad GUI. No Qt in here: the GUI imports this, and so do
|
|
the tests, so everything that matters can be checked without a display.
|
|
|
|
The pad is write-only, which shapes the whole design. There are two kinds
|
|
of truth and they live in separate files:
|
|
|
|
profile what you *want* on the pad. You edit it; the GUI edits it.
|
|
state what this tool last wrote *successfully*. Only the writer
|
|
touches it. It is our shadow copy of the pad's memory.
|
|
|
|
A control we never wrote, or whose write failed partway, is UNKNOWN. The
|
|
pad could hold anything there, and the UI must say so rather than guess.
|
|
|
|
Every byte on the wire still comes from macropad.py. This module decides
|
|
what to send and remembers what was sent; it never builds a report itself.
|
|
"""
|
|
|
|
import errno
|
|
import glob
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from dataclasses import dataclass, field, replace
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
try:
|
|
import macropad as proto
|
|
except ImportError: # running from inside the package dir
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
import macropad as proto
|
|
|
|
|
|
# ------------------------------------------------------------------ device
|
|
|
|
VID, PID = "1189", "8840"
|
|
DEVICE_ID = f"{VID}:{PID}"
|
|
|
|
DIAL_ACTIONS = ("left", "push", "right")
|
|
WRITE_GAP = 0.02 # same pause the CLI leaves between reports
|
|
|
|
|
|
def backends():
|
|
"""
|
|
The backend registry. Imported late because the backends import this
|
|
module for its data types, and a top level import either way would be
|
|
circular.
|
|
"""
|
|
from . import backends as registry
|
|
return registry
|
|
|
|
|
|
def backend_for(device):
|
|
"""Whichever family owns this device; the default when there is none."""
|
|
return backends().for_device(device)
|
|
|
|
|
|
def action_byte(control, backend=None):
|
|
"""
|
|
The slot number one control lives in.
|
|
|
|
Slot numbering belongs to the protocol, so this is the backend's answer.
|
|
The default is the CH57x family, which is what every caller that does
|
|
not have a device in hand wants.
|
|
"""
|
|
return (backend or backends().DEFAULT).action_byte(control)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Layout:
|
|
"""
|
|
The shape of one pad: how many keys and knobs, and how the keys are
|
|
arranged when you look at it.
|
|
|
|
keys and knobs come from the pad itself (the 0xFB query). Rows and
|
|
columns do not: the pad knows how many keys it has, not how they are
|
|
laid out, so that part is either known from the model or chosen by
|
|
whoever owns it.
|
|
|
|
Rows and columns describe the pad lying flat, the way its keys are
|
|
printed. Standing it upright just turns the drawing.
|
|
"""
|
|
keys: int = 12
|
|
knobs: int = 2
|
|
rows: int = 3
|
|
cols: int = 4
|
|
name: str = ""
|
|
places: tuple = () # ((row, col), ...) per key; empty means left to right
|
|
knob_side: str = "right" # which side of the keys the knobs sit on
|
|
|
|
def __post_init__(self):
|
|
if not 1 <= self.keys <= proto.KEY_SLOTS:
|
|
raise ValueError(f"a pad can have 1 to {proto.KEY_SLOTS} keys")
|
|
if not 0 <= self.knobs <= proto.KNOB_SLOTS:
|
|
raise ValueError(f"a pad can have up to {proto.KNOB_SLOTS} knobs")
|
|
if self.knob_side not in ("left", "right"):
|
|
raise ValueError("knobs sit on the left or the right")
|
|
if self.rows * self.cols < self.keys:
|
|
raise ValueError(f"{self.rows}x{self.cols} can't hold "
|
|
f"{self.keys} keys")
|
|
if self.places:
|
|
if len(self.places) != self.keys:
|
|
raise ValueError("places must give a spot for every key")
|
|
if len(set(self.places)) != len(self.places):
|
|
raise ValueError("two keys can't share a spot")
|
|
for r, c in self.places:
|
|
if not (0 <= r < self.rows and 0 <= c < self.cols):
|
|
raise ValueError(f"({r}, {c}) is outside {self.rows}x{self.cols}")
|
|
|
|
def spots(self):
|
|
"""[(key_id, row, col)] with the pad lying flat."""
|
|
if self.places:
|
|
return [(f"key{i + 1}", r, c) for i, (r, c) in enumerate(self.places)]
|
|
return [(f"key{i + 1}", i // self.cols, i % self.cols)
|
|
for i in range(self.keys)]
|
|
|
|
def rearranged(self, places):
|
|
"""The same pad with its keys in different spots."""
|
|
return replace(self, places=tuple((int(r), int(c)) for r, c in places))
|
|
|
|
def with_knobs_on(self, side):
|
|
"""
|
|
The same pad with its knobs down the other side.
|
|
|
|
Sides are named for the pad lying flat. Standing it upright turns
|
|
the whole board, so knobs on the right end up along the top and
|
|
knobs on the left along the bottom.
|
|
"""
|
|
return replace(self, knob_side=side)
|
|
|
|
def resized(self, rows=None, cols=None, keys=None, knobs=None):
|
|
"""
|
|
Change the grid, keeping any arrangement that still fits and
|
|
falling back to left to right for keys that no longer do.
|
|
"""
|
|
rows = self.rows if rows is None else rows
|
|
cols = self.cols if cols is None else cols
|
|
keys = self.keys if keys is None else keys
|
|
knobs = self.knobs if knobs is None else knobs
|
|
if rows * cols < keys: # grow to fit rather than refuse
|
|
rows = -(-keys // cols)
|
|
|
|
taken, places = set(), []
|
|
old = {i: p for i, p in enumerate(self.places)}
|
|
free = [(r, c) for r in range(rows) for c in range(cols)]
|
|
for i in range(keys):
|
|
spot = old.get(i)
|
|
if spot and spot[0] < rows and spot[1] < cols and spot not in taken:
|
|
places.append(spot)
|
|
taken.add(spot)
|
|
else:
|
|
places.append(None)
|
|
for i, spot in enumerate(places):
|
|
if spot is None:
|
|
nxt = next(p for p in free if p not in taken)
|
|
places[i] = nxt
|
|
taken.add(nxt)
|
|
return replace(self, rows=rows, cols=cols, keys=keys, knobs=knobs,
|
|
places=tuple(places) if self.places else ())
|
|
|
|
@property
|
|
def key_ids(self):
|
|
return tuple(f"key{i}" for i in range(1, self.keys + 1))
|
|
|
|
@property
|
|
def dial_ids(self):
|
|
return tuple(f"dial{i}" for i in range(1, self.knobs + 1))
|
|
|
|
@property
|
|
def control_ids(self):
|
|
return self.key_ids + tuple(f"{d}-{a}" for d in self.dial_ids
|
|
for a in DIAL_ACTIONS)
|
|
|
|
@property
|
|
def control_for_action(self):
|
|
return {action_byte(c): c for c in self.control_ids}
|
|
|
|
def describe(self):
|
|
knobs = (f" and {self.knobs} knob{'s' if self.knobs != 1 else ''}"
|
|
if self.knobs else "")
|
|
return f"{self.keys} keys in {self.rows}x{self.cols}{knobs}"
|
|
|
|
def to_json(self):
|
|
d = {"keys": self.keys, "knobs": self.knobs,
|
|
"rows": self.rows, "cols": self.cols, "name": self.name,
|
|
"knob_side": self.knob_side}
|
|
if self.places:
|
|
d["places"] = [list(p) for p in self.places]
|
|
return d
|
|
|
|
@classmethod
|
|
def from_json(cls, d):
|
|
# knob_side is newer than the file format. Files written before it
|
|
# existed describe pads with knobs on the right, which was the only
|
|
# thing the drawing could do.
|
|
return cls(keys=int(d["keys"]), knobs=int(d["knobs"]),
|
|
rows=int(d["rows"]), cols=int(d["cols"]),
|
|
name=d.get("name", ""),
|
|
knob_side=d.get("knob_side", "right"),
|
|
places=tuple(tuple(p) for p in d.get("places", ())))
|
|
|
|
|
|
# Pads we have laid eyes on. Anything else has to be detected or described
|
|
# by its owner, because a USB id alone does not pin down a layout: 1189:8840
|
|
# covers several different pads.
|
|
KNOWN_LAYOUTS = {
|
|
("1189", "8840"): Layout(12, 2, 3, 4, "12 keys, 2 knobs"),
|
|
("8808", "660a"): Layout(9, 0, 3, 3, "ITPROTECH MK-09, 9 keys"),
|
|
}
|
|
DEFAULT_LAYOUT = KNOWN_LAYOUTS[(VID, PID)]
|
|
|
|
# Kept so older code and the CLI keep working; the app uses a Layout.
|
|
KEYS = DEFAULT_LAYOUT.key_ids
|
|
DIALS = DEFAULT_LAYOUT.dial_ids
|
|
CONTROL_IDS = DEFAULT_LAYOUT.control_ids
|
|
CONTROL_FOR_ACTION = DEFAULT_LAYOUT.control_for_action
|
|
|
|
for _cid in CONTROL_IDS:
|
|
assert action_byte(_cid) == proto.CONTROLS[_cid], _cid
|
|
|
|
_DIAL_WORDS = {"left": "turn left", "push": "press", "right": "turn right"}
|
|
|
|
|
|
def control_name(control):
|
|
"""'key4' -> 'Key 4', 'dial1-left' -> 'Dial 1, turn left'."""
|
|
if control.startswith("key"):
|
|
return f"Key {control[3:]}"
|
|
dial, act = control.split("-")
|
|
return f"Dial {dial[4:]}, {_DIAL_WORDS[act]}"
|
|
|
|
|
|
# Media keys the GUI offers, in the order it offers them. Every id must be a
|
|
# name macropad.MEDIA_KEYS knows; the usage codes themselves live there.
|
|
MEDIA = [
|
|
("playpause", "Play / pause"), ("next", "Next track"),
|
|
("prev", "Previous track"), ("stop", "Stop"),
|
|
("volumeup", "Volume up"), ("volumedown", "Volume down"), ("mute", "Mute"),
|
|
("brightnessup", "Brightness up"), ("brightnessdown", "Brightness down"),
|
|
("calculator", "Calculator"), ("mail", "Email"),
|
|
("browser", "Browser"), ("mycomputer", "My computer"),
|
|
]
|
|
MEDIA_LABEL = dict(MEDIA)
|
|
for _m, _ in MEDIA:
|
|
assert _m in proto.MEDIA_KEYS, f"macropad.py has no media key {_m}"
|
|
_MEDIA_BY_USAGE = {}
|
|
for _m, _ in MEDIA:
|
|
_MEDIA_BY_USAGE.setdefault(proto.MEDIA_KEYS[_m], _m)
|
|
|
|
# Verified on 1189:8840 so far: single keys and shortcuts (capture + use),
|
|
# media keys (capture). Sequences and the delay field are upstream's layout
|
|
# and have not been seen on this pad - the GUI keeps them locked until a
|
|
# capture confirms them.
|
|
SEQUENCES_VERIFIED = False
|
|
|
|
_MOD_ORDER = [("ctrl", 0x01), ("shift", 0x02), ("alt", 0x04), ("super", 0x08),
|
|
("rctrl", 0x10), ("rshift", 0x20), ("ralt", 0x40), ("rwin", 0x80)]
|
|
_MOD_LABEL = {"ctrl": "Ctrl", "shift": "Shift", "alt": "Alt", "super": "Meta",
|
|
"win": "Meta", "lctrl": "Ctrl", "lshift": "Shift", "lalt": "Alt",
|
|
"lwin": "Meta", "rctrl": "Right Ctrl", "rshift": "Right Shift",
|
|
"ralt": "AltGr", "rwin": "Right Meta"}
|
|
_KEY_LABEL = {"pageup": "Page Up", "pagedown": "Page Down", "printscreen": "Print",
|
|
"scrolllock": "Scroll Lock", "capslock": "Caps Lock",
|
|
"backspace": "Backspace", "leftbracket": "[", "rightbracket": "]",
|
|
"backslash": "\\", "semicolon": ";", "quote": "'", "grave": "`",
|
|
"comma": ",", "period": ".", "slash": "/", "minus": "-", "equal": "=",
|
|
"esc": "Esc", "escape": "Esc", "return": "Enter"}
|
|
_KEYNAME_BY_CODE = {}
|
|
for _name, _code in proto.KEYCODES.items():
|
|
_KEYNAME_BY_CODE.setdefault(_code, _name)
|
|
|
|
|
|
def normalize_keys(text):
|
|
"""' Ctrl + C ,' -> 'ctrl+c'. Doesn't validate; Binding does that."""
|
|
alias = {"meta": "super"} # Plasma says Meta; macropad.py says super
|
|
steps = []
|
|
for chunk in text.lower().split(","):
|
|
parts = [alias.get(p.strip(), p.strip()) for p in chunk.split("+") if p.strip()]
|
|
if parts:
|
|
steps.append("+".join(parts))
|
|
return ",".join(steps)
|
|
|
|
|
|
def pretty_step(step):
|
|
parts = []
|
|
for p in step.split("+"):
|
|
p = p.strip().lower()
|
|
if p in _MOD_LABEL:
|
|
parts.append(_MOD_LABEL[p])
|
|
else:
|
|
parts.append(_KEY_LABEL.get(p, p.upper() if len(p) <= 3 else p.title()))
|
|
return "+".join(parts)
|
|
|
|
|
|
# ----------------------------------------------------------------- binding
|
|
|
|
@dataclass(frozen=True)
|
|
class Binding:
|
|
"""
|
|
What one control does. Stored in the CLI's own syntax, so every entry
|
|
in a profile is exactly a `macropad.py set <control> <binding>` command.
|
|
|
|
Binding(keys="ctrl+c")
|
|
Binding(keys="a,b,c", delay=50)
|
|
Binding(media="volumeup")
|
|
Binding(none=True) the control does nothing
|
|
|
|
The kind is explicit rather than guessed from the string, because
|
|
'pause' is both a keyboard key and a media key.
|
|
"""
|
|
keys: str = ""
|
|
media: str = ""
|
|
delay: int = 0
|
|
none: bool = False
|
|
|
|
def validate(self):
|
|
if sum(map(bool, (self.keys, self.media, self.none))) != 1:
|
|
raise ValueError("a binding is exactly one of keys, media or nothing")
|
|
if self.media and self.media not in proto.MEDIA_KEYS:
|
|
raise ValueError(f"unknown media key '{self.media}'")
|
|
if self.keys:
|
|
proto.parse_keystroke(self.keys) # raises ValueError
|
|
if not 0 <= self.delay <= 0xFFFF:
|
|
raise ValueError("delay must be 0-65535 ms")
|
|
if self.delay and not self.keys:
|
|
raise ValueError("delay only applies to key sequences")
|
|
return self
|
|
|
|
@property
|
|
def steps(self):
|
|
return len(proto.parse_keystroke(self.keys)) if self.keys else 1
|
|
|
|
@property
|
|
def is_nothing(self):
|
|
return self.none
|
|
|
|
def reports(self, report_id, action, layer=1):
|
|
"""The exact reports to send, built entirely by macropad.py."""
|
|
self.validate()
|
|
if self.none:
|
|
# A keystroke with no modifier and no key: count 0, pair 00 00.
|
|
# This is ch57x-keyboard-tool's encoding for a modifier-only key,
|
|
# with no modifier. Needs confirming on hardware (see README).
|
|
return proto.native_key(report_id, action, [(0, 0)], layer, 0)
|
|
if self.media:
|
|
return proto.native_media(report_id, action,
|
|
proto.MEDIA_KEYS[self.media], layer)
|
|
return proto.native_key(report_id, action,
|
|
proto.parse_keystroke(self.keys),
|
|
layer, self.delay)
|
|
|
|
@property
|
|
def is_sequence(self):
|
|
return bool(self.keys) and self.steps > 1
|
|
|
|
def label(self):
|
|
"""What a person reads on the key: 'Ctrl+Shift+N', 'Volume up'."""
|
|
if self.none:
|
|
return "Nothing"
|
|
if self.media:
|
|
return MEDIA_LABEL.get(self.media, self.media)
|
|
text = ", ".join(pretty_step(s) for s in self.keys.split(",") if s.strip())
|
|
return text + (f" ({self.delay} ms)" if self.delay else "")
|
|
|
|
def to_json(self):
|
|
if self.none:
|
|
return {"none": True}
|
|
if self.media:
|
|
return {"media": self.media}
|
|
d = {"keys": self.keys}
|
|
if self.delay:
|
|
d["delay"] = self.delay
|
|
return d
|
|
|
|
@classmethod
|
|
def from_json(cls, d):
|
|
return cls(keys=d.get("keys", ""), media=d.get("media", ""),
|
|
delay=int(d.get("delay", 0)),
|
|
none=bool(d.get("none", False))).validate()
|
|
|
|
|
|
# ------------------------------------------------------------------- files
|
|
|
|
def config_dir():
|
|
base = os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config")
|
|
return Path(base) / "macropad"
|
|
|
|
|
|
def _atomic_write_json(path, data):
|
|
"""Write-then-rename, so a crash mid-save never leaves half a file."""
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=path.name, suffix=".tmp")
|
|
try:
|
|
with os.fdopen(fd, "w") as f:
|
|
json.dump(data, f, indent=2, sort_keys=True)
|
|
f.write("\n")
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(tmp, path)
|
|
except BaseException:
|
|
try:
|
|
os.unlink(tmp)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
|
|
|
|
FORMAT = 2 # 1 had no layout, and one build saved rows and cols swapped
|
|
|
|
|
|
@dataclass
|
|
class Entry:
|
|
"""One control's line in the shadow state."""
|
|
binding: Binding = None # what we believe is on the pad
|
|
written: str = "" # ISO timestamp of the successful write
|
|
unknown: bool = True # True = pad contents not known
|
|
reason: str = "not written by this app yet"
|
|
|
|
|
|
class State:
|
|
"""
|
|
Shadow copy of the pad. Keyed by (layer, control). Only the writer
|
|
should call record()/mark_unknown(); the GUI just reads it.
|
|
"""
|
|
|
|
def __init__(self, path=None, layout=None, device_id=None):
|
|
# The default name is the one the app has always used, so upgrading
|
|
# never orphans somebody's bindings. A second family would pass its
|
|
# own id and get its own file.
|
|
self.device_id = device_id or DEVICE_ID
|
|
default = config_dir() / f"state-{self.device_id.replace(':', '-')}.json"
|
|
self.path = Path(path) if path else default
|
|
self.entries = {} # (layer, control) -> Entry
|
|
self.layout = layout or DEFAULT_LAYOUT
|
|
if self.path.exists():
|
|
self._load()
|
|
|
|
def _load(self):
|
|
data = json.loads(self.path.read_text())
|
|
version = data.get("format")
|
|
if version not in (1, FORMAT):
|
|
raise ValueError(f"{self.path}: unsupported format {version}")
|
|
# Format 1 files either have no layout or have one from the build that
|
|
# stored rows and columns the wrong way round. Ignore it either way;
|
|
# the bindings are still good.
|
|
if version == FORMAT and data.get("layout"):
|
|
self.layout = Layout.from_json(data["layout"])
|
|
for layer, controls in data.get("layers", {}).items():
|
|
for control, e in controls.items():
|
|
if e.get("unknown"):
|
|
entry = Entry(unknown=True, reason=e.get("reason", ""))
|
|
else:
|
|
entry = Entry(binding=Binding.from_json(e["binding"]),
|
|
written=e.get("written", ""), unknown=False,
|
|
reason="")
|
|
self.entries[(int(layer), control)] = entry
|
|
|
|
def save(self):
|
|
layers = {}
|
|
for (layer, control), e in sorted(self.entries.items()):
|
|
if e.unknown:
|
|
d = {"unknown": True, "reason": e.reason}
|
|
else:
|
|
d = {"binding": e.binding.to_json(), "written": e.written}
|
|
layers.setdefault(str(layer), {})[control] = d
|
|
_atomic_write_json(self.path, {"format": FORMAT, "device": self.device_id,
|
|
"layout": self.layout.to_json(),
|
|
"layers": layers})
|
|
|
|
def get(self, control, layer=1):
|
|
return self.entries.get((layer, control), Entry())
|
|
|
|
def record(self, control, binding, layer=1):
|
|
self.entries[(layer, control)] = Entry(
|
|
binding=binding, unknown=False, reason="",
|
|
written=datetime.now().isoformat(timespec="seconds"))
|
|
|
|
def mark_unknown(self, control, reason, layer=1):
|
|
self.entries[(layer, control)] = Entry(unknown=True, reason=reason)
|
|
|
|
|
|
# ---------------------------------------------------------------- diagnosis
|
|
|
|
@dataclass
|
|
class Device:
|
|
path: str
|
|
report_id: int
|
|
writable: bool
|
|
name: str = ""
|
|
vid: str = VID
|
|
pid: str = PID
|
|
|
|
|
|
@dataclass
|
|
class Diagnosis:
|
|
status: str # ready | unplugged | no-interface | no-permission | unsupported
|
|
headline: str
|
|
device: Device = None
|
|
detail: list = field(default_factory=list)
|
|
fix: list = field(default_factory=list) # works in bash, zsh and fish
|
|
warnings: list = field(default_factory=list)
|
|
|
|
@property
|
|
def ready(self):
|
|
return self.status == "ready"
|
|
|
|
|
|
RULES_DIRS = ("/etc/udev/rules.d", "/run/udev/rules.d")
|
|
SYSTEM_RULES = ("/usr/lib/udev/rules.d", "/lib/udev/rules.d")
|
|
SUGGESTED_RULE = "60-macropad.rules"
|
|
# Vendor wide on purpose. Pads in this family have several product ids, and
|
|
# a rule tied to one of them leaves everybody else unable to read their own
|
|
# hardware, which is exactly the point at which they give up.
|
|
def rule_lines():
|
|
"""One uaccess rule per vendor we support."""
|
|
return [b.udev_rule() for b in backends().REGISTRY]
|
|
|
|
|
|
RULE_LINE = (f'KERNEL=="hidraw*", ATTRS{{idVendor}}=="{VID}", '
|
|
f'TAG+="uaccess"')
|
|
|
|
|
|
def rule_command():
|
|
"""
|
|
The shell line that installs access rules for every family we support.
|
|
One vendor today, so it reads the same as it always has; a second
|
|
backend adds its own line here rather than needing the text edited.
|
|
"""
|
|
lines = rule_lines()
|
|
if len(lines) == 1:
|
|
return f"echo '{lines[0]}' | sudo tee /etc/udev/rules.d/{SUGGESTED_RULE}"
|
|
body = "\\n".join(lines)
|
|
return f"printf '{body}\\n' | sudo tee /etc/udev/rules.d/{SUGGESTED_RULE}"
|
|
|
|
|
|
|
|
def _uaccess_cutoff():
|
|
"""
|
|
The rules file that turns a uaccess tag into an actual ACL. On systemd
|
|
this is 73-seat-late.rules. udev runs every rules file in one lexical
|
|
order by filename, so a rule that adds the tag must sort before this one
|
|
- 99-foo.rules is tagged but never granted, and so is an unnumbered
|
|
foo.rules, since letters sort after digits.
|
|
"""
|
|
for d in SYSTEM_RULES:
|
|
for f in sorted(glob.glob(os.path.join(d, "*.rules"))):
|
|
try:
|
|
if 'RUN{builtin}+="uaccess"' in open(f).read():
|
|
return os.path.basename(f)
|
|
except OSError:
|
|
continue
|
|
return "73-seat-late.rules"
|
|
|
|
|
|
def _matching_rules():
|
|
"""User rule files that mention our vendor id and uaccess."""
|
|
hits = []
|
|
for d in RULES_DIRS:
|
|
for f in sorted(glob.glob(os.path.join(d, "*.rules"))):
|
|
try:
|
|
text = open(f).read()
|
|
except OSError:
|
|
continue
|
|
for line in text.splitlines():
|
|
s = line.strip()
|
|
if s.startswith("#"):
|
|
continue
|
|
if "uaccess" in s and any(
|
|
re.search(rf'idVendor\}}=="{v}"', s, re.I) for v in vendors()):
|
|
hits.append(f)
|
|
break
|
|
return hits
|
|
|
|
|
|
def _udev_tags(devpath):
|
|
try:
|
|
out = subprocess.run(["udevadm", "info", "--query=property",
|
|
f"--name={devpath}"],
|
|
capture_output=True, text=True, timeout=3).stdout
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return None
|
|
tags = set()
|
|
for line in out.splitlines():
|
|
if line.startswith(("TAGS=", "CURRENT_TAGS=")):
|
|
tags |= {t for t in line.split("=", 1)[1].split(":") if t}
|
|
return tags
|
|
|
|
|
|
def vendors():
|
|
"""Every USB vendor id any backend knows about."""
|
|
return {b.vid.lower() for b in backends().REGISTRY}
|
|
|
|
|
|
def discover():
|
|
"""Config-capable hidraw nodes belonging to any family we support."""
|
|
return proto.find_devices(vendors())
|
|
|
|
|
|
def _on_usb_bus():
|
|
want = vendors()
|
|
for d in glob.glob("/sys/bus/usb/devices/*"):
|
|
try:
|
|
v = open(os.path.join(d, "idVendor")).read().strip().lower()
|
|
except OSError:
|
|
continue
|
|
if v in want:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _keyd_warning():
|
|
"""keyd sits between the pad and the desktop and can rewrite its output."""
|
|
if not shutil.which("keyd"):
|
|
return None
|
|
try:
|
|
active = subprocess.run(["systemctl", "is-active", "keyd"],
|
|
capture_output=True, text=True,
|
|
timeout=3).stdout.strip() == "active"
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return None
|
|
if not active:
|
|
return None
|
|
for f in glob.glob("/etc/keyd/*.conf"):
|
|
try:
|
|
if DEVICE_ID in open(f).read().lower():
|
|
return (f"keyd, a key remapping service, is running and {f} "
|
|
"lists this pad. It can change keys on their way from "
|
|
"the pad to your desktop, so a key may not do exactly "
|
|
"what you set here. If you only set keyd up for this "
|
|
"pad, you can switch it off with: "
|
|
"sudo systemctl disable --now keyd")
|
|
except OSError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def on_linux():
|
|
return sys.platform.startswith("linux")
|
|
|
|
|
|
def diagnose(keyd=True):
|
|
"""keyd=False skips the keyd check, which shells out - the GUI polls."""
|
|
if not on_linux():
|
|
return Diagnosis(
|
|
"unsupported-os", f"MacroPad can't reach USB devices on {sys.platform}.",
|
|
detail=["The window works anywhere Qt does, but the part that talks "
|
|
"to the pad reads Linux's /dev/hidraw, which this system "
|
|
"doesn't have.",
|
|
"On Windows, the pad's own vendor software or "
|
|
"ch57x-keyboard-tool will configure it. Porting this to "
|
|
"Windows and macOS means swapping one module, and it's on "
|
|
"the roadmap."])
|
|
|
|
warnings = [w for w in (_keyd_warning() if keyd else None,) if w]
|
|
everything = discover()
|
|
# A device is "known" only when some backend is confirmed on its exact
|
|
# id. A sibling under the same vendor shows up as unsupported, which is
|
|
# what puts the detect button in front of its owner.
|
|
found = [d for d in everything
|
|
if backends().for_vid_pid(d["vid"], d["pid"])]
|
|
|
|
if not found and everything:
|
|
other = everything[0]
|
|
dev = Device(path=other["path"], report_id=other["report_id"],
|
|
writable=other["writable"], name=other.get("name", ""),
|
|
vid=other["vid"], pid=other["pid"])
|
|
d = Diagnosis(
|
|
"unsupported",
|
|
f"Found a {other['vid']}:{other['pid']} pad, which isn't one this "
|
|
"app knows.",
|
|
device=dev,
|
|
detail=["Pads in this family differ in both layout and protocol, "
|
|
"so the app won't write to one it hasn't identified.",
|
|
"It can ask the pad about itself first. That only reads, "
|
|
"and if the pad doesn't answer properly nothing else "
|
|
"happens."],
|
|
warnings=warnings)
|
|
if not dev.writable:
|
|
d.detail = ["This pad isn't readable yet. An older permission rule "
|
|
"may name a single product id; this one covers every "
|
|
"pad from the same maker.",
|
|
"Run these, then unplug the pad and plug it back in."]
|
|
d.fix = [rule_command(),
|
|
"sudo udevadm control --reload-rules && sudo udevadm trigger"]
|
|
return d
|
|
|
|
if not found:
|
|
if _on_usb_bus():
|
|
return Diagnosis(
|
|
"no-interface",
|
|
"Pad is plugged in, but its configuration interface is missing.",
|
|
detail=["The USB device is present, but no hidraw node with the "
|
|
"vendor-defined config interface was found.",
|
|
"Usually a replug fixes this. If not, run "
|
|
"macropad-probe.py and check the interface list."],
|
|
warnings=warnings)
|
|
return Diagnosis("unplugged", "No pad connected.",
|
|
detail=["Looking for " + ", ".join(
|
|
f"{v}:{'/'.join(p) or '*'}" for v, p, _ in backends().known_ids())
|
|
+ " on USB."],
|
|
warnings=warnings)
|
|
|
|
d = found[0]
|
|
dev = Device(path=d["path"], report_id=d["report_id"],
|
|
writable=d["writable"], name=d.get("name", ""),
|
|
vid=d["vid"], pid=d["pid"])
|
|
|
|
if dev.writable:
|
|
return Diagnosis("ready", f"Ready on {dev.path}.", device=dev,
|
|
detail=[f"report id {dev.report_id}"], warnings=warnings)
|
|
|
|
# Not writable: work out which of the three usual causes it is.
|
|
cutoff = _uaccess_cutoff()
|
|
rules = _matching_rules()
|
|
late = [r for r in rules if os.path.basename(r) >= cutoff]
|
|
early = [r for r in rules if os.path.basename(r) < cutoff]
|
|
tags = _udev_tags(dev.path)
|
|
tagged = bool(tags and "uaccess" in tags)
|
|
|
|
diag = Diagnosis("no-permission", f"{dev.path} isn't writable.",
|
|
device=dev, warnings=warnings)
|
|
|
|
if late and not early:
|
|
bad = late[0]
|
|
new = os.path.join(os.path.dirname(bad), SUGGESTED_RULE)
|
|
diag.detail = [
|
|
f"{bad} matches the pad, but it runs after {cutoff}, which is "
|
|
"where access is actually granted. The device gets tagged, "
|
|
"then nothing acts on the tag.",
|
|
"Rename it so it sorts earlier."]
|
|
diag.fix = [f"sudo mv {bad} {new}",
|
|
"sudo udevadm control --reload-rules && sudo udevadm trigger",
|
|
"# then unplug and replug the pad"]
|
|
elif early and not tagged:
|
|
diag.detail = [f"{early[0]} looks right, but the device hasn't picked "
|
|
"it up yet."]
|
|
diag.fix = ["sudo udevadm control --reload-rules && sudo udevadm trigger",
|
|
"# then unplug and replug the pad"]
|
|
elif early and tagged:
|
|
diag.detail = [
|
|
"The rule matched and the device is tagged, but no access was "
|
|
"granted. That happens when you aren't in an active local desktop "
|
|
"session - for example over SSH, or in a shell started with sudo."]
|
|
else:
|
|
diag.detail = ["No udev rule lets you write to the pad yet. Running "
|
|
"the installer again adds it, or paste these commands "
|
|
"into a terminal:"]
|
|
diag.fix = [rule_command(),
|
|
"sudo udevadm control --reload-rules && sudo udevadm trigger",
|
|
"# then unplug and replug the pad"]
|
|
return diag
|
|
|
|
|
|
# ------------------------------------------------------------------ writer
|
|
|
|
class WriteError(Exception):
|
|
"""
|
|
pad_touched is the thing the UI cares about: False means the pad is
|
|
exactly as it was; True means some reports landed and the control's
|
|
contents are now unknown.
|
|
"""
|
|
def __init__(self, message, pad_touched, cause=None):
|
|
super().__init__(message)
|
|
self.pad_touched = pad_touched
|
|
self.cause = cause
|
|
|
|
|
|
def _explain(err):
|
|
return {
|
|
errno.ENOENT: "the pad was unplugged",
|
|
errno.ENODEV: "the pad was unplugged",
|
|
errno.EACCES: "no permission to write to the pad",
|
|
errno.EPERM: "no permission to write to the pad",
|
|
errno.EPIPE: "the pad rejected the report",
|
|
errno.EIO: "the pad stopped responding (it may have reset)",
|
|
errno.ETIMEDOUT: "the pad didn't accept the report in time",
|
|
}.get(err.errno, err.strerror or str(err))
|
|
|
|
|
|
def write_binding(device, control, binding, state, layer=1, _write=None):
|
|
"""
|
|
Send one binding and update the shadow state. Raises WriteError; on
|
|
success the state file is already saved.
|
|
"""
|
|
return backend_for(device).write(device, control, binding, state, layer,
|
|
_write)
|
|
|
|
|
|
# --------------------------------------------------------------- decoding
|
|
|
|
def decode_info(report, backend=None):
|
|
"""(keys, knobs) from an info reply, or None."""
|
|
return (backend or backends().DEFAULT).decode_info(report)
|
|
|
|
|
|
def decode_config(report, layout=None, backend=None):
|
|
"""
|
|
The inverse of Binding.reports()[0]: (control, layer, Binding) from a
|
|
config report, or None if it isn't one we understand. Used to import
|
|
captures and to prove round trips in the tests.
|
|
"""
|
|
return (backend or backends().DEFAULT).decode_config(report, layout)
|
|
|
|
|
|
def is_commit(report):
|
|
return len(report) >= 4 and report[1] == proto.MAGIC_NATIVE and \
|
|
(report[2], report[3]) == proto.TERMINATOR
|
|
|
|
|
|
def read_capture(path, incoming=False):
|
|
"""
|
|
Reports from a macropad-capture.py output file, in order.
|
|
|
|
Captures can now hold both directions. By default we return only what
|
|
was sent TO the pad, since that is what a binding looks like; pass
|
|
incoming=True for what the pad sent back.
|
|
"""
|
|
out = []
|
|
want = " in " if incoming else " out "
|
|
keep = False
|
|
for line in Path(path).read_text().splitlines():
|
|
if line.startswith("["):
|
|
keep = want in f" {line.split(']', 1)[-1].strip()} "
|
|
elif line.startswith(" ") and keep:
|
|
try:
|
|
out.append(bytes.fromhex(line.strip()))
|
|
except ValueError:
|
|
continue
|
|
return out
|
|
|
|
|
|
def import_capture(path, state, layout=None):
|
|
"""
|
|
Record what a vendor-software session wrote, so the GUI knows what's on
|
|
the pad. Only bindings followed by a commit count - the vendor app can
|
|
send many configs and save once. Returns [(control, layer, Binding)].
|
|
"""
|
|
pending, done = [], []
|
|
for report in read_capture(path):
|
|
if is_commit(report):
|
|
done.extend(pending)
|
|
pending = []
|
|
continue
|
|
decoded = decode_config(report, layout or state.layout)
|
|
if decoded:
|
|
pending = [p for p in pending if p[:2] != decoded[:2]] + [decoded]
|
|
for control, layer, binding in done:
|
|
state.record(control, binding, layer)
|
|
if done:
|
|
state.save()
|
|
return done
|
|
|
|
|
|
# ---------------------------------------------------------------- reading
|
|
|
|
READ_TIMEOUT = 1.5 # total, per request
|
|
READ_QUIET = 0.25 # replies have stopped once this much silence passes
|
|
|
|
|
|
class ReadError(Exception):
|
|
pass
|
|
|
|
|
|
def _exchange(path, request, want=None, repeat=1):
|
|
"""
|
|
Send one request and collect replies until the pad goes quiet.
|
|
|
|
repeat sends the same request that many times, WRITE_GAP apart, for
|
|
firmware that acts on every other report (the ITPROTECH pads).
|
|
|
|
Reading is the one place we write to the pad without changing it: both
|
|
commands here are queries, byte for byte what the vendor software sends.
|
|
"""
|
|
import select
|
|
|
|
try:
|
|
fd = os.open(path, os.O_RDWR | os.O_NONBLOCK)
|
|
except OSError as e:
|
|
raise ReadError(_explain(e)) from e
|
|
|
|
replies = []
|
|
try:
|
|
while select.select([fd], [], [], 0)[0]: # drain anything stale
|
|
try:
|
|
os.read(fd, 128)
|
|
except OSError:
|
|
break
|
|
try:
|
|
for i in range(repeat):
|
|
if i:
|
|
time.sleep(WRITE_GAP)
|
|
os.write(fd, request)
|
|
except OSError as e:
|
|
raise ReadError(_explain(e)) from e
|
|
|
|
deadline = time.monotonic() + READ_TIMEOUT
|
|
last = time.monotonic()
|
|
while time.monotonic() < deadline:
|
|
if not select.select([fd], [], [], 0.05)[0]:
|
|
if replies and time.monotonic() - last > READ_QUIET:
|
|
break
|
|
continue
|
|
try:
|
|
data = os.read(fd, 128)
|
|
except BlockingIOError:
|
|
continue
|
|
except OSError as e:
|
|
raise ReadError(_explain(e)) from e
|
|
if not data:
|
|
continue
|
|
last = time.monotonic()
|
|
replies.append(data)
|
|
if want is not None and len(replies) >= want:
|
|
break
|
|
finally:
|
|
os.close(fd)
|
|
return replies
|
|
|
|
|
|
def read_info(device):
|
|
"""(keys, knobs) as the pad reports them, or None if it doesn't answer."""
|
|
return backend_for(device).info(device)
|
|
|
|
|
|
def read_layer(device, layer=1, layout=None):
|
|
"""{control: Binding} for one layer, as it is on the pad right now."""
|
|
return backend_for(device).read_layer(device, layer, layout)
|
|
|
|
|
|
def read_into_state(device, state, layer=1, layout=None):
|
|
"""Read the pad and make the shadow state match it. Returns what was read."""
|
|
bindings = read_layer(device, layer, layout or state.layout)
|
|
if not bindings:
|
|
raise ReadError("the pad didn't answer")
|
|
for control, binding in bindings.items():
|
|
state.record(control, binding, layer)
|
|
state.save()
|
|
return bindings
|
|
|
|
|
|
# -------------------------------------------------------------- detection
|
|
|
|
@dataclass
|
|
class Detection:
|
|
"""What a read-only interrogation of an unrecognised pad turned up."""
|
|
speaks: bool = False # it answered our protocol properly
|
|
keys: int = 0
|
|
knobs: int = 0
|
|
layers: int = 0
|
|
bindings: dict = field(default_factory=dict)
|
|
why: str = "" # when speaks is False, what went wrong
|
|
grid: tuple = () # (rows, cols) when the pad reports geometry
|
|
model: str = "" # what the pad calls itself, if it says
|
|
|
|
@property
|
|
def layout(self):
|
|
"""
|
|
The shape to draw. Use the pad's own geometry when it reports any,
|
|
and guess from the key count when it doesn't.
|
|
|
|
The CH57x only reports counts, so for that family this is a guess
|
|
the owner should confirm. Pads in it are wider than they are tall,
|
|
so the guess is the squarest split that keeps rows no greater than
|
|
columns: 6 keys give 2x3, 12 give 3x4, 15 give 3x5. The old
|
|
version had this the wrong way round and drew every pad on its
|
|
side.
|
|
"""
|
|
if not self.speaks:
|
|
return None
|
|
if self.grid and self.grid[0] * self.grid[1] == self.keys:
|
|
rows, cols = self.grid
|
|
return Layout(self.keys, self.knobs, rows, cols, self.model)
|
|
rows = max(r for r in range(1, int(self.keys ** 0.5) + 1)
|
|
if self.keys % r == 0)
|
|
return Layout(self.keys, self.knobs, rows, self.keys // rows,
|
|
self.model)
|
|
|
|
|
|
def detect(device):
|
|
"""
|
|
Ask an unknown pad what it is, without configuring anything. Which
|
|
questions get asked is the backend's business; what comes back is a
|
|
Detection either way.
|
|
"""
|
|
return backend_for(device).detect(device)
|
|
|
|
|
|
def detection_report(device, detected):
|
|
"""A block of text to paste into an issue when adding a new pad."""
|
|
from . import __version__
|
|
backend = backend_for(device)
|
|
lines = [f"MacroPad {__version__}",
|
|
f"Device: {device.vid}:{device.pid}",
|
|
f"Name: {device.name or 'not reported'}",
|
|
f"Node: {device.path}, report id {device.report_id}",
|
|
f"Tried: {backend.name} ({backend.key})", ""]
|
|
if not detected.speaks:
|
|
lines += [f"This pad did NOT answer the {backend.name} protocol.",
|
|
f"Reason: {detected.why}"]
|
|
return "\n".join(lines)
|
|
lines += [f"The pad reports {detected.keys} keys and {detected.knobs} knobs.",
|
|
f"It answered a layer read for every slot that implies.", "",
|
|
"Layer 1 as read from the pad:"]
|
|
for control, binding in sorted(detected.bindings.items(),
|
|
key=lambda kv: action_byte(kv[0])):
|
|
lines.append(f" 0x{action_byte(control):02x} {control:<12} {binding.label()}")
|
|
return "\n".join(lines)
|