| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342 |
- #!/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()
|