diff options
Diffstat (limited to 'core')
| -rw-r--r-- | core/__init__.py | 4 | ||||
| -rw-r--r-- | core/acquisition.py | 60 | ||||
| -rw-r--r-- | core/profile.py | 166 | ||||
| -rw-r--r-- | core/reference_registry.py | 74 | ||||
| -rw-r--r-- | core/signal_processor.py | 60 |
5 files changed, 240 insertions, 124 deletions
diff --git a/core/__init__.py b/core/__init__.py index 06fabd5..ba8c2b8 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,3 +1,3 @@ # core/__init__.py -from core.acquisition import AcquisitionEngine, ChannelBuffer -__all__ = ["AcquisitionEngine", "ChannelBuffer"] +from core.acquisition import AcquisitionEngine, SignalBuffer +__all__ = ["AcquisitionEngine", "SignalBuffer"] diff --git a/core/acquisition.py b/core/acquisition.py index f54f265..eeeb4c8 100644 --- a/core/acquisition.py +++ b/core/acquisition.py @@ -21,7 +21,7 @@ from devices.base_device import BaseDevice, DeviceStatus MAX_BUFFER = 20_000 -class ChannelBuffer: +class SignalBuffer: """Circular time-series buffer for one channel.""" def __init__(self, maxlen: int = MAX_BUFFER): @@ -70,10 +70,10 @@ class AcquisitionEngine(QObject): Signals ------- - new_data(timestamp, readings) # readings: list[(device_id, channel_id, value)] — + new_data(timestamp, readings) # readings: list[(device_id, signal_id, value)] — # one batched emit per poll tick, not one per channel device_status_changed(device_id, status_str) - channel_unit_changed(device_id, channel_id, new_unit) # e.g. Mark-10 gauge + channel_unit_changed(device_id, signal_id, new_unit) # e.g. Mark-10 gauge # switched lb/kgF/N/ozF on its own front panel log_started(filepath) log_stopped(filepath) @@ -89,7 +89,7 @@ class AcquisitionEngine(QObject): super().__init__() self._interval = poll_interval_ms / 1000.0 self._devices: List[BaseDevice] = [] - self._buffers: Dict[str, Dict[str, ChannelBuffer]] = {} + self._buffers: Dict[str, Dict[str, SignalBuffer]] = {} self._last_units: Dict[Tuple[str, str], str] = {} self._last_status: Dict[str, str] = {} self._running = False @@ -106,11 +106,11 @@ class AcquisitionEngine(QObject): def add_device(self, dev: BaseDevice): self._devices.append(dev) self._buffers[dev.info.device_id] = { - ch.channel_id: ChannelBuffer() - for ch in dev.info.channels + ch.signal_id: SignalBuffer() + for ch in dev.info.signals } - for ch in dev.info.channels: - self._last_units[(dev.info.device_id, ch.channel_id)] = ch.unit + for ch in dev.info.signals: + self._last_units[(dev.info.device_id, ch.signal_id)] = ch.unit def on_device_reconfigured(self, device_id: str): """Reconcile buffers with a device's current channel list — reconfigure() @@ -120,13 +120,13 @@ class AcquisitionEngine(QObject): if dev is None: return bufs = self._buffers.setdefault(device_id, {}) - valid_ids = {ch.channel_id for ch in dev.info.channels} + valid_ids = {ch.signal_id for ch in dev.info.signals} for stale_id in [cid for cid in bufs if cid not in valid_ids]: bufs.pop(stale_id, None) - for ch in dev.info.channels: - if ch.channel_id not in bufs: - bufs[ch.channel_id] = ChannelBuffer() - self._last_units[(device_id, ch.channel_id)] = ch.unit + for ch in dev.info.signals: + if ch.signal_id not in bufs: + bufs[ch.signal_id] = SignalBuffer() + self._last_units[(device_id, ch.signal_id)] = ch.unit for key in [k for k in self._last_units if k[0] == device_id and k[1] not in valid_ids]: self._last_units.pop(key, None) @@ -137,8 +137,22 @@ class AcquisitionEngine(QObject): for key in [k for k in self._last_units if k[0] == device_id]: self._last_units.pop(key, None) - def get_buffer(self, device_id: str, channel_id: str) -> Optional[ChannelBuffer]: - return self._buffers.get(device_id, {}).get(channel_id) + def get_signal_buffer(self, device_id: str, signal_id: str) -> Optional[SignalBuffer]: + return self._buffers.get(device_id, {}).get(signal_id) + + def rename_channel(self, device_id: str, old_id: str, new_id: str) -> None: + """Move a channel's buffer/unit-cache entry to a new id after a rename + — mirrors on_device_reconfigured's bookkeeping but for a single + in-place rename rather than a full channel-list diff, so history is + carried over instead of dropped.""" + if old_id == new_id: + return + bufs = self._buffers.get(device_id) + if bufs and old_id in bufs: + bufs[new_id] = bufs.pop(old_id) + key = (device_id, old_id) + if key in self._last_units: + self._last_units[(device_id, new_id)] = self._last_units.pop(key) def clear_history(self): for ch_map in self._buffers.values(): @@ -181,8 +195,8 @@ class AcquisitionEngine(QObject): self._csv_file = open(filepath, "w", newline="") headers = ["elapsed_s"] for dev in self._devices: - for ch in dev.info.channels: - headers.append(f"{dev.info.device_id}/{ch.channel_id}[{ch.unit}]") + for ch in dev.info.signals: + headers.append(f"{dev.info.device_id}/{ch.signal_id}[{ch.unit}]") self._csv_writer = csv.writer(self._csv_file) self._csv_writer.writerow(headers) self._logging = True @@ -225,21 +239,21 @@ class AcquisitionEngine(QObject): print(f"[Acq] {dev.info.device_id} read error: {e}") readings = {} - for ch in dev.info.channels: - key = (dev.info.device_id, ch.channel_id) + for ch in dev.info.signals: + key = (dev.info.device_id, ch.signal_id) if self._last_units.get(key) != ch.unit: self._last_units[key] = ch.unit - self.channel_unit_changed.emit(dev.info.device_id, ch.channel_id, ch.unit) + self.channel_unit_changed.emit(dev.info.device_id, ch.signal_id, ch.unit) - val = readings.get(ch.channel_id) + val = readings.get(ch.signal_id) if val is None or not ch.enabled: log_row.append("") continue fval = float(val) - buf = self._buffers.get(dev.info.device_id, {}).get(ch.channel_id) + buf = self._buffers.get(dev.info.device_id, {}).get(ch.signal_id) if buf is not None: buf.append(elapsed, fval) - batch.append((dev.info.device_id, ch.channel_id, fval)) + batch.append((dev.info.device_id, ch.signal_id, fval)) log_row.append(f"{fval:.5f}") if batch: diff --git a/core/profile.py b/core/profile.py index c0ae5f1..c77e646 100644 --- a/core/profile.py +++ b/core/profile.py @@ -3,20 +3,28 @@ core/profile.py Lab profile — saves and loads the complete operator configuration: - • Controls — all control widgets (type, title, device, channel, params) - • Channels — per-channel visibility, name overrides - • Signal pipelines — filter stacks per channel - • Derived channels — all virtual/computed channels - • Plot layout — pane arrangement, traces, axes, time window + • Controls — all control widgets (type, title, device, signal, params) + • Signals — per-signal visibility, name overrides (physical device channels) + • Signal pipelines — filter stacks per signal + • Channels — all derived/virtual channels (Signals + math -> a new stream) + • Plot layout — pane arrangement, traces, axes, time window Profiles are stored as human-readable JSON files (.labui). + +Schema v2 renamed two sections from v1 to match the Signal/Channel naming +split (devices/base_device.py, core/signal_processor.py): the old "channels" +section (physical overrides) is now "signals", and the old "derived" section +(virtual channels) is now "channels". `_migrate_legacy_profile()` upgrades a +v1 file transparently on load, including remapping Arduino analog Signal ids +that used to be the raw pin name (e.g. "A0") to the stable positional id +("ai0") introduced alongside this schema — see devices/arduino_device.py. """ from __future__ import annotations import json import os from dataclasses import dataclass, field, asdict -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple # ── Helpers ─────────────────────────────────────────────────────────────────── @@ -26,54 +34,80 @@ def _safe(d: dict, key: str, default=None): # ══════════════════════════════════════════════════════════════════════════════ -# Profile data model +# Legacy (pre-2.0) schema migration # ══════════════════════════════════════════════════════════════════════════════ -@dataclass -class ProfileChannelOverride: - """Per-channel settings that survive device reconnects.""" - device_id: str - channel_id: str - name: str = "" - unit: str = "" - enabled: bool = True - color: str = "" - - -@dataclass -class ProfilePipeline: - """Serialised filter pipeline for one channel.""" - device_id: str - channel_id: str - enabled: bool = True - filters: List[Dict[str, Any]] = field(default_factory=list) - # filters = [{"type": "low_pass", "alpha": 0.1}, ...] - +def _migrate_legacy_profile(raw: dict) -> dict: + """Upgrade a pre-2.0 profile dict to the current schema. Returns a new + dict; does not mutate the input. Two independent changes bundled + together since both only matter for old files: + + 1. Section rename: "channels" (physical overrides) -> "signals", with + its "channel_id" keys -> "signal_id"; "derived" (virtual channel + defs) -> "channels". The "channels" (ex-"derived") section's own + "channel_id" field is untouched — a Channel's own id field was + never renamed, only what it's called at the top level. + + 2. Arduino analog Signal ids used to BE the pin name ("A0") instead of + a stable positional id ("ai0") that survives a pin remap (see + devices/arduino_device.py::_build_channels). Every reference to an + old pin-name id is rewritten to the new positional id, using each + Arduino device's saved `analog_pins` order to know which pin became + which "ai<N>". + """ + raw = dict(raw) + + derived = raw.pop("derived", []) + signals = raw.pop("channels", []) + for s in signals: + if "channel_id" in s: + s["signal_id"] = s.pop("channel_id") + raw["signals"] = signals + raw["channels"] = derived + + pin_to_id: Dict[Tuple[str, str], str] = {} # (device_id, old_pin) -> new positional id + for dev_cfg in raw.get("devices", []): + if dev_cfg.get("device_type") != "arduino": + continue + dev_id = dev_cfg.get("device_id", "") + for i, pin in enumerate(dev_cfg.get("analog_pins", [])): + pin_to_id[(dev_id, pin)] = f"ai{i}" + + if pin_to_id: + def _remap(dev_id: str, ch_id: str) -> str: + return pin_to_id.get((dev_id, ch_id), ch_id) + + for s in raw["signals"]: + s["signal_id"] = _remap(s.get("device_id", ""), s.get("signal_id", "")) + for pd in raw.get("pipelines", []): + pd["channel_id"] = _remap(pd.get("device_id", ""), pd.get("channel_id", "")) + for dc in raw["channels"]: + dc["sources"] = [[d, _remap(d, c)] for d, c in dc.get("sources", [])] + for c in raw.get("controls", []): + c["channel_id"] = _remap(c.get("device_id", ""), c.get("channel_id", "")) + plot = raw.get("plot") + if plot: + for pane in plot.get("panes", []): + for tr in pane.get("traces", []): + tr["channel_id"] = _remap(tr.get("device_id", ""), tr.get("channel_id", "")) + + raw["version"] = "2.0" + return raw -@dataclass -class ProfileDerived: - """Serialised derived/virtual channel.""" - channel_id: str - name: str - unit: str = "" - color: str = "#f72585" - kind: str = "expression" - sources: List = field(default_factory=list) # [(dev_id, ch_id), ...] - expression: str = "" - script: str = "" - params: Dict = field(default_factory=dict) - enabled: bool = True +# ══════════════════════════════════════════════════════════════════════════════ +# Profile data model +# ══════════════════════════════════════════════════════════════════════════════ @dataclass class Profile: name: str = "Untitled Profile" - version: str = "1.0" + version: str = "2.0" devices: List[Dict] = field(default_factory=list) controls: List[Dict] = field(default_factory=list) - channels: List[Dict] = field(default_factory=list) + signals: List[Dict] = field(default_factory=list) pipelines: List[Dict] = field(default_factory=list) - derived: List[Dict] = field(default_factory=list) + channels: List[Dict] = field(default_factory=list) plot: Optional[Dict] = None # LayoutConfig.to_json() settings: Dict[str, Any] = field(default_factory=dict) plugin_state: Dict[str, Any] = field(default_factory=dict) @@ -91,14 +125,16 @@ class Profile: @staticmethod def from_json(s: str) -> "Profile": d = json.loads(s) + if not str(d.get("version", "1.0")).startswith("2."): + d = _migrate_legacy_profile(d) return Profile( name = d.get("name", ""), - version = d.get("version", "1.0"), + version = d.get("version", "2.0"), devices = d.get("devices", []), controls = d.get("controls", []), - channels = d.get("channels", []), + signals = d.get("signals", []), pipelines = d.get("pipelines", []), - derived = d.get("derived", []), + channels = d.get("channels", []), plot = d.get("plot"), settings = d.get("settings", {}), plugin_state = d.get("plugin_state", {}), @@ -171,12 +207,12 @@ class ProfileManager: # Controls p.controls = [spec.to_dict() for spec in control_specs] - # Channel overrides + # Signal overrides for dev in registry.all_instances(): - for ch in dev.info.channels: - p.channels.append({ + for ch in dev.info.signals: + p.signals.append({ "device_id": dev.info.device_id, - "channel_id": ch.channel_id, + "signal_id": ch.signal_id, "name": ch.name, "unit": ch.unit, "enabled": ch.enabled, @@ -192,9 +228,9 @@ class ProfileManager: "filters": [f.to_dict() for f in pipeline.filters], }) - # Derived channels + # Channels (derived) for dc in processor.get_derived(): - p.derived.append({ + p.channels.append({ "channel_id": dc.channel_id, "name": dc.name, "unit": dc.unit, @@ -263,7 +299,7 @@ class ProfileManager: """Apply a loaded profile to the live application state.""" from ui.control_editor import ControlSpec from core.signal_processor import ( - ChannelPipeline, DerivedChannel, + ChannelPipeline, Channel, filter_from_dict, ) @@ -284,18 +320,6 @@ class ProfileManager: _DEVICE_FACTORIES["serial"] = SerialDevice except Exception: pass - # Backward compatibility with old profiles - try: - from devices.analog_input import AnalogInputDevice - _DEVICE_FACTORIES["analog_input"] = AnalogInputDevice - except Exception: - pass - try: - from devices.digital_io import DigitalIODevice - _DEVICE_FACTORIES["digital_io"] = DigitalIODevice - except Exception: - pass - # Clear all existing devices — profile defines the complete device set for dev in list(registry.all_instances()): try: @@ -362,12 +386,12 @@ class ProfileManager: except Exception as e: print(f"[Profile] Could not restore device {dev_id}: {e}") - # ── Channel overrides ──────────────────────────────────────────── - for ch_data in profile.channels: + # ── Signal overrides ───────────────────────────────────────────── + for ch_data in profile.signals: dev = registry.get_instance(ch_data["device_id"]) if not dev: continue - ch = dev.get_channel(ch_data["channel_id"]) + ch = dev.get_signal(ch_data["signal_id"]) if not ch: continue ch.name = ch_data.get("name", ch.name) @@ -387,10 +411,10 @@ class ProfileManager: ) processor.set_pipeline(pipeline) - # ── Derived channels ───────────────────────────────────────────── - for dd in profile.derived: + # ── Channels (derived) ──────────────────────────────────────────── + for dd in profile.channels: sources = [tuple(s) for s in dd.get("sources", [])] - dc = DerivedChannel( + dc = Channel( channel_id=dd["channel_id"], name=dd.get("name", dd["channel_id"]), unit=dd.get("unit", ""), diff --git a/core/reference_registry.py b/core/reference_registry.py new file mode 100644 index 0000000..b6fcd0f --- /dev/null +++ b/core/reference_registry.py @@ -0,0 +1,74 @@ +""" +core/reference_registry.py + +Generic (device_id, channel_id) reference-rename propagation. + +A Signal (physical device channel) or a Channel (derived, device_id == +"derived") can be renamed by the user. Both kinds are referenced the same +way everywhere downstream — plot traces, control targets, filter +pipelines, other derived channels' sources — as a plain +(device_id, channel_id) pair. Renaming one without rewriting every +reference to it leaves those references dangling: they silently stop +receiving data (this is exactly the bug that motivated this module — an +Arduino analog pin remap changed a Signal's id and orphaned its plot +trace). + +propagate_rename() is the single place that walks every subsystem able to +hold such a reference and rewrites it. It expects a MainWindow instance +(or anything exposing the same .engine/.processor/._chart/._ctrl/ +._win_config attributes) since that's the one place all of them are +reachable from. +""" + +from typing import Tuple + +Ref = Tuple[str, str] + + +def propagate_rename(app, old_ref: Ref, new_ref: Ref) -> None: + if old_ref == new_ref: + return + + # Physical signals live in the acquisition engine's buffers; derived + # Channels don't (their buffers live in SignalProcessor._derived_bufs). + if old_ref[0] != "derived": + app.engine.rename_channel(old_ref[0], old_ref[1], new_ref[1]) + + # Filter pipelines, other derived channels' sources, latest-value cache, + # and (if old_ref is itself a derived Channel) its ring buffer. + app.processor.rename_reference(old_ref, new_ref) + + # Live plot — trace specs in the strip chart's current layout. + chart = getattr(app, "_chart", None) + changed = False + if chart is not None and chart._cfg is not None: + for pane in chart._cfg.panes: + for tr in pane.traces: + if (tr.device_id, tr.channel_id) == old_ref: + tr.device_id, tr.channel_id = new_ref + changed = True + if changed: + chart.refresh() + + # Control panel — saved specs (profile persistence) + live bound widgets + # (actual I/O routing — ControlWidget reads its own .device_id/.channel_id + # directly in _write()/_check_stale(), not from the spec). + ctrl = getattr(app, "_ctrl", None) + if ctrl is not None: + for spec in ctrl.get_specs(): + if (spec.device_id, spec.channel_id) == old_ref: + spec.device_id, spec.channel_id = new_ref + for widget in getattr(ctrl, "_widgets", []): + if (widget.device_id, widget.channel_id) == old_ref: + widget.device_id, widget.channel_id = new_ref + + # Config window tabs, if currently open — refresh so pickers/tables + # don't show the stale id until next full refresh. + win = getattr(app, "_win_config", None) + if win is not None: + if chart is not None: + win.refresh_plot(chart._cfg) + if hasattr(win, "refresh_derived"): + win.refresh_derived() + if hasattr(win, "refresh_devices"): + win.refresh_devices() diff --git a/core/signal_processor.py b/core/signal_processor.py index 4781e91..da615b6 100644 --- a/core/signal_processor.py +++ b/core/signal_processor.py @@ -230,7 +230,7 @@ def _make_src_names(sources: List[Tuple[str, str]], # ══════════════════════════════════════════════════════════════════════════════ @dataclass -class DerivedChannel: +class Channel: """ A virtual channel computed from one or more physical channels. @@ -393,7 +393,7 @@ class SignalProcessor(QObject): def __init__(self): super().__init__() self._pipelines: Dict[Tuple[str, str], ChannelPipeline] = {} - self._derived: List[DerivedChannel] = [] + self._derived: List[Channel] = [] self._lock = threading.Lock() # Latest raw values cache for derived evaluation @@ -421,7 +421,7 @@ class SignalProcessor(QObject): # ── Derived channel management ──────────────────────────────────────── - def add_derived(self, dc: DerivedChannel) -> Optional[str]: + def add_derived(self, dc: Channel) -> Optional[str]: """Add a derived channel. Returns compile error string or None.""" err = dc.compile() if err: @@ -437,36 +437,40 @@ class SignalProcessor(QObject): self._derived = [d for d in self._derived if d.channel_id != channel_id] self._derived_bufs.pop(channel_id, None) - def rename_derived(self, old_id: str, new_id: str): - """Propagate a Channel ID rename to every other derived channel's - sources — otherwise anything chained from it keeps a dangling - ("derived", old_id) reference (SignalProcessor._evaluate_derived - looks sources up by exact match) and silently stops updating. - Also carries the ring buffer and latest-value cache over under the - new key so the rename doesn't reset plotted history or blank the - Value column for a poll cycle. Does NOT touch source_names or - expression/script text — those are display labels / user-authored - code, out of scope for an automatic rewrite.""" - if old_id == new_id: + def rename_reference(self, old_ref: Tuple[str, str], new_ref: Tuple[str, str]): + """Propagate a (device_id, channel_id) reference rename to everything + this processor holds one of — otherwise anything reading it keeps a + dangling reference (_evaluate_derived/_process_one look sources and + pipelines up by exact match) and silently stops updating. Covers + both a physical Signal rename and a derived Channel rename with the + same code path — the renamed item's own identity (Channel. + channel_id, or a device's SignalConfig) is the caller's job; this + only fixes up *references to* old_ref held elsewhere. + Also carries the latest-value cache, and — when old_ref is itself a + derived Channel (device_id == "derived") — its ring buffer, over + under the new key so the rename doesn't reset plotted history or + blank the Value column for a poll cycle. Does NOT touch + source_names or expression/script text — those are display labels / + user-authored code, out of scope for an automatic rewrite.""" + if old_ref == new_ref: return with self._lock: + if old_ref in self._pipelines: + self._pipelines[new_ref] = self._pipelines.pop(old_ref) + for dc in self._derived: - if dc.channel_id == new_id: - continue - new_sources = [ - ("derived", new_id) if d == "derived" and c == old_id else (d, c) - for d, c in dc.sources - ] + new_sources = [new_ref if s == old_ref else s for s in dc.sources] if new_sources != dc.sources: dc.sources = new_sources - buf = self._derived_bufs.pop(old_id, None) - if buf is not None: - self._derived_bufs[new_id] = buf - - entry = self._latest.pop(("derived", old_id), None) + entry = self._latest.pop(old_ref, None) if entry is not None: - self._latest[("derived", new_id)] = entry + self._latest[new_ref] = entry + + if old_ref[0] == "derived": + buf = self._derived_bufs.pop(old_ref[1], None) + if buf is not None: + self._derived_bufs[new_ref[1]] = buf def reorder_derived(self, src_idx: int, dst_idx: int): """Move the derived channel at src_idx to dst_idx — list order is @@ -478,7 +482,7 @@ class SignalProcessor(QObject): dc = self._derived.pop(src_idx) self._derived.insert(dst_idx, dc) - def get_derived(self) -> List[DerivedChannel]: + def get_derived(self) -> List[Channel]: return list(self._derived) def get_derived_buffer(self, channel_id: str): @@ -575,7 +579,7 @@ class SignalProcessor(QObject): self.derived_error.emit(dc.channel_id, traceback.format_exc(limit=3)) - def _compute_derived(self, dc: DerivedChannel, inputs: List[float], + def _compute_derived(self, dc: Channel, inputs: List[float], t: float, latest: dict) -> Optional[float]: if dc.kind in ("expression", "function", "custom_script"): if dc._fn is None: |