diff options
| -rw-r--r-- | core/profile.py | 12 | ||||
| -rw-r--r-- | core/signal_processor.py | 90 | ||||
| -rw-r--r-- | ui/code_templates.py | 70 | ||||
| -rw-r--r-- | ui/control_editor.py | 58 | ||||
| -rw-r--r-- | ui/control_panel.py | 53 | ||||
| -rw-r--r-- | ui/main_window.py | 2 | ||||
| -rw-r--r-- | ui/windows/channels_window.py | 63 |
7 files changed, 320 insertions, 28 deletions
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 @@ -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 diff --git a/ui/code_templates.py b/ui/code_templates.py new file mode 100644 index 0000000..0863e62 --- /dev/null +++ b/ui/code_templates.py @@ -0,0 +1,70 @@ +""" +ui/code_templates.py + +Python code templates for built-in derived channel kinds. +Shown as read-only previews in the UI; clicking "Edit as Custom Script" +converts the channel to custom_script kind with the template pre-filled. + +Templates use the enhanced expression namespace: + x — list of source values in order + <name> — each source's channel_id as a named variable + t — elapsed time (seconds) + math — Python math module + vars — shared SignalProcessor._script_vars dict (read/write) + state — per-channel persistent dict (survives between evaluations) + result — assign the computed value here (expression-style exec) +""" + +BUILTIN_TEMPLATES: dict = { + + "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 +""", + + "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) + +cur_vel = (x[0] - prev_v) / (t - prev_t) if t != prev_t else 0.0 +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 +""", + + "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 +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) +""", +} diff --git a/ui/control_editor.py b/ui/control_editor.py index c582e6b..48de21d 100644 --- a/ui/control_editor.py +++ b/ui/control_editor.py @@ -18,8 +18,10 @@ from PyQt6.QtWidgets import ( QLabel, QLineEdit, QComboBox, QDoubleSpinBox, QSpinBox, QPushButton, QGroupBox, QWidget, QStackedWidget, QFrame, QCheckBox, QSizePolicy, + QTextEdit, QScrollArea, ) from PyQt6.QtCore import Qt +from PyQt6.QtGui import QFont, QFontMetrics class _DynStack(QStackedWidget): @@ -68,6 +70,8 @@ class ControlSpec: max_val: float = 100.0 step: float = 1.0 max_rpm: int = 3000 + # Optional Python snippet executed on every value change + on_action_script: str = "" def to_dict(self) -> dict: return asdict(self) @@ -177,8 +181,8 @@ class ControlEditorDialog(QDialog): self.result_spec: Optional[ControlSpec] = None self.setWindowTitle("Edit Control" if spec else "Add Control") - self.setMinimumSize(440, 460) - self.resize(460, 500) + self.setMinimumSize(440, 520) + self.resize(460, 580) self._build() self._load_spec() @@ -244,6 +248,40 @@ class ControlEditorDialog(QDialog): self._stack.addWidget(panel) params_lay.addWidget(self._stack) root.addWidget(self._params_grp) + + # ── On-action script ─────────────────────────────────────────── + script_grp = QGroupBox("On Action Script (optional)") + script_lay = QVBoxLayout(script_grp); script_lay.setContentsMargins(10, 12, 10, 10) + + _help = QLabel( + "Runs every time this control changes value.\n" + "Available: value, channel_id, device_id, vars (shared dict), " + "channels (latest values by channel_id), math" + ) + _help.setObjectName("traceSource"); _help.setWordWrap(True) + script_lay.addWidget(_help) + + _mono = QFont("IBM Plex Mono, Consolas, Monospace") + _mono.setStyleHint(QFont.StyleHint.Monospace) + self._script_edit = QTextEdit() + self._script_edit.setFont(_mono) + self._script_edit.setTabStopDistance(QFontMetrics(_mono).horizontalAdvance(" ") * 4) + self._script_edit.setMinimumHeight(80); self._script_edit.setMaximumHeight(160) + self._script_edit.setPlaceholderText( + "# Example: zero a position offset when button is pressed\n" + "# vars['zero_pos'] = channels.get('encoder_0', 0)" + ) + script_lay.addWidget(self._script_edit) + + self._script_status = QLabel("") + self._script_status.setObjectName("traceSource") + check_btn = QPushButton("Check Syntax") + check_btn.clicked.connect(self._check_script) + btn_row = QHBoxLayout() + btn_row.addWidget(self._script_status, 1) + btn_row.addWidget(check_btn) + script_lay.addLayout(btn_row) + root.addWidget(script_grp) root.addStretch() # ── Buttons ──────────────────────────────────────────────────── @@ -258,6 +296,18 @@ class ControlEditorDialog(QDialog): btn_row.addWidget(cancel); btn_row.addWidget(ok) root.addLayout(btn_row) + def _check_script(self): + src = self._script_edit.toPlainText().strip() + if not src: + self._script_status.setText(""); return + try: + compile(src, "<script>", "exec") + self._script_status.setText("✓ OK") + self._script_status.setStyleSheet("color:#22c55e;") + except SyntaxError as e: + self._script_status.setText(f"⚠ {e}") + self._script_status.setStyleSheet("color:#ef4444;") + def _load_spec(self): """Populate fields from self.spec.""" # Type @@ -287,6 +337,9 @@ class ControlEditorDialog(QDialog): for name, panel in self._param_panels.items(): panel.load(self.spec) + # Script + self._script_edit.setPlainText(self.spec.on_action_script) + def _on_type_changed(self, type_name: str): idx = list(_PARAM_PANELS.keys()).index(type_name) self._stack.setCurrentIndex(idx) @@ -340,6 +393,7 @@ class ControlEditorDialog(QDialog): device_id=self._dev_cb.currentData() or "", channel_id=channel_id, active_low=self._active_low_chk.isChecked(), + on_action_script=self._script_edit.toPlainText().strip(), ) # Save type-specific params panel = self._param_panels.get(ctype) diff --git a/ui/control_panel.py b/ui/control_panel.py index 7a3e416..6e78e89 100644 --- a/ui/control_panel.py +++ b/ui/control_panel.py @@ -20,6 +20,9 @@ an instance into ControlPanel.add_widget(). from __future__ import annotations from typing import Optional +import math as _math +from typing import Any + from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QSlider, QDoubleSpinBox, QComboBox, @@ -44,7 +47,9 @@ class ControlWidget(QFrame): def __init__(self, title: str, icon: str = "⚙", device_id: str = "", channel_id: str = "", registry: Optional[DeviceRegistry] = None, - active_low: bool = False): + active_low: bool = False, + processor=None, + on_action_script: str = ""): super().__init__() self.title = title self.icon = icon @@ -52,9 +57,20 @@ class ControlWidget(QFrame): self.channel_id = channel_id self.registry = registry self.active_low = active_low + self._processor = processor + self._on_action_fn = None + if on_action_script.strip(): + self._compile_action(on_action_script) self.setObjectName("controlWidget") self._build_frame() + def _compile_action(self, script: str): + try: + code = compile(script, "<control_script>", "exec") + self._on_action_fn = code + except SyntaxError as e: + print(f"[Control '{self.title}'] script syntax error: {e}") + def _logic_level(self, enabled: bool) -> float: """Map logical ON/OFF to electrical level, respecting active-low outputs.""" if self.active_low: @@ -112,6 +128,27 @@ class ControlWidget(QFrame): f"check device type and channel ID") self.value_changed.emit(self.channel_id, value) + if self._on_action_fn is not None: + try: + sv: dict = {} + channels: dict = {} + if self._processor is not None: + with self._processor._lock: + sv = self._processor._script_vars + channels = {ch_id: v + for (_, ch_id), (_, v) in self._processor._latest.items()} + ns: dict = { + "value": value, + "channel_id": self.channel_id, + "device_id": self.device_id, + "vars": sv, + "channels": channels, + "math": _math, + } + exec(self._on_action_fn, ns) + except Exception as e: + print(f"[Control '{self.title}'] script error: {e}") + # ══════════════════════════════════════════════════════════════════════════════ # On/Off Switch @@ -480,9 +517,10 @@ class ControlPanel(QWidget): controls_changed = pyqtSignal() # emitted whenever widgets are added/edited/removed - def __init__(self, registry: DeviceRegistry): + def __init__(self, registry: DeviceRegistry, processor=None): super().__init__() - self.registry = registry + self.registry = registry + self.processor = processor self._widgets: list[ControlWidget] = [] self._specs: list = [] # parallel list of ControlSpec self._build() @@ -560,7 +598,7 @@ class ControlPanel(QWidget): def _add_widget_from_spec(self, spec): """Instantiate a ControlWidget from a ControlSpec and add to panel.""" from ui.control_editor import ControlSpec as CS - widget = _build_widget_from_spec(spec, self.registry) + widget = _build_widget_from_spec(spec, self.registry, self.processor) if widget is None: return wrapper = self._make_wrapper(widget, spec) @@ -641,7 +679,7 @@ class ControlPanel(QWidget): self._widgets.pop(idx); self._specs.pop(idx) # Insert new one at same position - new_widget = _build_widget_from_spec(new_spec, self.registry) + new_widget = _build_widget_from_spec(new_spec, self.registry, self.processor) if new_widget is None: return new_wrapper = self._make_wrapper(new_widget, new_spec) @@ -674,7 +712,8 @@ class ControlPanel(QWidget): # ── Factory — build a ControlWidget from a ControlSpec ───────────────────────── -def _build_widget_from_spec(spec, registry: DeviceRegistry) -> Optional[ControlWidget]: +def _build_widget_from_spec(spec, registry: DeviceRegistry, + processor=None) -> Optional[ControlWidget]: """Instantiate the right ControlWidget subclass from a ControlSpec.""" # Base kwargs — icon is NOT included here; it's passed explicitly below # so subclasses never get a double-value collision. @@ -683,6 +722,8 @@ def _build_widget_from_spec(spec, registry: DeviceRegistry) -> Optional[ControlW channel_id=spec.channel_id, registry=registry, active_low=getattr(spec, "active_low", False), + processor=processor, + on_action_script=getattr(spec, "on_action_script", ""), ) t = spec.control_type ttl = spec.title diff --git a/ui/main_window.py b/ui/main_window.py index ebc7463..d0e0ac4 100644 --- a/ui/main_window.py +++ b/ui/main_window.py @@ -178,7 +178,7 @@ class MainWindow(QMainWindow): hsplit = QSplitter(Qt.Orientation.Horizontal); hsplit.setHandleWidth(3) - self._ctrl = ControlPanel(self.registry) + self._ctrl = ControlPanel(self.registry, processor=self.processor) self._ctrl.setMinimumWidth(200); self._ctrl.setMaximumWidth(340) self._ctrl._add_demo_widgets() hsplit.addWidget(self._ctrl) diff --git a/ui/windows/channels_window.py b/ui/windows/channels_window.py index f0ea34e..ecc7cbc 100644 --- a/ui/windows/channels_window.py +++ b/ui/windows/channels_window.py @@ -606,9 +606,21 @@ _KIND_HELP = { "rms": "Rolling RMS from 1 source (set window below)", "difference": "source[0] − source[1] from 2 sources", "sum": "Σ all source values", - "expression": "x = list of source values, t = elapsed time (s)\nExample: x[0] * 3.14159 / 180", - "function": "def compute(x, t):\n # x = source values list\n return x[0] * 2.0", - "custom_script":"Full Python. Must define: def compute(x, t): ...", + "expression": ( + "Single-line Python expression.\n" + "Available: x[0]… or use signal names directly (e.g. A0, encoder_0)\n" + " t = elapsed time (s), math, vars, state\n" + "Example: A0 * 3.14159 / 180" + ), + "function": ( + "Multi-line function body — auto-wrapped as def compute(x, t):\n" + "Available globals: x, signal names, t, math, vars, state\n" + "Example:\n offset = vars.get('zero', 0)\n return x[0] - offset" + ), + "custom_script": ( + "Full Python. Must define: def compute(x, t): ...\n" + "Available globals: x, signal names, t, math, vars, state" + ), } @@ -728,17 +740,28 @@ class DerivedBlock(QFrame): wrt_lay.addWidget(self._wrt_x_row) bl.addWidget(self._wrt_grp) - # Code editor + # Code editor (shown for all kinds — read-only for built-ins) self._code_grp = QGroupBox("Code") code_lay = QVBoxLayout(self._code_grp); code_lay.setContentsMargins(6, 4, 6, 4) + _mono = QFont("IBM Plex Mono, Consolas, Monospace"); _mono.setStyleHint(QFont.StyleHint.Monospace) self._code = QTextEdit(); self._code.setObjectName("codeEditor") self._code.setMinimumHeight(90); self._code.setMaximumHeight(180) - _mono = QFont("IBM Plex Mono, Consolas, Monospace"); _mono.setStyleHint(QFont.StyleHint.Monospace) self._code.setFont(_mono) self._code.setTabStopDistance(QFontMetrics(_mono).horizontalAdvance(" ") * 4) - txt = self.dc.expression if self.dc.kind == "expression" else self.dc.script + if self.dc.kind in _BUILTIN: + from ui.code_templates import BUILTIN_TEMPLATES + txt = BUILTIN_TEMPLATES.get(self.dc.kind, "") + elif self.dc.kind == "expression": + txt = self.dc.expression + else: + txt = self.dc.script self._code.setPlainText(txt) code_lay.addWidget(self._code) + # "Edit as Custom Script" button — only visible for built-in kinds + self._edit_as_script_btn = QPushButton("✎ Edit as Custom Script") + self._edit_as_script_btn.setObjectName("addTraceBtn") + self._edit_as_script_btn.clicked.connect(self._convert_to_custom_script) + code_lay.addWidget(self._edit_as_script_btn) bl.addWidget(self._code_grp) # Status + apply @@ -783,19 +806,43 @@ class DerivedBlock(QFrame): if it.widget(): it.widget().deleteLater() def _on_kind(self, idx: int): + from ui.code_templates import BUILTIN_TEMPLATES k = _ALL_KINDS[idx] self.dc.kind = k 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: + # Show the template as a read-only preview + self._code.setPlainText(BUILTIN_TEMPLATES.get(k, "")) self._update_visibility() def _on_wrt_changed(self, idx: int): self._wrt_x_row.setVisible(idx == 1) + def _convert_to_custom_script(self): + """Switch kind to custom_script, pre-fill with built-in template, make editable.""" + from ui.code_templates import BUILTIN_TEMPLATES + template = BUILTIN_TEMPLATES.get(self.dc.kind, "") + # Switch kind combo to custom_script + idx = _ALL_KINDS.index("custom_script") + self._kind_cb.blockSignals(True) + self._kind_cb.setCurrentIndex(idx) + self._kind_cb.blockSignals(False) + self.dc.kind = "custom_script" + self._help.setText(_KIND_HELP.get("custom_script", "")) + self.dc.script = template + self._code.setPlainText(template) + self._update_visibility() + def _update_visibility(self): - is_deriv = self.dc.kind in ("derivative", "second_derivative") - self._code_grp.setVisible(self.dc.kind in _SCRIPT) + is_deriv = self.dc.kind in ("derivative", "second_derivative") + is_builtin = self.dc.kind in _BUILTIN + is_script = self.dc.kind in _SCRIPT + # Code group shown for all kinds; read-only for built-ins + self._code_grp.setVisible(True) + self._code.setReadOnly(is_builtin) + self._edit_as_script_btn.setVisible(is_builtin) self._win_widget.setVisible(self.dc.kind == "rms") self._wrt_grp.setVisible(is_deriv) self._add_src_btn.setVisible(not is_deriv) |