#!/usr/bin/env python3 """ Framework Laptop 16 LED Matrix + ANSI RGB keyboard socket server. Linux-only, standard-library-only daemon. Devices: 0 left Framework LED Matrix (9 x 34 grayscale) 1 right Framework LED Matrix (9 x 34 grayscale) 2 Framework ANSI RGB keyboard (29 x 6 sparse RGB grid, 97 exact LEDs) All socket coordinates use a bottom-left origin: x increases rightward and y increases upward. The keyboard's 29 x 6 coordinate space is sparse because a keyboard is not a rectangle; use {"op":"keyboard_layout"} to discover valid coordinates. Exact keyboard LED IDs are also supported. The two LED Matrix modules use their standard USB serial protocol. True per-LED keyboard control requires Framework's HID LampArray/Dynamic Lighting firmware (the experimental Framework QMK branch named fl16-2025-hidlamp). If that interface is not present, the daemon still runs and the two matrices keep working; keyboard requests return a descriptive error. The server accepts newline-delimited JSON over TCP or a Unix-domain socket. Every request receives one newline-delimited JSON response. Keyboard animation writes are rate-limited and delta-compressed because the experimental firmware has a five-report request queue, while a full 97-LED frame requires thirteen HID reports. Matrix examples (old "module" addressing remains compatible): {"op":"pixel","module":0,"x":4,"y":10,"brightness":255} {"op":"pixel","device":1,"x":4,"y":10,"brightness":255} {"op":"clear","module":"all"} Keyboard examples: {"op":"pixel","device":2,"x":0,"y":0,"r":255,"g":0,"b":0} {"op":"pixels","device":2,"pixels":[[0,0,255,0,0],[2,0,0,255,0]]} {"op":"fill","device":2,"r":32,"g":0,"b":80} {"op":"led","device":2,"led":0,"r":255,"g":255,"b":255} {"op":"leds","device":2,"leds":[[0,255,0,0],[1,0,255,0]]} {"op":"keyboard_layout"} {"op":"keyboard_ascii"} {"op":"status"} """ from __future__ import annotations import argparse import errno import fcntl import glob import json import logging import os from pathlib import Path import signal import socketserver import struct import sys import termios import threading import time import tty from typing import Any, Iterable, Sequence MATRIX_WIDTH = 9 MATRIX_HEIGHT = 34 MATRIX_PIXEL_COUNT = MATRIX_WIDTH * MATRIX_HEIGHT KEYBOARD_DEVICE_INDEX = 2 KEYBOARD_GRID_WIDTH = 29 KEYBOARD_GRID_HEIGHT = 6 SERVER_VERSION = "2.5.0-stable-matrix-commits" FRAMEWORK_VID = "32ac" LED_MATRIX_PID = "0020" ANSI_KEYBOARD_PID = "0012" MATRIX_MAGIC = bytes((0x32, 0xAC)) MATRIX_CMD_BRIGHTNESS = 0x00 MATRIX_CMD_SLEEP = 0x03 MATRIX_CMD_STAGE_GREY_COLUMN = 0x07 MATRIX_CMD_FLUSH_GREY_COLUMNS = 0x08 LAMPARRAY_REPORT_ATTRIBUTES = 0x01 LAMPARRAY_REPORT_MULTI_UPDATE = 0x04 LAMPARRAY_REPORT_RANGE_UPDATE = 0x05 LAMPARRAY_REPORT_CONTROL = 0x06 LAMPARRAY_COMPLETE = 0x01 LAMPARRAY_MULTI_COUNT = 8 LAMPARRAY_LED_COUNT = 97 DEFAULT_KEYBOARD_REPORT_GAP_MS = 3.0 MAX_REQUEST_BYTES = 1024 * 1024 LOG = logging.getLogger("framework-input-server") # Official Framework ANSI RGB LED positions, in firmware LED-ID order. These # QMK coordinates have x=0..224 and y=0..64. On the installed ANSI keyboard, # increasing raw y runs physically downward, so the raw row index is inverted # when exposed through the socket API. The API therefore consistently uses a # bottom-left origin for the keyboard and both LED matrices. x / 8 gives an # intuitive half-key-ish 0..28 grid. The firmware's HID LampArray interface uses # the same LED IDs; keeping this fallback table also avoids 97 startup queries. ANSI_LED_POSITIONS: tuple[tuple[int, int], ...] = ( (11,23),(0,23),(57,22),(41,22),(73,22),(25,22),(121,22),(89,22),(105,22), (82,10),(114,10),(66,10),(50,10),(34,10),(98,10),(18,10),(2,10),(130,10), (56,0),(40,0),(72,0),(24,0),(88,0),(10,0),(104,0),(0,0),(120,0), (38,47),(22,47),(54,47),(5,48),(70,47),(1,48),(86,47),(0,60),(102,47), (30,34),(62,34),(7,35),(94,34),(110,34),(46,34),(13,36),(78,34),(1,36), (8,60),(38,59),(54,59),(174,34),(126,34),(142,34),(158,34),(190,34), (205,34),(202,22),(219,22),(223,36),(185,22),(137,22),(153,22),(169,22), (162,10),(194,10),(178,10),(209,11),(222,11),(181,0),(136,0),(168,0), (146,10),(199,0),(186,0),(213,0),(224,0),(152,0),(166,47),(118,47), (134,47),(150,47),(182,47),(197,48),(202,58),(203,48),(217,48),(150,59), (108,60),(121,60),(134,60),(166,59),(183,64),(202,62),(220,64),(222,48), (22,59),(70,60),(83,60),(95,60), ) # Nearest-row centers in the QMK physical coordinate system, bottom to top. ANSI_ROW_CENTERS = (0, 11, 23, 35, 48, 61) # Human-readable Framework US ANSI key geometry from the keyboard's QMK # info.json. The x values and widths are in QMK keyboard-layout units. Rows are # already expressed in the socket API's bottom-left coordinate system. # # A key may contain more than one independently addressable LED coordinate. # For example, Space and the Shift keys have several emitters. The half-height # Up and Down arrow keys collapse to the same approximate (x, y) coordinate in # the daemon's intentionally coarse 29 x 6 grid; exact LED addressing remains # available for clients that need to distinguish them. ANSI_LAYOUT_X_SCALE = 12.8 ANSI_KEY_ROWS: dict[int, tuple[tuple[str, float, float], ...]] = { 5: ( ("Esc", 0.0, 1.25), ("F1", 1.5, 1.0), ("F2", 2.75, 1.0), ("F3", 4.0, 1.0), ("F4", 5.25, 1.0), ("F5", 6.5, 1.0), ("F6", 7.75, 1.0), ("F7", 9.0, 1.0), ("F8", 10.25, 1.0), ("F9", 11.5, 1.0), ("F10", 12.75, 1.0), ("F11", 14.0, 1.0), ("F12", 15.25, 1.0), ("Delete", 16.5, 1.75), ), 4: ( ("Grave", 0.0, 1.0), ("1", 1.25, 1.0), ("2", 2.5, 1.0), ("3", 3.75, 1.0), ("4", 5.0, 1.0), ("5", 6.25, 1.0), ("6", 7.5, 1.0), ("7", 8.75, 1.0), ("8", 10.0, 1.0), ("9", 11.25, 1.0), ("0", 12.5, 1.0), ("Minus", 13.75, 1.0), ("Equals", 15.0, 1.0), ("Backspace", 16.25, 2.0), ), 3: ( ("Tab", 0.0, 1.5), ("Q", 1.75, 1.0), ("W", 3.0, 1.0), ("E", 4.25, 1.0), ("R", 5.5, 1.0), ("T", 6.75, 1.0), ("Y", 8.0, 1.0), ("U", 9.25, 1.0), ("I", 10.5, 1.0), ("O", 11.75, 1.0), ("P", 13.0, 1.0), ("LBracket", 14.25, 1.0), ("RBracket", 15.5, 1.0), ("Backslash", 16.75, 1.5), ), 2: ( ("Caps", 0.0, 1.75), ("A", 2.0, 1.0), ("S", 3.25, 1.0), ("D", 4.5, 1.0), ("F", 5.75, 1.0), ("G", 7.0, 1.0), ("H", 8.25, 1.0), ("J", 9.5, 1.0), ("K", 10.75, 1.0), ("L", 12.0, 1.0), ("Semicolon", 13.25, 1.0), ("Quote", 14.5, 1.0), ("Enter", 15.75, 2.5), ), 1: ( ("LShift", 0.0, 2.5), ("Z", 2.75, 1.0), ("X", 4.0, 1.0), ("C", 5.25, 1.0), ("V", 6.5, 1.0), ("B", 7.75, 1.0), ("N", 9.0, 1.0), ("M", 10.25, 1.0), ("Comma", 11.5, 1.0), ("Period", 12.75, 1.0), ("Slash", 14.0, 1.0), ("RShift", 15.25, 3.0), ), 0: ( ("LCtrl", 0.0, 1.25), ("Fn", 1.5, 1.0), ("Super", 2.75, 1.0), ("LAlt", 4.0, 1.0), ("Space", 5.25, 6.0), ("RAlt", 11.5, 1.0), ("RCtrl", 12.75, 1.0), ("Left", 14.0, 1.25), ("Up", 15.5, 1.25), ("Down", 15.5, 1.25), ("Right", 17.0, 1.25), ), } class ProtocolError(ValueError): """An invalid socket request.""" class KeyboardUnavailableError(OSError): """The per-LED keyboard interface is unavailable.""" def _read_text(path: Path) -> str | None: try: return path.read_text(encoding="ascii").strip().lower() except (OSError, UnicodeError): return None def _read_bytes(path: Path) -> bytes: try: return path.read_bytes() except OSError: return b"" def usb_ids_for_class_device(class_name: str, device: str) -> tuple[str | None, str | None]: name = os.path.basename(os.path.realpath(device)) sys_device = Path("/sys/class") / class_name / name / "device" try: current = sys_device.resolve(strict=True) except OSError: return None, None for parent in (current, *current.parents): vid = _read_text(parent / "idVendor") pid = _read_text(parent / "idProduct") if vid is not None and pid is not None: return vid, pid return None, None def usb_ids_for_tty(device: str) -> tuple[str | None, str | None]: return usb_ids_for_class_device("tty", device) def usb_ids_for_hidraw(device: str) -> tuple[str | None, str | None]: return usb_ids_for_class_device("hidraw", device) def is_framework_led_matrix(device: str) -> bool: vid, pid = usb_ids_for_tty(device) return vid == FRAMEWORK_VID and pid == LED_MATRIX_PID def discover_matrix_devices() -> list[str]: """Discover Framework LED Matrix serial devices in left-to-right order.""" preferred = sorted(glob.glob("/dev/serial/by-path/*")) fallback = sorted(glob.glob("/dev/ttyACM*")) found: list[str] = [] seen_real_paths: set[str] = set() for candidate in preferred + fallback: real = os.path.realpath(candidate) if real in seen_real_paths or not os.path.exists(candidate): continue if not os.path.basename(real).startswith("ttyACM"): continue if is_framework_led_matrix(candidate): found.append(candidate) seen_real_paths.add(real) # On the target Framework Laptop the stable by-path lexical order is the # reverse of physical left-to-right. Explicit --device order is untouched. return list(reversed(found)) def hidraw_report_descriptor(device: str) -> bytes: name = os.path.basename(os.path.realpath(device)) return _read_bytes(Path("/sys/class/hidraw") / name / "device" / "report_descriptor") def is_lamparray_descriptor(descriptor: bytes) -> bool: # HID Usage Page 0x59 (Lighting and Illumination), Usage 0x01 (LampArray). # The official Framework descriptor emits these short-form items. return b"\x05\x59" in descriptor and b"\x09\x01" in descriptor def discover_keyboard_hidraw() -> list[str]: found: list[str] = [] for candidate in sorted(glob.glob("/dev/hidraw*")): vid, pid = usb_ids_for_hidraw(candidate) if vid != FRAMEWORK_VID or pid != ANSI_KEYBOARD_PID: continue if is_lamparray_descriptor(hidraw_report_descriptor(candidate)): found.append(candidate) return found def configure_serial(fd: int) -> None: tty.setraw(fd, termios.TCSANOW) attrs = termios.tcgetattr(fd) attrs[4] = termios.B115200 attrs[5] = termios.B115200 attrs[2] |= termios.CLOCAL | termios.CREAD attrs[2] &= ~termios.PARENB attrs[2] &= ~termios.CSTOPB attrs[2] &= ~termios.CSIZE attrs[2] |= termios.CS8 attrs[6][termios.VMIN] = 0 attrs[6][termios.VTIME] = 0 termios.tcsetattr(fd, termios.TCSANOW, attrs) termios.tcflush(fd, termios.TCIOFLUSH) def write_all(fd: int, payload: bytes) -> None: view = memoryview(payload) while view: try: written = os.write(fd, view) except InterruptedError: continue if written <= 0: raise OSError(errno.EIO, "short hardware write") view = view[written:] # Linux _IOC helpers, matching and . def _ioc(direction: int, io_type: int, number: int, size: int) -> int: return (direction << 30) | (io_type << 8) | number | (size << 16) def hidiocsfeature(length: int) -> int: return _ioc(3, ord("H"), 0x06, length) # _IOC_READ | _IOC_WRITE def hidiocgfeature(length: int) -> int: return _ioc(3, ord("H"), 0x07, length) def hid_set_feature(fd: int, report: bytes | bytearray) -> None: data = bytearray(report) fcntl.ioctl(fd, hidiocsfeature(len(data)), data, True) def hid_get_feature(fd: int, report_id: int, length: int) -> bytes: data = bytearray(length) data[0] = report_id fcntl.ioctl(fd, hidiocgfeature(length), data, True) return bytes(data) class MatrixModule: def __init__(self, index: int, device: str): self.index = index self.device = device self.real_device = os.path.realpath(device) self.framebuffer = [[0 for _ in range(MATRIX_HEIGHT)] for _ in range(MATRIX_WIDTH)] self.global_brightness = 255 self.sleeping = False self._fd: int | None = None self._lock = threading.RLock() def _close_unlocked(self) -> None: if self._fd is not None: try: os.close(self._fd) except OSError: pass self._fd = None def close(self) -> None: with self._lock: self._close_unlocked() def _open_unlocked(self) -> int: if self._fd is not None: return self._fd fd = os.open(self.device, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK) try: flags = fcntl.fcntl(fd, fcntl.F_GETFL) fcntl.fcntl(fd, fcntl.F_SETFL, flags & ~os.O_NONBLOCK) configure_serial(fd) except Exception: os.close(fd) raise self._fd = fd self.real_device = os.path.realpath(self.device) return fd def _write_unlocked(self, payload: bytes) -> None: last_error: OSError | None = None for attempt in range(2): try: write_all(self._open_unlocked(), payload) return except OSError as exc: last_error = exc self._close_unlocked() if attempt == 0: continue assert last_error is not None raise last_error def _command_unlocked(self, command: int, parameters: Iterable[int] = ()) -> None: self._write_unlocked(MATRIX_MAGIC + bytes((command,)) + bytes(parameters)) def wake_unlocked(self) -> None: self._command_unlocked(MATRIX_CMD_SLEEP, (0,)) self.sleeping = False def wake(self) -> None: with self._lock: self.wake_unlocked() def sleep(self) -> None: with self._lock: self._command_unlocked(MATRIX_CMD_SLEEP, (1,)) self.sleeping = True def set_global_brightness(self, brightness: int) -> None: brightness = validate_byte(brightness, "brightness") with self._lock: self.wake_unlocked() self._command_unlocked(MATRIX_CMD_BRIGHTNESS, (brightness,)) self.global_brightness = brightness def _send_column_unlocked(self, x: int) -> None: payload = MATRIX_MAGIC + bytes((MATRIX_CMD_STAGE_GREY_COLUMN, x)) + bytes(self.framebuffer[x]) self._write_unlocked(payload) def _flush_unlocked(self) -> None: self._write_unlocked(MATRIX_MAGIC + bytes((MATRIX_CMD_FLUSH_GREY_COLUMNS, 0x00))) def set_pixels(self, pixels: Sequence[tuple[int, int, int]]) -> None: """ Update selected pixels, then restage the complete 9-column framebuffer. The LED Matrix firmware's greyscale staging area must not be treated as a durable partial-update buffer across commits. Committing only the columns touched by a mouse batch can leave untouched columns sourced from stale or cleared staging data, producing the visible whole-column blinking seen with the GUI. Always sending all nine columns before DrawGreyColBuffer matches the official full-frame greyscale update sequence and makes each commit self-contained and atomic. """ normalized: list[tuple[int, int, int]] = [] for x, y, brightness in pixels: x = validate_matrix_x(x) y = validate_matrix_y(y) brightness = validate_byte(brightness, "brightness") normalized.append((x, y, brightness)) if not normalized: return with self._lock: self.wake_unlocked() for x, y, brightness in normalized: self.framebuffer[x][MATRIX_HEIGHT - 1 - y] = brightness # Restage every column, not merely the dirty ones. for x in range(MATRIX_WIDTH): self._send_column_unlocked(x) self._flush_unlocked() def fill(self, brightness: int) -> None: brightness = validate_byte(brightness, "brightness") with self._lock: self.wake_unlocked() for x in range(MATRIX_WIDTH): self.framebuffer[x] = [brightness] * MATRIX_HEIGHT self._send_column_unlocked(x) self._flush_unlocked() def set_frame(self, rows: Sequence[Sequence[int]]) -> None: normalized = normalize_matrix_frame(rows) with self._lock: self.wake_unlocked() for x in range(MATRIX_WIDTH): self.framebuffer[x] = [ normalized[MATRIX_HEIGHT - 1 - hardware_row][x] for hardware_row in range(MATRIX_HEIGHT) ] self._send_column_unlocked(x) self._flush_unlocked() def info(self) -> dict[str, Any]: return { "device": self.index, "module": self.index, "type": "led_matrix", "side": "left" if self.index == 0 else "right" if self.index == 1 else None, "path": self.device, "real_path": self.real_device, "width": MATRIX_WIDTH, "height": MATRIX_HEIGHT, "channels": ["brightness"], "value_range": [0, 255], "coordinate_origin": "bottom-left", "global_brightness": self.global_brightness, "sleeping": self.sleeping, "available": True, } class LampArrayKeyboard: """Framework ANSI keyboard controlled through HID LampArray feature reports.""" def __init__( self, configured_path: str | None = None, enabled: bool = True, report_gap_ms: float = DEFAULT_KEYBOARD_REPORT_GAP_MS, ): self.index = KEYBOARD_DEVICE_INDEX self.configured_path = configured_path self.enabled = enabled self.path: str | None = configured_path self.real_path: str | None = os.path.realpath(configured_path) if configured_path else None self.lamp_count = LAMPARRAY_LED_COUNT if report_gap_ms < 0: raise ValueError("keyboard report gap must not be negative") self.report_gap_seconds = report_gap_ms / 1000.0 # None means the host-side cache is not known to match the keyboard. # Once a color is successfully sent, it is cached so unchanged LEDs can # be omitted from subsequent animation frames. self.framebuffer: list[tuple[int, int, int] | None] = [None] * LAMPARRAY_LED_COUNT self._fd: int | None = None self._lock = threading.RLock() self.last_error: str | None = None self.protocol_available = False self.grid_to_leds: dict[tuple[int, int], list[int]] = self._build_grid_map() self.led_to_grid: list[tuple[int, int]] = self._build_led_grid() if enabled: self._probe() else: self.last_error = "keyboard support disabled with --no-keyboard" @staticmethod def _row_for_raw_y(raw_y: int) -> int: return min(range(len(ANSI_ROW_CENTERS)), key=lambda row: abs(ANSI_ROW_CENTERS[row] - raw_y)) @classmethod def _build_led_grid(cls) -> list[tuple[int, int]]: # QMK's physical position table is top-origin on the installed ANSI # module. Convert its six row indices to the socket API's bottom-origin # coordinates. Exact LED IDs are deliberately unaffected. return [ ( int(round(raw_x / 8.0)), KEYBOARD_GRID_HEIGHT - 1 - cls._row_for_raw_y(raw_y), ) for raw_x, raw_y in ANSI_LED_POSITIONS ] @classmethod def _build_grid_map(cls) -> dict[tuple[int, int], list[int]]: result: dict[tuple[int, int], list[int]] = {} for led, coordinate in enumerate(cls._build_led_grid()): result.setdefault(coordinate, []).append(led) return result def _discover_path_unlocked(self) -> str | None: if self.configured_path: return self.configured_path paths = discover_keyboard_hidraw() return paths[0] if paths else None def _close_unlocked(self) -> None: if self._fd is not None: try: os.close(self._fd) except OSError: pass self._fd = None self.protocol_available = False # A reconnect or firmware restart invalidates our assumptions about the # colors currently displayed on the keyboard. self.framebuffer = [None] * self.lamp_count def close(self) -> None: with self._lock: self._close_unlocked() def _open_unlocked(self) -> int: if not self.enabled: raise KeyboardUnavailableError("keyboard support is disabled") if self._fd is not None: return self._fd path = self._discover_path_unlocked() if not path: raise KeyboardUnavailableError( "no Framework HID LampArray keyboard interface found; true per-key control requires " "Framework's experimental fl16-2025-hidlamp keyboard firmware" ) descriptor = hidraw_report_descriptor(path) if not is_lamparray_descriptor(descriptor): raise KeyboardUnavailableError(f"{path} is not a HID LampArray interface") vid, pid = usb_ids_for_hidraw(path) if vid != FRAMEWORK_VID or pid != ANSI_KEYBOARD_PID: raise KeyboardUnavailableError(f"{path} is not Framework ANSI keyboard USB {FRAMEWORK_VID}:{ANSI_KEYBOARD_PID}") self.path = path self.real_path = os.path.realpath(path) fd = os.open(path, os.O_RDWR | os.O_CLOEXEC) try: # Confirm the interface answers LampArrayAttributesReport. attributes = hid_get_feature(fd, LAMPARRAY_REPORT_ATTRIBUTES, 23) if not attributes or attributes[0] != LAMPARRAY_REPORT_ATTRIBUTES: raise OSError(errno.EPROTO, "unexpected LampArray attributes report") lamp_count = struct.unpack_from(" None: with self._lock: try: self._open_unlocked() except OSError as exc: self.last_error = str(exc) self._close_unlocked() def _feature_unlocked(self, report: bytes) -> None: last_error: OSError | None = None for attempt in range(2): try: hid_set_feature(self._open_unlocked(), report) self.protocol_available = True self.last_error = None return except OSError as exc: last_error = exc self.last_error = str(exc) self._close_unlocked() if attempt == 0: continue assert last_error is not None raise last_error def _require_ready_unlocked(self) -> int: try: return self._open_unlocked() except OSError as exc: self.last_error = str(exc) raise KeyboardUnavailableError(str(exc)) from exc @staticmethod def _multi_report(chunk: Sequence[tuple[int, int, int, int]], complete: bool) -> bytes: # report id, count, flags, 8 uint16 IDs, then 8 RGBA/intensity states. report = bytearray(51) report[0] = LAMPARRAY_REPORT_MULTI_UPDATE report[1] = len(chunk) report[2] = LAMPARRAY_COMPLETE if complete else 0 ids_offset = 3 colors_offset = ids_offset + (LAMPARRAY_MULTI_COUNT * 2) for slot, (led, red, green, blue) in enumerate(chunk): struct.pack_into(" None: normalized: dict[int, tuple[int, int, int]] = {} for led, red, green, blue in updates: led = validate_keyboard_led(led) normalized[led] = validate_rgb(red, green, blue) # Animation clients often submit all 97 LEDs every frame. Avoid sending # values that are already displayed; this substantially reduces USB and # firmware work, especially while a narrow wave is crossing the laptop. items = [ (led, *rgb) for led, rgb in sorted(normalized.items()) if self.framebuffer[led] != rgb ] if not items: return with self._lock: self._require_ready_unlocked() for start in range(0, len(items), LAMPARRAY_MULTI_COUNT): chunk = items[start : start + LAMPARRAY_MULTI_COUNT] complete = start + LAMPARRAY_MULTI_COUNT >= len(items) self._feature_unlocked(self._multi_report(chunk, complete)) # The experimental Framework firmware queues only five LampArray # reports, while a complete 97-LED frame needs thirteen. Do not # burst all reports back-to-back; yield time for QMK's main task to # drain its queue and continue scanning the keyboard matrix. if not complete and self.report_gap_seconds > 0: time.sleep(self.report_gap_seconds) for led, red, green, blue in items: self.framebuffer[led] = (red, green, blue) def set_pixels(self, pixels: Sequence[tuple[int, int, int, int, int]]) -> int: updates: list[tuple[int, int, int, int]] = [] addressed_leds: set[int] = set() for x, y, red, green, blue in pixels: x = validate_keyboard_x(x) y = validate_keyboard_y(y) rgb = validate_rgb(red, green, blue) leds = self.grid_to_leds.get((x, y)) if not leds: raise ProtocolError( f"keyboard coordinate ({x},{y}) has no LED; use op='keyboard_layout' for valid coordinates" ) for led in leds: updates.append((led, *rgb)) addressed_leds.add(led) self.set_leds(updates) return len(addressed_leds) def fill(self, red: int, green: int, blue: int) -> None: red, green, blue = validate_rgb(red, green, blue) # Feature report: id, complete flag, uint16 start/end, RGBA/intensity. report = struct.pack( " None: """Return lighting control to the keyboard's autonomous QMK effect.""" with self._lock: self._feature_unlocked(bytes((LAMPARRAY_REPORT_CONTROL, 1))) self.protocol_available = True def take_control(self) -> None: with self._lock: self._feature_unlocked(bytes((LAMPARRAY_REPORT_CONTROL, 0))) time.sleep(0.01) def layout(self) -> dict[str, Any]: rows: list[list[dict[str, Any]]] = [] for y in range(KEYBOARD_GRID_HEIGHT - 1, -1, -1): row: list[dict[str, Any]] = [] for (x_value, y_value), leds in sorted(self.grid_to_leds.items()): if y_value == y: row.append({"x": x_value, "y": y_value, "leds": leds}) rows.append(row) return { "ok": True, "device": KEYBOARD_DEVICE_INDEX, "type": "rgb_keyboard", "coordinate_origin": "bottom-left", "width": KEYBOARD_GRID_WIDTH, "height": KEYBOARD_GRID_HEIGHT, "sparse": True, "led_count": len(self.led_to_grid), "rows_top_to_bottom": rows, "leds": [ {"led": led, "x": coordinate[0], "y": coordinate[1]} for led, coordinate in enumerate(self.led_to_grid) ], } def _human_key_map(self) -> list[dict[str, Any]]: """Associate human key labels with the socket grid coordinates they light.""" assignments: dict[tuple[int, str], list[int]] = { (y, label): [] for y, keys in ANSI_KEY_ROWS.items() for label, _layout_x, _width in keys } for led, ((raw_x, raw_y), (_grid_x, api_y)) in enumerate( zip(ANSI_LED_POSITIONS, self.led_to_grid) ): keys = ANSI_KEY_ROWS[api_y] # Up and Down occupy upper/lower half-height positions but collapse # onto the same y=0 row in the intentionally coarse socket grid. if api_y == 0 and 198 <= raw_x <= 215: label = "Up" if raw_y < 60 else "Down" else: containing = [ key for key in keys if key[1] * ANSI_LAYOUT_X_SCALE - 0.01 <= raw_x <= (key[1] + key[2]) * ANSI_LAYOUT_X_SCALE + 0.01 ] candidates = containing or list(keys) label, _layout_x, _width = min( candidates, key=lambda key: ( key[2] if containing else 0.0, abs(raw_x - (key[1] + key[2] / 2.0) * ANSI_LAYOUT_X_SCALE), ), ) assignments[(api_y, label)].append(led) result: list[dict[str, Any]] = [] for y in range(KEYBOARD_GRID_HEIGHT - 1, -1, -1): for label, layout_x, width in ANSI_KEY_ROWS[y]: leds = assignments[(y, label)] coordinates = sorted({self.led_to_grid[led] for led in leds}) result.append( { "key": label, "row": y, "layout_x": layout_x, "layout_width": width, "coordinates": [ {"x": x, "y": coordinate_y} for x, coordinate_y in coordinates ], "leds": leds, } ) return result @staticmethod def _format_key_coordinates(key: dict[str, Any], include_leds: bool) -> str: coordinates = key["coordinates"] if not coordinates: coordinate_text = "unmapped" elif all(item["y"] == coordinates[0]["y"] for item in coordinates): x_values = "/".join(str(item["x"]) for item in coordinates) coordinate_text = f"{x_values},{coordinates[0]['y']}" else: coordinate_text = "/".join( f"{item['x']},{item['y']}" for item in coordinates ) text = f"{key['key']} {coordinate_text}" if include_leds: text += " led=" + "/".join(str(led) for led in key["leds"]) return f"[{text}]" def ascii_layout(self, include_leds: bool = False) -> dict[str, Any]: """Return a human-readable US ANSI key-to-coordinate diagram.""" keys = self._human_key_map() rows: dict[int, list[dict[str, Any]]] = { y: [key for key in keys if key["row"] == y] for y in range(KEYBOARD_GRID_HEIGHT) } lines = [ "Framework US ANSI RGB keyboard socket coordinate map", "Origin: bottom-left; x increases right; y increases up", "Each key shows every approximate (x,y) socket coordinate that lights it.", "Wide keys may have several coordinates separated by '/'.", "", ] for y in (5, 4, 3, 2, 1): lines.append( " ".join( self._format_key_coordinates(key, include_leds) for key in rows[y] ) ) bottom_main = [ key for key in rows[0] if key["key"] not in {"Left", "Up", "Down", "Right"} ] arrows = {key["key"]: key for key in rows[0] if key["key"] in {"Left", "Up", "Down", "Right"}} lines.append( " ".join( self._format_key_coordinates(key, include_leds) for key in bottom_main ) ) lines.append(" " * 72 + self._format_key_coordinates(arrows["Up"], include_leds)) lines.append( " " * 56 + " ".join( self._format_key_coordinates(arrows[label], include_leds) for label in ("Left", "Down", "Right") ) ) lines.extend( [ "", "Examples:", " [A 4,2] means: device 2, x=4, y=2 lights A.", " [Space 9/10/12/14/15/17,0] means Space has several separately addressable positions.", " Up and Down both map to (25,0) in the coarse grid; exact LED addressing distinguishes them.", ] ) public_keys = [] for key in keys: item = { "key": key["key"], "coordinates": key["coordinates"], } if include_leds: item["leds"] = key["leds"] public_keys.append(item) coordinate_to_keys: dict[str, list[str]] = {} for key in public_keys: for coordinate in key["coordinates"]: coordinate_to_keys.setdefault( f"{coordinate['x']},{coordinate['y']}", [] ).append(key["key"]) return { "ok": True, "device": KEYBOARD_DEVICE_INDEX, "type": "rgb_keyboard_ascii", "coordinate_origin": "bottom-left", "width": KEYBOARD_GRID_WIDTH, "height": KEYBOARD_GRID_HEIGHT, "cell_contents": "human key label and socket coordinate(s)", "include_leds": include_leds, "ascii": "\n".join(lines), "keys": public_keys, "shared_coordinates": { coordinate: labels for coordinate, labels in coordinate_to_keys.items() if len(labels) > 1 }, } def info(self, refresh: bool = False) -> dict[str, Any]: if refresh and self.enabled and not self.protocol_available: self._probe() return { "device": self.index, "type": "rgb_keyboard", "layout": "US English ANSI", "path": self.path, "real_path": self.real_path, "usb_id": f"{FRAMEWORK_VID}:{ANSI_KEYBOARD_PID}", "protocol": "HID LampArray", "firmware_requirement": "Framework QMK branch fl16-2025-hidlamp", "available": self.protocol_available, "error": self.last_error, "width": KEYBOARD_GRID_WIDTH, "height": KEYBOARD_GRID_HEIGHT, "sparse": True, "coordinate_origin": "bottom-left", "channels": ["red", "green", "blue"], "value_range": [0, 255], "led_count": self.lamp_count, "streaming_safety": { "delta_updates": True, "report_gap_ms": self.report_gap_seconds * 1000.0, "firmware_queue_reports": 5, "reports_for_full_frame": (self.lamp_count + LAMPARRAY_MULTI_COUNT - 1) // LAMPARRAY_MULTI_COUNT, }, } def require_int(value: Any, name: str) -> int: if isinstance(value, bool) or not isinstance(value, int): raise ProtocolError(f"{name} must be an integer") return value def validate_byte(value: Any, name: str) -> int: value = require_int(value, name) if not 0 <= value <= 255: raise ProtocolError(f"{name} must be between 0 and 255") return value def validate_rgb(red: Any, green: Any, blue: Any) -> tuple[int, int, int]: return ( validate_byte(red, "r"), validate_byte(green, "g"), validate_byte(blue, "b"), ) def validate_matrix_x(value: Any) -> int: value = require_int(value, "x") if not 0 <= value < MATRIX_WIDTH: raise ProtocolError(f"x must be between 0 and {MATRIX_WIDTH - 1} for a matrix") return value def validate_matrix_y(value: Any) -> int: value = require_int(value, "y") if not 0 <= value < MATRIX_HEIGHT: raise ProtocolError(f"y must be between 0 and {MATRIX_HEIGHT - 1} for a matrix") return value def validate_keyboard_x(value: Any) -> int: value = require_int(value, "x") if not 0 <= value < KEYBOARD_GRID_WIDTH: raise ProtocolError(f"x must be between 0 and {KEYBOARD_GRID_WIDTH - 1} for the keyboard") return value def validate_keyboard_y(value: Any) -> int: value = require_int(value, "y") if not 0 <= value < KEYBOARD_GRID_HEIGHT: raise ProtocolError(f"y must be between 0 and {KEYBOARD_GRID_HEIGHT - 1} for the keyboard") return value def validate_keyboard_led(value: Any) -> int: value = require_int(value, "led") if not 0 <= value < LAMPARRAY_LED_COUNT: raise ProtocolError(f"led must be between 0 and {LAMPARRAY_LED_COUNT - 1}") return value def normalize_matrix_frame(frame: Any) -> list[list[int]]: if not isinstance(frame, list): raise ProtocolError("frame must be a JSON list") if len(frame) == MATRIX_PIXEL_COUNT and all(not isinstance(v, list) for v in frame): rows = [frame[y * MATRIX_WIDTH : (y + 1) * MATRIX_WIDTH] for y in range(MATRIX_HEIGHT)] else: rows = frame if len(rows) != MATRIX_HEIGHT: raise ProtocolError( f"matrix frame must contain {MATRIX_HEIGHT} rows or {MATRIX_PIXEL_COUNT} flat values" ) normalized: list[list[int]] = [] for y, row in enumerate(rows): if not isinstance(row, list) or len(row) != MATRIX_WIDTH: raise ProtocolError(f"matrix frame row {y} must contain exactly {MATRIX_WIDTH} values") normalized.append([validate_byte(v, "brightness") for v in row]) return normalized def parse_matrix_pixel(item: Any) -> tuple[int, int, int]: if isinstance(item, list) and len(item) == 3: x, y, brightness = item elif isinstance(item, dict): try: x, y, brightness = item["x"], item["y"], item["brightness"] except KeyError as exc: raise ProtocolError(f"matrix pixel object is missing {exc.args[0]!r}") from exc else: raise ProtocolError("each matrix pixel must be [x,y,brightness] or an object") return validate_matrix_x(x), validate_matrix_y(y), validate_byte(brightness, "brightness") def parse_keyboard_pixel(item: Any) -> tuple[int, int, int, int, int]: if isinstance(item, list) and len(item) == 5: x, y, red, green, blue = item elif isinstance(item, dict): try: x, y = item["x"], item["y"] red, green, blue = item["r"], item["g"], item["b"] except KeyError as exc: raise ProtocolError(f"keyboard pixel object is missing {exc.args[0]!r}") from exc else: raise ProtocolError("each keyboard pixel must be [x,y,r,g,b] or an object") red, green, blue = validate_rgb(red, green, blue) return validate_keyboard_x(x), validate_keyboard_y(y), red, green, blue def parse_keyboard_led(item: Any) -> tuple[int, int, int, int]: if isinstance(item, list) and len(item) == 4: led, red, green, blue = item elif isinstance(item, dict): try: led = item["led"] red, green, blue = item["r"], item["g"], item["b"] except KeyError as exc: raise ProtocolError(f"keyboard LED object is missing {exc.args[0]!r}") from exc else: raise ProtocolError("each keyboard LED must be [led,r,g,b] or an object") red, green, blue = validate_rgb(red, green, blue) return validate_keyboard_led(led), red, green, blue class InputController: def __init__( self, matrix_devices: Sequence[str], keyboard_path: str | None, keyboard_enabled: bool, keyboard_report_gap_ms: float = DEFAULT_KEYBOARD_REPORT_GAP_MS, ): self.modules = [MatrixModule(i, path) for i, path in enumerate(matrix_devices)] self.keyboard = LampArrayKeyboard( keyboard_path, enabled=keyboard_enabled, report_gap_ms=keyboard_report_gap_ms, ) def close(self) -> None: for module in self.modules: module.close() self.keyboard.close() def select_matrices(self, value: Any, allow_all: bool = True) -> list[MatrixModule]: if value == "all": if not allow_all: raise ProtocolError("this operation requires one numeric matrix device") return list(self.modules) index = require_int(value, "module/device") if not 0 <= index < len(self.modules): maximum = len(self.modules) - 1 raise ProtocolError(f"matrix module/device must be between 0 and {maximum}, or 'all'") return [self.modules[index]] @staticmethod def request_target(request: dict[str, Any]) -> tuple[str, Any]: if "device" in request: value = request["device"] if value == KEYBOARD_DEVICE_INDEX: return "keyboard", value return "matrix", value if "module" in request: return "matrix", request["module"] raise ProtocolError("request requires 'device' (0, 1, or 2) or legacy field 'module'") def handle_keyboard(self, op: str, request: dict[str, Any]) -> dict[str, Any]: if op == "pixel": pixel = parse_keyboard_pixel(request) count = self.keyboard.set_pixels([pixel]) return {"ok": True, "updated_device": KEYBOARD_DEVICE_INDEX, "leds": count} if op == "pixels": items = request.get("pixels") if not isinstance(items, list): raise ProtocolError("pixels must be a list") pixels = [parse_keyboard_pixel(item) for item in items] count = self.keyboard.set_pixels(pixels) return {"ok": True, "updated_device": KEYBOARD_DEVICE_INDEX, "leds": count} if op in ("fill", "clear"): rgb = (0, 0, 0) if op == "clear" else validate_rgb( request.get("r"), request.get("g"), request.get("b") ) self.keyboard.fill(*rgb) return {"ok": True, "updated_device": KEYBOARD_DEVICE_INDEX, "rgb": list(rgb)} if op == "led": led = parse_keyboard_led(request) self.keyboard.set_leds([led]) return {"ok": True, "updated_device": KEYBOARD_DEVICE_INDEX, "leds": 1} if op == "leds": items = request.get("leds") if not isinstance(items, list): raise ProtocolError("leds must be a list") leds = [parse_keyboard_led(item) for item in items] self.keyboard.set_leds(leds) return {"ok": True, "updated_device": KEYBOARD_DEVICE_INDEX, "leds": len(leds)} if op == "take_control": self.keyboard.take_control() return {"ok": True, "updated_device": KEYBOARD_DEVICE_INDEX, "autonomous": False} if op == "release_control": self.keyboard.release_to_firmware() return {"ok": True, "updated_device": KEYBOARD_DEVICE_INDEX, "autonomous": True} raise ProtocolError(f"operation {op!r} is not supported for the RGB keyboard") def handle_matrix(self, op: str, target: Any, request: dict[str, Any]) -> dict[str, Any]: targets = self.select_matrices(target) if op == "pixel": pixel = parse_matrix_pixel(request) for module in targets: module.set_pixels([pixel]) return {"ok": True, "updated_devices": [m.index for m in targets], "pixels": 1} if op == "pixels": items = request.get("pixels") if not isinstance(items, list): raise ProtocolError("pixels must be a list") pixels = [parse_matrix_pixel(item) for item in items] for module in targets: module.set_pixels(pixels) return {"ok": True, "updated_devices": [m.index for m in targets], "pixels": len(pixels)} if op in ("fill", "clear"): brightness = 0 if op == "clear" else validate_byte(request.get("brightness"), "brightness") for module in targets: module.fill(brightness) return {"ok": True, "updated_devices": [m.index for m in targets]} if op == "frame": frame = normalize_matrix_frame(request.get("frame")) for module in targets: module.set_frame(frame) return {"ok": True, "updated_devices": [m.index for m in targets]} if op == "global_brightness": brightness = validate_byte(request.get("brightness"), "brightness") for module in targets: module.set_global_brightness(brightness) return { "ok": True, "updated_devices": [m.index for m in targets], "global_brightness": brightness, } if op == "sleep": for module in targets: module.sleep() return {"ok": True, "updated_devices": [m.index for m in targets]} if op == "wake": for module in targets: module.wake() return {"ok": True, "updated_devices": [m.index for m in targets]} raise ProtocolError(f"operation {op!r} is not supported for LED Matrix devices") def handle(self, request: Any) -> dict[str, Any]: if not isinstance(request, dict): raise ProtocolError("request must be a JSON object") op = request.get("op") if not isinstance(op, str): raise ProtocolError("request requires string field 'op'") if op == "status": devices: list[dict[str, Any]] = [module.info() for module in self.modules] devices.append(self.keyboard.info(refresh=bool(request.get("refresh")))) return { "ok": True, "server_version": SERVER_VERSION, "coordinate_origin": "bottom-left", "devices": devices, "modules": [module.info() for module in self.modules], "keyboard": self.keyboard.info(), } if op == "keyboard_layout": return self.keyboard.layout() if op == "keyboard_ascii": include_leds = request.get("include_leds", False) if not isinstance(include_leds, bool): raise ProtocolError("include_leds must be true or false") return self.keyboard.ascii_layout(include_leds=include_leds) target_kind, target = self.request_target(request) if target_kind == "keyboard": return self.handle_keyboard(op, request) return self.handle_matrix(op, target, request) class JsonLineHandler(socketserver.StreamRequestHandler): def handle(self) -> None: peer = self.client_address if self.client_address else "local" LOG.debug("client connected: %s", peer) while True: raw = self.rfile.readline(MAX_REQUEST_BYTES + 1) if not raw: break if len(raw) > MAX_REQUEST_BYTES: self.send_response({"ok": False, "error": "request too large"}) break if not raw.strip(): continue try: request = json.loads(raw.decode("utf-8")) response = self.server.controller.handle(request) # type: ignore[attr-defined] except UnicodeDecodeError: response = {"ok": False, "error": "request is not valid UTF-8"} except json.JSONDecodeError as exc: response = {"ok": False, "error": f"invalid JSON: {exc.msg}"} except (ProtocolError, KeyboardUnavailableError) as exc: response = {"ok": False, "error": str(exc)} except OSError as exc: LOG.exception("hardware error") response = {"ok": False, "error": f"hardware error: {exc}"} except Exception as exc: LOG.exception("unexpected request failure") response = {"ok": False, "error": f"internal error: {exc}"} self.send_response(response) def send_response(self, response: dict[str, Any]) -> None: payload = json.dumps(response, separators=(",", ":"), sort_keys=True).encode("utf-8") + b"\n" self.wfile.write(payload) self.wfile.flush() class ThreadingTCPServer(socketserver.ThreadingTCPServer): allow_reuse_address = True daemon_threads = True def __init__(self, address: tuple[str, int], handler: type[JsonLineHandler], controller: InputController): self.controller = controller super().__init__(address, handler) class ThreadingUnixServer(socketserver.ThreadingUnixStreamServer): daemon_threads = True def __init__(self, path: str, handler: type[JsonLineHandler], controller: InputController): self.controller = controller super().__init__(path, handler) def parse_listen(value: str) -> tuple[str, int]: if ":" not in value: raise argparse.ArgumentTypeError("listen address must be HOST:PORT") host, port_text = value.rsplit(":", 1) try: port = int(port_text) except ValueError as exc: raise argparse.ArgumentTypeError("port must be an integer") from exc if not 1 <= port <= 65535: raise argparse.ArgumentTypeError("port must be between 1 and 65535") return host, port def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument( "--device", action="append", default=[], help="LED Matrix serial device; repeat to set explicit left-to-right order", ) parser.add_argument( "--expected-modules", type=int, default=2, help="fail unless this many matrix modules are found (default: 2; use 0 for any count)", ) parser.add_argument( "--keyboard-hidraw", metavar="PATH", help="explicit HID LampArray /dev/hidrawN path (normally auto-detected)", ) parser.add_argument("--no-keyboard", action="store_true", help="disable RGB keyboard discovery/control") parser.add_argument( "--keyboard-report-gap-ms", type=float, default=DEFAULT_KEYBOARD_REPORT_GAP_MS, metavar="MS", help=( "delay between keyboard HID update chunks to protect the experimental " "firmware's small request queue (default: 3.0 ms; try 5-8 if key input " "still becomes unreliable)" ), ) parser.add_argument( "--require-keyboard", action="store_true", help="fail startup unless the HID LampArray keyboard interface is available", ) parser.add_argument("--list", action="store_true", help="list discovered matrix and keyboard interfaces and exit") socket_group = parser.add_mutually_exclusive_group() socket_group.add_argument( "--listen", type=parse_listen, default=("127.0.0.1", 8765), metavar="HOST:PORT", help="TCP listen address (default: 127.0.0.1:8765)", ) socket_group.add_argument("--unix-socket", metavar="PATH", help="listen on a Unix-domain socket instead") parser.add_argument("--keep-display", action="store_true", help="do not clear displays when starting") parser.add_argument("--verbose", action="store_true", help="enable debug logging") return parser def list_hardware(matrix_devices: Sequence[str], explicit_keyboard: str | None) -> int: if matrix_devices: for index, device in enumerate(matrix_devices): vid, pid = usb_ids_for_tty(device) print(f"device {index} matrix: {device} -> {os.path.realpath(device)} USB {vid}:{pid}") else: print("No Framework LED Matrix modules found.") paths = [explicit_keyboard] if explicit_keyboard else discover_keyboard_hidraw() paths = [path for path in paths if path] if paths: for path in paths: vid, pid = usb_ids_for_hidraw(path) print(f"device {KEYBOARD_DEVICE_INDEX} RGB keyboard LampArray: {path} USB {vid}:{pid}") else: print( "No Framework keyboard HID LampArray interface found. " "Stock firmware does not expose true per-key host control; use fl16-2025-hidlamp." ) return 0 if matrix_devices or paths else 1 def main() -> int: args = build_parser().parse_args() logging.basicConfig( level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(message)s", ) matrix_devices = args.device or discover_matrix_devices() if args.list: return list_hardware(matrix_devices, args.keyboard_hidraw) if not matrix_devices: LOG.error("No Framework LED Matrix modules found") return 1 if args.expected_modules and len(matrix_devices) != args.expected_modules: LOG.error( "Expected %d matrix module(s), found %d: %s", args.expected_modules, len(matrix_devices), matrix_devices, ) return 1 LOG.info("Framework input server %s", SERVER_VERSION) controller = InputController( matrix_devices, keyboard_path=args.keyboard_hidraw, keyboard_enabled=not args.no_keyboard, keyboard_report_gap_ms=args.keyboard_report_gap_ms, ) for module in controller.modules: LOG.info("device %d matrix: %s -> %s", module.index, module.device, module.real_device) keyboard_info = controller.keyboard.info() if keyboard_info["available"]: LOG.info("device %d RGB keyboard: %s", KEYBOARD_DEVICE_INDEX, keyboard_info["path"]) else: LOG.warning("device %d RGB keyboard unavailable: %s", KEYBOARD_DEVICE_INDEX, keyboard_info["error"]) if args.require_keyboard: controller.close() return 1 unix_path: str | None = args.unix_socket server: socketserver.BaseServer try: if not args.keep_display: LOG.info("clearing matrix displays") for module in controller.modules: module.fill(0) if controller.keyboard.protocol_available: LOG.info("clearing RGB keyboard") controller.keyboard.fill(0, 0, 0) if unix_path: unix_path = os.path.abspath(os.path.expanduser(unix_path)) parent = os.path.dirname(unix_path) if parent: os.makedirs(parent, exist_ok=True) try: os.unlink(unix_path) except FileNotFoundError: pass server = ThreadingUnixServer(unix_path, JsonLineHandler, controller) os.chmod(unix_path, 0o600) LOG.info("listening on unix socket %s", unix_path) else: server = ThreadingTCPServer(args.listen, JsonLineHandler, controller) host, port = server.server_address[:2] LOG.info("listening on tcp %s:%s", host, port) stop_event = threading.Event() def stop_server(signum: int, _frame: Any) -> None: if stop_event.is_set(): return stop_event.set() LOG.info("received signal %s; shutting down", signum) threading.Thread(target=server.shutdown, daemon=True).start() signal.signal(signal.SIGINT, stop_server) signal.signal(signal.SIGTERM, stop_server) server.serve_forever(poll_interval=0.5) server.server_close() return 0 finally: controller.close() if unix_path: try: os.unlink(unix_path) except FileNotFoundError: pass if __name__ == "__main__": sys.exit(main())