V8

5 files changed, 489 insertions, 58 deletions

diff --git a/core/__pycache__/__init__.cpython-314.pyc b/core/__pycache__/__init__.cpython-314.pyc
index 2a7cb7e..e90d2a9 100644
--- a/core/__pycache__/__init__.cpython-314.pyc
+++ b/core/__pycache__/__init__.cpython-314.pyc
diff --git a/core/__pycache__/acquisition.cpython-314.pyc b/core/__pycache__/acquisition.cpython-314.pyc
index d84b66b..c41e1a5 100644
--- a/core/__pycache__/acquisition.cpython-314.pyc
+++ b/core/__pycache__/acquisition.cpython-314.pyc
diff --git a/core/__pycache__/signal_processor.cpython-314.pyc b/core/__pycache__/signal_processor.cpython-314.pyc
new file mode 100644
index 0000000..dc61203
--- /dev/null
+++ b/core/__pycache__/signal_processor.cpython-314.pyc
diff --git a/core/acquisition.py b/core/acquisition.py
index 640e6d5..afccb56 100644
--- a/core/acquisition.py
+++ b/core/acquisition.py
@@ -3,7 +3,7 @@ core/acquisition.py
 
 Background acquisition engine.
 Polls all connected devices, buffers data, fires Qt signals,
-writes CSV logs, and checks alarm thresholds.
+and writes CSV logs.
 """
 
 import csv
@@ -18,7 +18,7 @@ from PyQt6.QtCore import QObject, pyqtSignal
 
 from devices.base_device import BaseDevice, DeviceStatus
 
-MAX_BUFFER = 20_000   # samples per channel
+MAX_BUFFER = 20_000
 
 
 class ChannelBuffer:
@@ -33,7 +33,6 @@ class ChannelBuffer:
         self.values.append(v)
 
     def window(self, seconds: float) -> Tuple[List[float], List[float]]:
-        """Return the last `seconds` worth of data."""
         if not self.times:
             return [], []
         cutoff = self.times[-1] - seconds
@@ -63,35 +62,29 @@ class AcquisitionEngine(QObject):
     Signals
     -------
     new_data(device_id, channel_id, timestamp, value)
-    alarm_triggered(device_id, channel_id, value, kind)   kind: "low"|"high"
     device_status_changed(device_id, status_str)
     log_started(filepath)
     log_stopped(filepath)
     """
 
-    new_data               = pyqtSignal(str, str, float, float)
-    alarm_triggered        = pyqtSignal(str, str, float, str)
-    device_status_changed  = pyqtSignal(str, str)
-    log_started            = pyqtSignal(str)
-    log_stopped            = pyqtSignal(str)
+    new_data              = pyqtSignal(str, str, float, float)
+    device_status_changed = pyqtSignal(str, str)
+    log_started           = pyqtSignal(str)
+    log_stopped           = pyqtSignal(str)
 
     def __init__(self, poll_interval_ms: int = 100):
         super().__init__()
         self._interval = poll_interval_ms / 1000.0
-        self._devices:  List[BaseDevice]                       = []
-        self._buffers:  Dict[str, Dict[str, ChannelBuffer]]    = {}
-        self._running   = False
-        self._thread:   Optional[threading.Thread]             = None
-        self._t0        = 0.0
-
-        # Logging
-        self._logging     = False
-        self._log_path    = ""
-        self._csv_file    = None
-        self._csv_writer  = None
-
-        # Alarm dedup
-        self._alarm_state: Dict[str, bool] = {}
+        self._devices: List[BaseDevice]                    = []
+        self._buffers: Dict[str, Dict[str, ChannelBuffer]] = {}
+        self._running  = False
+        self._thread:  Optional[threading.Thread]          = None
+        self._t0       = 0.0
+
+        self._logging    = False
+        self._log_path   = ""
+        self._csv_file   = None
+        self._csv_writer = None
 
     # ── Device management ────────────────────────────────────────────────
 
@@ -128,10 +121,11 @@ class AcquisitionEngine(QObject):
     # ── Logging ──────────────────────────────────────────────────────────
 
     def start_logging(self, filepath: str = "") -> str:
-        if not filepath:
-            os.makedirs("logs", exist_ok=True)
-            ts       = datetime.now().strftime("%Y%m%d_%H%M%S")
-            filepath = f"logs/daq_{ts}.csv"
+        if not filepath or filepath.endswith(os.sep) or filepath.endswith("/"):
+            dir_ = filepath if filepath else "logs"
+            os.makedirs(dir_, exist_ok=True)
+            ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+            filepath = os.path.join(dir_, f"daq_{ts}.csv")
         self._log_path  = filepath
         self._csv_file  = open(filepath, "w", newline="")
         headers = ["elapsed_s"]
