""" ui/control_panel.py Left-panel Control Panel. Contains modular output control widgets: • OnOffSwitch — latching power switch with indicator • MotorControl — speed dial (0–100 %), direction toggle, start/stop • SetpointControl — numeric target with ± step buttons and live readback • PwmControl — PWM duty cycle slider with frequency setting • GenericOutput — generic voltage/current analog output entry Each widget is self-contained and writes to a device channel via the DeviceRegistry when a device+channel is configured. New widget types can be added by subclassing ControlWidget and dropping an instance into ControlPanel.add_widget(). """ from __future__ import annotations from typing import Optional import math as _math import threading from typing import Any from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QSlider, QDoubleSpinBox, QComboBox, QScrollArea, QFrame, QSizePolicy, QGridLayout, QSpinBox, QGroupBox, ) from PyQt6.QtCore import Qt, pyqtSignal, QTimer, QMimeData from PyQt6.QtGui import QFont, QDrag, QPainter, QPen, QColor from devices.device_registry import DeviceRegistry from devices.base_device import DeviceStatus # ── Background write dispatch ─────────────────────────────────────────────── # # write_channel() is synchronous and, for some protocols (e.g. CML — see # api_layers/protocols/cml.py), does multiple blocking serial round-trips # per call — each one can cost up to the serial port's full read timeout if # the device doesn't reply to that particular command, so a single write # can take seconds. Control widgets call this directly from Qt signal # handlers (slider valueChanged fires on every tick of a drag), so on the # GUI thread that means each tick blocks the whole app. # # Writes are coalesced per (device, channel) — not just queued — keyed in # this dict rather than a plain FIFO queue: a slider drag fires far more # ticks than a multi-second write can keep up with, and a strict FIFO would # just work through an ever-growing backlog of stale values (observed: 30+ # queued while only ~4 had drained), so the motor keeps "moving" to # seconds-old positions long after the user stopped touching the slider — # looks exactly like a freeze. Re-writing the same key updates its pending # value in place without changing its position, so the worker always sends # the latest value once it's free, never a backlog of old ones. _write_lock = threading.Lock() _write_cond = threading.Condition(_write_lock) _pending: "dict" = {} # (device_id, channel_id) -> (dev, channel_id, value) def _enqueue_write(dev, channel_id, value): with _write_cond: _pending[(dev.info.device_id, channel_id)] = (dev, channel_id, value) _write_cond.notify() def _write_worker(): while True: with _write_cond: while not _pending: _write_cond.wait() key = next(iter(_pending)) dev, channel_id, value = _pending.pop(key) try: ok = dev.write_channel(channel_id, value) if not ok: print(f"[Control] write_channel({channel_id}, {value}) " f"returned False on {dev.info.device_id} — " f"check device type and channel ID") except Exception as e: print(f"[Control] write_channel({channel_id}, {value}) raised: {e}") threading.Thread(target=_write_worker, daemon=True, name="ControlWrite").start() # ══════════════════════════════════════════════════════════════════════════════ # Base widget # ══════════════════════════════════════════════════════════════════════════════ class _ControlHeader(QWidget): """Header bar — doubles as the drag handle for reordering. Clicks land here except where the Edit/Remove buttons (added on top) intercept them.""" def __init__(self): super().__init__() self._panel = None self._widget = None self.setCursor(Qt.CursorShape.OpenHandCursor) def set_drag_source(self, panel: "ControlPanel", widget: "ControlWidget"): self._panel = panel self._widget = widget def mousePressEvent(self, event): if event.button() == Qt.MouseButton.LeftButton and self._panel is not None: drag = QDrag(self) mime = QMimeData() mime.setText(str(self._panel._widgets.index(self._widget))) drag.setMimeData(mime) drag.exec(Qt.DropAction.MoveAction) super().mousePressEvent(event) class ControlWidget(QFrame): """Base class for all control panel widgets.""" value_changed = pyqtSignal(str, float) # channel_id, value def __init__(self, title: str, icon: str = "⚙", device_id: str = "", channel_id: str = "", registry: Optional[DeviceRegistry] = None, active_low: bool = False, processor=None, on_action_script: str = ""): super().__init__() self.title = title self.icon = icon self.device_id = device_id self.channel_id = channel_id self.registry = registry self.active_low = active_low self._processor = processor self._on_action_fn = None if on_action_script.strip(): self._compile_action(on_action_script) self.setObjectName("controlWidget") self._build_frame() self._stale = None self._check_stale() def _compile_action(self, script: str): try: code = compile(script, "", "exec") self._on_action_fn = code except SyntaxError as e: print(f"[Control '{self.title}'] script syntax error: {e}") def _check_stale(self): """Re-evaluate whether this control's target signal is currently reachable — unbound controls (no device/channel assigned) are never stale, only ones pointing at a signal that's since gone missing.""" stale = False if self.device_id and self.channel_id: dev = self.registry.get_instance(self.device_id) if self.registry else None if dev is None: stale = True else: ch = dev.get_channel(self.channel_id) if ch is None or not ch.enabled: stale = True elif dev.status not in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED): stale = True self._set_stale(stale) def _set_stale(self, stale: bool): if self._stale == stale: return self._stale = stale self._stale_lbl.setVisible(stale) self.setProperty("stale", "true" if stale else "false") self.style().unpolish(self); self.style().polish(self) def _logic_level(self, enabled: bool) -> float: """Map logical ON/OFF to electrical level, respecting active-low outputs.""" if self.active_low: return 0.0 if enabled else 1.0 return 1.0 if enabled else 0.0 def _build_frame(self): """Build the common outer frame; subclasses fill self._body.""" outer = QVBoxLayout(self) outer.setContentsMargins(0, 0, 0, 0) outer.setSpacing(0) # Header bar — also the drag handle; Edit/Remove buttons added on top by ControlPanel hdr = _ControlHeader() hdr.setObjectName("controlWidgetHeader") hdr.setFixedHeight(26) hdr_lay = QHBoxLayout(hdr) hdr_lay.setContentsMargins(8, 0, 4, 0) hdr_lay.setSpacing(4) icon_lbl = QLabel(self.icon) icon_lbl.setObjectName("controlWidgetIcon") icon_lbl.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents) title_lbl = QLabel(self.title.upper()) title_lbl.setObjectName("controlWidgetTitle") title_lbl.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents) hdr_lay.addWidget(icon_lbl) hdr_lay.addWidget(title_lbl, 1) self._stale_lbl = QLabel("⚠") self._stale_lbl.setObjectName("controlWidgetStale") self._stale_lbl.setToolTip("Target signal not connected") self._stale_lbl.setVisible(False) hdr_lay.addWidget(self._stale_lbl) self._header = hdr self._header_layout = hdr_lay outer.addWidget(hdr) # Body — subclasses populate this self._body = QWidget() self._body.setObjectName("controlWidgetBody") body_lay = QVBoxLayout(self._body) body_lay.setContentsMargins(10, 8, 10, 10) body_lay.setSpacing(6) self._body_layout = body_lay outer.addWidget(self._body) def _write(self, value: float): """ Write value to the linked device channel. Routing: DigitalIODevice → write_channel() → ArduinoLayer.digital_write() → W:Dxx:val AnalogInputDevice (arduino backend) → write_channel() → ArduinoLayer.digital_write() Any device with no write support → logs a warning, emits signal only The actual write is dispatched to a background worker (_enqueue_write) rather than called directly here — some protocols (CML) do multiple blocking serial round-trips per write, and this runs on the GUI thread (slider valueChanged fires on every tick of a drag), so a direct call here would freeze the whole app on every tick. """ if self.registry and self.device_id and self.channel_id: dev = self.registry.get_instance(self.device_id) if dev: ch = dev.get_channel(self.channel_id) if ch is not None and not ch.enabled: print(f"[Control] write_channel({self.channel_id}, {value}) skipped on " f"{self.device_id} — channel is disabled") elif dev.status not in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED): print(f"[Control] write_channel({self.channel_id}, {value}) skipped on " f"{self.device_id} — device status is {dev.status.value}, not connected") else: _enqueue_write(dev, self.channel_id, value) else: print(f"[Control] write_channel({self.channel_id}, {value}) skipped — " f"device '{self.device_id}' not found in registry") self.value_changed.emit(self.channel_id, value) if self._on_action_fn is not None: try: sv: dict = {} channels: dict = {} if self._processor is not None: with self._processor._lock: sv = self._processor._script_vars channels = {ch_id: v for (_, ch_id), (_, v) in self._processor._latest.items()} ns: dict = { "value": value, "channel_id": self.channel_id, "device_id": self.device_id, "vars": sv, "channels": channels, "math": _math, } exec(self._on_action_fn, ns) except Exception as e: print(f"[Control '{self.title}'] script error: {e}") def safe_stop(self): """ Called on every control when the master Stop is pressed. Default: zero the output. Widgets with a latched running/enabled state (OnOffSwitch, MotorControl, PwmControl) override this to go through their own toggle handler, so UI state and the write stay consistent. Widgets that only write on an explicit user action (SetpointControl, AnalogOutputControl) override with a no-op — there's no universally "safe" value to force onto an arbitrary process setpoint or analog output, so Stop leaves them alone rather than guessing. """ self._write(0.0) # ══════════════════════════════════════════════════════════════════════════════ # On/Off Switch # ══════════════════════════════════════════════════════════════════════════════ class OnOffSwitch(ControlWidget): """Large latching power switch with green/red indicator.""" def __init__(self, title: str = "Power", icon: str = "⏻", **kw): super().__init__(title, icon=icon, **kw) self._state = False self._build_body() def _build_body(self): lay = self._body_layout # Big toggle button self._btn = QPushButton("OFF") self._btn.setToolTip("Toggle this output on/off") self._btn.setObjectName("switchBtn") self._btn.setCheckable(True) self._btn.setMinimumHeight(52) self._btn.toggled.connect(self._on_toggle) lay.addWidget(self._btn) # Status indicator row ind_row = QHBoxLayout() self._indicator = QLabel("●") self._indicator.setObjectName("switchIndicatorOff") self._state_lbl = QLabel("INACTIVE") self._state_lbl.setObjectName("switchStateLbl") ind_row.addStretch() ind_row.addWidget(self._indicator) ind_row.addWidget(self._state_lbl) ind_row.addStretch() lay.addLayout(ind_row) def _on_toggle(self, checked: bool): self._state = checked if checked: self._btn.setText("ON") self._btn.setObjectName("switchBtnOn") self._indicator.setObjectName("switchIndicatorOn") self._state_lbl.setText("ACTIVE") else: self._btn.setText("OFF") self._btn.setObjectName("switchBtnOff") self._indicator.setObjectName("switchIndicatorOff") self._state_lbl.setText("INACTIVE") # Re-polish so QSS picks up new objectName for w in (self._btn, self._indicator, self._state_lbl): w.style().unpolish(w); w.style().polish(w) self._write(self._logic_level(checked)) def safe_stop(self): self._btn.setChecked(False) # routes through _on_toggle: updates UI + writes off # ══════════════════════════════════════════════════════════════════════════════ # Motor Control # ══════════════════════════════════════════════════════════════════════════════ class MotorControl(ControlWidget): """Motor speed (0–100 %), direction toggle, start/stop.""" def __init__(self, title: str = "Motor", icon: str = "⟳", max_rpm: int = 3000, **kw): super().__init__(title, icon=icon, **kw) self.max_rpm = max_rpm self._running = False self._fwd = True self._build_body() def _build_body(self): lay = self._body_layout # Speed readout readout_row = QHBoxLayout() self._speed_lbl = QLabel("0") self._speed_lbl.setObjectName("motorSpeedLbl") unit_lbl = QLabel("RPM") unit_lbl.setObjectName("motorUnitLbl") readout_row.addStretch() readout_row.addWidget(self._speed_lbl) readout_row.addWidget(unit_lbl) readout_row.addStretch() lay.addLayout(readout_row) # Speed slider self._slider = QSlider(Qt.Orientation.Horizontal) self._slider.setRange(0, 100) self._slider.setValue(0) self._slider.setObjectName("motorSlider") self._slider.valueChanged.connect(self._on_speed) lay.addWidget(self._slider) # Pct label self._pct_lbl = QLabel("0 %") self._pct_lbl.setObjectName("motorPctLbl") self._pct_lbl.setAlignment(Qt.AlignmentFlag.AlignCenter) lay.addWidget(self._pct_lbl) # Direction + start/stop row btn_row = QHBoxLayout() self._dir_btn = QPushButton("▶ FWD") self._dir_btn.setToolTip("Toggle motor direction (forward/reverse)") self._dir_btn.setObjectName("motorDirBtn") self._dir_btn.setCheckable(True) self._dir_btn.toggled.connect(self._on_dir) btn_row.addWidget(self._dir_btn) self._run_btn = QPushButton("START") self._run_btn.setToolTip("Start/stop the motor at the speed set by the slider above") self._run_btn.setObjectName("motorRunBtnOff") self._run_btn.setCheckable(True) self._run_btn.toggled.connect(self._on_run) btn_row.addWidget(self._run_btn) lay.addLayout(btn_row) def _on_speed(self, pct: int): rpm = int(pct / 100.0 * self.max_rpm) self._speed_lbl.setText(str(rpm)) self._pct_lbl.setText(f"{pct} %") if self._running: self._write(float(rpm) * (1 if self._fwd else -1)) def _on_dir(self, rev: bool): self._fwd = not rev self._dir_btn.setText("◀ REV" if rev else "▶ FWD") self._dir_btn.setObjectName("motorDirBtnRev" if rev else "motorDirBtn") self._dir_btn.style().unpolish(self._dir_btn) self._dir_btn.style().polish(self._dir_btn) self._on_speed(self._slider.value()) def _on_run(self, running: bool): self._running = running self._run_btn.setText("STOP" if running else "START") self._run_btn.setObjectName("motorRunBtnOn" if running else "motorRunBtnOff") self._run_btn.style().unpolish(self._run_btn) self._run_btn.style().polish(self._run_btn) if not running: self._write(0.0) else: self._on_speed(self._slider.value()) def safe_stop(self): self._run_btn.setChecked(False) # routes through _on_run: stops + writes 0 self._slider.setValue(0) # ══════════════════════════════════════════════════════════════════════════════ # Setpoint Control # ══════════════════════════════════════════════════════════════════════════════ class SetpointControl(ControlWidget): """Numeric setpoint with ± step buttons and live process-value readback.""" def __init__(self, title: str = "Setpoint", icon: str = "◎", unit: str = "", min_val: float = 0.0, max_val: float = 100.0, step: float = 1.0, **kw): super().__init__(title, icon=icon, **kw) self.unit = unit self.min_val = min_val self.max_val = max_val self.step = step self._pv = 0.0 self._build_body() def _build_body(self): lay = self._body_layout # SP row sp_row = QHBoxLayout() sp_lbl = QLabel("SP") sp_lbl.setObjectName("spLabel") sp_lbl.setFixedWidth(24) self._minus_btn = QPushButton("−") self._minus_btn.setToolTip(f"Decrease setpoint by {self.step} {self.unit}") self._minus_btn.setObjectName("spStepBtn") self._minus_btn.setFixedWidth(32) self._minus_btn.clicked.connect(self._decrement) self._sp_spin = QDoubleSpinBox() self._sp_spin.setRange(self.min_val, self.max_val) self._sp_spin.setSingleStep(self.step) self._sp_spin.setSuffix(f" {self.unit}") self._sp_spin.setObjectName("spSpinbox") self._sp_spin.valueChanged.connect(self._on_sp_changed) self._plus_btn = QPushButton("+") self._plus_btn.setToolTip(f"Increase setpoint by {self.step} {self.unit}") self._plus_btn.setObjectName("spStepBtn") self._plus_btn.setFixedWidth(32) self._plus_btn.clicked.connect(self._increment) sp_row.addWidget(sp_lbl) sp_row.addWidget(self._minus_btn) sp_row.addWidget(self._sp_spin, 1) sp_row.addWidget(self._plus_btn) lay.addLayout(sp_row) # PV readback row pv_row = QHBoxLayout() pv_lbl = QLabel("PV") pv_lbl.setObjectName("pvLabel") pv_lbl.setFixedWidth(24) self._pv_lbl = QLabel(f"— {self.unit}") self._pv_lbl.setObjectName("pvValue") pv_row.addWidget(pv_lbl) pv_row.addWidget(self._pv_lbl, 1) lay.addLayout(pv_row) # Error bar err_row = QHBoxLayout() err_lbl = QLabel("ERR") err_lbl.setObjectName("errLabel") err_lbl.setFixedWidth(24) self._err_lbl = QLabel("—") self._err_lbl.setObjectName("errValue") err_row.addWidget(err_lbl) err_row.addWidget(self._err_lbl, 1) lay.addLayout(err_row) def update_pv(self, value: float): """Call this with live process-variable readings.""" self._pv = value self._pv_lbl.setText(f"{value:.3f} {self.unit}") err = self._sp_spin.value() - value self._err_lbl.setText(f"{err:+.3f} {self.unit}") self._err_lbl.setStyleSheet( "color:#ef4444;" if abs(err) > 1.0 else "color:#22c55e;" ) def _on_sp_changed(self, v: float): self._write(v) def _increment(self): self._sp_spin.setValue(self._sp_spin.value() + self.step) def _decrement(self): self._sp_spin.setValue(self._sp_spin.value() - self.step) def safe_stop(self): pass # no safe universal value for an arbitrary process setpoint — leave it # ══════════════════════════════════════════════════════════════════════════════ # PWM Control # ══════════════════════════════════════════════════════════════════════════════ class PwmControl(ControlWidget): """PWM duty cycle slider + frequency setting.""" def __init__(self, title: str = "PWM Output", icon: str = "⊓", **kw): super().__init__(title, icon=icon, **kw) self._build_body() def _build_body(self): lay = self._body_layout # Duty cycle dc_row = QHBoxLayout() dc_lbl = QLabel("Duty:") dc_lbl.setObjectName("pwmLabel") self._dc_lbl = QLabel("0 %") self._dc_lbl.setObjectName("pwmValue") dc_row.addWidget(dc_lbl) dc_row.addStretch() dc_row.addWidget(self._dc_lbl) lay.addLayout(dc_row) self._dc_slider = QSlider(Qt.Orientation.Horizontal) self._dc_slider.setRange(0, 100) self._dc_slider.setValue(0) self._dc_slider.setObjectName("pwmSlider") self._dc_slider.valueChanged.connect(self._on_dc) lay.addWidget(self._dc_slider) # Frequency freq_row = QHBoxLayout() freq_lbl = QLabel("Freq:") freq_lbl.setObjectName("pwmLabel") self._freq_spin = QSpinBox() self._freq_spin.setRange(1, 100_000) self._freq_spin.setValue(1000) self._freq_spin.setSuffix(" Hz") self._freq_spin.setObjectName("pwmFreqSpin") self._freq_spin.valueChanged.connect(self._on_freq) freq_row.addWidget(freq_lbl) freq_row.addWidget(self._freq_spin) lay.addLayout(freq_row) # Enable self._en_btn = QPushButton("ENABLE") self._en_btn.setObjectName("motorRunBtnOff") self._en_btn.setCheckable(True) self._en_btn.toggled.connect(self._on_enable) lay.addWidget(self._en_btn) def _on_dc(self, v: int): self._dc_lbl.setText(f"{v} %") if self._en_btn.isChecked(): self._write(float(v)) def _on_freq(self, hz: int): pass # TODO: send freq command to hardware def _on_enable(self, en: bool): self._en_btn.setText("DISABLE" if en else "ENABLE") self._en_btn.setObjectName("motorRunBtnOn" if en else "motorRunBtnOff") self._en_btn.style().unpolish(self._en_btn) self._en_btn.style().polish(self._en_btn) self._write(float(self._dc_slider.value()) if en else 0.0) def safe_stop(self): self._en_btn.setChecked(False) # routes through _on_enable: disables + writes 0 self._dc_slider.setValue(0) # ══════════════════════════════════════════════════════════════════════════════ # Generic Analog Output # ══════════════════════════════════════════════════════════════════════════════ class AnalogOutputControl(ControlWidget): """Generic voltage/current analog output with spinbox + send button.""" def __init__(self, title: str = "Analog Out", icon: str = "↗", unit: str = "V", min_val: float = 0.0, max_val: float = 10.0, **kw): super().__init__(title, icon=icon, **kw) self.unit = unit self.min_val = min_val self.max_val = max_val self._build_body() def _build_body(self): lay = self._body_layout val_row = QHBoxLayout() self._spin = QDoubleSpinBox() self._spin.setRange(self.min_val, self.max_val) self._spin.setSingleStep(0.01) self._spin.setSuffix(f" {self.unit}") self._spin.setObjectName("spSpinbox") val_row.addWidget(self._spin, 1) send_btn = QPushButton("SET") send_btn.setObjectName("applyButton") send_btn.clicked.connect(lambda: self._write(self._spin.value())) val_row.addWidget(send_btn) lay.addLayout(val_row) # Slider mirror self._slider = QSlider(Qt.Orientation.Horizontal) self._slider.setRange(0, 1000) self._slider.setObjectName("motorSlider") self._slider.valueChanged.connect(self._on_slider) self._spin.valueChanged.connect(self._sync_slider) lay.addWidget(self._slider) def _on_slider(self, v: int): mapped = self.min_val + (v / 1000.0) * (self.max_val - self.min_val) self._spin.blockSignals(True) self._spin.setValue(mapped) self._spin.blockSignals(False) def _sync_slider(self, v: float): rng = self.max_val - self.min_val norm = int(((v - self.min_val) / rng) * 1000) if rng else 0 self._slider.blockSignals(True) self._slider.setValue(max(0, min(1000, norm))) self._slider.blockSignals(False) def safe_stop(self): pass # only writes on explicit SET click — no safe universal value to force # ══════════════════════════════════════════════════════════════════════════════ # Drag-to-reorder helper # ══════════════════════════════════════════════════════════════════════════════ class _DropContainer(QWidget): """Holds the stacked control widgets; drops reorder them via the panel.""" def __init__(self, panel: "ControlPanel"): super().__init__() self._panel = panel self._indicator_y = None self.setAcceptDrops(True) def dragEnterEvent(self, event): if event.mimeData().hasText(): event.acceptProposedAction() self._show_indicator(event.position().y()) def dragMoveEvent(self, event): event.acceptProposedAction() self._show_indicator(event.position().y()) def dragLeaveEvent(self, event): self._indicator_y = None self.update() def dropEvent(self, event): self._indicator_y = None self.update() try: src_idx = int(event.mimeData().text()) except ValueError: return self._panel._reorder(src_idx, self._panel._drop_index(event.position().y())) event.acceptProposedAction() def _show_indicator(self, y: float): self._indicator_y = self._panel._indicator_y_for(self._panel._drop_index(y)) self.update() def paintEvent(self, event): if self._indicator_y is not None: painter = QPainter(self) painter.setPen(QPen(QColor("#3b82f6"), 2)) painter.drawLine(6, self._indicator_y, self.width() - 6, self._indicator_y) painter.end() # ══════════════════════════════════════════════════════════════════════════════ # Control Panel container # ══════════════════════════════════════════════════════════════════════════════ class ControlPanel(QWidget): """ Left panel — output control widgets. Header has Add button. Each widget has Edit, Remove, and drag-to-reorder. """ controls_changed = pyqtSignal() # emitted whenever widgets are added/edited/removed def __init__(self, registry: DeviceRegistry, processor=None, engine=None): super().__init__() self.registry = registry self.processor = processor self.engine = engine self._widgets: list[ControlWidget] = [] self._specs: list = [] # parallel list of ControlSpec self._build() if engine is not None: engine.device_status_changed.connect(self._on_device_status_changed) def _on_device_status_changed(self, device_id: str, status: str): for w in self._widgets: w._check_stale() # ── Layout ─────────────────────────────────────────────────────────────── def _build(self): layout = QVBoxLayout(self) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(0) scroll = QScrollArea() scroll.setWidgetResizable(True) scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff) scroll.setObjectName("deviceScroll") self._container = _DropContainer(self) self._inner = QVBoxLayout(self._container) self._inner.setContentsMargins(6, 6, 6, 6) self._inner.setSpacing(8) self._inner.addStretch() scroll.setWidget(self._container) layout.addWidget(scroll) # ── Widget management ───────────────────────────────────────────────────── def safe_stop_all(self): """Master Stop — tell every control widget to go to a safe state.""" for w in self._widgets: try: w.safe_stop() except Exception as e: print(f"[Control '{w.title}'] safe_stop failed: {e}") def _wire_widget(self, widget: ControlWidget, spec): """Hook up the widget's header as a drag source and add Edit/Remove buttons top-right in that same header — the whole bar drags to reorder except where these two buttons intercept the click.""" widget._header.set_drag_source(self, widget) edit_btn = QPushButton("✎ Edit"); edit_btn.setObjectName("configButton") edit_btn.setFixedHeight(20) rm_btn = QPushButton("✕"); rm_btn.setObjectName("traceRemoveBtn") rm_btn.setFixedSize(20, 20); rm_btn.setToolTip("Remove") edit_btn.clicked.connect(lambda: self._on_edit(widget, spec)) rm_btn.clicked.connect( lambda: self._on_remove(widget, spec)) widget._header_layout.addWidget(edit_btn) widget._header_layout.addWidget(rm_btn) def _add_widget_from_spec(self, spec): """Instantiate a ControlWidget from a ControlSpec and add to panel.""" widget = _build_widget_from_spec(spec, self.registry, self.processor) if widget is None: return self._wire_widget(widget, spec) self._widgets.append(widget) self._specs.append(spec) self._inner.insertWidget(self._inner.count() - 1, widget) self.controls_changed.emit() def add_widget(self, widget: ControlWidget, spec=None): """Legacy API — add a pre-built widget directly.""" from ui.control_editor import ControlSpec if spec is None: spec = ControlSpec( title=widget.title, icon=widget.icon, device_id=widget.device_id, channel_id=widget.channel_id, active_low=getattr(widget, "active_low", False), ) self._wire_widget(widget, spec) self._widgets.append(widget) self._specs.append(spec) self._inner.insertWidget(self._inner.count() - 1, widget) def clear_widgets(self): while self._inner.count() > 1: item = self._inner.takeAt(0) if item and item.widget(): item.widget().deleteLater() self._widgets.clear() self._specs.clear() def get_specs(self) -> list: """Return list of ControlSpec for all current widgets (for profile save).""" return list(self._specs) def load_specs(self, specs: list): """Load a list of ControlSpec objects (from profile restore).""" self.clear_widgets() for spec in specs: self._add_widget_from_spec(spec) def _drop_index(self, y: float) -> int: """Insertion index for a drop at container-relative y.""" n = self._inner.count() - 1 # exclude trailing stretch for i in range(n): w = self._inner.itemAt(i).widget() if y < w.y() + w.height() / 2: return i return n def _indicator_y_for(self, idx: int) -> int: """Container-relative y for the drop-line at insertion index idx.""" n = self._inner.count() - 1 # exclude trailing stretch spacing = self._inner.spacing() if n == 0: return self._inner.contentsMargins().top() if idx <= 0: w = self._inner.itemAt(0).widget() return w.y() - spacing // 2 if idx >= n: w = self._inner.itemAt(n - 1).widget() return w.y() + w.height() + spacing // 2 prev = self._inner.itemAt(idx - 1).widget() nxt = self._inner.itemAt(idx).widget() return (prev.y() + prev.height() + nxt.y()) // 2 def _reorder(self, src_idx: int, dst_idx: int): """Move the widget at src_idx to dst_idx (drop-target index, pre-removal).""" if dst_idx > src_idx: dst_idx -= 1 if dst_idx == src_idx: return spec = self._specs.pop(src_idx) widget = self._widgets.pop(src_idx) self._specs.insert(dst_idx, spec) self._widgets.insert(dst_idx, widget) self._inner.removeWidget(widget) self._inner.insertWidget(dst_idx, widget) self.controls_changed.emit() # ── Slots ───────────────────────────────────────────────────────────────── def _on_add(self): from ui.control_editor import ControlEditorDialog dlg = ControlEditorDialog(self.registry, parent=self) if dlg.exec() and dlg.result_spec: self._add_widget_from_spec(dlg.result_spec) def _on_edit(self, widget: ControlWidget, spec): from ui.control_editor import ControlEditorDialog dlg = ControlEditorDialog(self.registry, spec=spec, parent=self) if not (dlg.exec() and dlg.result_spec): return new_spec = dlg.result_spec idx = self._specs.index(spec) # Remove old widget self._inner.removeWidget(widget); widget.deleteLater() self._widgets.pop(idx); self._specs.pop(idx) # Insert new one at same position new_widget = _build_widget_from_spec(new_spec, self.registry, self.processor) if new_widget is None: return self._wire_widget(new_widget, new_spec) self._widgets.insert(idx, new_widget) self._specs.insert(idx, new_spec) self._inner.insertWidget(idx, new_widget) self.controls_changed.emit() def _on_remove(self, widget: ControlWidget, spec): idx = self._specs.index(spec) self._inner.removeWidget(widget); widget.deleteLater() self._widgets.pop(idx); self._specs.pop(idx) self.controls_changed.emit() def _add_demo_widgets(self): """Default demo configuration.""" from ui.control_editor import ControlSpec demos = [ ControlSpec("On/Off Switch", "Pump Power", "⏻", "dio_0", "do0"), ControlSpec("On/Off Switch", "Heater", "⏻", "ard_0", "D7"), ControlSpec("Motor Control", "Drive Motor", "⟳", "ard_0", "A0", max_rpm=3000), ControlSpec("Setpoint", "Temp Setpoint","◎", "", "", unit="°C", min_val=0, max_val=300, step=0.5), ControlSpec("Setpoint", "Flow Rate", "◎", "", "", unit="mL/min",min_val=0, max_val=500, step=5.0), ControlSpec("PWM Output", "PWM Ch 1", "⊓", "ard_0", "D9"), ControlSpec("Analog Output", "Analog Out", "↗", "", "", unit="V", min_val=0, max_val=10), ] for spec in demos: self._add_widget_from_spec(spec) # ── Factory — build a ControlWidget from a ControlSpec ───────────────────────── def _build_widget_from_spec(spec, registry: DeviceRegistry, processor=None) -> Optional[ControlWidget]: """Instantiate the right ControlWidget subclass from a ControlSpec.""" # Base kwargs — icon is NOT included here; it's passed explicitly below # so subclasses never get a double-value collision. base_kw = dict( device_id=spec.device_id, channel_id=spec.channel_id, registry=registry, active_low=getattr(spec, "active_low", False), processor=processor, on_action_script=getattr(spec, "on_action_script", ""), ) t = spec.control_type ttl = spec.title ico = spec.icon if t == "On/Off Switch": return OnOffSwitch(title=ttl, icon=ico, **base_kw) elif t == "Motor Control": return MotorControl(title=ttl, icon=ico, max_rpm=spec.max_rpm, **base_kw) elif t == "PWM Output": return PwmControl(title=ttl, icon=ico, **base_kw) elif t == "Setpoint": return SetpointControl(title=ttl, icon=ico, unit=spec.unit, min_val=spec.min_val, max_val=spec.max_val, step=spec.step, **base_kw) elif t == "Analog Output": return AnalogOutputControl(title=ttl, icon=ico, unit=spec.unit, min_val=spec.min_val, max_val=spec.max_val, **base_kw) return None