diff options
Diffstat (limited to 'core/signal_processor.py')
| -rw-r--r-- | core/signal_processor.py | 90 |
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 |