From 91b0b613f0b58498315ef91f6a541dd1089b88de Mon Sep 17 00:00:00 2001 From: Christian Kolset Date: Tue, 9 Jun 2026 23:18:43 -0600 Subject: Derived channels: signal display names as vars, auto-update template code, None op MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Variable names derived from signal display name (e.g. "Acceleration X" → acceleration_x) instead of raw channel ID; all auto-names lowercase with underscores - Template code auto-updates x[0]/x[1] refs to named vars as sources are added/changed; stops updating once user edits the code manually - Derivative wrt-channel mode uses separate template with x[1] substituted to channel var - Added "None" passthrough operation (result = x[0]) - source_names persisted in .labdaq profiles Co-Authored-By: Claude Sonnet 4.6 --- ui/code_templates.py | 41 +++++++----- ui/windows/channels_window.py | 145 +++++++++++++++++++++++++----------------- 2 files changed, 111 insertions(+), 75 deletions(-) (limited to 'ui') diff --git a/ui/code_templates.py b/ui/code_templates.py index 32332e9..ffa1662 100644 --- a/ui/code_templates.py +++ b/ui/code_templates.py @@ -23,17 +23,27 @@ Namespace available in all kinds: BUILTIN_TEMPLATES: dict = { + "none": """\ +result = x[0] +""", + "derivative": """\ -# dy/dt — backward finite difference prev_v = state.get("v", x[0]) prev_t = state.get("t", t) state["v"] = x[0] state["t"] = t result = (x[0] - prev_v) / (t - prev_t) if t != prev_t else 0.0 +""", + + "derivative_channel": """\ +prev_v = state.get("v", x[0]) +prev_x = state.get("x", x[1]) +state["v"] = x[0] +state["x"] = x[1] +result = (x[0] - prev_v) / (x[1] - prev_x) if x[1] != prev_x else 0.0 """, "second_derivative": """\ -# d²y/dt² — two-pass finite difference prev_v = state.get("v", x[0]) prev_t = state.get("t", t) prev_vel = state.get("vel", 0.0) @@ -44,33 +54,40 @@ result = (cur_vel - prev_vel) / (t - prev_t) if t != prev_t else 0.0 state["v"] = x[0] state["t"] = t state["vel"] = cur_vel +""", + + "second_derivative_channel": """\ +prev_v = state.get("v", x[0]) +prev_x = state.get("x", x[1]) +prev_vel = state.get("vel", 0.0) + +cur_vel = (x[0] - prev_v) / (x[1] - prev_x) if x[1] != prev_x else 0.0 +result = (cur_vel - prev_vel) / (x[1] - prev_x) if x[1] != prev_x else 0.0 + +state["v"] = x[0] +state["x"] = x[1] +state["vel"] = cur_vel """, "rms": """\ -# Rolling RMS over a sliding window import math as _math buf = state.setdefault("buf", []) buf.append(x[0]) -window = 20 # adjust window size (samples) here +window = 20 if len(buf) > window: buf.pop(0) result = _math.sqrt(sum(v * v for v in buf) / len(buf)) """, "power": """\ -# Electrical power P = V × I -# source 0 = voltage, source 1 = current result = x[0] * x[1] """, "difference": """\ -# Difference A − B -# source 0 = A, source 1 = B result = x[0] - x[1] """, "sum": """\ -# Sum of all sources result = sum(x) """, } @@ -82,17 +99,11 @@ SCRIPT_TEMPLATES: dict = { "expression": "x[0]", "function": """\ -# Function body — use 'return' to output the value. -# Named source vars, t, math, vars, state all available. offset = vars.get("offset", 0) return x[0] - offset """, "custom_script": """\ -# Assign computed value to 'result'. -# Named source vars, t, math, vars, state all available. - -# Example: running zero-offset zero = vars.get("zero", 0) result = x[0] - zero """, diff --git a/ui/windows/channels_window.py b/ui/windows/channels_window.py index 597b0b2..381800a 100644 --- a/ui/windows/channels_window.py +++ b/ui/windows/channels_window.py @@ -21,6 +21,7 @@ Classes: """ from __future__ import annotations +import re from typing import List, Optional, Set, Tuple from PyQt6.QtWidgets import ( @@ -47,6 +48,23 @@ _FILTER_KEYS = [k for k in FILTER_CLASSES if k not in ("derivative", "integral", _MATH_KEYS = ["derivative", "integral"] +def _channel_display_name(registry: DeviceRegistry, + processor: Optional[SignalProcessor], + dev_id: str, ch_id: str) -> str: + """Return human display name for a source; falls back to ch_id.""" + if dev_id == "derived" and processor: + for dc in processor.get_derived(): + if dc.channel_id == ch_id: + return dc.name + else: + for dev in registry.all_instances(): + if dev.info.device_id == dev_id: + for ch in dev.info.channels: + if ch.channel_id == ch_id: + return ch.name if ch.name else ch_id + return ch_id + + def _channel_combo(registry: DeviceRegistry, processor: Optional[SignalProcessor] = None, include_derived: bool = False, @@ -593,13 +611,15 @@ class PipelineTab(QWidget): # Derived editor for one DerivedChannel # ══════════════════════════════════════════════════════════════════════════════ -_BUILTIN = ["derivative", "second_derivative", "power", "rms", "difference", "sum"] +_BUILTIN = ["none", "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_DISPLAY["none"] = "None" _KIND_HELP = { + "none": "Pass source through unchanged", "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)", @@ -624,6 +644,14 @@ _KIND_HELP = { } +def _sub_xrefs(code: str, names: list) -> str: + """Replace x[0], x[1], ... with named source vars where index is in range.""" + def repl(m): + i = int(m.group(1)) + return names[i] if i < len(names) else m.group(0) + return re.sub(r'\bx\[(\d+)\]', repl, code) + + class DerivedBlock(QFrame): removed = pyqtSignal(object) applied = pyqtSignal(object) # DerivedChannel @@ -707,8 +735,8 @@ class DerivedBlock(QFrame): 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) + 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) @@ -754,15 +782,21 @@ class DerivedBlock(QFrame): self._code.setMinimumHeight(90); self._code.setMaximumHeight(180) self._code.setFont(_mono) self._code.setTabStopDistance(QFontMetrics(_mono).horizontalAdvance(" ") * 4) + from ui.code_templates import BUILTIN_TEMPLATES, SCRIPT_TEMPLATES if self.dc.kind in _BUILTIN: - from ui.code_templates import BUILTIN_TEMPLATES txt = BUILTIN_TEMPLATES.get(self.dc.kind, "") + self._template_base = txt + self._last_auto_code = txt elif self.dc.kind == "expression": - from ui.code_templates import SCRIPT_TEMPLATES - txt = self.dc.expression or SCRIPT_TEMPLATES.get("expression", "") + existing = self.dc.expression + txt = existing or SCRIPT_TEMPLATES.get("expression", "") + self._template_base = SCRIPT_TEMPLATES.get("expression", "") + self._last_auto_code = txt if not existing else None else: - from ui.code_templates import SCRIPT_TEMPLATES - txt = self.dc.script or SCRIPT_TEMPLATES.get(self.dc.kind, "") + existing = self.dc.script + txt = existing or SCRIPT_TEMPLATES.get(self.dc.kind, "") + self._template_base = SCRIPT_TEMPLATES.get(self.dc.kind, "") + self._last_auto_code = txt if not existing else None self._code.setPlainText(txt) code_lay.addWidget(self._code) # "Edit as Custom Script" button — only visible for built-in kinds @@ -818,13 +852,11 @@ class DerivedBlock(QFrame): self._update_vars_hint() def _update_vars_hint(self): - """Refresh the variable-name hint and code comment whenever sources change.""" + """Refresh the variable-name hint label whenever sources change.""" if not hasattr(self, "_vars_lbl"): return from core.signal_processor import _make_src_names - from ui.code_templates import BUILTIN_TEMPLATES - # Build source list — for derivatives in channel mode, include wrt channel sources = [cb.currentData() for cb in self._src_combos if cb.currentData()] is_deriv = self.dc.kind in ("derivative", "second_derivative") if (is_deriv and hasattr(self, "_wrt_cb") @@ -836,45 +868,30 @@ class DerivedBlock(QFrame): if not sources: self._vars_lbl.setText("No sources — add sources above to use named variables") - self._inject_code_comment("") - return - - names = _make_src_names(sources) - parts = [f"{name} = x[{i}]" for i, name in enumerate(names)] - comment = "# Variables: " + " · ".join(parts) - self._vars_lbl.setText(comment) - self._inject_code_comment(comment) - - def _inject_code_comment(self, comment: str): - """Insert or update a '# Variables:' first line in the code editor.""" - if not hasattr(self, "_code"): return - from ui.code_templates import BUILTIN_TEMPLATES - if self.dc.kind in _BUILTIN: - # Read-only template — prepend comment to base template each time - template = BUILTIN_TEMPLATES.get(self.dc.kind, "") - new_text = (comment + "\n" + template) if comment else template - if self._code.toPlainText() != new_text: - self._code.setPlainText(new_text) + display_names = [_channel_display_name(self.registry, self.processor, d, c) + for d, c in sources] + names = _make_src_names(sources, display_names) + parts = [f"{name} (x[{i}])" for i, name in enumerate(names)] + self._vars_lbl.setText("Variables: " + " · ".join(parts)) - elif self.dc.kind in ("function", "custom_script"): + # Update code editor if it still matches the last auto-generated text + if (getattr(self, "_last_auto_code", None) is not None + and hasattr(self, "_code") + and getattr(self, "_template_base", "")): current = self._code.toPlainText() - lines = current.splitlines() - if lines and lines[0].startswith("# Variables:"): - # Replace existing auto-comment - rest = "\n".join(lines[1:]).lstrip("\n") - new_text = (comment + "\n" + rest) if comment else rest - else: - new_text = (comment + "\n" + current) if comment else current - if self._code.toPlainText() != new_text: - # Preserve cursor position - cursor = self._code.textCursor() - pos = cursor.position() + len(comment) + 1 - self._code.setPlainText(new_text) - cursor.setPosition(min(pos, len(new_text))) - self._code.setTextCursor(cursor) - # expression kind: label-only, no code injection (single-line expr) + if current == self._last_auto_code: + new_code = _sub_xrefs(self._template_base, names) + self._last_auto_code = new_code + if current != new_code: + self._code.blockSignals(True) + cursor_pos = self._code.textCursor().position() + self._code.setPlainText(new_code) + c = self._code.textCursor() + c.setPosition(min(cursor_pos, len(new_code))) + self._code.setTextCursor(c) + self._code.blockSignals(False) def _on_kind(self, idx: int): from ui.code_templates import BUILTIN_TEMPLATES, SCRIPT_TEMPLATES @@ -883,17 +900,26 @@ class DerivedBlock(QFrame): self._help.setText(_KIND_HELP.get(k, "")) if k in ("derivative", "second_derivative") and not self._src_combos: self._add_src() - if k in _BUILTIN: - # Reset to bare template; _update_vars_hint will prepend variable comment - self._code.setPlainText(BUILTIN_TEMPLATES.get(k, "")) - elif k in _SCRIPT: - # Show starter template for script kinds (always replace on kind switch) - self._code.setPlainText(SCRIPT_TEMPLATES.get(k, "")) + if k in ("derivative", "second_derivative"): + wrt_ch = (hasattr(self, "_wrt_cb") and self._wrt_cb.currentIndex() == 1) + key = f"{k}_channel" if wrt_ch else k + base = BUILTIN_TEMPLATES.get(key, BUILTIN_TEMPLATES.get(k, "")) + elif k in _BUILTIN: + base = BUILTIN_TEMPLATES.get(k, "") + else: + base = SCRIPT_TEMPLATES.get(k, "") + self._template_base = base + self._last_auto_code = base + self._code.setPlainText(base) self._update_vars_hint() self._update_visibility() def _on_wrt_changed(self, idx: int): self._wrt_x_row.setVisible(idx == 1) + if self.dc.kind in ("derivative", "second_derivative"): + from ui.code_templates import BUILTIN_TEMPLATES + key = f"{self.dc.kind}_channel" if idx == 1 else self.dc.kind + self._template_base = BUILTIN_TEMPLATES.get(key, BUILTIN_TEMPLATES.get(self.dc.kind, "")) def _convert_to_custom_script(self): """Switch kind to custom_script, pre-fill with built-in template, make editable.""" @@ -906,9 +932,11 @@ class DerivedBlock(QFrame): self._kind_cb.blockSignals(False) self.dc.kind = "custom_script" self._help.setText(_KIND_HELP.get("custom_script", "")) - self.dc.script = template + self._template_base = template + self._last_auto_code = template self._code.setPlainText(template) - self._update_vars_hint() # prepend variable comment to template + self.dc.script = template + self._update_vars_hint() self._update_visibility() def _update_visibility(self): @@ -935,15 +963,12 @@ class DerivedBlock(QFrame): if x_src: sources.append(x_src) self.dc.sources = sources + self.dc.source_names = [_channel_display_name(self.registry, self.processor, d, c) + for d, c in sources] if self.dc.kind == "expression": self.dc.expression = self._code.toPlainText().strip() elif self.dc.kind in ("function", "custom_script"): - # Strip auto-generated Variables comment before saving - raw = self._code.toPlainText() - lines = raw.splitlines() - if lines and lines[0].startswith("# Variables:"): - raw = "\n".join(lines[1:]).lstrip("\n") - self.dc.script = raw + self.dc.script = self._code.toPlainText() err = self.processor.add_derived(self.dc) if err: self._st.setText(f"⚠ {err[:100]}") -- cgit v1.2.3