From 7aec470acd69e9c1810b40aa5eea9b96b9eb73ad Mon Sep 17 00:00:00 2001 From: Christian Kolset Date: Tue, 9 Jun 2026 22:14:20 -0600 Subject: 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 --- core/profile.py | 12 ++++++ core/signal_processor.py | 90 +++++++++++++++++++++++++++++++++++++------ ui/code_templates.py | 70 +++++++++++++++++++++++++++++++++ ui/control_editor.py | 58 +++++++++++++++++++++++++++- ui/control_panel.py | 53 ++++++++++++++++++++++--- ui/main_window.py | 2 +- ui/windows/channels_window.py | 63 ++++++++++++++++++++++++++---- 7 files changed, 320 insertions(+), 28 deletions(-) create mode 100644 ui/code_templates.py diff --git a/core/profile.py b/core/profile.py index 39cb3ca..16c923f 100644 --- a/core/profile.py +++ b/core/profile.py @@ -80,6 +80,8 @@ class Profile: # plugin_state = {plugin_id: plugin.get_save_state()} plugins_enabled: List[str] = field(default_factory=list) # plugins_enabled = [plugin_id, ...] — which plugins were on when saved + script_vars: Dict[str, Any] = field(default_factory=dict) + # script_vars — shared vars dict accessible as `vars` in all expressions def to_json(self) -> str: return json.dumps(asdict(self), indent=2) @@ -99,6 +101,7 @@ class Profile: settings = d.get("settings", {}), plugin_state = d.get("plugin_state", {}), plugins_enabled = d.get("plugins_enabled", []), + script_vars = d.get("script_vars", {}), ) def save(self, path: str): @@ -213,6 +216,10 @@ class ProfileManager: p.settings = dict(settings) + # Shared script variables + with processor._lock: + p.script_vars = dict(processor._script_vars) + # Plugin state + enabled list if plugin_manager is not None: p.plugins_enabled = plugin_manager.get_enabled_ids() @@ -349,6 +356,11 @@ class ProfileManager: specs = [ControlSpec.from_dict(d) for d in profile.controls] control_panel.load_specs(specs) + # ── Script vars ────────────────────────────────────────────────── + if profile.script_vars: + with processor._lock: + processor._script_vars.update(profile.script_vars) + # ── Settings ───────────────────────────────────────────────────── settings_ref.update(profile.settings) 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 @@ -195,6 +196,31 @@ def unregister_filter_class(name: str) -> None: FILTER_CLASSES.pop(name, 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 + — 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}", "", "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, "