diff options
Diffstat (limited to 'ui/signal_builder.py')
| -rw-r--r-- | ui/signal_builder.py | 605 |
1 files changed, 605 insertions, 0 deletions
diff --git a/ui/signal_builder.py b/ui/signal_builder.py new file mode 100644 index 0000000..a84d861 --- /dev/null +++ b/ui/signal_builder.py @@ -0,0 +1,605 @@ +""" +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() |