#!/usr/bin/env python3 """LuxStats Configurator v3 - PyQt5 applet for the Tiny 2040 status-LED firmware. v3 changes ---------- * Known source types are enforced: picking cpu/eth/wifi/vpn/cam/*_temp/etc locks the zone to its correct type so displays can't be misconfigured * Temperature sources report RAW Celsius; temp zones map them onto a gradient via per-zone min/max degC * Ethernet & Wi-Fi are positional: one color per position (down/10M/100M/ 1G/2.5G/10G and down/2.4/5/6 GHz) instead of an interpolated range; eth reports 0 unless the link is actually active (carrier up) * Camera detection ignores PipeWire/WirePlumber/PulseAudio holding the device open, and camera state is edge-triggered (on AND off are sent) * Camera glow uses assigned LEDs with range syntax ("1-5,9-13") and the same animation presets as ambient zones (solid/breathe/rainbow/sparkle) * Any start/stop test (zone range, camera glow, raw fill) pauses automated polling and resumes it when the test ends; fake-status buttons pause polling for 10 s so the fake value stays visible * Polling interval is set in 100 ms steps (minimum 100 ms; slow probes are internally cached so 10 Hz polling stays cheap) * Config persists everything on the host - zones, device identity ("Tiny 2040 (8MB) - CircuitPython CDC2 control" on any /dev/ttyACM#), blinker period/count, notification-watcher toggle, camera-automation toggle - and the app auto-connects to the named device on startup Run: python3 luxstats_configurator.py """ import copy import json import os import sys import threading import time import serial import serial.tools.list_ports from PyQt5.QtCore import Qt, QThread, QTimer, pyqtSignal from PyQt5.QtGui import QColor from PyQt5.QtWidgets import ( QApplication, QCheckBox, QColorDialog, QComboBox, QDialog, QDialogButtonBox, QDoubleSpinBox, QFileDialog, QFormLayout, QGridLayout, QGroupBox, QHBoxLayout, QLabel, QLineEdit, QMainWindow, QMessageBox, QPushButton, QSlider, QSpinBox, QStatusBar, QTableWidget, QTabWidget, QTextEdit, QVBoxLayout, QWidget, ) import system_monitor RP2040_VID = 0x2E8A DEFAULT_DEVICE_NAME = "Tiny 2040 (8MB) - CircuitPython CDC2 control" SEG_TYPES = ["percent", "temp", "link", "wifi", "vpn", "bool", "custom", "ambient"] GRADIENT_TYPES = {"percent", "temp"} PALETTE_TYPES = {"link", "wifi"} PIXEL_ORDERS = ["GRBW", "RGBW", "GRB", "RGB"] AMBIENT_EFFECTS = ["off", "solid", "breathe", "rainbow", "sparkle"] CAMERA_EFFECTS = ["solid", "breathe", "rainbow", "sparkle"] LINK_POSITIONS = ["down", "10M", "100M", "1G", "2.5G", "10G"] DEFAULT_LINK_COLORS = [[0, 0, 0, 0], [255, 0, 0, 0], [255, 90, 0, 0], [0, 255, 0, 0], [0, 255, 180, 0], [80, 0, 255, 0]] WIFI_POSITIONS = ["down", "2.4 GHz", "5 GHz", "6 GHz"] DEFAULT_WIFI_COLORS = [[0, 0, 0, 0], [255, 120, 0, 0], [0, 255, 60, 0], [40, 80, 255, 0]] DEFAULT_GRADIENT = [[0.0, [0, 0, 255, 0]], [0.5, [0, 255, 0, 0]], [1.0, [255, 0, 0, 0]]] # Per-source default degC windows for temp zones TEMP_DEFAULTS = {"cpu_temp": (20, 90), "gpu_temp": (20, 95)} DISK_TEMP_DEFAULT = (20, 70) DEFAULT_SEGMENTS = [ {"name": "cpu", "type": "percent", "start": 0, "count": 2, "fill": True, "gradient": copy.deepcopy(DEFAULT_GRADIENT)}, {"name": "eth", "type": "link", "start": 2, "count": 1}, {"name": "wifi", "type": "wifi", "start": 3, "count": 1}, {"name": "vpn", "type": "vpn", "start": 4, "count": 1}, {"name": "glow", "type": "ambient", "start": 5, "count": 3, "effect": "breathe", "color": [10, 10, 30, 0], "speed": 1.0}, ] def known_type(name): """Fixed type for a known source name (None = user's choice).""" name = (name or "").strip() if name in ("cpu", "ram", "gpu"): return "percent" if name == "eth": return "link" if name == "wifi": return "wifi" if name == "vpn": return "vpn" if name == "cam": return "bool" if name.endswith("_temp") or name == "disk_temp": return "temp" if name.startswith("disk"): return "percent" return None def default_temp_range(name): if name in TEMP_DEFAULTS: return TEMP_DEFAULTS[name] if name.startswith("disk"): return DISK_TEMP_DEFAULT return (20, 90) def parse_ranges(text): """'1-5,9-13,20' -> [1,2,3,4,5,9,10,11,12,13,20] (LED indices).""" out = [] for part in (text or "").replace(";", ",").split(","): part = part.strip() if not part: continue if "-" in part: a, _, b = part.partition("-") if a.strip().isdigit() and b.strip().isdigit(): lo, hi = int(a), int(b) if lo > hi: lo, hi = hi, lo out.extend(range(lo, hi + 1)) elif part.isdigit(): out.append(int(part)) return sorted(set(out)) def format_ranges(indices): """[1,2,3,4,5,9] -> '1-5,9'.""" indices = sorted(set(int(i) for i in indices)) if not indices: return "" parts, a, b = [], indices[0], indices[0] for i in indices[1:]: if i == b + 1: b = i else: parts.append(f"{a}-{b}" if b > a else f"{a}") a = b = i parts.append(f"{a}-{b}" if b > a else f"{a}") return ",".join(parts) def app_config_path(): if sys.platform == "win32": base = os.environ.get("APPDATA", os.path.expanduser("~")) else: base = os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config")) d = os.path.join(base, "luxstats") os.makedirs(d, exist_ok=True) return os.path.join(d, "config.json") # -------------------------------------------------------------------------- # Serial link # -------------------------------------------------------------------------- class SerialLink: def __init__(self, log): self.ser = None self.log = log self.lock = threading.Lock() self.last_activity = 0.0 @staticmethod def list_ports(): ports = list(serial.tools.list_ports.comports()) ports.sort(key=lambda p: (p.vid != RP2040_VID, p.device)) return ports @staticmethod def find_device(device_name): """Locate the named CDC2 control port on any /dev/ttyACM#.""" matches = [] for p in serial.tools.list_ports.comports(): desc = " ".join(filter(None, [p.description, p.interface or ""])) if device_name and device_name.lower() in desc.lower(): matches.append(p) if not matches: return None # Prefer the interface explicitly named as the CDC2 data channel, # otherwise the highest-numbered interface of the board. for p in matches: joined = f"{p.description} {p.interface or ''}".lower() if "cdc2" in joined or "data" in joined: return p.device matches.sort(key=lambda p: (p.location or "", p.device)) return matches[-1].device def open(self, device): self.close() self.ser = serial.Serial(device, 115200, timeout=0.5, write_timeout=1) time.sleep(0.2) self.ser.reset_input_buffer() self.last_activity = time.monotonic() def close(self): if self.ser: try: self.ser.close() except OSError: pass self.ser = None @property def connected(self): return self.ser is not None and self.ser.is_open def _drain(self): while self.ser.in_waiting: self.ser.readline() def send(self, obj, expect_reply=True): if not self.connected: return None with self.lock: if not self.connected: return None try: self._drain() self.ser.write((json.dumps(obj) + "\n").encode()) self.last_activity = time.monotonic() if not expect_reply: return None reply = self.ser.readline().decode(errors="replace").strip() if reply: self.last_activity = time.monotonic() try: return json.loads(reply) except ValueError: return {"raw": reply} except (OSError, serial.SerialException) as e: self.log(f"serial error: {e}") self.close() return None # -------------------------------------------------------------------------- # Background automation thread (pausable) # -------------------------------------------------------------------------- class MonitorThread(QThread): statuses = pyqtSignal(dict) def __init__(self): super().__init__() self.interval = 2.0 self.enabled_probes = None self.running = True self.paused = False def run(self): system_monitor.psutil.cpu_percent(interval=None) while self.running: if not self.paused: try: self.statuses.emit( system_monitor.collect(self.enabled_probes)) except Exception as e: self.statuses.emit({"_error": str(e)}) t = 0.0 while self.running and t < self.interval: self.msleep(50) t += 0.05 def stop(self): self.running = False # -------------------------------------------------------------------------- # Small widgets # -------------------------------------------------------------------------- class ColorButton(QPushButton): changed = pyqtSignal() def __init__(self, rgbw=(0, 60, 255, 0)): super().__init__() self._rgbw = list(rgbw)[:4] + [0] * (4 - len(rgbw)) self.clicked.connect(self.pick) self._refresh() def _refresh(self): r, g, b, _ = self._rgbw lum = 0.299 * r + 0.587 * g + 0.114 * b fg = "black" if lum > 128 else "white" self.setStyleSheet( f"background-color: rgb({r},{g},{b}); color:{fg};" "min-width: 72px;") self.setText(f"({r},{g},{b},{self._rgbw[3]})") def pick(self): r, g, b, _ = self._rgbw c = QColorDialog.getColor(QColor(r, g, b), self, "Pick color") if c.isValid(): self._rgbw[:3] = [c.red(), c.green(), c.blue()] self._refresh() self.changed.emit() def set_white(self, w): self._rgbw[3] = int(w) self._refresh() self.changed.emit() def rgbw(self): return list(self._rgbw) def set_rgbw(self, rgbw): self._rgbw = list(rgbw)[:4] + [0] * (4 - len(rgbw)) self._refresh() # -------------------------------------------------------------------------- # Per-zone settings dialog # -------------------------------------------------------------------------- class SegmentDialog(QDialog): def __init__(self, seg, parent=None): super().__init__(parent) self.seg = seg stype = seg.get("type", "percent") name = seg.get("name", "") self.setWindowTitle(f"Zone settings - {name} ({stype})") lay = QVBoxLayout(self) self.grad_table = None self.palette_btns = None self.effect_combo = None self.on_btn = self.off_btn = self.custom_btn = None self.min_c = self.max_c = None if stype == "temp": form = QFormLayout() lo, hi = default_temp_range(name) self.min_c = QDoubleSpinBox() self.min_c.setRange(-50, 200); self.min_c.setSuffix(" °C") self.min_c.setValue(float(seg.get("min_c", lo))) self.max_c = QDoubleSpinBox() self.max_c.setRange(-50, 200); self.max_c.setSuffix(" °C") self.max_c.setValue(float(seg.get("max_c", hi))) form.addRow("Gradient start at:", self.min_c) form.addRow("Gradient end at:", self.max_c) lay.addLayout(form) lay.addWidget(QLabel( "Raw Celsius values map onto the gradient between these " "temperatures (clamped outside).")) if stype in GRADIENT_TYPES: lay.addWidget(QLabel("Gradient color stops (position 0.0-1.0). " "Remove all stops for the built-in " "blue→green→red.")) self.grad_table = QTableWidget(0, 3) self.grad_table.setHorizontalHeaderLabels( ["Position", "Color", "White"]) self.grad_table.horizontalHeader().setStretchLastSection(True) for pos, color in seg.get("gradient", []) or []: self._add_stop(pos, color) lay.addWidget(self.grad_table) row = QHBoxLayout() b_add = QPushButton("Add stop") b_add.clicked.connect(lambda: self._add_stop(1.0, [255, 0, 0, 0])) b_del = QPushButton("Remove selected") b_del.clicked.connect(lambda: self.grad_table.removeRow( self.grad_table.currentRow())) b_def = QPushButton("Load default gradient") b_def.clicked.connect(self._load_default) row.addWidget(b_add); row.addWidget(b_del); row.addWidget(b_def) row.addStretch(1) lay.addLayout(row) elif stype in PALETTE_TYPES: positions = LINK_POSITIONS if stype == "link" else WIFI_POSITIONS defaults = (DEFAULT_LINK_COLORS if stype == "link" else DEFAULT_WIFI_COLORS) lay.addWidget(QLabel( "One color per position (no blending between positions).")) form = QFormLayout() colors = seg.get("colors") or copy.deepcopy(defaults) self.palette_btns = [] for i, label in enumerate(positions): c = colors[i] if i < len(colors) else defaults[i] btn = ColorButton(c) w = QSpinBox(); w.setRange(0, 255) w.setValue(int(c[3]) if len(c) > 3 else 0) w.valueChanged.connect(btn.set_white) row = QHBoxLayout() row.addWidget(btn); row.addWidget(QLabel("W:")) row.addWidget(w); row.addStretch(1) form.addRow(f"{label}:", row) self.palette_btns.append(btn) lay.addLayout(form) elif stype == "ambient": form = QFormLayout() self.effect_combo = QComboBox() self.effect_combo.addItems(AMBIENT_EFFECTS) self.effect_combo.setCurrentText(seg.get("effect", "breathe")) self.amb_color = ColorButton(seg.get("color", [10, 10, 30, 0])) self.amb_white = QSpinBox(); self.amb_white.setRange(0, 255) col = seg.get("color", [0, 0, 0, 0]) self.amb_white.setValue(int(col[3]) if len(col) > 3 else 0) self.amb_white.valueChanged.connect(self.amb_color.set_white) self.amb_speed = QDoubleSpinBox() self.amb_speed.setRange(0.1, 5.0); self.amb_speed.setSingleStep(0.1) self.amb_speed.setValue(float(seg.get("speed", 1.0))) crow = QHBoxLayout() crow.addWidget(self.amb_color) crow.addWidget(QLabel("W:")); crow.addWidget(self.amb_white) form.addRow("Effect:", self.effect_combo) form.addRow("Color:", crow) form.addRow("Speed:", self.amb_speed) lay.addLayout(form) elif stype in ("vpn", "bool"): form = QFormLayout() on_def = [170, 0, 255, 0] if stype == "vpn" else [0, 255, 0, 0] off_def = [6, 6, 6, 0] if stype == "vpn" else [4, 0, 0, 0] self.on_btn = ColorButton(seg.get("on_color", on_def)) self.off_btn = ColorButton(seg.get("off_color", off_def)) form.addRow("Color when ON:", self.on_btn) form.addRow("Color when OFF:", self.off_btn) lay.addLayout(form) elif stype == "custom": form = QFormLayout() self.custom_btn = ColorButton(seg.get("color", [0, 0, 0, 0])) form.addRow("Fallback color (until a value is pushed):", self.custom_btn) lay.addLayout(form) buttons = QDialogButtonBox( QDialogButtonBox.Ok | QDialogButtonBox.Cancel) buttons.accepted.connect(self.accept) buttons.rejected.connect(self.reject) lay.addWidget(buttons) def _add_stop(self, pos, color): r = self.grad_table.rowCount() self.grad_table.insertRow(r) spin = QDoubleSpinBox() spin.setRange(0.0, 1.0); spin.setSingleStep(0.05) spin.setDecimals(2); spin.setValue(float(pos)) btn = ColorButton(color) w = QSpinBox(); w.setRange(0, 255) w.setValue(int(color[3]) if len(color) > 3 else 0) w.valueChanged.connect(btn.set_white) self.grad_table.setCellWidget(r, 0, spin) self.grad_table.setCellWidget(r, 1, btn) self.grad_table.setCellWidget(r, 2, w) def _load_default(self): self.grad_table.setRowCount(0) for pos, color in copy.deepcopy(DEFAULT_GRADIENT): self._add_stop(pos, color) def apply_to_segment(self): if self.min_c is not None: self.seg["min_c"] = self.min_c.value() self.seg["max_c"] = self.max_c.value() if self.grad_table is not None: stops = [] for r in range(self.grad_table.rowCount()): stops.append([self.grad_table.cellWidget(r, 0).value(), self.grad_table.cellWidget(r, 1).rgbw()]) stops.sort(key=lambda s: s[0]) self.seg["gradient"] = stops if self.palette_btns is not None: self.seg["colors"] = [b.rgbw() for b in self.palette_btns] if self.effect_combo is not None: self.seg["effect"] = self.effect_combo.currentText() self.seg["color"] = self.amb_color.rgbw() self.seg["speed"] = self.amb_speed.value() if self.on_btn is not None: self.seg["on_color"] = self.on_btn.rgbw() self.seg["off_color"] = self.off_btn.rgbw() if self.custom_btn is not None: self.seg["color"] = self.custom_btn.rgbw() # -------------------------------------------------------------------------- # Main window # -------------------------------------------------------------------------- class MainWindow(QMainWindow): KEEPALIVE_S = 10 RECONNECT_S = 3 FAKE_PAUSE_S = 10 # fake-status buttons pause polling this long notification_seen = pyqtSignal() def __init__(self): super().__init__() self.setWindowTitle("LuxStats Configurator") self.resize(960, 720) self.link = SerialLink(self.log) self.monitor = MonitorThread() self.monitor.statuses.connect(self.on_statuses) self.notify_watcher = None self.notification_seen.connect(self.fire_notification) self.segments = copy.deepcopy(DEFAULT_SEGMENTS) self._building = False self._active_test_row = None self._raw_test = False self._last_cam = None self.device_name = DEFAULT_DEVICE_NAME self._last_port = None self._user_disconnected = False # auto-connect from the start self.push_timer = QTimer(self) self.push_timer.setSingleShot(True) self.push_timer.setInterval(400) self.push_timer.timeout.connect(self._auto_push) self.fake_resume_timer = QTimer(self) self.fake_resume_timer.setSingleShot(True) self.fake_resume_timer.setInterval(self.FAKE_PAUSE_S * 1000) self.fake_resume_timer.timeout.connect( lambda: self.resume_polling("fake status")) self.keepalive_timer = QTimer(self) self.keepalive_timer.setInterval(self.RECONNECT_S * 1000) self.keepalive_timer.timeout.connect(self._keepalive) self.keepalive_timer.start() tabs = QTabWidget() tabs.addTab(self._build_connection_tab(), "Connection") tabs.addTab(self._build_device_tab(), "Device") tabs.addTab(self._build_segments_tab(), "Zones && Mapping") tabs.addTab(self._build_automation_tab(), "Automation") tabs.addTab(self._build_test_tab(), "Test") self.setCentralWidget(tabs) self.setStatusBar(QStatusBar()) self._wire_autopush() self.load_app_config() self.refresh_ports() self.rebuild_seg_table() QTimer.singleShot(300, self._startup_connect) # ---------------- Connection tab ---------------- def _build_connection_tab(self): w = QWidget() lay = QVBoxLayout(w) row0 = QHBoxLayout() self.device_name_edit = QLineEdit(DEFAULT_DEVICE_NAME) row0.addWidget(QLabel("Device name (auto-connect):")) row0.addWidget(self.device_name_edit, 1) lay.addLayout(row0) row = QHBoxLayout() self.port_combo = QComboBox() btn_refresh = QPushButton("Refresh") btn_refresh.clicked.connect(self.refresh_ports) self.btn_connect = QPushButton("Connect") self.btn_connect.clicked.connect(self.toggle_connect) row.addWidget(QLabel("Serial port:")) row.addWidget(self.port_combo, 1) row.addWidget(btn_refresh) row.addWidget(self.btn_connect) lay.addLayout(row) hint = QLabel( "The app looks for the named device on any /dev/ttyACM# port at " "startup and whenever the link drops, and re-pushes the stored " "config on every connect. Config lives on this computer only.") hint.setWordWrap(True) lay.addWidget(hint) self.log_box = QTextEdit() self.log_box.setReadOnly(True) lay.addWidget(self.log_box, 1) row2 = QHBoxLayout() for label, fn in [ ("Ping", self.do_ping), ("Push config now", self.push_config), ("Export profile...", self.save_profile), ("Import profile...", self.load_profile), ]: b = QPushButton(label) b.clicked.connect(fn) row2.addWidget(b) lay.addLayout(row2) return w def refresh_ports(self): self.port_combo.clear() for p in SerialLink.list_ports(): tag = " [RP2040]" if p.vid == RP2040_VID else "" self.port_combo.addItem( f"{p.device} - {p.description}{tag}", p.device) def _startup_connect(self): if self.link.connected or self._user_disconnected: return dev = SerialLink.find_device(self.device_name_edit.text()) \ or self._last_port if dev and self._connect_to(dev, quiet=True): self.log("auto-connected on startup") def toggle_connect(self): if self.link.connected: self._user_disconnected = True self.link.close() self.btn_connect.setText("Connect") self.log("disconnected") return dev = self.port_combo.currentData() \ or SerialLink.find_device(self.device_name_edit.text()) if not dev: self.log("no port selected and named device not found") return self._user_disconnected = False self._connect_to(dev) def _connect_to(self, dev, quiet=False): try: self.link.open(dev) except (OSError, serial.SerialException) as e: if not quiet: self.log(f"open failed: {e}") return False self._last_port = dev self.btn_connect.setText("Disconnect") self.log(f"connected to {dev}") reply = self.link.send({"cmd": "ping"}) if reply and reply.get("pong"): self.log("device answered ping - pushing config") self.push_config() return True self.log("no pong - is this the DATA port and is code.py running?") return False def _keepalive(self): if self._user_disconnected: return if self.link.connected: if time.monotonic() - self.link.last_activity >= self.KEEPALIVE_S: reply = self.link.send({"cmd": "ping"}) if not (reply and reply.get("pong")): self.log("keepalive ping failed - link lost, will " "auto-reconnect") self.link.close() return self.btn_connect.setText("Connect") dev = SerialLink.find_device(self.device_name_edit.text()) \ or self._last_port if dev and self._connect_to(dev, quiet=True): self.log("auto-reconnected") def do_ping(self): self.log(f"ping -> {self.link.send({'cmd': 'ping'})}") def send_logged(self, obj): self.log(f"{obj.get('cmd')} -> {self.link.send(obj)}") def log(self, msg): self.log_box.append(f"[{time.strftime('%H:%M:%S')}] {msg}") self.statusBar().showMessage(msg, 4000) # ---------------- Device tab ---------------- def _build_device_tab(self): w = QWidget() lay = QVBoxLayout(w) gb = QGroupBox("Strip") form = QFormLayout(gb) self.num_pixels = QSpinBox(); self.num_pixels.setRange(1, 1024) self.num_pixels.setValue(8) self.pin_edit = QLineEdit("GP0") self.order_combo = QComboBox(); self.order_combo.addItems(PIXEL_ORDERS) self.fps_spin = QSpinBox(); self.fps_spin.setRange(5, 120) self.fps_spin.setValue(30) self.brightness = QSlider(Qt.Horizontal) self.brightness.setRange(1, 100); self.brightness.setValue(10) self.bright_label = QLabel("10%") self.brightness.valueChanged.connect( lambda v: self.bright_label.setText(f"{v}%")) brow = QHBoxLayout(); brow.addWidget(self.brightness, 1) brow.addWidget(self.bright_label) form.addRow("Number of LEDs:", self.num_pixels) form.addRow("Data pin:", self.pin_edit) form.addRow("Pixel order:", self.order_combo) form.addRow("Frame rate:", self.fps_spin) form.addRow("Brightness:", brow) lay.addWidget(gb) lay.addWidget(QLabel("Changes push to the device automatically.")) lay.addStretch(1) return w # ---------------- Zones & mapping tab ---------------- def _build_segments_tab(self): w = QWidget() lay = QVBoxLayout(w) lay.addWidget(QLabel( "Ordered zones. Known sources lock to their correct type " "(temps use raw °C, eth/wifi use one color per position). " "Zones may not overlap; unassigned pixels stay off. Test blinks " "a zone's range and pauses polling until stopped.")) self.seg_table = QTableWidget(0, 7) self.seg_table.setHorizontalHeaderLabels( ["Source / name", "Type", "Start", "Count", "Fill", "Settings", "Test"]) self.seg_table.horizontalHeader().setStretchLastSection(True) lay.addWidget(self.seg_table, 1) self.overlap_label = QLabel("") self.overlap_label.setStyleSheet("color: #c02020; font-weight: bold;") self.overlap_label.setWordWrap(True) lay.addWidget(self.overlap_label) row = QHBoxLayout() for label, fn in [("Add zone", self.add_segment), ("Remove selected", self.remove_segment), ("Move up", lambda: self.move_segment(-1)), ("Move down", lambda: self.move_segment(+1))]: b = QPushButton(label) b.clicked.connect(fn) row.addWidget(b) row.addStretch(1) lay.addLayout(row) gb = QGroupBox("Camera glow (overlay on assigned LEDs)") form = QFormLayout(gb) self.cam_pixels = QLineEdit("5-7") self.cam_pixels.setPlaceholderText("e.g. 1-5,9-13") self.cam_effect = QComboBox(); self.cam_effect.addItems(CAMERA_EFFECTS) self.cam_effect.setCurrentText("breathe") self.cam_speed = QDoubleSpinBox() self.cam_speed.setRange(0.1, 5.0); self.cam_speed.setSingleStep(0.1) self.cam_speed.setValue(1.0) self.cam_color = ColorButton((255, 40, 0, 0)) self.cam_test_btn = QPushButton("Test glow (start/stop)") self.cam_test_btn.setCheckable(True) self.cam_test_btn.toggled.connect(self.toggle_camera_test) form.addRow("LEDs (ranges, e.g. 1-5,9-13):", self.cam_pixels) form.addRow("Animation:", self.cam_effect) form.addRow("Speed:", self.cam_speed) form.addRow("Glow color:", self.cam_color) form.addRow("", self.cam_test_btn) lay.addWidget(gb) gb2 = QGroupBox("Notification blinker (overlay)") form2 = QFormLayout(gb2) self.notify_start = QSpinBox(); self.notify_start.setRange(0, 1023) self.notify_count = QSpinBox(); self.notify_count.setRange(0, 1024) self.notify_count.setValue(0) self.notify_color = ColorButton((0, 60, 255, 0)) self.notify_period = QDoubleSpinBox() self.notify_period.setRange(0.05, 5.0) self.notify_period.setValue(0.4); self.notify_period.setSingleStep(0.05) self.notify_blinks = QSpinBox() self.notify_blinks.setRange(-1, 100); self.notify_blinks.setValue(6) nrow = QHBoxLayout() nrow.addWidget(QLabel("Start:")); nrow.addWidget(self.notify_start) nrow.addWidget(QLabel("Count (0 = all):")) nrow.addWidget(self.notify_count) nrow.addStretch(1) form2.addRow("Range:", nrow) form2.addRow("Default color:", self.notify_color) form2.addRow("Blink period (s):", self.notify_period) form2.addRow("Blink count (-1 = forever):", self.notify_blinks) lay.addWidget(gb2) return w # ----- segment model <-> table ----- def rebuild_seg_table(self): self._building = True sources = system_monitor.discover_sources() + ["glow", "custom1"] self.seg_table.setRowCount(0) for idx, seg in enumerate(self.segments): # enforce known types on load/rebuild (requirement 7) fixed = known_type(seg.get("name", "")) if fixed and seg.get("type") != fixed: seg["type"] = fixed r = self.seg_table.rowCount() self.seg_table.insertRow(r) name = QComboBox(); name.setEditable(True) name.addItems(sources) name.setCurrentText(str(seg.get("name", ""))) name.currentTextChanged.connect( lambda v, i=idx: self._seg_edit(i, "name", v)) stype = QComboBox(); stype.addItems(SEG_TYPES) stype.setCurrentText(seg.get("type", "percent")) stype.setEnabled(fixed is None) stype.setToolTip("Type is fixed for known sources" if fixed else "") stype.currentTextChanged.connect( lambda v, i=idx: self._seg_edit(i, "type", v)) start = QSpinBox(); start.setRange(0, 1023) start.setValue(int(seg.get("start", 0))) start.valueChanged.connect( lambda v, i=idx: self._seg_edit(i, "start", v)) count = QSpinBox(); count.setRange(1, 1024) count.setValue(int(seg.get("count", 1))) count.valueChanged.connect( lambda v, i=idx: self._seg_edit(i, "count", v)) fill = QCheckBox() fill.setChecked(bool(seg.get("fill", False))) fill.setEnabled(seg.get("type") == "percent") fill.toggled.connect( lambda v, i=idx: self._seg_edit(i, "fill", v)) settings = QPushButton("Settings...") settings.clicked.connect(lambda _, i=idx: self.open_settings(i)) test = QPushButton("Test") test.setCheckable(True) test.setChecked(self._active_test_row == idx) test.toggled.connect(lambda on, i=idx: self.toggle_seg_test(i, on)) for col, widget in enumerate( [name, stype, start, count, fill, settings, test]): self.seg_table.setCellWidget(r, col, widget) self._building = False self.validate_segments() def _seg_edit(self, idx, key, value): if self._building or idx >= len(self.segments): return seg = self.segments[idx] seg[key] = value if key == "name": fixed = known_type(value) if fixed and seg.get("type") != fixed: seg["type"] = fixed seg.pop("gradient", None) seg.pop("colors", None) self.rebuild_seg_table() else: type_combo = self.seg_table.cellWidget(idx, 1) if type_combo: type_combo.setEnabled(fixed is None) if key == "type": seg.pop("gradient", None) seg.pop("colors", None) fill = self.seg_table.cellWidget(idx, 4) if fill: fill.setEnabled(value == "percent") self.schedule_push() self.validate_segments() if self._active_test_row == idx and key in ("start", "count"): self.link.send({"cmd": "test", "start": seg["start"], "count": seg["count"], "active": True}, expect_reply=False) def add_segment(self): end = max((s.get("start", 0) + s.get("count", 1) for s in self.segments), default=0) self.segments.append({"name": "custom1", "type": "custom", "start": end, "count": 1}) self.rebuild_seg_table() self.schedule_push() def remove_segment(self): r = self.seg_table.currentRow() if 0 <= r < len(self.segments): if self._active_test_row == r: self._stop_range_test() del self.segments[r] self._active_test_row = None self.rebuild_seg_table() self.schedule_push() def move_segment(self, delta): r = self.seg_table.currentRow() nr = r + delta if 0 <= r < len(self.segments) and 0 <= nr < len(self.segments): self.segments[r], self.segments[nr] = \ self.segments[nr], self.segments[r] if self._active_test_row == r: self._active_test_row = nr elif self._active_test_row == nr: self._active_test_row = r self.rebuild_seg_table() self.seg_table.selectRow(nr) self.schedule_push() def open_settings(self, idx): if idx >= len(self.segments): return dlg = SegmentDialog(self.segments[idx], self) if dlg.exec_() == QDialog.Accepted: dlg.apply_to_segment() self.schedule_push() # ----- polling pause/resume around tests (requirement 8) ----- def pause_polling(self, reason): if not self.monitor.paused: self.monitor.paused = True self.log(f"polling paused ({reason})") def resume_polling(self, reason): if self.monitor.paused and self._active_test_row is None \ and not self.cam_test_btn.isChecked() and not self._raw_test \ and not self.fake_resume_timer.isActive(): self.monitor.paused = False self.log(f"polling resumed ({reason} done)") # ----- range tests ----- def toggle_seg_test(self, idx, on): if self._building: return if on: if self._active_test_row is not None \ and self._active_test_row != idx: old = self.seg_table.cellWidget(self._active_test_row, 6) if old: old.blockSignals(True) old.setChecked(False) old.blockSignals(False) self._active_test_row = idx self.pause_polling("zone test") seg = self.segments[idx] self.send_logged({"cmd": "test", "start": seg.get("start", 0), "count": seg.get("count", 1), "active": True}) else: if self._active_test_row == idx: self._stop_range_test() def _stop_range_test(self): self._active_test_row = None self.link.send({"cmd": "test", "active": False}, expect_reply=False) self.resume_polling("zone test") def toggle_camera_test(self, on): if on: self.pause_polling("camera glow test") self.send_logged({"cmd": "camera", "active": bool(on)}) self.cam_test_btn.setText( "Stop glow test" if on else "Test glow (start/stop)") if not on: self._last_cam = None # re-sync real camera state self.resume_polling("camera glow test") # ----- overlap validation ----- def validate_segments(self): n = self.num_pixels.value() owner = {} problems = [] for idx, seg in enumerate(self.segments): s, c = int(seg.get("start", 0)), int(seg.get("count", 1)) label = f"#{idx + 1} {seg.get('name', '?')}" if s + c > n: problems.append(f"{label} runs past the strip " f"(LED {s + c - 1} > {n - 1})") for i in range(s, min(s + c, n)): if i in owner: problems.append(f"{label} overlaps {owner[i]} at LED {i}") break owner[i] = label bad_cam = [i for i in parse_ranges(self.cam_pixels.text()) if i >= n] if hasattr(self, "cam_pixels") else [] if bad_cam: problems.append(f"camera glow LEDs {bad_cam} past the strip") if problems: self.overlap_label.setText( "Not pushed - fix these first:\n" + "\n".join(problems)) else: self.overlap_label.setText("") return not problems # ---------------- Automation tab ---------------- def _build_automation_tab(self): w = QWidget() lay = QVBoxLayout(w) self.probe_group = QGroupBox( "Live status sources (drives auto-detected)") self.probe_grid = QGridLayout(self.probe_group) self.probe_checks = {} self._rebuild_probes() lay.addWidget(self.probe_group) row = QHBoxLayout() b_re = QPushButton("Re-detect sources") b_re.clicked.connect(self._rebuild_probes) self.interval_spin = QDoubleSpinBox() self.interval_spin.setRange(0.1, 60.0) # 100 ms steps (req 9) self.interval_spin.setSingleStep(0.1) self.interval_spin.setDecimals(1) self.interval_spin.setSuffix(" s") self.interval_spin.setValue(1.0) self.btn_monitor = QPushButton("Start automation") self.btn_monitor.setCheckable(True) self.btn_monitor.toggled.connect(self.toggle_monitor) row.addWidget(b_re) row.addWidget(QLabel("Poll interval (100 ms steps):")) row.addWidget(self.interval_spin) row.addWidget(self.btn_monitor) row.addStretch(1) lay.addLayout(row) gb2 = QGroupBox("Desktop notification watcher (Linux/DBus)") v2 = QVBoxLayout(gb2) self.notify_watch_chk = QCheckBox( "Blink the notification zone on any desktop notification") self.notify_watch_chk.toggled.connect(self.toggle_notify_watcher) v2.addWidget(self.notify_watch_chk) lay.addWidget(gb2) gb3 = QGroupBox("Camera automation") v3 = QVBoxLayout(gb3) self.cam_auto_chk = QCheckBox( "Automatically glow while a real app uses the camera " "(media daemons like PipeWire are ignored)") self.cam_auto_chk.setChecked(True) v3.addWidget(self.cam_auto_chk) lay.addWidget(gb3) self.live_label = QLabel("Live values: -") self.live_label.setWordWrap(True) lay.addWidget(self.live_label) lay.addStretch(1) return w def _rebuild_probes(self): prev = {k: cb.isChecked() for k, cb in self.probe_checks.items()} while self.probe_grid.count(): item = self.probe_grid.takeAt(0) if item.widget(): item.widget().deleteLater() self.probe_checks = {} for i, src in enumerate(system_monitor.discover_sources()): cb = QCheckBox(src) cb.setChecked(prev.get(src, True)) cb.toggled.connect(self._probes_changed) self.probe_checks[src] = cb self.probe_grid.addWidget(cb, i // 4, i % 4) def _probes_changed(self): self.monitor.enabled_probes = { k for k, cb in self.probe_checks.items() if cb.isChecked()} self.save_app_config() def toggle_monitor(self, on): if on: self.monitor.interval = self.interval_spin.value() self._probes_changed() if not self.monitor.isRunning(): self.monitor.running = True self.monitor.start() self.btn_monitor.setText("Stop automation") self.log("automation started") else: self.monitor.stop() self.btn_monitor.setText("Start automation") self.log("automation stopped") def on_statuses(self, values): if "_error" in values: self.log(f"monitor error: {values['_error']}") return self.live_label.setText("Live values: " + json.dumps(values)) if not self.link.connected: return cam = values.pop("cam", None) self.link.send({"cmd": "status", "values": values}, expect_reply=False) # edge-triggered camera state: send BOTH transitions (fixes stuck-on) if cam is not None and self.cam_auto_chk.isChecked() \ and not self.cam_test_btn.isChecked(): if cam != self._last_cam: self._last_cam = cam self.send_logged({"cmd": "camera", "active": bool(cam)}) def toggle_notify_watcher(self, on): if on: if self.notify_watcher is None: self.notify_watcher = system_monitor.NotificationWatcher( self.notification_seen.emit) self.notify_watcher.start() self.log("notification watcher started (needs dbus-monitor)") elif self.notify_watcher: self.notify_watcher.stop() self.notify_watcher = None self.log("notification watcher stopped") self.save_app_config() def fire_notification(self): self.link.send({"cmd": "notify", "color": self.notify_color.rgbw(), "period": self.notify_period.value(), "count": self.notify_blinks.value()}, expect_reply=False) # ---------------- Test tab ---------------- def _build_test_tab(self): w = QWidget() lay = QVBoxLayout(w) lay.addWidget(QLabel( "Raw fill pauses polling until you exit test mode; fake-status " f"buttons pause polling for {self.FAKE_PAUSE_S} s so the value " "stays visible.")) self.test_color = ColorButton((0, 255, 0, 0)) row = QHBoxLayout() row.addWidget(QLabel("Test color:")) row.addWidget(self.test_color) row.addStretch(1) lay.addLayout(row) grid = QGridLayout() fakes = [ ("Fake CPU 90% / GPU 60%", {"cpu": 0.9, "gpu": 0.6}), ("Fake CPU 85 °C / GPU 70 °C", {"cpu_temp": 85.0, "gpu_temp": 70.0}), ("Fake 10G link", {"eth": 10000}), ("Fake eth inactive", {"eth": 0}), ("Fake Wi-Fi 6 GHz", {"wifi": 6}), ("Fake VPN on", {"vpn": True}), ("Fake disk 85% full", {"disk": 0.85}), ("Fake disk 55 °C", {"disk_temp": 55.0}), ] buttons = [ ("Fire notification", self.fire_notification), ("Clear notification", lambda: self.send_logged({"cmd": "notify_clear"})), ("Fill all (test color)", self.test_fill), ("Exit raw/test mode", self.test_raw_off), ] + [(label, lambda _=None, v=vals: self.fake_status(v)) for label, vals in fakes] for i, (label, fn) in enumerate(buttons): b = QPushButton(label) b.clicked.connect(fn) grid.addWidget(b, i // 2, i % 2) lay.addLayout(grid) lay.addStretch(1) return w def fake_status(self, values): self.pause_polling("fake status") self.fake_resume_timer.start() self.send_logged({"cmd": "status", "values": values, "ack": 1}) def test_fill(self): self._raw_test = True self.pause_polling("raw fill") self.send_logged({"cmd": "raw", "pixels": [self.test_color.rgbw()] * self.num_pixels.value()}) def test_raw_off(self): self.send_logged({"cmd": "raw_off"}) self._raw_test = False self.resume_polling("raw fill") # ---------------- Config assembly / persistence ---------------- def build_config(self): return { "num_pixels": self.num_pixels.value(), "pin": self.pin_edit.text().strip() or "GP0", "pixel_order": self.order_combo.currentText(), "brightness": self.brightness.value() / 100.0, "fps": self.fps_spin.value(), "segments": copy.deepcopy(self.segments), "camera": {"pixels": parse_ranges(self.cam_pixels.text()), "color": self.cam_color.rgbw(), "effect": self.cam_effect.currentText(), "speed": self.cam_speed.value()}, "notify": {"start": self.notify_start.value(), "count": self.notify_count.value(), "blinks": self.notify_blinks.value(), "default_color": self.notify_color.rgbw(), "default_period": self.notify_period.value()}, } def apply_config_to_ui(self, cfg): self._building = True self.num_pixels.setValue(int(cfg.get("num_pixels", 8))) self.pin_edit.setText(cfg.get("pin", "GP0")) self.order_combo.setCurrentText(cfg.get("pixel_order", "GRBW")) self.brightness.setValue(int(float(cfg.get("brightness", 0.1)) * 100)) self.fps_spin.setValue(int(cfg.get("fps", 30))) self.segments = copy.deepcopy(cfg.get("segments", DEFAULT_SEGMENTS)) cam = cfg.get("camera", {}) pixels = cam.get("pixels") if pixels is None and "start" in cam: # migrate v2 start/count pixels = list(range(int(cam.get("start", 5)), int(cam.get("start", 5)) + int(cam.get("count", 3)))) self.cam_pixels.setText(format_ranges(pixels or [5, 6, 7])) self.cam_color.set_rgbw(cam.get("color", [255, 40, 0, 0])) self.cam_effect.setCurrentText(cam.get( "effect", "breathe" if cam.get("breathe", True) else "solid")) self.cam_speed.setValue(float(cam.get("speed", 1.0))) nt = cfg.get("notify", {}) self.notify_start.setValue(int(nt.get("start", 0))) self.notify_count.setValue(int(nt.get("count", 0))) self.notify_blinks.setValue(int(nt.get("blinks", 6))) self.notify_color.set_rgbw(nt.get("default_color", [0, 60, 255, 0])) self.notify_period.setValue(float(nt.get("default_period", 0.4))) self._building = False self.rebuild_seg_table() def _wire_autopush(self): for w in (self.num_pixels, self.fps_spin, self.notify_start, self.notify_count, self.notify_blinks): w.valueChanged.connect(self.schedule_push) for w in (self.notify_period, self.cam_speed): w.valueChanged.connect(self.schedule_push) self.brightness.valueChanged.connect(self.schedule_push) self.order_combo.currentIndexChanged.connect(self.schedule_push) self.cam_effect.currentIndexChanged.connect(self.schedule_push) self.pin_edit.editingFinished.connect(self.schedule_push) self.cam_pixels.editingFinished.connect(self.schedule_push) self.device_name_edit.editingFinished.connect(self.save_app_config) for cb in (self.cam_color, self.notify_color): cb.changed.connect(self.schedule_push) self.interval_spin.valueChanged.connect( lambda v: (setattr(self.monitor, "interval", v), self.save_app_config())) self.cam_auto_chk.toggled.connect(lambda _: self.save_app_config()) def schedule_push(self, *_): if self._building: return self.push_timer.start() def _auto_push(self): self.save_app_config() if not self.validate_segments(): self.log("auto-push blocked: fix the flagged zone issues") return if self.link.connected: reply = self.link.send( {"cmd": "config", "config": self.build_config()}) self.log(f"auto-push -> {reply}") def push_config(self): if not self.validate_segments(): self.log("push blocked: fix the flagged zone issues") return self.log("push config -> " f"{self.link.send({'cmd': 'config', 'config': self.build_config()})}") self.save_app_config() def save_app_config(self): data = { "device": self.build_config(), "connection": { "device_name": self.device_name_edit.text().strip() or DEFAULT_DEVICE_NAME, "last_port": self._last_port, }, "ui": { "interval": self.interval_spin.value(), "probes": {k: cb.isChecked() for k, cb in self.probe_checks.items()}, "cam_auto": self.cam_auto_chk.isChecked(), "notify_watch": self.notify_watch_chk.isChecked(), }, } try: with open(app_config_path(), "w") as f: json.dump(data, f, indent=2) except OSError as e: self.log(f"config save failed: {e}") def load_app_config(self): try: with open(app_config_path()) as f: data = json.load(f) except (OSError, ValueError): return self.apply_config_to_ui(data.get("device", {})) conn = data.get("connection", {}) self.device_name_edit.setText( conn.get("device_name") or DEFAULT_DEVICE_NAME) self._last_port = conn.get("last_port") ui = data.get("ui", {}) self.interval_spin.setValue(float(ui.get("interval", 1.0))) for k, on in ui.get("probes", {}).items(): if k in self.probe_checks: self.probe_checks[k].setChecked(bool(on)) self.cam_auto_chk.setChecked(bool(ui.get("cam_auto", True))) self.notify_watch_chk.setChecked(bool(ui.get("notify_watch", False))) self.log(f"loaded config from {app_config_path()}") def save_profile(self): path, _ = QFileDialog.getSaveFileName( self, "Export profile", "luxstats_profile.json", "JSON (*.json)") if path: with open(path, "w") as f: json.dump(self.build_config(), f, indent=2) self.log(f"profile exported: {path}") def load_profile(self): path, _ = QFileDialog.getOpenFileName( self, "Import profile", "", "JSON (*.json)") if path: try: with open(path) as f: self.apply_config_to_ui(json.load(f)) self.schedule_push() self.log(f"profile imported: {path}") except (OSError, ValueError) as e: QMessageBox.warning(self, "Import failed", str(e)) def closeEvent(self, event): self.save_app_config() self._stop_range_test() self.monitor.stop() self.monitor.wait(1500) if self.notify_watcher: self.notify_watcher.stop() self.link.close() event.accept() def main(): app = QApplication(sys.argv) win = MainWindow() win.show() sys.exit(app.exec_()) if __name__ == "__main__": main()