commit 2f4af50ea6b223ead96dd3b2646a45af2a45402c Author: HeySmokers420! Date: Tue Sep 8 00:05:52 2026 -0500 ASRock Polychrome system LED indicator: HID driver, daemon, systemd unit, udev rule diff --git a/99-asrock-led.rules b/99-asrock-led.rules new file mode 100644 index 0000000..7b1e431 --- /dev/null +++ b/99-asrock-led.rules @@ -0,0 +1 @@ +KERNEL=="hidraw*", SUBSYSTEM=="hidraw", ATTRS{idVendor}=="26ce", ATTRS{idProduct}=="01a2", GROUP="plugdev", MODE="0660" \ No newline at end of file diff --git a/__pycache__/asrock_hid.cpython-311.pyc b/__pycache__/asrock_hid.cpython-311.pyc new file mode 100644 index 0000000..85e9fad Binary files /dev/null and b/__pycache__/asrock_hid.cpython-311.pyc differ diff --git a/asrock-indicator.py b/asrock-indicator.py new file mode 100755 index 0000000..e323013 --- /dev/null +++ b/asrock-indicator.py @@ -0,0 +1,267 @@ +#!/usr/bin/env python3 +import math +import os +import select +import socket +import struct +import sys +import time + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import asrock_hid as h + +SOCK_PATH = os.environ.get( + "ASROCK_SOCK", os.path.join(os.environ.get("XDG_RUNTIME_DIR", "/tmp"), "asrock-led.sock") +) +FRAME = 0.05 + +NATIVE_TAKES_COLOR = { + "static": (h.A_STATIC, True), + "breath": (h.A_BREATH, True), + "strobe": (h.A_STROBE, True), + "direct": (h.A_DIRECT, True), + "cycle": (h.A_CYCLE, False), + "random": (h.A_RANDOM, False), + "music": (h.A_MUSIC, False), + "wave": (h.A_WAVE, False), + "spring": (h.A_SPRING, False), + "stack": (h.A_STACK, False), + "cram": (h.A_CRAM, False), + "scan": (h.A_SCAN, False), + "neon": (h.A_NEON, False), + "water": (h.A_WATER, False), + "rainbow": (h.A_RAINBOW, False), +} + +COLORS = { + "black": "000000", "off": "000000", "white": "ffffff", + "red": "ff0000", "green": "00ff00", "blue": "0000ff", + "yellow": "ffff00", "cyan": "00ffff", "magenta": "ff00ff", + "orange": "ff8000", "purple": "8000ff", "pink": "ff0080", + "amber": "ffbf00", +} + + +def parse_color(s): + if s in COLORS: + s = COLORS[s] + s = s.lstrip("#") + if len(s) == 6 and all(c in "0123456789abcdefABCDEF" for c in s): + return struct.unpack("BBB", bytes.fromhex(s)) + parts = s.replace(",", " ").split() + if len(parts) == 3: + vals = [int(p, 0) for p in parts] + if all(0 <= v <= 255 for v in vals): + return tuple(vals) + raise ValueError("bad color: %s" % s) + + +def pulse_frames(rgb, count, period): + n = max(2, int(round(period / FRAME))) + frames = [] + for _ in range(count): + for i in range(n): + br = math.sin(math.pi * i / (n - 1)) + frames.append((int(rgb[0] * br), int(rgb[1] * br), int(rgb[2] * br))) + return frames + + +def blink_frames(rgb, count, period): + half = max(1, int(round(period / 2 / FRAME))) + frames = [] + for _ in range(count): + frames.extend([rgb] * half) + frames.extend([(0, 0, 0)] * half) + return frames + + +class Daemon: + def __init__(self, dev=h.DEFAULT_DEV): + self.fd = h.open_dev(dev) + self.zone = 0 + self.allzone = 1 + self.idle = {"kind": "native", "mode": h.A_OFF, "rgb": (0, 0, 0), "speed": 0, "written": True} + self.current = {"kind": "native", "mode": h.A_OFF, "rgb": (0, 0, 0), "speed": 0, "written": False} + + def set_zone(self, zone): + if zone >= 0: + self.zone = zone + self.allzone = 0 + else: + self.zone = 0 + self.allzone = 1 + + def write(self, mode, rgb, speed): + r, g, b = rgb + return h.write_set(self.fd, self.zone, mode, r, g, b, speed, self.allzone) + + def native(self, mode, rgb, speed): + self.current = {"kind": "native", "mode": mode, "rgb": rgb, "speed": speed, "written": False} + return "ok" + + def frames(self, frames): + self.current = {"kind": "frames", "frames": frames, "idx": 0} + return "ok" + + def apply(self, line): + toks = line.split() + if not toks: + return "err empty" + if toks[0] == "zone": + self.set_zone(-1 if toks[1] == "all" else int(toks[1], 0)) + toks = toks[2:] + if not toks: + return "ok" + effect = toks[0] + args = toks[1:] + + if effect == "idle": + self.idle = {"kind": "native", "mode": h.A_STATIC, "rgb": parse_color(args[0]), + "speed": int(args[1], 0) if len(args) > 1 else 0xE0, "written": False} + self.current = dict(self.idle, written=False) + return "ok idle set" + + if effect == "off": + return self.native(h.A_OFF, (0, 0, 0), 0) + + if effect == "static": + rgb = parse_color(args[0]) + speed = int(args[1], 0) if len(args) > 1 else 0xE0 + return self.native(h.A_STATIC, rgb, speed) + + if effect == "pulse": + rgb = parse_color(args[0]) + count = int(args[1], 0) if len(args) > 1 else 1 + period = float(args[2]) if len(args) > 2 else 0.8 + return self.frames(pulse_frames(rgb, count, period)) + + if effect == "blink" or effect == "flash": + rgb = parse_color(args[0]) + count = int(args[1], 0) if len(args) > 1 else 1 + period = float(args[2]) if len(args) > 2 else (0.2 if effect == "flash" else 0.4) + return self.frames(blink_frames(rgb, count, period)) + + if effect in NATIVE_TAKES_COLOR: + mode, takes_color = NATIVE_TAKES_COLOR[effect] + speed = 0xE0 + rgb = (255, 255, 255) + n = 0 + if args: + try: + speed = int(args[0], 0) + n = 1 + except ValueError: + pass + if n < len(args) and takes_color: + rgb = parse_color(args[n]) + n += 1 + elif takes_color and len(args) == 1: + try: + rgb = parse_color(args[0]) + except ValueError: + pass + return self.native(mode, rgb, speed) + + if effect == "query": + st = h.read_zone(self.fd, self.zone) + if not st: + return "err no reply" + return "ok z%d mode=0x%02x rgb=(%d,%d,%d) speed=%d all=%d" % ( + st["zone"], st["mode"], st["r"], st["g"], st["b"], st["speed"], st["all"]) + + if effect == "state": + c = self.current + if c["kind"] == "native": + desc = "native mode=0x%02x rgb=%s speed=%d" % (c["mode"], c["rgb"], c["speed"]) + else: + nxt = c["frames"][c["idx"]] if c["idx"] < len(c["frames"]) else None + desc = "frames %d/%d next=%s" % (c["idx"], len(c["frames"]), nxt) + return "ok zone=%d all=%d | current=%s | idle=%s" % ( + self.zone, self.allzone, desc, self.idle) + + if effect == "commit": + h.commit(self.fd) + return "ok committed" + + if effect == "help": + return ("ok static|off|idle | pulse [n] [s] | blink [n] [s] | flash | " + + "|".join(NATIVE_TAKES_COLOR) + " [speed] [color] | query | commit | zone ") + + raise ValueError("unknown effect: %s (try 'help')" % effect) + + def render(self): + cur = self.current + if cur is None: + return + if cur["kind"] == "native": + if not cur["written"]: + self.write(cur["mode"], cur["rgb"], cur["speed"]) + cur["written"] = True + elif cur["kind"] == "frames": + if cur["idx"] < len(cur["frames"]): + self.write(h.A_STATIC, cur["frames"][cur["idx"]], 0xE0) + cur["idx"] += 1 + else: + self.current = dict(self.idle) + + def run(self): + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + os.unlink(SOCK_PATH) + except FileNotFoundError: + pass + sock.bind(SOCK_PATH) + os.chmod(SOCK_PATH, 0o600) + sock.listen(16) + sock.settimeout(FRAME) + next_tick = time.monotonic() + while True: + try: + conn, _ = sock.accept() + except socket.timeout: + conn = None + if conn is not None: + with conn: + data = b"" + while b"\n" not in data: + chunk = conn.recv(1024) + if not chunk: + break + data += chunk + try: + ack = self.apply(data.decode().strip()) + except Exception as e: + ack = "err %s" % e + conn.sendall(ack.encode() + b"\n") + now = time.monotonic() + if now >= next_tick: + self.render() + next_tick = now + FRAME + + +def ctl(args): + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.settimeout(1.5) + sock.connect(SOCK_PATH) + sock.sendall(" ".join(args).encode() + b"\n") + data = b"" + while b"\n" not in data: + chunk = sock.recv(1024) + if not chunk: + break + data += chunk + print(data.decode().strip()) + + +def main(): + args = sys.argv[1:] + if not args or args[0] == "daemon": + Daemon().run() + elif args[0] == "ctl": + ctl(args[1:]) + else: + sys.exit("usage: asrock-indicator.py daemon | ctl ") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/asrock-led.py b/asrock-led.py new file mode 100755 index 0000000..68f3e75 --- /dev/null +++ b/asrock-led.py @@ -0,0 +1,54 @@ +#!/usr/bin/env python3 +import sys +import asrock_hid as h + +MODES = { + "off": 0x00, "static": 0x01, "breath": 0x02, "strobe": 0x03, + "cycle": 0x04, "random": 0x05, "music": 0x06, "wave": 0x07, + "spring": 0x08, "stack": 0x09, "cram": 0x0A, "scan": 0x0B, + "neon": 0x0C, "water": 0x0D, "rainbow": 0x0E, "direct": 0x0F, +} + + +def to_hex(data): + return " ".join("%02x" % b for b in data) if data else "no reply" + + +def main(): + args = sys.argv[1:] + if not args: + sys.exit( + "usage: asrock-led.py read | set [all|] [speed] | commit" + ) + fd = h.open_dev() + try: + cmd_name = args[0].lower() + if cmd_name == "read": + zone = int(args[1], 0) + state = h.read_zone(fd, zone) + print(state if state else "no reply") + elif cmd_name == "set": + is_all = args[1] == "all" + zone = 0 if is_all else int(args[1], 0) + mode = MODES[args[2]] if args[2] in MODES else int(args[2], 0) + r = int(args[3], 0) + g = int(args[4], 0) + b = int(args[5], 0) + speed = int(args[6], 0) if len(args) > 6 else 0xE0 + reply = h.write_set(fd, zone, mode, r, g, b, speed, 1 if is_all else 0) + print(to_hex(reply)) + elif cmd_name == "commit": + print(to_hex(h.commit(fd))) + else: + sys.exit(f"unknown command: {cmd_name}") + finally: + os_close(fd) + + +def os_close(fd): + import os + os.close(fd) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/asrock-led.service b/asrock-led.service new file mode 100644 index 0000000..c1341b4 --- /dev/null +++ b/asrock-led.service @@ -0,0 +1,12 @@ +[Unit] +Description=ASRock Polychrome system indicator daemon +After=default.target + +[Service] +Type=simple +ExecStart=/usr/bin/python3 /home/owen/asrock-indicator.py daemon +Restart=always +RestartSec=2 + +[Install] +WantedBy=default.target \ No newline at end of file diff --git a/asrock_hid.py b/asrock_hid.py new file mode 100644 index 0000000..d618780 --- /dev/null +++ b/asrock_hid.py @@ -0,0 +1,76 @@ +import os +import select +import time + +DEFAULT_DEV = os.environ.get("ASROCK_HID", "/dev/hidraw0") + +A_OFF = 0x00 +A_STATIC = 0x01 +A_BREATH = 0x02 +A_STROBE = 0x03 +A_CYCLE = 0x04 +A_RANDOM = 0x05 +A_MUSIC = 0x06 +A_WAVE = 0x07 +A_SPRING = 0x08 +A_STACK = 0x09 +A_CRAM = 0x0A +A_SCAN = 0x0B +A_NEON = 0x0C +A_WATER = 0x0D +A_RAINBOW = 0x0E +A_DIRECT = 0x0F + +WRITE = 0x10 +READ = 0x11 +COMMIT = 0x12 + + +def _cmd(c, zone=0, mode=0, r=0, g=0, b=0, speed=0, ff=0, allzone=0): + m = bytearray(64) + m[0] = c + m[2] = zone + m[3] = mode + m[4] = r + m[5] = g + m[6] = b + m[7] = speed + m[8] = ff + m[9] = allzone + return m + + +def open_dev(path=DEFAULT_DEV): + return os.open(path, os.O_RDWR) + + +def xfer(fd, msg): + os.write(fd, bytes(msg)) + time.sleep(0.01) + r, _, _ = select.select([fd], [], [], 0.2) + if r: + return os.read(fd, 256) + return None + + +def write_set(fd, zone, mode, r, g, b, speed, allzone): + return xfer(fd, _cmd(WRITE, zone, mode, r, g, b, speed, 0xFF, allzone)) + + +def read_zone(fd, zone): + reply = xfer(fd, _cmd(READ, zone=zone)) + if not reply or len(reply) < 11: + return None + return { + "zone": reply[3], + "mode": reply[4], + "r": reply[5], + "g": reply[6], + "b": reply[7], + "speed": reply[8], + "all": reply[10], + } + + +def commit(fd): + return xfer(fd, _cmd(COMMIT)) \ No newline at end of file