""" ui/windows/channels_window.py CHANNELS window — toolbar section 3. Layout (QSplitter, horizontal): Physical Channels — per-channel filter pipeline; add/remove channels, configure filters (moving average, low-pass, etc.), math ops (derivative, integral), calibrate. Virtual Channels — create derived channels (RMS, power, difference, sum, Python expression/script, etc.) Classes: ChannelsWindow — top-level floating window PipelineTab — splitter content (physical + virtual sections) ChannelPipelineBlock — collapsible filter+math stack for one physical channel ChannelPickerDialog — checkbox dialog for selecting channels to add DerivedBlock — editor for one virtual/derived channel FilterRow — single filter stage row inside a pipeline block SignalsListTab — unused; kept for profile-load compatibility """ from __future__ import annotations from typing import List, Optional, Set, Tuple from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QScrollArea, QFrame, QComboBox, QLineEdit, QDoubleSpinBox, QSpinBox, QCheckBox, QTextEdit, QGroupBox, QToolButton, QSplitter, QDialog, QDialogButtonBox, ) from PyQt6.QtCore import Qt, pyqtSignal from PyQt6.QtGui import QCloseEvent, QFont, QFontMetrics from devices.device_registry import DeviceRegistry from core.signal_processor import ( SignalProcessor, ChannelPipeline, DerivedChannel, FilterBase, FILTER_CLASSES, ) _DERIVED_COLORS = [ "#f72585","#7209b7","#4361ee","#4cc9f0", "#f77f00","#d62828","#588157","#e9c46a", ] _FILTER_KEYS = [k for k in FILTER_CLASSES if k not in ("derivative", "integral", "scale_offset")] _MATH_KEYS = ["derivative", "integral"] def _channel_combo(registry: DeviceRegistry, processor: Optional[SignalProcessor] = None, include_derived: bool = False, show_unit: bool = False) -> QComboBox: cb = QComboBox(); cb.setObjectName("channelPickerCb") for dev in registry.all_instances(): for ch in dev.info.channels: if not ch.enabled: continue label = f"{dev.info.device_id} / {ch.channel_id} ({ch.name})" if show_unit and ch.unit: label += f" [{ch.unit}]" cb.addItem(label, userData=(dev.info.device_id, ch.channel_id)) if include_derived and processor: for dc in processor.get_derived(): label = f"[derived] {dc.channel_id} ({dc.name})" if show_unit and dc.unit: label += f" [{dc.unit}]" cb.addItem(label, userData=("derived", dc.channel_id)) return cb # ══════════════════════════════════════════════════════════════════════════════ # Channel Picker Dialog # ══════════════════════════════════════════════════════════════════════════════ class ChannelPickerDialog(QDialog): def __init__(self, registry: DeviceRegistry, already_shown: Set[Tuple[str, str]], parent=None): super().__init__(parent) self.setWindowTitle("Add Channels") self.setMinimumWidth(380) self.resize(420, 320) self._checks: List[Tuple[QCheckBox, str, str]] = [] lay = QVBoxLayout(self) scroll = QScrollArea(); scroll.setWidgetResizable(True) scroll.setObjectName("deviceScroll") cont = QWidget() cl = QVBoxLayout(cont); cl.setContentsMargins(8, 8, 8, 8); cl.setSpacing(4) any_available = False for dev in registry.all_instances(): for ch in dev.info.channels: if not ch.enabled: continue if (dev.info.device_id, ch.channel_id) in already_shown: continue any_available = True label = f"{dev.info.device_id} / {ch.channel_id} ({ch.name})" if ch.unit: label += f" [{ch.unit}]" chk = QCheckBox(label) self._checks.append((chk, dev.info.device_id, ch.channel_id)) cl.addWidget(chk) if not any_available: cl.addWidget(QLabel("No additional signals available.")) cl.addStretch() scroll.setWidget(cont) lay.addWidget(scroll, 1) btns = QDialogButtonBox( QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel ) btns.accepted.connect(self.accept) btns.rejected.connect(self.reject) lay.addWidget(btns) def selected(self) -> List[Tuple[str, str]]: return [(d, c) for chk, d, c in self._checks if chk.isChecked()] # ══════════════════════════════════════════════════════════════════════════════ # Filter row # ══════════════════════════════════════════════════════════════════════════════ _PARAM_SPECS = { "moving_average": [("window", "int", 10, 1, 5000)], "median": [("window", "int", 5, 1, 500)], "low_pass": [("alpha", "float", 0.1, 0.001, 1.0)], "high_pass": [("alpha", "float", 0.9, 0.001, 1.0)], "scale_offset": [("scale", "float", 1.0, -1e6, 1e6), ("offset", "float", 0.0, -1e9, 1e9)], "derivative": [], "integral": [], } class FilterRow(QFrame): removed = pyqtSignal(object) changed = pyqtSignal() 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) lbl = QLabel(self.filt.name.replace("_", " ").title()) lbl.setObjectName("traceSource"); lbl.setMinimumWidth(120) lay.addWidget(lbl) for pname, ptype, default, lo, hi in _PARAM_SPECS.get(self.filt.name, []): pl = QLabel(f"{pname}:"); pl.setObjectName("pwmLabel") lay.addWidget(pl) if ptype == "int": w = QSpinBox(); w.setRange(int(lo), int(hi)) w.setValue(int(self.filt.params.get(pname, default))) w.setFixedWidth(70); w.setObjectName("traceWidthSpin") w.valueChanged.connect(lambda v, p=pname: self._upd(p, v)) else: w = QDoubleSpinBox(); w.setRange(lo, hi); w.setDecimals(4) w.setValue(float(self.filt.params.get(pname, default))) w.setFixedWidth(88); w.setObjectName("traceWidthSpin") w.valueChanged.connect(lambda v, p=pname: self._upd(p, v)) lay.addWidget(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 _upd(self, name, value): self.filt.params[name] = value cls = FILTER_CLASSES.get(self.filt.name) if cls: nf = cls(**self.filt.params); self.filt.__dict__.update(nf.__dict__) self.changed.emit() # monkey-patch QLabel.also for inline objectName from PyQt6.QtWidgets import QLabel as _QLabel def _also(self, fn): fn(self); return self _QLabel.also = _also # ══════════════════════════════════════════════════════════════════════════════ # Channel Pipeline Block — collapsible filter+math stack for one channel # ══════════════════════════════════════════════════════════════════════════════ class ChannelPipelineBlock(QFrame): changed = pyqtSignal() removed = pyqtSignal(object) def __init__(self, dev_id: str, ch_id: str, ch_name: str, unit: str, processor: SignalProcessor): super().__init__() self.dev_id = dev_id self.ch_id = ch_id self.processor = processor self._filter_rows: List[FilterRow] = [] self._cal_rows: List[FilterRow] = [] self._cur_pipeline: Optional[ChannelPipeline] = None self._expanded = False self.setObjectName("plotBlock") self._build(ch_name, unit) self._load() def _build(self, ch_name: str, unit: str): self._ch_name = ch_name self._unit = unit 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) title = f"{self.dev_id} / {self.ch_id} ({ch_name})" if unit: title += f" [{unit}]" self._title_lbl = QLabel(title); self._title_lbl.setObjectName("traceSource") hl.addWidget(self._title_lbl, 1) rm_btn = QToolButton(); rm_btn.setText("✕") rm_btn.setObjectName("traceRemoveBtn"); rm_btn.setFixedSize(22, 22) rm_btn.clicked.connect(lambda: self.removed.emit(self)) hl.addWidget(rm_btn) self._toggle_btn = QToolButton(); self._toggle_btn.setText("▶") self._toggle_btn.setObjectName("traceRemoveBtn"); self._toggle_btn.setFixedSize(22, 22) self._toggle_btn.clicked.connect(self._toggle) hl.addWidget(self._toggle_btn) outer.addWidget(hdr) hdr.mousePressEvent = lambda e: self._toggle() # Body self._body = QWidget(); self._body.setObjectName("plotBlockSettings") bl = QVBoxLayout(self._body) bl.setContentsMargins(10, 8, 10, 10); bl.setSpacing(6) # ── Filters section ─────────────────────────────────────────────────── filt_lbl = QLabel("Filters"); filt_lbl.setObjectName("pwmLabel") bl.addWidget(filt_lbl) self._filter_area = QWidget() self._filter_lay = QVBoxLayout(self._filter_area) self._filter_lay.setContentsMargins(0, 0, 0, 0); self._filter_lay.setSpacing(3) bl.addWidget(self._filter_area) filt_add_bar = QHBoxLayout() self._filt_cb = QComboBox(); self._filt_cb.setObjectName("channelPickerCb") self._filt_cb.addItems([k.replace("_", " ").title() for k in _FILTER_KEYS]) filt_add_bar.addWidget(self._filt_cb, 1) filt_add_btn = QPushButton("＋ Add Filter"); filt_add_btn.setObjectName("addTraceBtn") filt_add_btn.clicked.connect(self._add_filter) filt_add_bar.addWidget(filt_add_btn) bl.addLayout(filt_add_bar) div = QFrame(); div.setFrameShape(QFrame.Shape.HLine) div.setObjectName("devWindowDivider") bl.addWidget(div) # ── Calibration section ─────────────────────────────────────────────── cal_lbl = QLabel("Calibration"); cal_lbl.setObjectName("pwmLabel") bl.addWidget(cal_lbl) self._cal_area = QWidget() self._cal_lay = QVBoxLayout(self._cal_area) self._cal_lay.setContentsMargins(0, 0, 0, 0); self._cal_lay.setSpacing(3) bl.addWidget(self._cal_area) cal_add_bar = QHBoxLayout() cal_add_btn = QPushButton("＋ Add Calibration"); cal_add_btn.setObjectName("addTraceBtn") cal_add_btn.clicked.connect(self._add_calibrate) cal_add_bar.addWidget(cal_add_btn) cal_add_bar.addStretch() bl.addLayout(cal_add_bar) ap_row = QHBoxLayout(); ap_row.addStretch() ap_btn = QPushButton("✓ Apply"); ap_btn.setObjectName("applyButton") ap_btn.clicked.connect(self._push) ap_row.addWidget(ap_btn) bl.addLayout(ap_row) outer.addWidget(self._body) self._body.setVisible(False) def _refresh_title(self): title = f"{self.dev_id} / {self.ch_id} ({self._ch_name})" if self._unit: title += f" [{self._unit}]" self._title_lbl.setText(title) def update_name(self, ch_name: str): self._ch_name = ch_name self._refresh_title() def update_unit(self, unit: str): self._unit = unit self._refresh_title() def _toggle(self): self._expanded = not self._expanded self._body.setVisible(self._expanded) self._toggle_btn.setText("▼" if self._expanded else "▶") def _load(self): pipeline = self.processor.get_pipeline(self.dev_id, self.ch_id) if pipeline is None: pipeline = ChannelPipeline(device_id=self.dev_id, channel_id=self.ch_id) self.processor.set_pipeline(pipeline) self._cur_pipeline = pipeline pipeline.enabled = True for f in pipeline.filters: if f.name == "scale_offset": self._add_cal_row(f) elif f.name not in _MATH_KEYS: self._add_filter_row(f) def _add_calibrate(self): filt = FILTER_CLASSES["scale_offset"]() self._add_cal_row(filt) self._push() if not self._expanded: self._toggle() def _add_filter(self): key = _FILTER_KEYS[self._filt_cb.currentIndex()] filt = FILTER_CLASSES[key]() self._add_filter_row(filt) self._push() def _add_filter_row(self, filt: FilterBase): row = FilterRow(filt) row.removed.connect(self._remove_row) row.changed.connect(self._push) self._filter_rows.append(row) self._filter_lay.addWidget(row) def _add_cal_row(self, filt: FilterBase): row = FilterRow(filt) row.removed.connect(self._remove_row) row.changed.connect(self._push) self._cal_rows.append(row) self._cal_lay.addWidget(row) def _remove_row(self, row: FilterRow): if row in self._filter_rows: self._filter_rows.remove(row) self._filter_lay.removeWidget(row) elif row in self._cal_rows: self._cal_rows.remove(row) self._cal_lay.removeWidget(row) row.deleteLater() self._push() def _push(self): if self._cur_pipeline: self._cur_pipeline.filters = ( [r.filt for r in self._filter_rows] + [r.filt for r in self._cal_rows] ) self.processor.set_pipeline(self._cur_pipeline) self.changed.emit() # ══════════════════════════════════════════════════════════════════════════════ # Pipeline Tab — filter pipelines per channel + virtual channel creation # ══════════════════════════════════════════════════════════════════════════════ class PipelineTab(QWidget): changed = pyqtSignal() derived_changed = pyqtSignal() def __init__(self, registry: DeviceRegistry, processor: SignalProcessor): super().__init__() self.registry = registry self.processor = processor self._shown: Set[Tuple[str, str]] = set() self._blocks: dict[Tuple[str, str], ChannelPipelineBlock] = {} self._derived_blocks: List[DerivedBlock] = [] self._color_idx = 0 self._build() def _build(self): root = QVBoxLayout(self); root.setContentsMargins(0, 0, 0, 0); root.setSpacing(0) splitter = QSplitter(Qt.Orientation.Horizontal) splitter.setObjectName("channelSplitter") root.addWidget(splitter) # ── Physical Channels pane ───────────────────────────────────────────── phys_w = QWidget() phys_l = QVBoxLayout(phys_w); phys_l.setContentsMargins(0, 0, 0, 0); phys_l.setSpacing(0) phys_bar = QWidget(); phys_bar.setObjectName("cfgGlobalBar") pb_lay = QHBoxLayout(phys_bar); pb_lay.setContentsMargins(10, 7, 10, 7); pb_lay.setSpacing(6) pb_lbl = QLabel("Signals"); pb_lbl.setObjectName("devWindowTitle") pb_lay.addWidget(pb_lbl, 1) add_btn = QPushButton("＋ Add"); add_btn.setObjectName("addTraceBtn") add_btn.clicked.connect(self._on_add_channel) pb_lay.addWidget(add_btn) phys_l.addWidget(phys_bar) div1 = QFrame(); div1.setFrameShape(QFrame.Shape.HLine) div1.setObjectName("devWindowDivider"); phys_l.addWidget(div1) phys_scroll = QScrollArea(); phys_scroll.setWidgetResizable(True) phys_scroll.setObjectName("deviceScroll") phys_scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff) self._phys_cont = QWidget() self._phys_inner = QVBoxLayout(self._phys_cont) self._phys_inner.setContentsMargins(10, 10, 10, 10); self._phys_inner.setSpacing(6) self._phys_inner.addStretch() phys_scroll.setWidget(self._phys_cont) phys_l.addWidget(phys_scroll, 1) splitter.addWidget(phys_w) # ── Virtual Channels pane ────────────────────────────────────────────── virt_w = QWidget() virt_l = QVBoxLayout(virt_w); virt_l.setContentsMargins(0, 0, 0, 0); virt_l.setSpacing(0) virt_bar = QWidget(); virt_bar.setObjectName("cfgGlobalBar") vb_lay = QHBoxLayout(virt_bar); vb_lay.setContentsMargins(10, 7, 10, 7); vb_lay.setSpacing(6) vb_lbl = QLabel("Channels"); vb_lbl.setObjectName("devWindowTitle") vb_lay.addWidget(vb_lbl, 1) add_virt = QPushButton("＋ Add Channel"); add_virt.setObjectName("addTraceBtn") add_virt.clicked.connect(self._add_virtual) vb_lay.addWidget(add_virt) virt_l.addWidget(virt_bar) div2 = QFrame(); div2.setFrameShape(QFrame.Shape.HLine) div2.setObjectName("devWindowDivider"); virt_l.addWidget(div2) virt_scroll = QScrollArea(); virt_scroll.setWidgetResizable(True) virt_scroll.setObjectName("deviceScroll") virt_scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff) self._virt_cont = QWidget() self._virt_inner = QVBoxLayout(self._virt_cont) self._virt_inner.setContentsMargins(10, 10, 10, 10); self._virt_inner.setSpacing(6) self._virt_inner.addStretch() virt_scroll.setWidget(self._virt_cont) virt_l.addWidget(virt_scroll, 1) splitter.addWidget(virt_w) splitter.setStretchFactor(0, 0) splitter.setStretchFactor(1, 1) def _init_split(): min_w = splitter.widget(0).minimumSizeHint().width() w = max(min_w, 280) splitter.setSizes([w, max(1, splitter.width() - w)]) from PyQt6.QtCore import QTimer QTimer.singleShot(0, _init_split) for dev in self.registry.all_instances(): for ch in dev.info.channels: if ch.enabled: self._add_block(dev.info.device_id, ch.channel_id, ch.name, ch.unit) # ── Physical channel block management ────────────────────────────────────── def _add_block(self, dev_id: str, ch_id: str, ch_name: str, unit: str): if (dev_id, ch_id) in self._shown: return self._shown.add((dev_id, ch_id)) block = ChannelPipelineBlock(dev_id, ch_id, ch_name, unit, self.processor) block.changed.connect(self.changed) block.removed.connect(self._on_remove_block) self._blocks[(dev_id, ch_id)] = block self._phys_inner.insertWidget(self._phys_inner.count() - 1, block) def on_channel_enabled_changed(self, dev_id: str, ch_id: str, enabled: bool): key = (dev_id, ch_id) if enabled: if key not in self._shown: dev = self.registry.get_instance(dev_id) if dev: ch = dev.get_channel(ch_id) if ch: self._add_block(dev_id, ch_id, ch.name, ch.unit) else: block = self._blocks.get(key) if block: block.setVisible(True) else: block = self._blocks.get(key) if block: block.setVisible(False) def on_device_removed(self, dev_id: str): keys = [(d, c) for (d, c) in list(self._shown) if d == dev_id] for key in keys: block = self._blocks.get(key) if block: self._on_remove_block(block) def on_device_reconfigured(self, dev_id: str): dev = self.registry.get_instance(dev_id) if dev is None: self.on_device_removed(dev_id) return valid = {ch.channel_id for ch in dev.info.channels} keys = [(d, c) for (d, c) in list(self._shown) if d == dev_id and c not in valid] for key in keys: block = self._blocks.get(key) if block: self._on_remove_block(block) def on_device_added(self): pass def on_channel_name_changed(self, dev_id: str, ch_id: str, ch_name: str): key = (dev_id, ch_id) if key in self._blocks: self._blocks[key].update_name(ch_name) else: dev = self.registry.get_instance(dev_id) if dev: ch = dev.get_channel(ch_id) if ch and ch.enabled: self._add_block(dev_id, ch_id, ch_name, ch.unit) def on_channel_unit_changed(self, dev_id: str, ch_id: str, unit: str): key = (dev_id, ch_id) if key in self._blocks: self._blocks[key].update_unit(unit) def _on_add_channel(self): dlg = ChannelPickerDialog(self.registry, self._shown, self) if dlg.exec() != QDialog.DialogCode.Accepted: return for dev_id, ch_id in dlg.selected(): dev = self.registry.get_instance(dev_id) if not dev: continue ch = dev.get_channel(ch_id) if not ch: continue self._add_block(dev_id, ch_id, ch.name, ch.unit) def _on_remove_block(self, block: ChannelPipelineBlock): key = (block.dev_id, block.ch_id) self._shown.discard(key) self._blocks.pop(key, None) self._phys_inner.removeWidget(block) block.deleteLater() # ── Virtual channel management ───────────────────────────────────────────── def _add_virtual(self): color = _DERIVED_COLORS[self._color_idx % len(_DERIVED_COLORS)] self._color_idx += 1 n = len(self._derived_blocks) + 1 dc = DerivedChannel(channel_id=f"CH_{n}", name=f"Channel {n}", kind="expression", color=color) blk = self._make_derived_block(dc) self._virt_inner.insertWidget(self._virt_inner.count() - 1, blk) def _make_derived_block(self, dc: DerivedChannel) -> DerivedBlock: blk = DerivedBlock(dc, self.registry, self.processor) blk.removed.connect(self._rm_virtual) blk.applied.connect(lambda _: self.derived_changed.emit()) self._derived_blocks.append(blk) return blk def _rm_virtual(self, blk: DerivedBlock): self.processor.remove_derived(blk.dc.channel_id) self._derived_blocks.remove(blk) self._virt_inner.removeWidget(blk); blk.deleteLater() self.derived_changed.emit() def refresh_derived(self): """Rebuild DerivedBlock widgets from processor state (called after profile load).""" for blk in self._derived_blocks: self._virt_inner.removeWidget(blk); blk.deleteLater() self._derived_blocks.clear() self._color_idx = 0 for dc in self.processor.get_derived(): blk = self._make_derived_block(dc) self._virt_inner.insertWidget(self._virt_inner.count() - 1, blk) self._color_idx = len(self._derived_blocks) # ══════════════════════════════════════════════════════════════════════════════ # Derived editor for one DerivedChannel # ══════════════════════════════════════════════════════════════════════════════ _BUILTIN = ["derivative", "second_derivative", "power", "rms", "difference", "sum"] _SCRIPT = ["expression", "function", "custom_script"] _ALL_KINDS = _BUILTIN + _SCRIPT _KIND_DISPLAY = {k: k.replace("_", " ").title() for k in _ALL_KINDS} _KIND_HELP = { "derivative": "Backward difference dy/dt from 1 source", "second_derivative": "Second derivative d²y/dt² from 1 source", "power": "V × I from 2 sources (voltage, current)", "rms": "Rolling RMS from 1 source (set window below)", "difference": "source[0] − source[1] from 2 sources", "sum": "Σ all source values", "expression": "x = list of source values, t = elapsed time (s)\nExample: x[0] * 3.14159 / 180", "function": "def compute(x, t):\n # x = source values list\n return x[0] * 2.0", "custom_script":"Full Python. Must define: def compute(x, t): ...", } class DerivedBlock(QFrame): removed = pyqtSignal(object) applied = pyqtSignal(object) # DerivedChannel def __init__(self, dc: DerivedChannel, registry: DeviceRegistry, processor: SignalProcessor): super().__init__() self.dc = dc; self.registry = registry; self.processor = processor self.setObjectName("plotBlock") self._src_combos: List[QComboBox] = [] self._expanded = False 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 = QLineEdit(self.dc.name); self._name.setObjectName("plotBlockTitle") self._name.textChanged.connect(lambda t: setattr(self.dc, "name", t)) hl.addWidget(self._name, 1) rm = QToolButton(); rm.setText("✕"); rm.setObjectName("plotRemoveBtn") rm.setFixedSize(22, 22); rm.clicked.connect(lambda: self.removed.emit(self)) hl.addWidget(rm) self._toggle_btn = QToolButton(); self._toggle_btn.setText("▶") self._toggle_btn.setObjectName("traceRemoveBtn"); self._toggle_btn.setFixedSize(22, 22) self._toggle_btn.clicked.connect(self._toggle) hl.addWidget(self._toggle_btn) outer.addWidget(hdr) hdr.mousePressEvent = lambda e: self._toggle() # Body body = QWidget(); body.setObjectName("plotBlockSettings") bl = QVBoxLayout(body); bl.setContentsMargins(10, 8, 10, 10); bl.setSpacing(6) # ID / unit row meta = QHBoxLayout() meta.addWidget(QLabel("ID:")) self._id = QLineEdit(self.dc.channel_id) self._id.setObjectName("traceLabel"); self._id.setFixedWidth(100) self._id.textChanged.connect(lambda t: setattr(self.dc, "channel_id", t)) meta.addWidget(self._id) meta.addWidget(QLabel("Unit:")) self._unit = QLineEdit(self.dc.unit) self._unit.setObjectName("traceLabel"); self._unit.setFixedWidth(60) self._unit.textChanged.connect(lambda t: setattr(self.dc, "unit", t)) meta.addWidget(self._unit); meta.addStretch() bl.addLayout(meta) # Kind combo + help k_row = QHBoxLayout(); k_row.addWidget(QLabel("Operation:")) self._kind_cb = QComboBox(); self._kind_cb.setObjectName("channelPickerCb") self._kind_cb.addItems([_KIND_DISPLAY[k] for k in _ALL_KINDS]) cur_idx = _ALL_KINDS.index(self.dc.kind) if self.dc.kind in _ALL_KINDS else 0 self._kind_cb.setCurrentIndex(cur_idx) self._kind_cb.currentIndexChanged.connect(self._on_kind) k_row.addWidget(self._kind_cb); k_row.addStretch() bl.addLayout(k_row) self._help = QLabel(_KIND_HELP.get(self.dc.kind, "")) self._help.setObjectName("traceSource"); self._help.setWordWrap(True) bl.addWidget(self._help) # RMS window self._win_row = QHBoxLayout() self._win_row.addWidget(QLabel("Window samples:")) self._win_sp = QSpinBox(); self._win_sp.setRange(2, 10000) self._win_sp.setValue(self.dc.params.get("window", 20)) self._win_sp.setObjectName("traceWidthSpin"); self._win_sp.setFixedWidth(80) self._win_sp.valueChanged.connect(lambda v: self.dc.params.update({"window": v})) self._win_row.addWidget(self._win_sp); self._win_row.addStretch() self._win_widget = QWidget(); self._win_widget.setLayout(self._win_row) bl.addWidget(self._win_widget) # Sources — include derived channels as valid sources src_grp = QGroupBox("Sources"); src_grp.setObjectName("cfgGlobalBar") src_lay = QVBoxLayout(src_grp); src_lay.setContentsMargins(6, 4, 6, 4); src_lay.setSpacing(3) self._src_cont = QWidget(); self._src_vlay = QVBoxLayout(self._src_cont) self._src_vlay.setContentsMargins(0, 0, 0, 0); self._src_vlay.setSpacing(3) _is_deriv = self.dc.kind in ("derivative", "second_derivative") if _is_deriv: self._add_src(self.dc.sources[0] if self.dc.sources else None) else: for s in self.dc.sources: self._add_src(s) self._add_src_btn = QPushButton("＋ Add Source"); self._add_src_btn.setObjectName("addTraceBtn") self._add_src_btn.clicked.connect(lambda: self._add_src()) src_lay.addWidget(self._src_cont); src_lay.addWidget(self._add_src_btn) bl.addWidget(src_grp) # W.r.t. — only for derivative/second_derivative self._wrt_grp = QGroupBox("With Respect To"); self._wrt_grp.setObjectName("cfgGlobalBar") wrt_lay = QVBoxLayout(self._wrt_grp); wrt_lay.setContentsMargins(6, 6, 6, 6); wrt_lay.setSpacing(4) wrt_top = QHBoxLayout(); wrt_top.addWidget(QLabel("Variable:")) self._wrt_cb = QComboBox(); self._wrt_cb.setObjectName("channelPickerCb") self._wrt_cb.addItems(["Time", "Channel"]) _cur_wrt = self.dc.params.get("deriv_wrt", "time") self._wrt_cb.setCurrentIndex(1 if _cur_wrt == "channel" else 0) self._wrt_cb.currentIndexChanged.connect(self._on_wrt_changed) wrt_top.addWidget(self._wrt_cb); wrt_top.addStretch() wrt_lay.addLayout(wrt_top) self._wrt_x_row = QWidget() wx_lay = QHBoxLayout(self._wrt_x_row); wx_lay.setContentsMargins(0, 0, 0, 0); wx_lay.setSpacing(6) wx_lay.addWidget(QLabel("Channel (x):")) self._wrt_x_cb = _channel_combo(self.registry, self.processor, include_derived=True, show_unit=True) self._wrt_x_cb.setObjectName("channelPickerCb") if _cur_wrt == "channel" and len(self.dc.sources) > 1: x_src = self.dc.sources[1] for i in range(self._wrt_x_cb.count()): if self._wrt_x_cb.itemData(i) == x_src: self._wrt_x_cb.setCurrentIndex(i); break wx_lay.addWidget(self._wrt_x_cb, 1) wrt_lay.addWidget(self._wrt_x_row) bl.addWidget(self._wrt_grp) # Code editor self._code_grp = QGroupBox("Code") code_lay = QVBoxLayout(self._code_grp); code_lay.setContentsMargins(6, 4, 6, 4) self._code = QTextEdit(); self._code.setObjectName("codeEditor") self._code.setMinimumHeight(90); self._code.setMaximumHeight(180) _mono = QFont("IBM Plex Mono, Consolas, Monospace"); _mono.setStyleHint(QFont.StyleHint.Monospace) self._code.setFont(_mono) self._code.setTabStopDistance(QFontMetrics(_mono).horizontalAdvance(" ") * 4) txt = self.dc.expression if self.dc.kind == "expression" else self.dc.script self._code.setPlainText(txt) code_lay.addWidget(self._code) bl.addWidget(self._code_grp) # Status + apply ap_row = QHBoxLayout() self._st = QLabel(""); self._st.setObjectName("traceSource") ap_row.addWidget(self._st, 1) ap = QPushButton("✓ Apply"); ap.setObjectName("applyButton") ap.clicked.connect(self._apply) ap_row.addWidget(ap) bl.addLayout(ap_row) body.setVisible(False) outer.addWidget(body) self._body = body self._update_visibility() def _toggle(self): self._expanded = not self._expanded self._body.setVisible(self._expanded) self._toggle_btn.setText("▼" if self._expanded else "▶") def _add_src(self, src=None): row = QHBoxLayout() cb = _channel_combo(self.registry, self.processor, include_derived=True, show_unit=True) 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._rm_src(row, cb)) row.addWidget(cb, 1); row.addWidget(rm) self._src_vlay.addLayout(row) self._src_combos.append(cb) def _rm_src(self, row, cb): if cb in self._src_combos: self._src_combos.remove(cb) while row.count(): it = row.takeAt(0) if it.widget(): it.widget().deleteLater() def _on_kind(self, idx: int): k = _ALL_KINDS[idx] self.dc.kind = k self._help.setText(_KIND_HELP.get(k, "")) if k in ("derivative", "second_derivative") and not self._src_combos: self._add_src() self._update_visibility() def _on_wrt_changed(self, idx: int): self._wrt_x_row.setVisible(idx == 1) def _update_visibility(self): is_deriv = self.dc.kind in ("derivative", "second_derivative") self._code_grp.setVisible(self.dc.kind in _SCRIPT) self._win_widget.setVisible(self.dc.kind == "rms") self._wrt_grp.setVisible(is_deriv) self._add_src_btn.setVisible(not is_deriv) if is_deriv: self._wrt_x_row.setVisible(self._wrt_cb.currentIndex() == 1) def _apply(self): sources = [cb.currentData() for cb in self._src_combos if cb.currentData()] if self.dc.kind in ("derivative", "second_derivative"): wrt = "channel" if self._wrt_cb.currentIndex() == 1 else "time" self.dc.params["deriv_wrt"] = wrt if wrt == "channel": x_src = self._wrt_x_cb.currentData() if x_src: sources.append(x_src) self.dc.sources = sources if self.dc.kind == "expression": self.dc.expression = self._code.toPlainText().strip() elif self.dc.kind in ("function", "custom_script"): self.dc.script = self._code.toPlainText() err = self.processor.add_derived(self.dc) if err: self._st.setText(f"⚠ {err[:100]}") self._st.setStyleSheet("color:#ef4444;") else: self._st.setText("✓ Applied") self._st.setStyleSheet("color:#22c55e;") self.applied.emit(self.dc) # ══════════════════════════════════════════════════════════════════════════════ # Signals List Tab — flat list of all signals with enable/disable # ══════════════════════════════════════════════════════════════════════════════ class SignalsListTab(QWidget): visibility_changed = pyqtSignal() channel_toggled = pyqtSignal(str, str, bool) # dev_id, ch_id, enabled def __init__(self, registry: DeviceRegistry, processor: SignalProcessor): super().__init__() self.registry = registry self.processor = processor self._build() def _build(self): scroll = QScrollArea(); scroll.setWidgetResizable(True) scroll.setObjectName("deviceScroll") scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff) self._cont = QWidget() self._lay = QVBoxLayout(self._cont) self._lay.setContentsMargins(8, 8, 8, 8); self._lay.setSpacing(2) scroll.setWidget(self._cont) root = QVBoxLayout(self); root.setContentsMargins(0, 0, 0, 0) root.addWidget(scroll) self.refresh() def refresh(self): while self._lay.count(): it = self._lay.takeAt(0) if it.widget(): it.widget().deleteLater() for dev in self.registry.all_instances(): for ch in dev.info.channels: dev_id_cap = dev.info.device_id ch_id_cap = ch.channel_id self._lay.addWidget(self._make_row( color = ch.color, name = f"{ch.channel_id} — {ch.name}", unit = ch.unit, source = dev_id_cap, enabled = ch.enabled, on_toggle = lambda v, c=ch, d=dev_id_cap, ci=ch_id_cap: ( setattr(c, "enabled", v), self.visibility_changed.emit(), self.channel_toggled.emit(d, ci, v), ), )) derived = self.processor.get_derived() if self.registry.all_instances() and derived: div = QFrame(); div.setFrameShape(QFrame.Shape.HLine) div.setObjectName("devWindowDivider") self._lay.addWidget(div) for dc in derived: self._lay.addWidget(self._make_row( color = dc.color, name = f"{dc.channel_id} — {dc.name}", unit = dc.unit, source = "virtual", enabled = dc.enabled, on_toggle = lambda v, d=dc: (setattr(d, "enabled", v), self.visibility_changed.emit()), )) self._lay.addStretch() def _make_row(self, color: str, name: str, unit: str, source: str, enabled: bool, on_toggle) -> QFrame: row = QFrame(); row.setObjectName("traceRow") lay = QHBoxLayout(row); lay.setContentsMargins(8, 4, 8, 4); lay.setSpacing(8) dot = QLabel("●"); dot.setStyleSheet(f"color:{color};"); dot.setFixedWidth(14) lay.addWidget(dot) chk = QCheckBox(name); chk.setChecked(enabled) chk.toggled.connect(on_toggle) lay.addWidget(chk, 1) if unit: ul = QLabel(unit); ul.setObjectName("traceSource") ul.setFixedWidth(52) ul.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter) lay.addWidget(ul) sl = QLabel(source); sl.setObjectName("traceSource") sl.setFixedWidth(72) sl.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter) lay.addWidget(sl) return row # ══════════════════════════════════════════════════════════════════════════════ # ChannelsWindow — main # ══════════════════════════════════════════════════════════════════════════════ class ChannelsWindow(QWidget): pipeline_changed = pyqtSignal() derived_changed = pyqtSignal() closed = pyqtSignal() def __init__(self, registry: DeviceRegistry, processor: SignalProcessor, parent=None): super().__init__(parent, Qt.WindowType.Window) self.registry = registry self.processor = processor self.setWindowTitle("Channels") self.setMinimumSize(800, 520) self.resize(1400, 820) self._build() def _build(self): root = QVBoxLayout(self); root.setContentsMargins(0, 0, 0, 0); root.setSpacing(0) hdr = QWidget(); hdr.setObjectName("devWindowTitleBar"); hdr.setFixedHeight(44) hl = QHBoxLayout(hdr); hl.setContentsMargins(14, 0, 14, 0) hl.addWidget(QLabel("CHANNELS").also(lambda w: w.setObjectName("devWindowTitle")), 1) root.addWidget(hdr) div = QFrame(); div.setFrameShape(QFrame.Shape.HLine) div.setObjectName("devWindowDivider"); root.addWidget(div) self._pip_tab = PipelineTab(self.registry, self.processor) self._pip_tab.changed.connect(self.pipeline_changed) self._pip_tab.derived_changed.connect(self.derived_changed) root.addWidget(self._pip_tab, 1) self.refresh_derived() def refresh_derived(self): self._pip_tab.refresh_derived() def on_channel_enabled_changed(self, dev_id: str, ch_id: str, enabled: bool): self._pip_tab.on_channel_enabled_changed(dev_id, ch_id, enabled) def on_device_removed(self, dev_id: str): self._pip_tab.on_device_removed(dev_id) def on_device_reconfigured(self, dev_id: str): self._pip_tab.on_device_reconfigured(dev_id) def on_device_added(self): self._pip_tab.on_device_added() def on_channel_name_changed(self, dev_id: str, ch_id: str, ch_name: str): self._pip_tab.on_channel_name_changed(dev_id, ch_id, ch_name) def on_channel_unit_changed(self, dev_id: str, ch_id: str, unit: str): self._pip_tab.on_channel_unit_changed(dev_id, ch_id, unit) def closeEvent(self, e: QCloseEvent): self.closed.emit(); e.accept()