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 --- core/profile.py | 14 ++-- core/signal_processor.py | 31 +++++---- ui/code_templates.py | 41 +++++++----- ui/windows/channels_window.py | 145 +++++++++++++++++++++++++----------------- 4 files changed, 139 insertions(+), 92 deletions(-) diff --git a/core/profile.py b/core/profile.py index 16c923f..d3c7e3a 100644 --- a/core/profile.py +++ b/core/profile.py @@ -197,12 +197,13 @@ class ProfileManager: "unit": dc.unit, "color": dc.color, "kind": dc.kind, - "sources": [list(s) for s in dc.sources], - "expression": dc.expression, - "script": dc.script, - "params": {k: v for k, v in dc.params.items() - if not k.startswith("_")}, - "enabled": dc.enabled, + "sources": [list(s) for s in dc.sources], + "source_names": list(dc.source_names), + "expression": dc.expression, + "script": dc.script, + "params": {k: v for k, v in dc.params.items() + if not k.startswith("_")}, + "enabled": dc.enabled, }) # Plot layout @@ -345,6 +346,7 @@ class ProfileManager: color=dd.get("color", "#f72585"), kind=dd.get("kind", "expression"), sources=sources, + source_names=dd.get("source_names", []), expression=dd.get("expression", ""), script=dd.get("script", ""), params=dd.get("params", {}), diff --git a/core/signal_processor.py b/core/signal_processor.py index e2752cd..bfb7ba3 100644 --- a/core/signal_processor.py +++ b/core/signal_processor.py @@ -200,24 +200,28 @@ def unregister_filter_class(name: str) -> None: # Helpers # ══════════════════════════════════════════════════════════════════════════════ -def _make_src_names(sources: List[Tuple[str, str]]) -> List[str]: +def _make_src_names(sources: List[Tuple[str, str]], + source_names: Optional[List[str]] = None) -> List[str]: """ - Build a list of Python-safe variable names from (device_id, channel_id) pairs. - Uses channel_id alone when unambiguous; prefixes device_id when two sources - share the same channel_id from different devices. + Build Python-safe variable names for each source. + Priority: user-defined name > channel_id (with device prefix if ambiguous). + User names are lowercased and spaces replaced with underscores. """ - seen: Dict[str, str] = {} # ch_id -> dev_id of first occurrence + seen: Dict[str, str] = {} names = [] - for dev_id, ch_id in sources: - if ch_id in seen and seen[ch_id] != dev_id: + for i, (dev_id, ch_id) in enumerate(sources): + user = source_names[i].strip() if (source_names and i < len(source_names)) else "" + if user: + raw = user.lower().replace(" ", "_") + elif ch_id in seen and seen[ch_id] != dev_id: raw = f"{dev_id}_{ch_id}" else: - seen[ch_id] = dev_id raw = ch_id - safe = re.sub(r"\W", "_", raw) + seen[ch_id] = dev_id + safe = re.sub(r"\W", "_", raw.lower()) if safe and safe[0].isdigit(): safe = "_" + safe - names.append(safe or "_ch") + names.append(safe or f"_ch{i}") return names @@ -252,6 +256,8 @@ class DerivedChannel: expression: str = "" # single-line: "x[0] * 2" script: str = "" # multi-line function body enabled: bool = True + # User-defined names for each source (e.g. ["force", "moment"]) + source_names: List[str] = field(default_factory=list) # Runtime: compiled callable and per-channel script state (not serialised) _fn: Optional[Callable] = field(default=None, repr=False, compare=False) _exec_state: dict = field(default_factory=dict, repr=False, compare=False) @@ -270,7 +276,7 @@ class DerivedChannel: state — per-channel persistent dict (survives between evaluations) """ try: - src_names = _make_src_names(self.sources) + src_names = _make_src_names(self.sources, self.source_names) if self.kind == "expression": code = compile(f"__result__ = {self.expression}", "", "exec") @@ -585,4 +591,7 @@ class SignalProcessor(QObject): elif dc.kind == "sum": return sum(inputs) + elif dc.kind == "none": + return inputs[0] if inputs else None + return None 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