""" 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 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 from PyQt6.QtGui import QFont from devices.device_registry import DeviceRegistry from devices.base_device import DeviceStatus # ══════════════════════════════════════════════════════════════════════════════ # Base widget # ══════════════════════════════════════════════════════════════════════════════ 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() 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 _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 hdr = QWidget() hdr.setObjectName("controlWidgetHeader") hdr.setFixedHeight(26) hdr_lay = QHBoxLayout(hdr) hdr_lay.setContentsMargins(8, 0, 8, 0) icon_lbl = QLabel(self.icon) icon_lbl.setObjectName("controlWidgetIcon") title_lbl = QLabel(self.title.upper()) title_lbl.setObjectName("controlWidgetTitle") hdr_lay.addWidget(icon_lbl) hdr_lay.addWidget(title_lbl, 1) 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 """ written = False if self.registry and self.device_id and self.channel_id: dev = self.registry.get_instance(self.device_id) if dev: if 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: ok = dev.write_channel(self.channel_id, value) if ok: written = True else: print(f"[Control] write_channel({self.channel_id}, {value}) " f"returned False on {self.device_id} — " f"check device type and channel ID") 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.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.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.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.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.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 # ══════════════════════════════════════════════════════════════════════════════ # Control Panel container # ══════════════════════════════════════════════════════════════════════════════ class ControlPanel(QWidget): """ Left panel — output control widgets. Header has Add button. Each widget has Edit, Remove, and reorder buttons. """ controls_changed = pyqtSignal() # emitted whenever widgets are added/edited/removed def __init__(self, registry: DeviceRegistry, processor=None): super().__init__() self.registry = registry self.processor = processor self._widgets: list[ControlWidget] = [] self._specs: list = [] # parallel list of ControlSpec self._build() # ── Layout ─────────────────────────────────────────────────────────────── def _build(self): layout = QVBoxLayout(self) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(0) # Header with Add button hdr_widget = QWidget(); hdr_widget.setObjectName("controlPanelHeader") hdr_widget.setFixedHeight(30) hdr_lay = QHBoxLayout(hdr_widget) hdr_lay.setContentsMargins(8, 0, 6, 0) hdr_lbl = QLabel("CONTROLS"); hdr_lbl.setObjectName("panelHeader") hdr_lay.addWidget(hdr_lbl, 1) add_btn = QPushButton("＋"); add_btn.setObjectName("devicesSmallBtn") add_btn.setFixedSize(24, 22); add_btn.setToolTip("Add control widget") add_btn.clicked.connect(self._on_add) hdr_lay.addWidget(add_btn) layout.addWidget(hdr_widget) scroll = QScrollArea() scroll.setWidgetResizable(True) scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff) scroll.setObjectName("deviceScroll") self._container = QWidget() 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 _make_wrapper(self, widget: ControlWidget, spec) -> QFrame: """Wrap a ControlWidget with Edit / Remove / reorder buttons.""" wrapper = QFrame(); wrapper.setObjectName("controlWidgetWrapper") wl = QVBoxLayout(wrapper); wl.setContentsMargins(0, 0, 0, 0); wl.setSpacing(0) wl.addWidget(widget) # Button row below each widget btn_row = QHBoxLayout(); btn_row.setContentsMargins(2, 1, 2, 1) up_btn = QPushButton("↑"); up_btn.setObjectName("traceRemoveBtn") up_btn.setFixedSize(20, 20); up_btn.setToolTip("Move up") dn_btn = QPushButton("↓"); dn_btn.setObjectName("traceRemoveBtn") dn_btn.setFixedSize(20, 20); dn_btn.setToolTip("Move down") up_btn.clicked.connect(lambda: self._move(spec, -1)) dn_btn.clicked.connect(lambda: self._move(spec, +1)) btn_row.addWidget(up_btn) btn_row.addWidget(dn_btn) btn_row.addStretch() edit_btn = QPushButton("✎ Edit"); edit_btn.setObjectName("configButton") edit_btn.setFixedHeight(20) rm_btn = QPushButton("✕"); rm_btn.setObjectName("traceRemoveBtn") rm_btn.setFixedSize(20, 20) edit_btn.clicked.connect(lambda: self._on_edit(widget, spec, wrapper)) rm_btn.clicked.connect( lambda: self._on_remove(widget, spec, wrapper)) btn_row.addWidget(edit_btn) btn_row.addWidget(rm_btn) wl.addLayout(btn_row) return wrapper def _add_widget_from_spec(self, spec): """Instantiate a ControlWidget from a ControlSpec and add to panel.""" from ui.control_editor import ControlSpec as CS widget = _build_widget_from_spec(spec, self.registry, self.processor) if widget is None: return wrapper = self._make_wrapper(widget, spec) self._widgets.append(widget) self._specs.append(spec) self._inner.insertWidget(self._inner.count() - 1, wrapper) 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), ) wrapper = self._make_wrapper(widget, spec) self._widgets.append(widget) self._specs.append(spec) self._inner.insertWidget(self._inner.count() - 1, wrapper) 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 _move(self, spec, direction: int): """Move a control up (-1) or down (+1) in the list.""" try: idx = self._specs.index(spec) except ValueError: return new_idx = idx + direction if new_idx < 0 or new_idx >= len(self._specs): return self._specs.insert(new_idx, self._specs.pop(idx)) self._widgets.insert(new_idx, self._widgets.pop(idx)) wrapper = self._inner.itemAt(idx).widget() self._inner.removeWidget(wrapper) self._inner.insertWidget(new_idx, wrapper) 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, wrapper: QFrame): 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 wrapper self._inner.removeWidget(wrapper); wrapper.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 new_wrapper = self._make_wrapper(new_widget, new_spec) self._widgets.insert(idx, new_widget) self._specs.insert(idx, new_spec) self._inner.insertWidget(idx, new_wrapper) self.controls_changed.emit() def _on_remove(self, widget: ControlWidget, spec, wrapper: QFrame): idx = self._specs.index(spec) self._inner.removeWidget(wrapper); wrapper.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