summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorChristian Kolset <christian.kolset@gmail.com>2026-06-09 23:18:43 -0600
committerChristian Kolset <christian.kolset@gmail.com>2026-06-09 23:18:43 -0600
commit91b0b613f0b58498315ef91f6a541dd1089b88de (patch)
tree65b378d6ad2706ac1e29a7e7888c78770d4c945a
parentd396692de571445537e617ac31627d7bd7ab84aa (diff)
Derived channels: signal display names as vars, auto-update template code, None op
- 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 <noreply@anthropic.com>
-rw-r--r--core/profile.py14
-rw-r--r--core/signal_processor.py31
-rw-r--r--ui/code_templates.py41
-rw-r--r--ui/windows/channels_window.py145
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}", "<expr>", "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,8 +23,11 @@ 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]
@@ -32,8 +35,15 @@ 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)
@@ -46,31 +56,38 @@ 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]}")