Scripting: named signal vars, shared vars dict, control scripts, code templates

- Named variables in expressions: source channel IDs injected as Python
  identifiers alongside x[] (e.g. write `A0 * 2` instead of `x[0] * 2`)
- Shared `vars` dict on SignalProcessor accessible from all expressions,
  functions, and control scripts; persisted in .labdaq profiles
- Per-derived-channel `state` dict for persistent computation state in
  custom scripts
- Control widgets gain optional on_action_script (Python snippet with
  value, vars, channels, math in scope); syntax-checked in editor dialog
- Built-in derived kinds (derivative, rms, power, etc.) now show their
  Python implementation as read-only code previews; "Edit as Custom Script"
  converts to custom_script kind with template pre-filled

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

1 files changed, 79 insertions, 11 deletions

diff --git a/core/signal_processor.py b/core/signal_processor.py
index df80dbb..8ac4d8c 100644
--- a/core/signal_processor.py
+++ b/core/signal_processor.py
@@ -26,6 +26,7 @@ strip chart can query history just like physical channels.
 from __future__ import annotations
 
 import math
+import re
 import threading
 import traceback
 from collections import deque
@@ -196,6 +197,31 @@ def unregister_filter_class(name: str) -> None:
 
 
 # ══════════════════════════════════════════════════════════════════════════════
+#  Helpers
+# ══════════════════════════════════════════════════════════════════════════════
+
+def _make_src_names(sources: List[Tuple[str, str]]) -> 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.
+    """
+    seen: Dict[str, str] = {}  # ch_id -> dev_id of first occurrence
+    names = []
+    for dev_id, ch_id in sources:
+        if 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)
+        if safe and safe[0].isdigit():
+            safe = "_" + safe
+        names.append(safe or "_ch")
+    return names
+
+
+# ══════════════════════════════════════════════════════════════════════════════
 #  Derived channel definitions
 # ══════════════════════════════════════════════════════════════════════════════
 
@@ -226,36 +252,62 @@ class DerivedChannel:
     expression:   str          = ""             # single-line: "x[0] * 2"
     script:       str          = ""             # multi-line function body
     enabled:      bool         = True
-    # Runtime: compiled callable (not serialised)
+    # 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)
 
     def compile(self) -> Optional[str]:
         """
         Compile expression/script into self._fn.
         Returns None on success, or error string on failure.
+
+        Namespace available in all expression/script kinds:
+          x        — list of source values in declaration order
+          <name>   — each source channel_id as a named variable (x[0] == first source name)
+          t        — elapsed time in seconds
+          math     — Python math module
+          vars     — shared SignalProcessor._script_vars dict (read/write)
+          state    — per-channel persistent dict (survives between evaluations)
         """
         try:
+            src_names = _make_src_names(self.sources)
+
             if self.kind == "expression":
-                # Single-line: inputs are x (list of latest values), t (time)
                 code = compile(f"__result__ = {self.expression}", "<expr>", "exec")
-                def _expr_fn(inputs, t, _code=code):
-                    ns = {"x": inputs, "t": t, "math": math}
+                _state = self._exec_state
+
+                def _expr_fn(inputs, t, sv, _code=code, _names=src_names, _st=_state):
+                    ns = {"x": inputs, "t": t, "math": math, "vars": sv, "state": _st}
+                    for i, name in enumerate(_names):
+                        if i < len(inputs):
+                            ns[name] = inputs[i]
                     exec(_code, ns)
                     return float(ns["__result__"])
+
                 self._fn = _expr_fn
 
             elif self.kind in ("function", "custom_script"):
-                # User provides a def compute(x, t): ... body
-                # We wrap it in a module namespace
                 src = self.script
                 if not src.strip().startswith("def compute"):
                     src = "def compute(x, t):\n" + "\n".join(
                         "    " + ln for ln in src.splitlines()
                     )
-                ns: dict = {"math": math}
-                exec(compile(src, "<script>", "exec"), ns)
-                fn = ns["compute"]
-                self._fn = lambda inputs, t, _f=fn: float(_f(inputs, t))
+                _exec_ns: dict = {"math": math}
+                exec(compile(src, "<script>", "exec"), _exec_ns)
+                _fn_ref = _exec_ns["compute"]
+                _state  = self._exec_state
+
+                def _script_fn(inputs, t, sv,
+                               _f=_fn_ref, _ns=_exec_ns, _st=_state, _names=src_names):
+                    # refresh mutable globals before each call so scripts see live values
+                    _ns["vars"]  = sv
+                    _ns["state"] = _st
+                    for i, name in enumerate(_names):
+                        if i < len(inputs):
+                            _ns[name] = inputs[i]
+                    return float(_f(inputs, t))
+
+                self._fn = _script_fn
 
             else:
                 # Built-in kinds handled in SignalProcessor._compute_derived
@@ -328,6 +380,9 @@ class SignalProcessor(QObject):
         # Ring buffers for derived channels (so strip chart can query history)
         self._derived_bufs: Dict[str, Tuple[deque, deque]] = {}
 
+        # Shared variable store — accessible as `vars` in all expressions/scripts
+        self._script_vars: Dict[str, Any] = {}
+
     # ── Pipeline management ───────────────────────────────────────────────
 
     def set_pipeline(self, pipeline: ChannelPipeline):
@@ -377,6 +432,17 @@ class SignalProcessor(QObject):
                 v_buf.clear()
             self._latest.clear()
 
+    # ── Shared script variables ───────────────────────────────────────────
+
+    def set_var(self, name: str, value: Any) -> None:
+        """Set a named variable accessible as `vars[name]` in all scripts."""
+        with self._lock:
+            self._script_vars[name] = value
+
+    def get_var(self, name: str, default: Any = 0.0) -> Any:
+        with self._lock:
+            return self._script_vars.get(name, default)
+
     # ── Main data path ────────────────────────────────────────────────────
 
     def on_raw_data(self, device_id: str, channel_id: str,
@@ -439,7 +505,9 @@ class SignalProcessor(QObject):
         if dc.kind in ("expression", "function", "custom_script"):
             if dc._fn is None:
                 return None
-            return dc._fn(inputs, t)
+            with self._lock:
+                sv = self._script_vars
+            return dc._fn(inputs, t, sv)
 
         elif dc.kind == "derivative":
             if not dc.sources: return None