@@ -163,9 +157,9 @@ class AcquisitionEngine(QObject):
 
     def _loop(self):
         while self._running:
-            t_start   = time.time()
-            elapsed   = t_start - self._t0
-            log_row   = [f"{elapsed:.4f}"]
+            t_start = time.time()
+            elapsed = t_start - self._t0
+            log_row = [f"{elapsed:.4f}"]
 
             for dev in list(self._devices):
                 active = dev.status in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED)
@@ -180,15 +174,11 @@ class AcquisitionEngine(QObject):
                     if val is None:
                         log_row.append("")
                         continue
-                    # Buffer
                     buf = self._buffers.get(dev.info.device_id, {}).get(ch.channel_id)
                     if buf is not None:
                         buf.append(elapsed, float(val))
-                    # Signal
                     self.new_data.emit(dev.info.device_id, ch.channel_id, elapsed, float(val))
                     log_row.append(f"{val:.5f}")
-                    # Alarms
-                    self._check_alarm(dev.info.device_id, ch, float(val))
 
             if self._logging and self._csv_writer:
                 try:
@@ -196,30 +186,7 @@ class AcquisitionEngine(QObject):
                 except Exception:
                     pass
 
-            # Sleep remainder of interval
-            dt = time.time() - t_start
+            dt    = time.time() - t_start
             sleep = self._interval - dt
             if sleep > 0:
                 time.sleep(sleep)
