""" ui/signal_builder.py Signal Builder — operator-facing window for: Tab 1: FILTERS — add/remove/reorder filter stages per channel Tab 2: DERIVED — create virtual channels (velocity, acceleration, power, RMS, expression, custom Python) Tab 3: CHANNELS — show/hide raw channels (replaces readout_panel settings) All changes are applied immediately to the SignalProcessor. """ from __future__ import annotations import traceback from copy import deepcopy from typing import List, Optional, Tuple from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QScrollArea, QFrame, QTabWidget, QComboBox, QLineEdit, QDoubleSpinBox, QSpinBox, QCheckBox, QTextEdit, QGroupBox, QFormLayout, QToolButton, QSizePolicy, QMessageBox, QSplitter, QListWidget, QListWidgetItem, ) from PyQt6.QtCore import Qt, pyqtSignal, QSize from PyQt6.QtGui import QFont, QColor, QCloseEvent from devices.device_registry import DeviceRegistry from core.signal_processor import ( SignalProcessor, ChannelPipeline, DerivedChannel, FilterBase, MovingAverageFilter, MedianFilter, LowPassFilter, HighPassFilter, ScaleOffsetFilter, DerivativeFilter, IntegralFilter, FILTER_CLASSES, ) # ── Color palette for new derived channels ───────────────────────────────── _DERIVED_COLORS = [ "#f72585", "#7209b7", "#3a0ca3", "#4361ee", "#4cc9f0", "#f77f00", "#d62828", "#588157", ] # ══════════════════════════════════════════════════════════════════════════════ # Helpers # ══════════════════════════════════════════════════════════════════════════════ def _make_channel_combo(registry: DeviceRegistry, include_derived: bool = False, processor: Optional[SignalProcessor] = None) -> QComboBox: cb = QComboBox() for dev in registry.all_instances(): for ch in dev.info.channels: cb.addItem(f"{dev.info.device_id} / {ch.channel_id} ({ch.name})", userData=(dev.info.device_id, ch.channel_id)) if include_derived and processor: for dc in processor.get_derived(): cb.addItem(f"[derived] {dc.channel_id} ({dc.name})", userData=("derived", dc.channel_id)) return cb # ══════════════════════════════════════════════════════════════════════════════ # Filter stage row # ══════════════════════════════════════════════════════════════════════════════ class FilterRow(QFrame): removed = pyqtSignal(object) changed = pyqtSignal() _PARAM_SPECS = { "moving_average": [("window", "int", 10)], "median": [("window", "int", 5)], "low_pass": [("alpha", "float", 0.1)], "high_pass": [("alpha", "float", 0.9)], "scale_offset": [("scale", "float", 1.0), ("offset", "float", 0.0)], "derivative": [], "integral": [], } def __init__(self, filt: FilterBase): super().__init__() self.filt = filt self.setObjectName("traceRow") self._build() def _build(self): lay = QHBoxLayout(self) lay.setContentsMargins(6, 4, 6, 4) lay.setSpacing(8) type_lbl = QLabel(self.filt.name.replace("_", " ").title()) type_lbl.setObjectName("traceSource") type_lbl.setMinimumWidth(110) lay.addWidget(type_lbl) self._param_widgets = {} for pname, ptype, default in self._PARAM_SPECS.get(self.filt.name, []): lbl = QLabel(f"{pname}:") lbl.setObjectName("pwmLabel") lay.addWidget(lbl) if ptype == "int": w = QSpinBox(); w.setRange(1, 10000); w.setValue(int(self.filt.params.get(pname, default))) w.setFixedWidth(64); w.setObjectName("traceWidthSpin") w.valueChanged.connect(lambda v, p=pname: self._update_param(p, v)) else: w = QDoubleSpinBox(); w.setRange(-1e9, 1e9); w.setDecimals(4) w.setValue(float(self.filt.params.get(pname, default))) w.setFixedWidth(80); w.setObjectName("traceWidthSpin") w.valueChanged.connect(lambda v, p=pname: self._update_param(p, v)) lay.addWidget(w) self._param_widgets[pname] = w lay.addStretch() rm = QToolButton(); rm.setText("✕"); rm.setObjectName("traceRemoveBtn") rm.setFixedSize(22, 22); rm.clicked.connect(lambda: self.removed.emit(self)) lay.addWidget(rm) def _update_param(self, name: str, value): self.filt.params[name] = value # Re-instantiate filter with new params cls = FILTER_CLASSES.get(self.filt.name) if cls: new_filt = cls(**self.filt.params) self.filt.__dict__.update(new_filt.__dict__) self.changed.emit() # ══════════════════════════════════════════════════════════════════════════════ # Filter pipeline editor for one channel # ══════════════════════════════════════════════════════════════════════════════ class PipelineEditor(QWidget): pipeline_changed = pyqtSignal() def __init__(self, pipeline: ChannelPipeline, processor: SignalProcessor): super().__init__() self.pipeline = pipeline self.processor = processor self._rows: List[FilterRow] = [] self._build() def _build(self): root = QVBoxLayout(self); root.setContentsMargins(0, 0, 0, 6) root.setSpacing(4) # Enable toggle top = QHBoxLayout() self._en_chk = QCheckBox("Pipeline enabled") self._en_chk.setChecked(self.pipeline.enabled) self._en_chk.toggled.connect(self._on_enable) top.addWidget(self._en_chk) top.addStretch() root.addLayout(top) # Filter rows self._rows_widget = QWidget() self._rows_lay = QVBoxLayout(self._rows_widget) self._rows_lay.setContentsMargins(0, 0, 0, 0); self._rows_lay.setSpacing(3) for f in self.pipeline.filters: self._add_row(f) root.addWidget(self._rows_widget) # Add filter bar add_bar = QHBoxLayout() self._filter_cb = QComboBox() self._filter_cb.addItems([k.replace("_"," ").title() for k in FILTER_CLASSES]) self._filter_cb.setObjectName("channelPickerCb") add_bar.addWidget(self._filter_cb, 1) add_btn = QPushButton("＋ Add Filter") add_btn.setObjectName("addTraceBtn") add_btn.clicked.connect(self._on_add_filter) add_bar.addWidget(add_btn) root.addLayout(add_bar) def _add_row(self, filt: FilterBase): row = FilterRow(filt) row.removed.connect(self._remove_row) row.changed.connect(self._push) self._rows.append(row) self._rows_lay.addWidget(row) def _remove_row(self, row: FilterRow): self.pipeline.filters.remove(row.filt) self._rows.remove(row) self._rows_lay.removeWidget(row); row.deleteLater() self._push() def _on_add_filter(self): key = list(FILTER_CLASSES.keys())[self._filter_cb.currentIndex()] filt = FILTER_CLASSES[key]() self.pipeline.filters.append(filt) self._add_row(filt); self._push() def _on_enable(self, v: bool): self.pipeline.enabled = v; self._push() def _push(self): self.processor.set_pipeline(self.pipeline) self.pipeline_changed.emit() # ══════════════════════════════════════════════════════════════════════════════ # Derived channel editor row # ══════════════════════════════════════════════════════════════════════════════ class DerivedEditor(QFrame): removed = pyqtSignal(object) changed = pyqtSignal(object) # DerivedChannel _BUILTIN_KINDS = ["velocity", "acceleration", "power", "rms", "difference", "sum"] _SCRIPT_KINDS = ["expression", "function", "custom_script"] _ALL_KINDS = _BUILTIN_KINDS + _SCRIPT_KINDS _KIND_HELP = { "velocity": "1 source: displacement channel → dy/dt", "acceleration": "1 source: displacement channel → d²y/dt²", "power": "2 sources: voltage, current → V×I", "rms": "1 source → rolling RMS (set window in params)", "difference": "2 sources → source[0] − source[1]", "sum": "N sources → Σ sources", "expression": 'Single line. x = list of source values, t = time.\nExample: x[0] * 2 + x[1]', "function": "def compute(x, t):\n # x = list of source values\n return x[0] * 2", "custom_script": "Full script. Must define:\ndef compute(x, t):\n ...\n return value", } def __init__(self, dc: DerivedChannel, registry: DeviceRegistry, processor: SignalProcessor, color_idx: int = 0): super().__init__() self.dc = dc self.registry = registry self.processor = processor self._color_idx = color_idx self.setObjectName("plotBlock") self._build() def _build(self): outer = QVBoxLayout(self); outer.setContentsMargins(0, 0, 0, 0); outer.setSpacing(0) # ── Header ───────────────────────────────────────────────────── hdr = QWidget(); hdr.setObjectName("plotBlockHeader"); hdr.setFixedHeight(32) hl = QHBoxLayout(hdr); hl.setContentsMargins(8, 0, 6, 0) self._name_edit = QLineEdit(self.dc.name) self._name_edit.setObjectName("plotBlockTitle") self._name_edit.textChanged.connect(lambda t: setattr(self.dc, "name", t)) hl.addWidget(self._name_edit, 1) self._en_chk = QCheckBox() self._en_chk.setChecked(self.dc.enabled) self._en_chk.setToolTip("Enable/disable") self._en_chk.toggled.connect(self._on_enable) hl.addWidget(self._en_chk) rm = QToolButton(); rm.setText("✕"); rm.setObjectName("plotRemoveBtn") rm.setFixedSize(22, 22); rm.clicked.connect(lambda: self.removed.emit(self)) hl.addWidget(rm) outer.addWidget(hdr) # ── Body ─────────────────────────────────────────────────────── body = QWidget(); body.setObjectName("plotBlockSettings") bl = QVBoxLayout(body); bl.setContentsMargins(10, 8, 10, 10); bl.setSpacing(6) # Meta row: id, unit, color meta = QHBoxLayout() meta.addWidget(QLabel("ID:")) self._id_edit = QLineEdit(self.dc.channel_id) self._id_edit.setObjectName("traceLabel"); self._id_edit.setFixedWidth(90) self._id_edit.textChanged.connect(lambda t: setattr(self.dc, "channel_id", t)) meta.addWidget(self._id_edit) meta.addWidget(QLabel("Unit:")) self._unit_edit = QLineEdit(self.dc.unit) self._unit_edit.setObjectName("traceLabel"); self._unit_edit.setFixedWidth(60) self._unit_edit.textChanged.connect(lambda t: setattr(self.dc, "unit", t)) meta.addWidget(self._unit_edit) meta.addStretch() bl.addLayout(meta) # Kind selector + help kind_row = QHBoxLayout() kind_row.addWidget(QLabel("Kind:")) self._kind_cb = QComboBox(); self._kind_cb.setObjectName("channelPickerCb") self._kind_cb.addItems(self._ALL_KINDS) self._kind_cb.setCurrentText(self.dc.kind) self._kind_cb.currentTextChanged.connect(self._on_kind_changed) kind_row.addWidget(self._kind_cb) kind_row.addStretch() bl.addLayout(kind_row) self._help_lbl = QLabel(self._KIND_HELP.get(self.dc.kind, "")) self._help_lbl.setObjectName("traceSource") self._help_lbl.setWordWrap(True) bl.addWidget(self._help_lbl) # Sources src_grp = QGroupBox("Sources"); src_grp.setObjectName("cfgGlobalBar") src_lay = QVBoxLayout(src_grp); src_lay.setContentsMargins(6,4,6,4); src_lay.setSpacing(4) self._src_rows: List[QHBoxLayout] = [] self._src_combos: List[QComboBox] = [] self._src_container = QWidget() self._src_lay = QVBoxLayout(self._src_container) self._src_lay.setContentsMargins(0,0,0,0); self._src_lay.setSpacing(3) for src in self.dc.sources: self._add_source_row(src) add_src_btn = QPushButton("＋ Add Source") add_src_btn.setObjectName("addTraceBtn") add_src_btn.clicked.connect(self._add_source_row) src_lay.addWidget(self._src_container) src_lay.addWidget(add_src_btn) bl.addWidget(src_grp) # Expression / Script editor (shown for script kinds only) self._script_grp = QGroupBox("Code") script_lay = QVBoxLayout(self._script_grp); script_lay.setContentsMargins(6,4,6,4) self._code_edit = QTextEdit() self._code_edit.setObjectName("codeEditor") self._code_edit.setPlaceholderText("Enter expression or function body…") self._code_edit.setMinimumHeight(80) self._code_edit.setMaximumHeight(160) if self.dc.kind == "expression": self._code_edit.setPlainText(self.dc.expression) else: self._code_edit.setPlainText(self.dc.script) script_lay.addWidget(self._code_edit) bl.addWidget(self._script_grp) # Compile / Apply apply_row = QHBoxLayout() self._status_lbl = QLabel("") self._status_lbl.setObjectName("traceSource") apply_row.addWidget(self._status_lbl, 1) apply_btn = QPushButton("✓ Compile & Apply") apply_btn.setObjectName("applyButton") apply_btn.clicked.connect(self._apply) apply_row.addWidget(apply_btn) bl.addLayout(apply_row) outer.addWidget(body) self._update_script_visibility() def _add_source_row(self, src: Tuple[str, str] = None): row = QHBoxLayout() cb = _make_channel_combo(self.registry) cb.setObjectName("channelPickerCb") if src: for i in range(cb.count()): if cb.itemData(i) == src: cb.setCurrentIndex(i); break rm = QToolButton(); rm.setText("✕"); rm.setObjectName("traceRemoveBtn") rm.setFixedSize(22, 22) rm.clicked.connect(lambda: self._remove_source_row(row, cb)) row.addWidget(cb, 1); row.addWidget(rm) self._src_lay.addLayout(row) self._src_combos.append(cb) self._src_rows.append(row) def _remove_source_row(self, row, cb): if cb in self._src_combos: self._src_combos.remove(cb) if row in self._src_rows: self._src_rows.remove(row) while row.count(): item = row.takeAt(0) if item.widget(): item.widget().deleteLater() def _on_kind_changed(self, kind: str): self.dc.kind = kind self._help_lbl.setText(self._KIND_HELP.get(kind, "")) self._update_script_visibility() def _update_script_visibility(self): show = self.dc.kind in self._SCRIPT_KINDS self._script_grp.setVisible(show) def _on_enable(self, v: bool): self.dc.enabled = v self._apply() def _apply(self): # Collect sources from combos self.dc.sources = [cb.currentData() for cb in self._src_combos if cb.currentData() is not None] # Collect code if self.dc.kind == "expression": self.dc.expression = self._code_edit.toPlainText().strip() elif self.dc.kind in ("function", "custom_script"): self.dc.script = self._code_edit.toPlainText() err = self.processor.add_derived(self.dc) if err: self._status_lbl.setText(f"⚠ {err[:80]}") self._status_lbl.setStyleSheet("color:#ef4444;") else: self._status_lbl.setText("✓ Applied") self._status_lbl.setStyleSheet("color:#22c55e;") self.changed.emit(self.dc) # ══════════════════════════════════════════════════════════════════════════════ # Channel visibility tab # ══════════════════════════════════════════════════════════════════════════════ class ChannelVisibilityTab(QWidget): visibility_changed = pyqtSignal() def __init__(self, registry: DeviceRegistry, processor: SignalProcessor): super().__init__() self.registry = registry self.processor = processor self._checks = {} # (dev_id, ch_id) -> QCheckBox self._build() def _build(self): scroll = QScrollArea(); scroll.setWidgetResizable(True) scroll.setObjectName("deviceScroll") container = QWidget() lay = QVBoxLayout(container); lay.setContentsMargins(10,10,10,10); lay.setSpacing(6) for dev in self.registry.all_instances(): grp = QGroupBox(f"{dev.info.icon} {dev.info.name} [{dev.info.device_id}]") grp_lay = QVBoxLayout(grp); grp_lay.setSpacing(3) for ch in dev.info.channels: row = QHBoxLayout() chk = QCheckBox(f"{ch.channel_id} — {ch.name}") chk.setChecked(ch.enabled) chk.setStyleSheet(f"color:{ch.color};") unit_lbl = QLabel(ch.unit); unit_lbl.setObjectName("traceSource") chk.toggled.connect(lambda v, c=ch: setattr(c, "enabled", v) or self.visibility_changed.emit()) row.addWidget(chk, 1); row.addWidget(unit_lbl) grp_lay.addLayout(row) self._checks[(dev.info.device_id, ch.channel_id)] = chk lay.addWidget(grp) lay.addStretch() scroll.setWidget(container) root = QVBoxLayout(self); root.setContentsMargins(0,0,0,0) root.addWidget(scroll) # ══════════════════════════════════════════════════════════════════════════════ # SignalBuilderWindow — main floating window # ══════════════════════════════════════════════════════════════════════════════ class SignalBuilderWindow(QWidget): """ Floating window with three tabs: Filters | Derived Channels | Channel Visibility """ pipeline_changed = pyqtSignal() derived_changed = pyqtSignal() visibility_changed = pyqtSignal() closed = pyqtSignal() def __init__(self, registry: DeviceRegistry, processor: SignalProcessor, parent=None): super().__init__(parent, Qt.WindowType.Window | Qt.WindowType.Tool) self.registry = registry self.processor = processor self._pipeline_editors: dict = {} self._derived_editors: List[DerivedEditor] = [] self._color_idx = 0 self.setWindowTitle("Signal Builder") self.setMinimumSize(640, 520) self.resize(720, 620) self._build() def _build(self): root = QVBoxLayout(self); root.setContentsMargins(0,0,0,0); root.setSpacing(0) # Title bar tb = QWidget(); tb.setObjectName("devWindowTitleBar"); tb.setFixedHeight(40) tbl = QHBoxLayout(tb); tbl.setContentsMargins(12,0,12,0) QLabel_t = QLabel("⚗ SIGNAL BUILDER"); QLabel_t.setObjectName("devWindowTitle") tbl.addWidget(QLabel_t) root.addWidget(tb) tabs = QTabWidget(); tabs.setObjectName("signalBuilderTabs") root.addWidget(tabs, 1) # ── Tab 1: Filters ──────────────────────────────────────────── tabs.addTab(self._build_filters_tab(), " Filters ") # ── Tab 2: Derived Channels ─────────────────────────────────── tabs.addTab(self._build_derived_tab(), " Derived ") # ── Tab 3: Channel Visibility ───────────────────────────────── vis_tab = ChannelVisibilityTab(self.registry, self.processor) vis_tab.visibility_changed.connect(self.visibility_changed) tabs.addTab(vis_tab, " Channels ") # ── Filters tab ─────────────────────────────────────────────────────── def _build_filters_tab(self): w = QWidget() lay = QVBoxLayout(w); lay.setContentsMargins(0,0,0,0); lay.setSpacing(0) # Channel selector top = QWidget(); top.setObjectName("cfgGlobalBar") tl = QHBoxLayout(top); tl.setContentsMargins(10,6,10,6) tl.addWidget(QLabel("Channel:")) self._filter_ch_cb = _make_channel_combo(self.registry) self._filter_ch_cb.setObjectName("channelPickerCb") self._filter_ch_cb.currentIndexChanged.connect(self._on_filter_channel_changed) tl.addWidget(self._filter_ch_cb, 1) lay.addWidget(top) div = QFrame(); div.setFrameShape(QFrame.Shape.HLine); div.setObjectName("devWindowDivider") lay.addWidget(div) # Pipeline editor area scroll = QScrollArea(); scroll.setWidgetResizable(True); scroll.setObjectName("deviceScroll") self._pipeline_container = QWidget() self._pipeline_lay = QVBoxLayout(self._pipeline_container) self._pipeline_lay.setContentsMargins(10,10,10,10); self._pipeline_lay.setSpacing(0) self._pipeline_lay.addStretch() scroll.setWidget(self._pipeline_container) lay.addWidget(scroll, 1) # Trigger initial load self._on_filter_channel_changed(0) return w def _on_filter_channel_changed(self, _idx: int): data = self._filter_ch_cb.currentData() if not data: return dev_id, ch_id = data # Clear old editor while self._pipeline_lay.count() > 1: item = self._pipeline_lay.takeAt(0) if item.widget(): item.widget().deleteLater() # Get or create pipeline pipeline = self.processor.get_pipeline(dev_id, ch_id) if pipeline is None: pipeline = ChannelPipeline(device_id=dev_id, channel_id=ch_id) self.processor.set_pipeline(pipeline) editor = PipelineEditor(pipeline, self.processor) editor.pipeline_changed.connect(self.pipeline_changed) self._pipeline_lay.insertWidget(0, editor) # ── Derived tab ─────────────────────────────────────────────────────── def _build_derived_tab(self): w = QWidget() lay = QVBoxLayout(w); lay.setContentsMargins(0,0,0,0); lay.setSpacing(0) # Add derived button bar top = QWidget(); top.setObjectName("cfgGlobalBar") tl = QHBoxLayout(top); tl.setContentsMargins(10,6,10,6) tl.addWidget(QLabel("New derived channel:")) self._new_kind_cb = QComboBox(); self._new_kind_cb.setObjectName("channelPickerCb") self._new_kind_cb.addItems(DerivedEditor._ALL_KINDS) tl.addWidget(self._new_kind_cb) add_btn = QPushButton("＋ Create") add_btn.setObjectName("addDeviceButton") add_btn.clicked.connect(self._add_derived) tl.addWidget(add_btn) lay.addWidget(top) div = QFrame(); div.setFrameShape(QFrame.Shape.HLine); div.setObjectName("devWindowDivider") lay.addWidget(div) # Derived editors scroll area scroll = QScrollArea(); scroll.setWidgetResizable(True); scroll.setObjectName("deviceScroll") self._derived_container = QWidget() self._derived_lay = QVBoxLayout(self._derived_container) self._derived_lay.setContentsMargins(10,10,10,10); self._derived_lay.setSpacing(10) self._derived_lay.addStretch() scroll.setWidget(self._derived_container) lay.addWidget(scroll, 1) # Load existing derived channels for dc in self.processor.get_derived(): self._insert_derived_editor(dc) return w def _add_derived(self): kind = self._new_kind_cb.currentText() color = _DERIVED_COLORS[self._color_idx % len(_DERIVED_COLORS)] self._color_idx += 1 dc = DerivedChannel( channel_id=f"derived_{len(self._derived_editors)}", name=f"Derived {len(self._derived_editors) + 1}", kind=kind, color=color, ) self._insert_derived_editor(dc) def _insert_derived_editor(self, dc: DerivedChannel): editor = DerivedEditor(dc, self.registry, self.processor, self._color_idx) editor.removed.connect(self._remove_derived_editor) editor.changed.connect(lambda _: self.derived_changed.emit()) self._derived_editors.append(editor) self._derived_lay.insertWidget(self._derived_lay.count() - 1, editor) def _remove_derived_editor(self, editor: DerivedEditor): self.processor.remove_derived(editor.dc.channel_id) self._derived_editors.remove(editor) self._derived_lay.removeWidget(editor); editor.deleteLater() self.derived_changed.emit() def closeEvent(self, event: QCloseEvent): self.closed.emit(); event.accept()