|
|
@@ -0,0 +1,1342 @@
|
|
|
+#!/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()
|
ArtyomV2X
1 ヶ月 前