-
-    # ── Alarm logic ───────────────────────────────────────────────────────
-
-    def _check_alarm(self, device_id: str, ch, val: float):
-        lo_key = f"{device_id}.{ch.channel_id}.lo"
-        hi_key = f"{device_id}.{ch.channel_id}.hi"
-
-        if ch.alarm_low is not None:
-            if val < ch.alarm_low:
-                if not self._alarm_state.get(lo_key):
-                    self._alarm_state[lo_key] = True
-                    self.alarm_triggered.emit(device_id, ch.channel_id, val, "low")
-            else:
-                self._alarm_state[lo_key] = False
-
-        if ch.alarm_high is not None:
-            if val > ch.alarm_high:
-                if not self._alarm_state.get(hi_key):
-                    self._alarm_state[hi_key] = True
-                    self.alarm_triggered.emit(device_id, ch.channel_id, val, "high")
-            else:
-                self._alarm_state[hi_key] = False
diff --git a/core/signal_processor.py b/core/signal_processor.py
new file mode 100644
index 0000000..4161d4d
--- /dev/null
+++ b/core/signal_processor.py
@@ -0,0 +1,464 @@
+"""
+core/signal_processor.py
+
+Signal processing pipeline engine.
+
+Provides two capabilities:
+
+1. FILTERS — applied to a raw channel buffer before plotting:
+     LowPass, HighPass, MovingAverage, Median, Derivative, Integral, Scale+Offset
+
+2. DERIVED CHANNELS — virtual channels computed from one or more physical
+   channels. Each derived channel runs a user-defined function every time
+   new data arrives. Built-ins: velocity/acceleration from displacement,
+   power from voltage+current, RMS, etc.  Custom: arbitrary Python snippet.
+
+Architecture
+------------
+   SignalProcessor sits between AcquisitionEngine and StripChartWidget.
+   engine.new_data  →  SignalProcessor.process(dev, ch, t, val)
+                    →  emits processed_data(virtual_or_real_id, ch_id, t, val)
+
+The processor maintains its own ring buffers for derived channels so the
+strip chart can query history just like physical channels.
+"""
+
+from __future__ import annotations
+
+import math
+import threading
+import traceback
+from collections import deque
+from dataclasses import dataclass, field
+from typing import Callable, Dict, List, Optional, Tuple, Any
+
+from PyQt6.QtCore import QObject, pyqtSignal
+
+# ── Constants ──────────────────────────────────────────────────────────────
+
+MAX_BUF = 20_000
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+#  Filter definitions
+# ══════════════════════════════════════════════════════════════════════════════
+
+class FilterBase:
+    """All filters implement __call__(value: float) -> float."""
+    name:    str = "identity"
+    params:  dict = {}
+
+    def __call__(self, value: float) -> float:
+        return value
+
+    def reset(self): pass
+
+    def to_dict(self) -> dict:
+        return {"type": self.name, **self.params}
+
+
+class MovingAverageFilter(FilterBase):
+    name = "moving_average"
+    def __init__(self, window: int = 10):
+        self.params = {"window": window}
+        self._buf   = deque(maxlen=window)
+
+    def __call__(self, v: float) -> float:
+        self._buf.append(v)
+        return sum(self._buf) / len(self._buf)
+
+    def reset(self): self._buf.clear()
+
+
+class MedianFilter(FilterBase):
+    name = "median"
+    def __init__(self, window: int = 5):
+        self.params = {"window": window}
+        self._buf   = deque(maxlen=window)
+
+    def __call__(self, v: float) -> float:
+        self._buf.append(v)
+        s = sorted(self._buf)
+        n = len(s)
+        return s[n // 2] if n % 2 else (s[n//2 - 1] + s[n//2]) / 2
+
+    def reset(self): self._buf.clear()
+
+
+class LowPassFilter(FilterBase):
+    """Exponential moving average (single-pole IIR low-pass)."""
+    name = "low_pass"
+    def __init__(self, alpha: float = 0.1):
+        """alpha=0.0 → no change, 1.0 → unfiltered."""
+        self.params = {"alpha": alpha}
+        self._prev  = None
+
+    def __call__(self, v: float) -> float:
+        if self._prev is None:
+            self._prev = v
+        self._prev = self._prev + self.params["alpha"] * (v - self._prev)
+        return self._prev
+
+    def reset(self): self._prev = None
+
+
+class HighPassFilter(FilterBase):
+    """Simple single-pole IIR high-pass (compliment of low-pass)."""
+    name = "high_pass"
+    def __init__(self, alpha: float = 0.9):
+        self.params  = {"alpha": alpha}
+        self._prev_v = None
+        self._prev_y = 0.0
+
+    def __call__(self, v: float) -> float:
+        if self._prev_v is None:
+            self._prev_v = v
+        y = self.params["alpha"] * (self._prev_y + v - self._prev_v)
+        self._prev_y = y
+        self._prev_v = v
+        return y
+
+    def reset(self): self._prev_v = None; self._prev_y = 0.0
+
+
+class ScaleOffsetFilter(FilterBase):
+    """y = scale * x + offset  (unit conversion, calibration)."""
+    name = "scale_offset"
+    def __init__(self, scale: float = 1.0, offset: float = 0.0):
+        self.params = {"scale": scale, "offset": offset}
+
+    def __call__(self, v: float) -> float:
+        return self.params["scale"] * v + self.params["offset"]
+
+
+class DerivativeFilter(FilterBase):
+    """Numerical first derivative  dy/dt."""
+    name = "derivative"
+    def __init__(self): self.params = {}; self._prev_v = None; self._prev_t = None
+
+    def process_with_t(self, v: float, t: float) -> float:
+        if self._prev_t is None or t == self._prev_t:
+            self._prev_v = v; self._prev_t = t; return 0.0
+        dy = (v - self._prev_v) / (t - self._prev_t)
+        self._prev_v = v; self._prev_t = t
+        return dy
+
+    def __call__(self, v: float) -> float:
+        return 0.0   # use process_with_t for real output
+
+    def reset(self): self._prev_v = None; self._prev_t = None
+
+
+class IntegralFilter(FilterBase):
+    """Numerical integration (trapezoidal rule)."""
+    name = "integral"
+    def __init__(self): self.params = {}; self._sum = 0.0; self._prev_v = None; self._prev_t = None
+
+    def process_with_t(self, v: float, t: float) -> float:
+        if self._prev_t is not None and t != self._prev_t:
+            self._sum += 0.5 * (v + self._prev_v) * (t - self._prev_t)
+        self._prev_v = v; self._prev_t = t
+        return self._sum
+
+    def __call__(self, v: float) -> float:
+        return self._sum
+
+    def reset(self): self._sum = 0.0; self._prev_v = None; self._prev_t = None
+
+
+FILTER_CLASSES = {
+    "moving_average": MovingAverageFilter,
+    "median":         MedianFilter,
+    "low_pass":       LowPassFilter,
+    "high_pass":      HighPassFilter,
+    "scale_offset":   ScaleOffsetFilter,
+    "derivative":     DerivativeFilter,
+    "integral":       IntegralFilter,
+}
+
+
+def filter_from_dict(d: dict) -> FilterBase:
+    cls = FILTER_CLASSES.get(d.get("type", ""))
+    if cls is None:
+        return FilterBase()
+    params = {k: v for k, v in d.items() if k != "type"}
+    return cls(**params)
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+#  Derived channel definitions
+# ══════════════════════════════════════════════════════════════════════════════
+
+@dataclass
+class DerivedChannel:
+    """
+    A virtual channel computed from one or more physical channels.
+
+    kind options:
+      "velocity"       — derivative of a displacement source
+      "acceleration"   — second derivative of a displacement source
+      "power"          — voltage_source * current_source
+      "rms"            — rolling RMS of a source (window samples)
+      "expression"     — arbitrary Python expression string
+      "function"       — multi-line Python function body (def compute(...))
+      "custom_script"  — full Python script, must define compute(inputs, t)
+    """
+    channel_id:   str                           # virtual ID, e.g. "vel_0"
+    name:         str                           # display name
+    unit:         str          = ""
+    color:        str          = "#f72585"
+    kind:         str          = "expression"   # see above
+    # Source channel references  [("dev_id", "ch_id"), ...]
+    sources:      List[Tuple[str, str]] = field(default_factory=list)
+    # For built-in kinds
+    params:       Dict[str, Any] = field(default_factory=dict)
+    # For expression / function / custom_script
+    expression:   str          = ""             # single-line: "x[0] * 2"
+    script:       str          = ""             # multi-line function body
+    enabled:      bool         = True
+    # Runtime: compiled callable (not serialised)
+    _fn:          Optional[Callable] = field(default=None, repr=False, compare=False)
+
+    def compile(self) -> Optional[str]:
+        """
+        Compile expression/script into self._fn.
+        Returns None on success, or error string on failure.
+        """
+        try:
+            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}
+                    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))
+
+            else:
+                # Built-in kinds handled in SignalProcessor._compute_derived
+                self._fn = None
+
+            return None
+        except Exception as e:
+            self._fn = None
+            return str(e)
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+#  ChannelPipeline — per-channel filter stack
+# ══════════════════════════════════════════════════════════════════════════════
+
+@dataclass
+class ChannelPipeline:
+    device_id:  str
+    channel_id: str
+    filters:    List[FilterBase] = field(default_factory=list)
+    enabled:    bool = True
+
+    def process(self, value: float, timestamp: float) -> float:
+        if not self.enabled:
+            return value
+        v = value
+        for f in self.filters:
+            if isinstance(f, (DerivativeFilter, IntegralFilter)):
+                v = f.process_with_t(v, timestamp)
+            else:
+                v = f(v)
+        return v
+
+    def reset(self):
+        for f in self.filters:
+            f.reset()
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+#  SignalProcessor
+# ══════════════════════════════════════════════════════════════════════════════
+
+class SignalProcessor(QObject):
+    """
+    Sits between AcquisitionEngine and StripChartWidget.
+
+    Applies filter pipelines to raw channel data, then evaluates all
+    derived channels and emits processed_data for everything.
+
+    Connect:  engine.new_data  →  processor.on_raw_data
+    Connect:  processor.processed_data  →  chart.on_new_data
+    """
+
+    processed_data = pyqtSignal(str, str, float, float)
+    # device_id, channel_id, timestamp, value
+    # For derived channels: device_id = "derived", channel_id = derived.channel_id
+
+    derived_error  = pyqtSignal(str, str)   # channel_id, error_message
+
+    def __init__(self):
+        super().__init__()
+        self._pipelines: Dict[Tuple[str, str], ChannelPipeline] = {}
+        self._derived:   List[DerivedChannel] = []
+        self._lock       = threading.Lock()
+
+        # Latest raw values cache for derived evaluation
+        # (dev_id, ch_id) -> (timestamp, value)
+        self._latest: Dict[Tuple[str, str], Tuple[float, float]] = {}
+
+        # Ring buffers for derived channels (so strip chart can query history)
+        self._derived_bufs: Dict[str, Tuple[deque, deque]] = {}
+
+    # ── Pipeline management ───────────────────────────────────────────────
+
+    def set_pipeline(self, pipeline: ChannelPipeline):
+        with self._lock:
+            self._pipelines[(pipeline.device_id, pipeline.channel_id)] = pipeline
+
+    def remove_pipeline(self, device_id: str, channel_id: str):
+        with self._lock:
+            self._pipelines.pop((device_id, channel_id), None)
+
+    def get_pipeline(self, device_id: str, channel_id: str) -> Optional[ChannelPipeline]:
+        return self._pipelines.get((device_id, channel_id))
+
+    # ── Derived channel management ────────────────────────────────────────
+
+    def add_derived(self, dc: DerivedChannel) -> Optional[str]:
+        """Add a derived channel. Returns compile error string or None."""
+        err = dc.compile()
+        if err:
+            return err
+        with self._lock:
+            self._derived = [d for d in self._derived if d.channel_id != dc.channel_id]
+            self._derived.append(dc)
+            self._derived_bufs[dc.channel_id] = (deque(maxlen=MAX_BUF), deque(maxlen=MAX_BUF))
+        return None
+
+    def remove_derived(self, channel_id: str):
+        with self._lock:
+            self._derived = [d for d in self._derived if d.channel_id != channel_id]
+            self._derived_bufs.pop(channel_id, None)
+
+    def get_derived(self) -> List[DerivedChannel]:
+        return list(self._derived)
+
+    def get_derived_buffer(self, channel_id: str):
+        """Returns (times_deque, values_deque) or None."""
+        return self._derived_bufs.get(channel_id)
+
+    def all_virtual_channel_ids(self) -> List[str]:
+        return list(self._derived_bufs.keys())
+
+    # ── Main data path ────────────────────────────────────────────────────
+
+    def on_raw_data(self, device_id: str, channel_id: str,
+                    timestamp: float, value: float):
+        """Slot: receive raw data, apply filters, emit processed, update derived."""
+        key = (device_id, channel_id)
+
+        # Apply filter pipeline
+        with self._lock:
+            pipeline = self._pipelines.get(key)
+        processed = pipeline.process(value, timestamp) if pipeline else value
+
+        # Cache latest processed value
+        with self._lock:
+            self._latest[key] = (timestamp, processed)
+
+        # Emit processed physical channel
+        self.processed_data.emit(device_id, channel_id, timestamp, processed)
+
+        # Evaluate all derived channels whose sources include this channel
+        self._evaluate_derived(timestamp)
+
+    def _evaluate_derived(self, timestamp: float):
+        with self._lock:
+            derived = list(self._derived)
+            latest  = dict(self._latest)
+
+        for dc in derived:
+            if not dc.enabled:
+                continue
+            # Check all sources have recent data
+            inputs = []
+            for src in dc.sources:
+                entry = latest.get(src)
+                if entry is None:
+                    break
+                inputs.append(entry[1])   # value only
+            else:
+                # All sources present
+                try:
+                    result = self._compute_derived(dc, inputs, timestamp, latest)
+                    if result is not None:
+                        bufs = self._derived_bufs.get(dc.channel_id)
+                        if bufs:
+                            bufs[0].append(timestamp)
+                            bufs[1].append(result)
+                        self.processed_data.emit("derived", dc.channel_id,
+                                                  timestamp, result)
+                except Exception as e:
+                    self.derived_error.emit(dc.channel_id,
+                                            traceback.format_exc(limit=3))
+
+    def _compute_derived(self, dc: DerivedChannel, inputs: List[float],
+                          t: float, latest: dict) -> Optional[float]:
+        if dc.kind in ("expression", "function", "custom_script"):
+            if dc._fn is None:
+                return None
+            return dc._fn(inputs, t)
+
+        elif dc.kind == "velocity":
+            # derivative of source[0]
+            src = dc.sources[0] if dc.sources else None
+            if src is None: return None
+            state = dc.params.setdefault("_state", {})
+            prev_t = state.get("t"); prev_v = state.get("v")
+            state["t"] = t; state["v"] = inputs[0]
+            if prev_t is None or t == prev_t: return 0.0
+            return (inputs[0] - prev_v) / (t - prev_t)
+
+        elif dc.kind == "acceleration":
+            # second derivative  — derivative of velocity
+            src = dc.sources[0] if dc.sources else None
+            if src is None: return None
+            state = dc.params.setdefault("_state", {})
+            # First get velocity
+            prev_t = state.get("t"); prev_v = state.get("v")
+            cur_vel = 0.0
+            if prev_t is not None and t != prev_t:
+                cur_vel = (inputs[0] - prev_v) / (t - prev_t)
+            prev_vel = state.get("vel", 0.0)
+            state["t"] = t; state["v"] = inputs[0]; state["vel"] = cur_vel
+            if prev_t is None or t == prev_t: return 0.0
+            return (cur_vel - prev_vel) / (t - prev_t)
+
+        elif dc.kind == "power":
+            # V * I
+            if len(inputs) < 2: return None
+            return inputs[0] * inputs[1]
+
+        elif dc.kind == "rms":
+            window = dc.params.get("window", 20)
+            buf = dc.params.setdefault("_buf", deque(maxlen=window))
+            buf.append(inputs[0])
+            return math.sqrt(sum(x*x for x in buf) / len(buf))
+
+        elif dc.kind == "difference":
+            if len(inputs) < 2: return None
+            return inputs[0] - inputs[1]
+
+        elif dc.kind == "sum":
+            return sum(inputs)
+
+        return None