diff options
29 files changed, 561 insertions, 2677 deletions
diff --git a/api_layers/nidaqmx_layer.py b/api_layers/nidaqmx_layer.py index ba40efb..e0fc670 100644 --- a/api_layers/nidaqmx_layer.py +++ b/api_layers/nidaqmx_layer.py @@ -6,8 +6,8 @@ NI-DAQmx API abstraction layer. • When nidaqmx package + NI runtime are present → uses real hardware • Otherwise → falls back to simulation -Swap this layer by changing the `backend` parameter on AnalogInputDevice -or by subclassing NidaqmxLayer and overriding _hw_read(). +Used by devices/nidaqmx_device.py::NidaqmxDevice. Swap it out by +subclassing NidaqmxLayer and overriding _hw_read(). Usage example: from api_layers.nidaqmx_layer import NidaqmxLayer diff --git a/api_layers/port_registry.py b/api_layers/port_registry.py index e3c85fd..4328bb2 100644 --- a/api_layers/port_registry.py +++ b/api_layers/port_registry.py @@ -3,10 +3,10 @@ api_layers/port_registry.py PortRegistry — ensures one ArduinoLayer per physical serial port. -Problem: if both an AnalogInputDevice and a DigitalIODevice are -configured for the same Arduino (/dev/ttyUSB0), they would each -create their own ArduinoLayer and both try to open the same serial -port → "device reports readiness to read but returned no data". +Problem: if two ArduinoDevice instances are configured for the same +physical Arduino (/dev/ttyUSB0), they would each create their own +ArduinoLayer and both try to open the same serial port → "device +reports readiness to read but returned no data". Solution: all devices call PortRegistry.get_layer(port, baud) instead of constructing ArduinoLayer directly. The registry returns an @@ -14,9 +14,9 @@ existing layer if one is already open on that port, or creates a new one. All reads/writes go through the single shared layer. The registry also merges analog_pins across devices so the layer -caches values for all channels on the Arduino. +caches values for all signals on the Arduino. -Usage (handled automatically by AnalogInputDevice and DigitalIODevice): +Usage (handled automatically by ArduinoDevice): from api_layers.port_registry import port_registry layer = port_registry.get_layer("/dev/ttyUSB0", 115200, simulate=False) 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: diff --git a/devices/__init__.py b/devices/__init__.py index 7cee4ac..d175390 100644 --- a/devices/__init__.py +++ b/devices/__init__.py @@ -1,4 +1,4 @@ -# devices/__init__.py from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus +# devices/__init__.py from devices.base_device import BaseDevice, SignalConfig, DeviceInfo, DeviceStatus from devices.device_registry import DeviceRegistry -__all__ = ["BaseDevice", "ChannelConfig", "DeviceInfo", "DeviceStatus", "DeviceRegistry"] +__all__ = ["BaseDevice", "SignalConfig", "DeviceInfo", "DeviceStatus", "DeviceRegistry"] diff --git a/devices/analog_input.py b/devices/analog_input.py deleted file mode 100644 index 8938bf2..0000000 --- a/devices/analog_input.py +++ /dev/null @@ -1,461 +0,0 @@ -""" -devices/analog_input.py - -Analog Input device. Backends: NI-DAQmx or Arduino. -""" - -from typing import Any, Dict, List - -from PyQt6.QtWidgets import ( - QWidget, QVBoxLayout, QHBoxLayout, QFormLayout, QGroupBox, - QComboBox, QDoubleSpinBox, QCheckBox, QLineEdit, QLabel, - QPushButton, QListWidget, QListWidgetItem, QFrame, QTextEdit, -) -from PyQt6.QtCore import Qt, QThread, pyqtSignal -from PyQt6.QtGui import QFont - -from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus -from api_layers.nidaqmx_layer import NidaqmxLayer -from api_layers.arduino_layer import ArduinoLayer - -_COLORS = [ - "#00d4ff", "#ff6b35", "#7fff6e", "#ffcc00", - "#c77dff", "#ff4d6d", "#4cc9f0", "#f72585", - "#38b000", "#e9c46a", "#a8dadc", "#e63946", - "#90e0ef", "#fb8500", "#b5e48c", "#d62828", -] - - -class AnalogInputDevice(BaseDevice): - DEVICE_TYPE = "analog_input" - ICON = "〜" - - def __init__( - self, - device_id: str = "ai_0", - num_channels: int = 4, - simulate: bool = True, - backend: str = "nidaqmx", - ni_device: str = "Dev1", - ni_min_v: float = -10.0, - ni_max_v: float = 10.0, - ard_port: str = "COM3", - ard_baud: int = 115200, - ): - self.backend = backend - self.simulate = simulate - - if backend == "arduino": - pins = ArduinoLayer.DEFAULT_ANALOG_PINS[:num_channels] - channels = [ - ChannelConfig(channel_id=p, name=p, unit="V", - min_value=0.0, max_value=5.0, - color=_COLORS[i % len(_COLORS)]) - for i, p in enumerate(pins) - ] - else: - ni_pins = [f"ai{i}" for i in range(num_channels)] - channels = [ - ChannelConfig(channel_id=p, name=p.upper(), unit="V", - min_value=ni_min_v, max_value=ni_max_v, - color=_COLORS[i % len(_COLORS)]) - for i, p in enumerate(ni_pins) - ] - - info = DeviceInfo( - device_id=device_id, - name=f"Analog Input ({backend.upper()})", - device_type=self.DEVICE_TYPE, - description=f"Multi-channel analog input via {backend}", - manufacturer="NI" if backend == "nidaqmx" else "Arduino", - icon=self.ICON, - channels=channels, - ) - super().__init__(info) - - self._ni_device = ni_device - self._ni_min_v = ni_min_v - self._ni_max_v = ni_max_v - self._ard_port = ard_port - self._ard_baud = ard_baud - self._num_channels = num_channels - self._last_error = "" - - self._layer = self._make_layer( - backend, simulate, ni_device, ni_min_v, ni_max_v, - ard_port, ard_baud, - [ch.channel_id for ch in channels], - ) - - def _make_layer(self, backend, simulate, - ni_device, ni_min_v, ni_max_v, - ard_port, ard_baud, channel_ids): - if backend == "arduino": - # Use shared layer so two devices on the same port don't conflict - from api_layers.port_registry import port_registry - return port_registry.get_layer( - port=ard_port, - baud=ard_baud, - simulate=simulate, - extra_pins=channel_ids, - ) - else: - return NidaqmxLayer( - device_name=ni_device, - channels=channel_ids, - min_val=ni_min_v, - max_val=ni_max_v, - simulate=simulate, - ) - - # ── BaseDevice ────────────────────────────────────────────────────────── - - def connect(self) -> bool: - self._last_error = "" - - # ArduinoLayer: use connect_if_needed so shared layers aren't - # opened twice when multiple devices share the same port. - # NidaqmxLayer: uses start() - if hasattr(self._layer, "connect"): - if self._layer.is_connected: - ok = True # already open — shared with another device - else: - ok = self._layer.connect() - else: - ok = self._layer.start() - - if not ok and hasattr(self._layer, "last_error"): - self._last_error = self._layer.last_error - - if self.simulate: - self.status = DeviceStatus.SIMULATED - else: - self.status = DeviceStatus.CONNECTED if ok else DeviceStatus.ERROR - - return ok - - def disconnect(self) -> None: - # For shared Arduino layers, release our ref count via port_registry. - # The layer stays open until the last device using it disconnects. - if self.backend == "arduino" and not self.simulate: - from api_layers.port_registry import port_registry - port_registry.release(self._ard_port, self._ard_baud) - else: - try: - if hasattr(self._layer, "stop"): - self._layer.stop() - elif hasattr(self._layer, "disconnect"): - self._layer.disconnect() - except Exception: - pass - self.status = DeviceStatus.DISCONNECTED - - def read_channels(self) -> Dict[str, float]: - return self._layer.read() - - def write_channel(self, channel_id: str, value: Any) -> bool: - """ - Route output commands through the Arduino layer. - - channel_id examples: - "D7" → digital write W:D7:1 or W:D7:0 - "D9" → PWM write P:D9:<duty> (if value is 0-255) - "HEAT" → named param C:HEAT:1.0 - """ - if self.backend != "arduino" or not hasattr(self._layer, "digital_write"): - return False - ch = channel_id.strip() - # Digital pin (D7, D13, etc.) - if ch.upper().startswith("D") and ch[1:].isdigit(): - return self._layer.digital_write(ch, int(bool(value))) - # Named parameter / setpoint - return self._layer.set_parameter(ch, value) - - def get_save_config(self) -> dict: - return { - "device_type": self.DEVICE_TYPE, - "device_id": self.info.device_id, - "name": self.info.name, - "num_channels": self._num_channels, - "simulate": self.simulate, - "backend": self.backend, - "ni_device": self._ni_device, - "ni_min_v": self._ni_min_v, - "ni_max_v": self._ni_max_v, - "ard_port": self._ard_port, - "ard_baud": self._ard_baud, - } - - def get_config_widget(self) -> QWidget: - return AnalogInputConfigWidget(self) - - def switch_backend(self, backend: str, simulate: bool, - ni_device: str, min_val: float, max_val: float, - port: str, baud: int) -> None: - was_running = self.status in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED) - if was_running: - self.disconnect() - - self.backend = backend - self.simulate = simulate - self._ni_device = ni_device - self._ni_min_v = min_val - self._ni_max_v = max_val - self._ard_port = port - self._ard_baud = baud - - self._layer = self._make_layer( - backend, simulate, ni_device, min_val, max_val, - port, baud, - [ch.channel_id for ch in self.info.channels], - ) - - if was_running: - self.connect() - - -# ── Port scanner thread ─────────────────────────────────────────────────────── - -class _PortScanThread(QThread): - done = pyqtSignal(list) - - def run(self): - ports = ArduinoLayer.list_ports() # returns [(device, desc), ...] - self.done.emit(ports) - - -# ── Config widget ───────────────────────────────────────────────────────────── - -class AnalogInputConfigWidget(QWidget): - def __init__(self, device: AnalogInputDevice): - super().__init__() - self.device = device - self._scanner = None - self._build() - - def _build(self): - root = QVBoxLayout(self) - root.setContentsMargins(10, 14, 10, 10) - root.setSpacing(12) - self.setMinimumWidth(460) - - # ── Backend / Simulate ─────────────────────────────────────────── - be_grp = QGroupBox("Backend Selection") - be_form = QFormLayout(be_grp) - be_form.setContentsMargins(10, 16, 10, 10) - - self.backend_cb = QComboBox() - self.backend_cb.addItems(["nidaqmx", "arduino"]) - self.backend_cb.setCurrentText(self.device.backend) - be_form.addRow("Backend:", self.backend_cb) - - self.sim_check = QCheckBox("Simulation Mode (no hardware)") - self.sim_check.setChecked(self.device.simulate) - from core.app_settings import is_developer_mode - self.sim_check.setVisible(is_developer_mode()) - be_form.addRow(self.sim_check) - root.addWidget(be_grp) - - # ── NI-DAQmx ──────────────────────────────────────────────────── - self.ni_grp = QGroupBox("NI-DAQmx Settings") - ni_form = QFormLayout(self.ni_grp) - ni_form.setContentsMargins(10, 16, 10, 10) - - self.ni_dev_edit = QLineEdit(self.device._ni_device) - ni_form.addRow("NI Device:", self.ni_dev_edit) - - self.ni_min_spin = QDoubleSpinBox() - self.ni_min_spin.setRange(-100, 0) - self.ni_min_spin.setValue(self.device._ni_min_v) - self.ni_min_spin.setSuffix(" V") - ni_form.addRow("Min Voltage:", self.ni_min_spin) - - self.ni_max_spin = QDoubleSpinBox() - self.ni_max_spin.setRange(0, 100) - self.ni_max_spin.setValue(self.device._ni_max_v) - self.ni_max_spin.setSuffix(" V") - ni_form.addRow("Max Voltage:", self.ni_max_spin) - - detect_btn = QPushButton("Detect NI Devices") - detect_btn.setObjectName("configButton") - detect_btn.clicked.connect(self._detect_ni) - ni_form.addRow(detect_btn) - self.ni_detect_lbl = QLabel("") - self.ni_detect_lbl.setObjectName("traceSource") - ni_form.addRow(self.ni_detect_lbl) - root.addWidget(self.ni_grp) - - # ── Arduino ────────────────────────────────────────────────────── - self.ard_grp = QGroupBox("Arduino Serial Settings") - ard_lay = QVBoxLayout(self.ard_grp) - ard_lay.setContentsMargins(10, 16, 10, 10) - ard_lay.setSpacing(8) - - ard_form = QFormLayout() - self.ard_port_edit = QLineEdit(self.device._ard_port) - self.ard_port_edit.setPlaceholderText("e.g. /dev/ttyUSB0 or COM3") - ard_form.addRow("Port:", self.ard_port_edit) - - self.ard_baud_cb = QComboBox() - self.ard_baud_cb.addItems(["9600", "19200", "57600", "115200", "230400"]) - self.ard_baud_cb.setCurrentText(str(self.device._ard_baud)) - ard_form.addRow("Baud Rate:", self.ard_baud_cb) - ard_lay.addLayout(ard_form) - - # Scan row - scan_row = QHBoxLayout() - self._scan_btn = QPushButton("🔍 Scan Ports") - self._scan_btn.setObjectName("addTraceBtn") - self._scan_btn.clicked.connect(self._scan_ports) - self._scan_lbl = QLabel("") - self._scan_lbl.setObjectName("traceSource") - scan_row.addWidget(self._scan_btn) - scan_row.addWidget(self._scan_lbl, 1) - ard_lay.addLayout(scan_row) - - # Port list - self._port_list = QListWidget() - self._port_list.setObjectName("portList") - self._port_list.setMaximumHeight(90) - self._port_list.itemClicked.connect(self._on_port_selected) - ard_lay.addWidget(self._port_list) - - hint = QLabel("↑ Click a port to fill the field above") - hint.setObjectName("traceSource") - ard_lay.addWidget(hint) - - # pyserial availability notice - if not ArduinoLayer.is_pyserial_available(): - warn = QLabel("⚠ pyserial not installed — run: pip install pyserial") - warn.setStyleSheet("color:#f59e0b; font-weight:600;") - ard_lay.addWidget(warn) - - root.addWidget(self.ard_grp) - - # ── Diagnostics ─────────────────────────────────────────────────── - self.diag_grp = QGroupBox("Connection Diagnostics") - diag_lay = QVBoxLayout(self.diag_grp) - diag_lay.setContentsMargins(10, 16, 10, 10) - - self._diag_box = QTextEdit() - self._diag_box.setObjectName("codeEditor") - self._diag_box.setReadOnly(True) - self._diag_box.setMaximumHeight(160) - self._diag_box.setPlaceholderText("Connection log will appear here…") - diag_lay.addWidget(self._diag_box) - - # Show any existing error - if self.device._last_error: - self._diag_box.setPlainText(f"Last error:\n{self.device._last_error}") - - root.addWidget(self.diag_grp) - - # ── Apply ───────────────────────────────────────────────────────── - apply_btn = QPushButton("Apply & Reconnect") - apply_btn.setObjectName("applyButton") - apply_btn.clicked.connect(self._apply) - root.addWidget(apply_btn) - root.addStretch() - - self._update_visibility() - self.backend_cb.currentTextChanged.connect(self._update_visibility) - self._refresh_diag() - - # ── Helpers ────────────────────────────────────────────────────────────── - - def _update_visibility(self): - be = self.backend_cb.currentText() - self.ni_grp.setVisible(be == "nidaqmx") - self.ard_grp.setVisible(be == "arduino") - - def _refresh_diag(self): - lines = [] - lines.append(f"Backend: {self.device.backend}") - lines.append(f"Simulate: {self.device.simulate}") - lines.append(f"Status: {self.device.status.value}") - if self.device.backend == "arduino": - lines.append(f"Port: {self.device._ard_port}") - lines.append(f"Baud: {self.device._ard_baud}") - lines.append(f"pyserial: {'available' if ArduinoLayer.is_pyserial_available() else 'NOT INSTALLED'}") - # Show last raw lines received from Arduino - if hasattr(self.device._layer, "raw_lines"): - raw = self.device._layer.raw_lines - if raw: - lines.append("\nLast lines from Arduino:") - for r in raw: - lines.append(f" {r}") - if self.device._last_error: - lines.append(f"\nError:\n{self.device._last_error}") - self._diag_box.setPlainText("\n".join(lines)) - - def _detect_ni(self): - devs = NidaqmxLayer.list_devices() - self.ni_detect_lbl.setText( - ", ".join(devs) if devs else "No NI devices found (nidaqmx package required)" - ) - - def _scan_ports(self): - self._scan_btn.setEnabled(False) - self._scan_lbl.setText("Scanning…") - self._port_list.clear() - self._scanner = _PortScanThread() - self._scanner.done.connect(self._on_scan_done) - self._scanner.start() - - def _on_scan_done(self, ports): - self._scan_btn.setEnabled(True) - self._port_list.clear() - - if not ArduinoLayer.is_pyserial_available(): - self._scan_lbl.setText("pyserial not installed") - item = QListWidgetItem(" Install pyserial: pip install pyserial") - item.setFlags(item.flags() & ~Qt.ItemFlag.ItemIsSelectable) - self._port_list.addItem(item) - return - - if not ports: - self._scan_lbl.setText("No ports detected") - item = QListWidgetItem( - " No serial ports found.\n" - " • Check USB cable\n" - " • On Linux: sudo usermod -aG dialout $USER then re-login\n" - " • On Arch: sudo usermod -aG uucp $USER then re-login" - ) - item.setFlags(item.flags() & ~Qt.ItemFlag.ItemIsSelectable) - self._port_list.addItem(item) - else: - self._scan_lbl.setText(f"{len(ports)} port(s) found — click to select") - for device, desc in ports: - label = f" {device}" - if desc and desc.strip() and desc.strip() != device: - label += f" — {desc}" - item = QListWidgetItem(label) - item.setData(Qt.ItemDataRole.UserRole, device) - self._port_list.addItem(item) - - def _on_port_selected(self, item: QListWidgetItem): - port = item.data(Qt.ItemDataRole.UserRole) - if port: - self.ard_port_edit.setText(port) - - def _apply(self): - backend = self.backend_cb.currentText() - simulate = self.sim_check.isChecked() - port = self.ard_port_edit.text().strip() - baud = int(self.ard_baud_cb.currentText()) - - self._diag_box.setPlainText( - f"Connecting...\nBackend: {backend}\nSimulate: {simulate}\n" - + (f"Port: {port}\nBaud: {baud}" if backend == "arduino" else "") - ) - - self.device.switch_backend( - backend=backend, - simulate=simulate, - ni_device=self.ni_dev_edit.text().strip() or "Dev1", - min_val=self.ni_min_spin.value(), - max_val=self.ni_max_spin.value(), - port=port, - baud=baud, - ) - self._refresh_diag() diff --git a/devices/arduino_device.py b/devices/arduino_device.py index 1641eaa..5c954c8 100644 --- a/devices/arduino_device.py +++ b/devices/arduino_device.py @@ -15,7 +15,7 @@ from PyQt6.QtWidgets import ( ) from PyQt6.QtCore import Qt, QThread, pyqtSignal -from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus +from devices.base_device import BaseDevice, SignalConfig, DeviceInfo, DeviceStatus from api_layers.arduino_layer import ArduinoLayer _ANALOG_COLORS = [ @@ -91,26 +91,11 @@ class ArduinoDevice(BaseDevice): self._di_pins: List[str] = list(di_pins) self._do_pins: List[str] = list(do_pins) - channels = [] - for i, pin in enumerate(self._analog_pins): - channels.append(ChannelConfig( - channel_id=pin, name=pin, unit="V", - min_value=0.0, max_value=5.0, - color=_ANALOG_COLORS[i % len(_ANALOG_COLORS)], - )) - for i in range(len(self._di_pins)): - channels.append(ChannelConfig( - channel_id=f"di{i}", name=f"DI {i}", unit="", - min_value=0.0, max_value=1.0, - color=_DI_COLORS[i % len(_DI_COLORS)], - )) - for i in range(len(self._do_pins)): - channels.append(ChannelConfig( - channel_id=f"do{i}", name=f"DO {i}", unit="", - min_value=0.0, max_value=1.0, - writable=True, - color=_DO_COLORS[i % len(_DO_COLORS)], - )) + self._ai_map = {f"ai{i}": pin for i, pin in enumerate(self._analog_pins)} + self._di_map = {f"di{i}": pin for i, pin in enumerate(self._di_pins)} + self._do_map = {f"do{i}": pin for i, pin in enumerate(self._do_pins)} + + channels = self._build_channels() _desc = ArduinoLayer.describe_port(port) if not simulate else "" if _desc: @@ -121,22 +106,45 @@ class ArduinoDevice(BaseDevice): device_id=device_id, name=_name, device_type=self.DEVICE_TYPE, - description=(f"Arduino — {len(self._analog_pins)} analog, " - f"{len(self._di_pins)} DI, {len(self._do_pins)} DO"), + description=self._description(), manufacturer="Arduino", icon=self.ICON, - channels=channels, + signals=channels, ) super().__init__(info) self._output_state: Dict[str, int] = {f"do{i}": 0 for i in range(len(self._do_pins))} - self._di_state: Dict[str, int] = {f"di{i}": 0 for i in range(len(self._di_pins))} - - self._di_map = {f"di{i}": pin for i, pin in enumerate(self._di_pins)} - self._do_map = {f"do{i}": pin for i, pin in enumerate(self._do_pins)} + self._di_state: Dict[str, int] = {f"di{i}": 0 for i in range(len(self._di_pins))} self._layer = self._make_layer() + def _description(self) -> str: + return (f"Arduino — {len(self._analog_pins)} analog, " + f"{len(self._di_pins)} DI, {len(self._do_pins)} DO") + + def _build_channels(self) -> List[SignalConfig]: + channels = [] + for i, pin in enumerate(self._analog_pins): + channels.append(SignalConfig( + signal_id=f"ai{i}", name=pin, unit="V", + min_value=0.0, max_value=5.0, + color=_ANALOG_COLORS[i % len(_ANALOG_COLORS)], + )) + for i in range(len(self._di_pins)): + channels.append(SignalConfig( + signal_id=f"di{i}", name=f"DI {i}", unit="", + min_value=0.0, max_value=1.0, + color=_DI_COLORS[i % len(_DI_COLORS)], + )) + for i in range(len(self._do_pins)): + channels.append(SignalConfig( + signal_id=f"do{i}", name=f"DO {i}", unit="", + min_value=0.0, max_value=1.0, + writable=True, + color=_DO_COLORS[i % len(_DO_COLORS)], + )) + return channels + def _make_layer(self): from api_layers.port_registry import port_registry return port_registry.get_layer( @@ -177,10 +185,12 @@ class ArduinoDevice(BaseDevice): def read_channels(self) -> Dict[str, float]: raw = self._layer.read() result = {} - for ch in self.info.channels: - cid = ch.channel_id - if cid in raw: - result[cid] = raw[cid] + for ch in self.info.signals: + cid = ch.signal_id + if cid.startswith("ai"): + pin = self._ai_map.get(cid) + if pin and pin in raw: + result[cid] = raw[pin] elif cid.startswith("di"): pin = self._di_map.get(cid) if pin and pin in raw: @@ -244,15 +254,25 @@ class ArduinoDevice(BaseDevice): di_pins: List[str] = None, do_pins: List[str] = None, ) -> None: - """Update active pin mapping and send config commands if hardware-connected.""" + """Update active pin mapping, rebuild channels, and send config + commands if hardware-connected. Caller is responsible for emitting + a `device_reconfigured` signal so the acquisition engine, profile, + and UI pick up the new channel list (see ui/windows/config_window.py).""" if analog_pins is not None: self._analog_pins = list(analog_pins) + self._ai_map = {f"ai{i}": p for i, p in enumerate(analog_pins)} if di_pins is not None: self._di_pins = list(di_pins) self._di_map = {f"di{i}": p for i, p in enumerate(di_pins)} if do_pins is not None: self._do_pins = list(do_pins) self._do_map = {f"do{i}": p for i, p in enumerate(do_pins)} + + self.info.signals = self._build_channels() + self.info.description = self._description() + self._output_state = {f"do{i}": self._output_state.get(f"do{i}", 0) for i in range(len(self._do_pins))} + self._di_state = {f"di{i}": self._di_state.get(f"di{i}", 0) for i in range(len(self._di_pins))} + if self.status == DeviceStatus.CONNECTED: self._layer.configure_pins( di_pins=self._di_pins, @@ -329,27 +349,40 @@ class ArduinoConfigWidget(QWidget): # ── Pin Mapping ─────────────────────────────────────────────────── pin_grp = QGroupBox("Pin Mapping") - pin_form = QFormLayout(pin_grp) - pin_form.setContentsMargins(10, 16, 10, 10) + pin_outer = QVBoxLayout(pin_grp) + pin_outer.setContentsMargins(10, 16, 10, 10) + pin_outer.setSpacing(8) + + pin_cols = QHBoxLayout() + pin_cols.setSpacing(12) + analog_grp = QGroupBox("Analog Pins") + analog_form = QFormLayout(analog_grp) self.analog_pins_edit = QLineEdit(", ".join(p[1:] for p in self.device._analog_pins)) - self.analog_pins_edit.setPlaceholderText("e.g. 0, 1, 2, 3 (indices into A0–A5)") - pin_form.addRow("Analog Input Pins:", self.analog_pins_edit) + self.analog_pins_edit.setPlaceholderText("e.g. 0, 1, 2, 3") + analog_form.addRow("Input:", self.analog_pins_edit) + pin_cols.addWidget(analog_grp) + digital_grp = QGroupBox("Digital Pins") + digital_form = QFormLayout(digital_grp) self.di_pins_edit = QLineEdit(", ".join(p[1:] for p in self.device._di_pins)) self.di_pins_edit.setPlaceholderText("e.g. 2, 3, 8") - pin_form.addRow("Digital Input Pins:", self.di_pins_edit) - + digital_form.addRow("Input:", self.di_pins_edit) self.do_pins_edit = QLineEdit(", ".join(p[1:] for p in self.device._do_pins)) - self.do_pins_edit.setPlaceholderText("e.g. 5, 6, 7, 9") - pin_form.addRow("Digital Output Pins:", self.do_pins_edit) + self.do_pins_edit.setPlaceholderText("e.g. 5, 6, 7") + digital_form.addRow("Output:", self.do_pins_edit) + pin_cols.addWidget(digital_grp) + + pin_outer.addLayout(pin_cols) hint = QLabel("Changes take effect on Apply & Reconnect.\n" "Analog: enter A-pin indices (0=A0, 1=A1, …). " - "Digital: enter pin numbers.") + "Digital: enter pin numbers. " + "For analog-style output, drive a PWM-capable digital " + "output pin (e.g. D9) with a duty-cycle value.") hint.setObjectName("traceSource") hint.setWordWrap(True) - pin_form.addRow(hint) + pin_outer.addWidget(hint) root.addWidget(pin_grp) @@ -417,19 +450,18 @@ class ArduinoConfigWidget(QWidget): self.port_edit.setText(port) def _apply(self): - # Parse pin fields and update device before reconnect so connect() sends correct APIN/DPIN + # Parse pin fields and remap (rebuilds device.info.signals) before + # reconnect, so connect() sends correct APIN/DPIN and the acquisition + # engine picks up the new channel list once this dialog closes. analog = _parse_analog_pin_edit(self.analog_pins_edit.text()) di = _parse_pin_edit(self.di_pins_edit.text()) do = _parse_pin_edit(self.do_pins_edit.text()) - if analog: - self.device._analog_pins = analog - if di: - self.device._di_pins = di - self.device._di_map = {f"di{i}": p for i, p in enumerate(di)} - if do: - self.device._do_pins = do - self.device._do_map = {f"do{i}": p for i, p in enumerate(do)} + self.device.remap_pins( + analog_pins=analog or None, + di_pins=di or None, + do_pins=do or None, + ) self.device.switch_backend( simulate=self.sim_chk.isChecked(), port=self.port_edit.text().strip() or self.device._port, diff --git a/devices/base_device.py b/devices/base_device.py index 9803867..e3eed65 100644 --- a/devices/base_device.py +++ b/devices/base_device.py @@ -20,17 +20,19 @@ class DeviceStatus(Enum): @dataclass -class ChannelConfig: - """Configuration for a single I/O channel.""" - channel_id: str +class SignalConfig: + """Configuration for a single physical Signal — one I/O stream coming + off a Device (e.g. an Arduino analog pin, a digital line). Contrast + with a Channel (core/signal_processor.py), which takes one or more + Signals and applies math/filters to produce a new derived stream.""" + signal_id: str name: str unit: str = "" min_value: float = 0.0 max_value: float = 100.0 enabled: bool = True - is_output: bool = False color: str = "#00d4ff" - writable: bool = False # True if this channel accepts write_channel() calls + writable: bool = False # True if this signal accepts write_channel() calls extra: Dict[str, Any] = field(default_factory=dict) @@ -39,13 +41,13 @@ class DeviceInfo: """Metadata describing a device module.""" device_id: str name: str - device_type: str # "analog_input" | "digital_io" | "serial" | "temperature" + device_type: str # "arduino" | "serial" | "nidaqmx" | ... description: str = "" manufacturer: str = "" model: str = "" version: str = "1.0.0" icon: str = "⚙" - channels: List[ChannelConfig] = field(default_factory=list) + signals: List[SignalConfig] = field(default_factory=list) class BaseDevice(ABC): @@ -75,7 +77,7 @@ class BaseDevice(ABC): @abstractmethod def read_channels(self) -> Dict[str, float]: - """Return {channel_id: value} for all enabled channels.""" + """Return {signal_id: value} for all enabled signals.""" @abstractmethod def write_channel(self, channel_id: str, value: Any) -> bool: @@ -97,8 +99,8 @@ class BaseDevice(ABC): except Exception: pass - def get_channel(self, channel_id: str) -> Optional[ChannelConfig]: - return next((c for c in self.info.channels if c.channel_id == channel_id), None) + def get_signal(self, signal_id: str) -> Optional[SignalConfig]: + return next((s for s in self.info.signals if s.signal_id == signal_id), None) def __repr__(self): return f"<{self.__class__.__name__} id={self.info.device_id} status={self.status.value}>" diff --git a/devices/digital_io.py b/devices/digital_io.py deleted file mode 100644 index 7cda61f..0000000 --- a/devices/digital_io.py +++ /dev/null @@ -1,334 +0,0 @@ -""" -devices/digital_io.py - -Digital I/O device module. Backends: NI-DAQmx or Arduino. - -NI backend – uses nidaqmx digital line tasks (P0.0..P0.7) -Arduino – uses ArduinoLayer digital pin reads; writes via W:Dxx:val -""" - -import random -import time -from typing import Any, Dict - -from PyQt6.QtWidgets import ( - QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, - QCheckBox, QPushButton, QLabel, QFormLayout, - QComboBox, QLineEdit, -) - -from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus - -_IN_COLORS = ["#00d4ff", "#4cc9f0", "#90e0ef", "#caf0f8", - "#0077b6", "#023e8a", "#48cae4", "#ade8f4"] -_OUT_COLORS = ["#ff6b35", "#ffcc00", "#f77f00", "#fcbf49", - "#d62828", "#e63946", "#fb8500", "#ffd166"] - - -class DigitalIODevice(BaseDevice): - DEVICE_TYPE = "digital_io" - ICON = "⬛" - - def __init__( - self, - device_id: str = "dio_0", - num_inputs: int = 8, - num_outputs: int = 8, - simulate: bool = True, - backend: str = "nidaqmx", # "nidaqmx" | "arduino" - ni_device: str = "Dev1", - ard_port: str = "COM3", - ard_baud: int = 115200, - ): - self.simulate = simulate - self.backend = backend - self._ni_device = ni_device - self._ard_port = ard_port - self._ard_baud = ard_baud - self._num_inputs = num_inputs - self._num_outputs = num_outputs - - channels = [] - for i in range(num_inputs): - channels.append(ChannelConfig( - channel_id=f"di{i}", name=f"DI {i}", unit="", - min_value=0.0, max_value=1.0, - color=_IN_COLORS[i % len(_IN_COLORS)], - )) - for i in range(num_outputs): - channels.append(ChannelConfig( - channel_id=f"do{i}", name=f"DO {i}", unit="", - min_value=0.0, max_value=1.0, - color=_OUT_COLORS[i % len(_OUT_COLORS)], - writable=True, - )) - - info = DeviceInfo( - device_id=device_id, - name=f"Digital I/O ({backend.upper()})", - device_type=self.DEVICE_TYPE, - description="Digital input/output module", - icon=self.ICON, - channels=channels, - ) - super().__init__(info) - - self._output_state: Dict[str, int] = { - f"do{i}": 0 for i in range(num_outputs) - } - self._sim_toggle: Dict[str, int] = {} - self._sim_state: Dict[str, int] = {} - - # Hardware task placeholders - self._ni_in_task = None - self._ni_out_task = None - self._ard_layer = None - - # ── BaseDevice ────────────────────────────────────────────────────── - - def connect(self) -> bool: - if self.simulate: - self.status = DeviceStatus.SIMULATED - return True - - if self.backend == "nidaqmx": - return self._ni_connect() - else: - return self._ard_connect() - - def _ni_connect(self) -> bool: - try: - import nidaqmx # type: ignore - from nidaqmx.constants import LineGrouping # type: ignore - n_in = sum(1 for c in self.info.channels if c.channel_id.startswith("di")) - n_out = sum(1 for c in self.info.channels if c.channel_id.startswith("do")) - - if n_in: - self._ni_in_task = nidaqmx.Task() - for i in range(n_in): - self._ni_in_task.di_channels.add_di_chan( - f"{self._ni_device}/port0/line{i}", - line_grouping=LineGrouping.CHAN_PER_LINE, - ) - self._ni_in_task.start() - - if n_out: - self._ni_out_task = nidaqmx.Task() - for i in range(n_out): - self._ni_out_task.do_channels.add_do_chan( - f"{self._ni_device}/port1/line{i}", - line_grouping=LineGrouping.CHAN_PER_LINE, - ) - self._ni_out_task.start() - - self.status = DeviceStatus.CONNECTED - return True - except Exception as e: - print(f"[DigitalIODevice] NI connect failed: {e}") - self.status = DeviceStatus.ERROR - return False - - def _ard_connect(self) -> bool: - from api_layers.port_registry import port_registry - # Get shared layer — won't open a second connection if AnalogInputDevice - # is already connected on the same port - self._ard_layer = port_registry.get_layer( - port=self._ard_port, - baud=self._ard_baud, - simulate=False, - ) - if self._ard_layer.is_connected: - # Port already open — just register ourselves and return OK - self.status = DeviceStatus.CONNECTED - return True - ok = self._ard_layer.connect() - self.status = DeviceStatus.CONNECTED if ok else DeviceStatus.ERROR - return ok - - def disconnect(self) -> None: - if self._ni_in_task: - try: self._ni_in_task.stop(); self._ni_in_task.close() - except Exception: pass - if self._ni_out_task: - try: self._ni_out_task.stop(); self._ni_out_task.close() - except Exception: pass - if self._ard_layer and self.backend == "arduino" and not self.simulate: - from api_layers.port_registry import port_registry - port_registry.release(self._ard_port, self._ard_baud) - self._ard_layer = None - self.status = DeviceStatus.DISCONNECTED - - def read_channels(self) -> Dict[str, float]: - if self.simulate: - return self._sim_read() - if self.backend == "nidaqmx": - return self._ni_read() - return self._ard_read() - - def _ni_read(self) -> Dict[str, float]: - result = {} - try: - if self._ni_in_task: - vals = self._ni_in_task.read() - for i, ch in enumerate(c for c in self.info.channels if c.channel_id.startswith("di")): - result[ch.channel_id] = float(vals[i] if isinstance(vals, list) else vals) - except Exception as e: - print(f"[DigitalIODevice] NI read failed: {e}") - for ch in (c for c in self.info.channels if c.channel_id.startswith("do")): - result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) - return result - - def _ard_read(self) -> Dict[str, float]: - if not self._ard_layer: - return {} - raw = self._ard_layer.read() - result = {} - for ch in self.info.channels: - if ch.channel_id in raw: - result[ch.channel_id] = raw[ch.channel_id] - elif ch.channel_id.startswith("do"): - result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) - return result - - def _sim_read(self) -> Dict[str, float]: - result = {} - for ch in self.info.channels: - if ch.channel_id.startswith("di"): - cnt = self._sim_toggle.get(ch.channel_id, 0) + 1 - if cnt >= random.randint(8, 40): - self._sim_state[ch.channel_id] = 1 - self._sim_state.get(ch.channel_id, 0) - cnt = 0 - self._sim_toggle[ch.channel_id] = cnt - result[ch.channel_id] = float(self._sim_state.get(ch.channel_id, 0)) - else: - result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) - return result - - def write_channel(self, channel_id: str, value: Any) -> bool: - self._output_state[channel_id] = int(bool(value)) - if not self.simulate: - if self.backend == "nidaqmx" and self._ni_out_task: - try: - out_chs = [c for c in self.info.channels if c.channel_id.startswith("do")] - idx = next((i for i, c in enumerate(out_chs) if c.channel_id == channel_id), None) - if idx is not None: - states = [self._output_state.get(c.channel_id, 0) for c in out_chs] - self._ni_out_task.write(states) - except Exception as e: - print(f"[DigitalIODevice] NI write failed: {e}") - elif self.backend == "arduino" and self._ard_layer: - # Map channel ID to Arduino pin name: - # do0→D5, do1→D6, do2→D7... (matches DIGITAL_OUT in firmware) - # Or the channel_id itself if it already looks like D5 - pin = channel_id if channel_id.upper().startswith("D") else f"D{5 + int(channel_id.replace('do',''))}" - self._ard_layer.digital_write(pin, int(bool(value))) - return True - - def pwm_channel(self, channel_id: str, duty_pct: float) -> bool: - """Send a PWM command to an Arduino output pin (0–100%).""" - if self.backend == "arduino" and self._ard_layer and not self.simulate: - pin = channel_id if channel_id.upper().startswith("D") else f"D{9 + int(channel_id.replace('do',''))}" - return self._ard_layer.pwm_write_pct(pin, duty_pct) - return False - - def set_parameter(self, name: str, value: Any) -> bool: - """Send a named parameter/setpoint to the Arduino.""" - if self.backend == "arduino" and self._ard_layer and not self.simulate: - return self._ard_layer.set_parameter(name, value) - return False - - def switch_backend(self, backend: str, simulate: bool, - ni_device: str, ard_port: str, ard_baud: int) -> None: - was_running = self.status in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED) - if was_running: - self.disconnect() - - self.backend = backend - self.simulate = simulate - self._ni_device = ni_device - self._ard_port = ard_port - self._ard_baud = ard_baud - self._ard_layer = None - self._ni_in_task = None - self._ni_out_task = None - - if was_running: - self.connect() - - def get_save_config(self) -> dict: - return { - "device_type": self.DEVICE_TYPE, - "device_id": self.info.device_id, - "name": self.info.name, - "num_inputs": self._num_inputs, - "num_outputs": self._num_outputs, - "simulate": self.simulate, - "backend": self.backend, - "ni_device": self._ni_device, - "ard_port": self._ard_port, - "ard_baud": self._ard_baud, - } - - def get_config_widget(self) -> QWidget: - return DigitalIOConfigWidget(self) - - -class DigitalIOConfigWidget(QWidget): - def __init__(self, device: DigitalIODevice): - super().__init__() - self.device = device - self._build() - - def _build(self): - root = QVBoxLayout(self) - root.setContentsMargins(8, 8, 8, 8) - self.setMinimumWidth(380) - - be_grp = QGroupBox("Backend") - be_form = QFormLayout(be_grp) - self.be_cb = QComboBox() - self.be_cb.addItems(["nidaqmx", "arduino"]) - self.be_cb.setCurrentText(self.device.backend) - be_form.addRow("Backend:", self.be_cb) - self.sim_chk = QCheckBox("Simulate") - self.sim_chk.setChecked(self.device.simulate) - from core.app_settings import is_developer_mode - self.sim_chk.setVisible(is_developer_mode()) - be_form.addRow(self.sim_chk) - - self.ni_dev_edit = QLineEdit(self.device._ni_device) - be_form.addRow("NI Device:", self.ni_dev_edit) - - self.ard_port_edit = QLineEdit(self.device._ard_port) - self.ard_port_edit.setPlaceholderText("e.g. /dev/ttyUSB0 or COM3") - be_form.addRow("Arduino Port:", self.ard_port_edit) - - self.ard_baud_cb = QComboBox() - self.ard_baud_cb.addItems(["9600", "19200", "57600", "115200", "230400"]) - self.ard_baud_cb.setCurrentText(str(self.device._ard_baud)) - be_form.addRow("Baud Rate:", self.ard_baud_cb) - - apply_btn = QPushButton("Apply & Reconnect") - apply_btn.setObjectName("applyButton") - apply_btn.clicked.connect(self._apply) - be_form.addRow(apply_btn) - - self.be_cb.currentTextChanged.connect(self._update_visibility) - root.addWidget(be_grp) - self._update_visibility(self.device.backend) - root.addStretch() - - def _update_visibility(self, backend: str = ""): - backend = backend or self.be_cb.currentText() - self.ni_dev_edit.setVisible(backend == "nidaqmx") - self.ard_port_edit.setVisible(backend == "arduino") - self.ard_baud_cb.setVisible(backend == "arduino") - - def _apply(self): - self.device.switch_backend( - backend=self.be_cb.currentText(), - simulate=self.sim_chk.isChecked(), - ni_device=self.ni_dev_edit.text().strip() or "Dev1", - ard_port=self.ard_port_edit.text().strip(), - ard_baud=int(self.ard_baud_cb.currentText()), - ) diff --git a/devices/nidaqmx_device.py b/devices/nidaqmx_device.py index 9ebec0f..652b7e2 100644 --- a/devices/nidaqmx_device.py +++ b/devices/nidaqmx_device.py @@ -13,7 +13,7 @@ from PyQt6.QtWidgets import ( ) from PyQt6.QtCore import Qt, QThread, pyqtSignal -from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus +from devices.base_device import BaseDevice, SignalConfig, DeviceInfo, DeviceStatus from api_layers.nidaqmx_layer import NidaqmxLayer _ANALOG_COLORS = [ @@ -53,20 +53,20 @@ class NidaqmxDevice(BaseDevice): channels = [] for i in range(num_analog): - channels.append(ChannelConfig( - channel_id=f"ai{i}", name=f"AI{i}", unit="V", + channels.append(SignalConfig( + signal_id=f"ai{i}", name=f"AI{i}", unit="V", min_value=min_v, max_value=max_v, color=_ANALOG_COLORS[i % len(_ANALOG_COLORS)], )) for i in range(num_di): - channels.append(ChannelConfig( - channel_id=f"di{i}", name=f"DI {i}", unit="", + channels.append(SignalConfig( + signal_id=f"di{i}", name=f"DI {i}", unit="", min_value=0.0, max_value=1.0, color=_DI_COLORS[i % len(_DI_COLORS)], )) for i in range(num_do): - channels.append(ChannelConfig( - channel_id=f"do{i}", name=f"DO {i}", unit="", + channels.append(SignalConfig( + signal_id=f"do{i}", name=f"DO {i}", unit="", min_value=0.0, max_value=1.0, color=_DO_COLORS[i % len(_DO_COLORS)], writable=True, @@ -79,7 +79,7 @@ class NidaqmxDevice(BaseDevice): description=f"NI-DAQmx — {num_analog} AI, {num_di} DI, {num_do} DO", manufacturer="National Instruments", icon=self.ICON, - channels=channels, + signals=channels, ) super().__init__(info) @@ -99,8 +99,8 @@ class NidaqmxDevice(BaseDevice): ok = True try: if self._num_analog > 0: - ai_ids = [ch.channel_id for ch in self.info.channels - if ch.channel_id.startswith("ai")] + ai_ids = [ch.signal_id for ch in self.info.signals + if ch.signal_id.startswith("ai")] self._ai_layer = NidaqmxLayer( device_name=self._ni_device, channels=ai_ids, @@ -179,14 +179,14 @@ class NidaqmxDevice(BaseDevice): try: if self._ni_in_task: vals = self._ni_in_task.read() - di_chs = [c for c in self.info.channels if c.channel_id.startswith("di")] + di_chs = [c for c in self.info.signals if c.signal_id.startswith("di")] for i, ch in enumerate(di_chs): - result[ch.channel_id] = float(vals[i] if isinstance(vals, list) else vals) + result[ch.signal_id] = float(vals[i] if isinstance(vals, list) else vals) except Exception: pass - for ch in (c for c in self.info.channels if c.channel_id.startswith("do")): - result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) + for ch in (c for c in self.info.signals if c.signal_id.startswith("do")): + result[ch.signal_id] = float(self._output_state.get(ch.signal_id, 0)) return result @@ -194,12 +194,12 @@ class NidaqmxDevice(BaseDevice): import math, time t = time.time() result = {} - for i, ch in enumerate([c for c in self.info.channels if c.channel_id.startswith("ai")]): - result[ch.channel_id] = math.sin(t + i) * (self._max_v * 0.5) - for ch in (c for c in self.info.channels if c.channel_id.startswith("di")): - result[ch.channel_id] = 0.0 - for ch in (c for c in self.info.channels if c.channel_id.startswith("do")): - result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) + for i, ch in enumerate([c for c in self.info.signals if c.signal_id.startswith("ai")]): + result[ch.signal_id] = math.sin(t + i) * (self._max_v * 0.5) + for ch in (c for c in self.info.signals if c.signal_id.startswith("di")): + result[ch.signal_id] = 0.0 + for ch in (c for c in self.info.signals if c.signal_id.startswith("do")): + result[ch.signal_id] = float(self._output_state.get(ch.signal_id, 0)) return result def write_channel(self, channel_id: str, value: Any) -> bool: @@ -208,8 +208,8 @@ class NidaqmxDevice(BaseDevice): self._output_state[cid] = int(bool(value)) if not self.simulate and self._ni_out_task: try: - do_chs = [c for c in self.info.channels if c.channel_id.startswith("do")] - states = [self._output_state.get(c.channel_id, 0) for c in do_chs] + do_chs = [c for c in self.info.signals if c.signal_id.startswith("do")] + states = [self._output_state.get(c.signal_id, 0) for c in do_chs] self._ni_out_task.write(states) except Exception as e: print(f"[NidaqmxDevice] write failed: {e}") diff --git a/devices/serial_device.py b/devices/serial_device.py index d402f09..86d5e80 100644 --- a/devices/serial_device.py +++ b/devices/serial_device.py @@ -31,7 +31,7 @@ from PyQt6.QtWidgets import ( QStackedWidget, QTableWidget, QTableWidgetItem, QTextEdit, QVBoxLayout, QWidget, ) -from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus +from devices.base_device import BaseDevice, SignalConfig, DeviceInfo, DeviceStatus from api_layers.arduino_layer import ArduinoLayer _COLORS = ["#7fff6e", "#4cc9f0", "#f72585", "#00d4ff", @@ -145,7 +145,7 @@ class SerialDevice(BaseDevice): device_type=self.DEVICE_TYPE, description=self._description(), icon=self.ICON, - channels=channels, + signals=channels, ) super().__init__(info) self._layer = self._make_layer() @@ -175,21 +175,21 @@ class SerialDevice(BaseDevice): # keep the channel's unit in sync so new plots/pickers/CSV log # headers pick up the currently-selected unit instead of a fixed # default. Does not relabel the axis of an already-open plot pane. - force_ch = self.get_channel("force") + force_ch = self.get_signal("force") if force_ch is not None: force_ch.unit = self._layer.current_unit if self._fmt not in _GENERIC_FORMATS: - # Protocol layers already key by channel_id — pass through untouched. + # Protocol layers already key by signal_id — pass through untouched. # (Write-only channels, e.g. CML "M1_VS", never appear in raw — correctly dropped.) return raw # Generic (ArduinoLayer) may use arbitrary names — remap by position. mapped: Dict[str, float] = {} raw_vals = list(raw.values()) - for i, ch in enumerate(self.info.channels): - if ch.channel_id in raw: - mapped[ch.channel_id] = raw[ch.channel_id] + for i, ch in enumerate(self.info.signals): + if ch.signal_id in raw: + mapped[ch.signal_id] = raw[ch.signal_id] elif i < len(raw_vals): - mapped[ch.channel_id] = raw_vals[i] + mapped[ch.signal_id] = raw_vals[i] return mapped def write_channel(self, channel_id: str, value: Any) -> bool: @@ -217,9 +217,9 @@ class SerialDevice(BaseDevice): cfg["port_serial"] = ArduinoLayer.port_serial_number(self._port) if self._fmt in _GENERIC_FORMATS: cfg.update({ - "num_channels": len(self.info.channels), - "channel_names": [c.channel_id for c in self.info.channels], - "units": [c.unit for c in self.info.channels], + "num_channels": len(self.info.signals), + "channel_names": [c.signal_id for c in self.info.signals], + "units": [c.unit for c in self.info.signals], }) elif self._fmt == "scpi": cfg["scpi_channels"] = self._scpi_channels @@ -274,7 +274,7 @@ class SerialDevice(BaseDevice): self._mb_channels = modbus_channels or self._mb_channels self._motors = motors or self._motors - self.info.channels = self._build_channel_configs() + self.info.signals = self._build_channel_configs() self.info.description = self._description() self._layer = self._make_layer() @@ -287,14 +287,14 @@ class SerialDevice(BaseDevice): label = _LABEL_FOR.get(self._fmt, self._fmt) return f"{self._port} @ {self._baud} [{label}]" - def _build_channel_configs(self) -> List[ChannelConfig]: + def _build_channel_configs(self) -> List[SignalConfig]: fmt = self._fmt if fmt in _GENERIC_FORMATS: names = self._chan_names or [f"sig{i}" for i in range(self._num_channels)] _units = self._units or [""] * len(names) return [ - ChannelConfig( - channel_id=names[i], name=names[i], unit=_units[i], + SignalConfig( + signal_id=names[i], name=names[i], unit=_units[i], min_value=0.0, max_value=1023.0, color=_COLORS[i % len(_COLORS)], ) @@ -302,8 +302,8 @@ class SerialDevice(BaseDevice): ] elif fmt == "scpi": return [ - ChannelConfig( - channel_id=sc["channel_id"], + SignalConfig( + signal_id=sc["channel_id"], name=sc.get("name", sc["channel_id"]), unit=sc.get("unit", ""), min_value=-1e9, max_value=1e9, @@ -314,8 +314,8 @@ class SerialDevice(BaseDevice): ] elif fmt == "modbus_rtu": return [ - ChannelConfig( - channel_id=mc["channel_id"], + SignalConfig( + signal_id=mc["channel_id"], name=mc.get("name", mc["channel_id"]), unit=mc.get("unit", ""), min_value=-1e9, max_value=1e9, @@ -328,15 +328,15 @@ class SerialDevice(BaseDevice): elif fmt == "mark10": from api_layers.protocols.mark10 import UNITS as _MARK10_UNITS return [ - ChannelConfig("force", "Force", "N", -5000.0, 5000.0, color=_COLORS[0]), - ChannelConfig("unit_code", "Unit Code", "", 0.0, float(len(_MARK10_UNITS) - 1), color=_COLORS[1]), + SignalConfig("force", "Force", "N", -5000.0, 5000.0, color=_COLORS[0]), + SignalConfig("unit_code", "Unit Code", "", 0.0, float(len(_MARK10_UNITS) - 1), color=_COLORS[1]), # Write-only action channels (no reading — for control buttons) - ChannelConfig("zero", "Zero Gauge", "", 0.0, 1.0, color=_COLORS[2], writable=True), - ChannelConfig("cycle_units", "Cycle Units", "", 0.0, 1.0, color=_COLORS[3], writable=True), + SignalConfig("zero", "Zero Gauge", "", 0.0, 1.0, color=_COLORS[2], writable=True), + SignalConfig("cycle_units", "Cycle Units", "", 0.0, 1.0, color=_COLORS[3], writable=True), ] elif fmt == "dscusb": return [ - ChannelConfig("extension", "Extension", "", -50.0, 50.0, color=_COLORS[0]), + SignalConfig("extension", "Extension", "", -50.0, 50.0, color=_COLORS[0]), ] elif fmt == "cml": _CMD_UNITS = {"TP": "counts", "TV": "counts/s", "TC": "%×10", "TS": "flags"} @@ -346,8 +346,8 @@ class SerialDevice(BaseDevice): for motor in self._motors: mid = motor.get("motor_id", "M1") for cmd in motor.get("read_cmds", ["TP", "TV", "TC"]): - channels.append(ChannelConfig( - channel_id=f"{mid}_{cmd}", + channels.append(SignalConfig( + signal_id=f"{mid}_{cmd}", name=f"{mid} {cmd}", unit=_CMD_UNITS.get(cmd, ""), min_value=-1e6, max_value=1e6, @@ -356,8 +356,8 @@ class SerialDevice(BaseDevice): color_idx += 1 # Write-only setpoint channels (no reading — for control widgets) for cmd in ("VS", "MA"): - channels.append(ChannelConfig( - channel_id=f"{mid}_{cmd}", + channels.append(SignalConfig( + signal_id=f"{mid}_{cmd}", name=f"{mid} {cmd}", unit=_WRITE_UNITS.get(cmd, ""), min_value=-1e6, max_value=1e6, @@ -367,8 +367,8 @@ class SerialDevice(BaseDevice): color_idx += 1 # Write-only action channels — motor must be enabled (ME) before VS/MA take effect for cmd, label in (("ME", "Enable"), ("MD", "Disable"), ("ST", "Stop")): - channels.append(ChannelConfig( - channel_id=f"{mid}_{cmd}", + channels.append(SignalConfig( + signal_id=f"{mid}_{cmd}", name=f"{mid} {label}", unit="", min_value=0.0, max_value=1.0, @@ -376,13 +376,13 @@ class SerialDevice(BaseDevice): writable=True, )) color_idx += 1 - return channels or [ChannelConfig("M1_TP", "M1 TP", "counts", color=_COLORS[0])] + return channels or [SignalConfig("M1_TP", "M1 TP", "counts", color=_COLORS[0])] return [] def _make_layer(self): fmt = self._fmt if fmt in _GENERIC_FORMATS: - names = [ch.channel_id for ch in self.info.channels] + names = [ch.signal_id for ch in self.info.signals] return ArduinoLayer( port=self._port, baud=self._baud, analog_pins=names, simulate=self.simulate, @@ -632,7 +632,7 @@ class _GenericPanel(QWidget): self._ch_spin = QSpinBox() self._ch_spin.setRange(1, 32) - self._ch_spin.setValue(len(device.info.channels) if device._fmt in _GENERIC_FORMATS else 4) + self._ch_spin.setValue(len(device.info.signals) if device._fmt in _GENERIC_FORMATS else 4) lay.addRow("Signals:", self._ch_spin) # ── Diagnostics ─────────────────────────────────────────────────── diff --git a/plugins/motion_capture/device.py b/plugins/motion_capture/device.py index fc1c588..ec2ca22 100644 --- a/plugins/motion_capture/device.py +++ b/plugins/motion_capture/device.py @@ -18,7 +18,7 @@ from __future__ import annotations from typing import Any, Dict, Optional from devices.base_device import ( - BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus, + BaseDevice, SignalConfig, DeviceInfo, DeviceStatus, ) @@ -38,17 +38,17 @@ class CameraDevice(BaseDevice): device_type = "camera", description = "Camera point-tracking — x/y position in pixels.", icon = "📷", - channels = [ - ChannelConfig( - channel_id = "x_pos", + signals = [ + SignalConfig( + signal_id = "x_pos", name = "X Position", unit = "px", min_value = 0.0, max_value = 1920.0, color = "#00d4ff", ), - ChannelConfig( - channel_id = "y_pos", + SignalConfig( + signal_id = "y_pos", name = "Y Position", unit = "px", min_value = 0.0, diff --git a/ui/config_dialog.py b/ui/config_dialog.py index cefee2d..a1ad625 100644 --- a/ui/config_dialog.py +++ b/ui/config_dialog.py @@ -74,5 +74,5 @@ class DeviceConfigDialog(QDialog): form.addRow("Model:", _ro(info.model)) form.addRow("Version:", _ro(info.version)) form.addRow("Status:", _ro(self.device.status.value)) - form.addRow("Signals:", _ro(len(info.channels))) + form.addRow("Signals:", _ro(len(info.signals))) return w diff --git a/ui/control_editor.py b/ui/control_editor.py index 624e4c1..40cde7e 100644 --- a/ui/control_editor.py +++ b/ui/control_editor.py @@ -214,11 +214,11 @@ class ControlEditorDialog(QDialog): self._target_cb = QComboBox() self._target_cb.addItem("— none —", userData=("", "")) for dev in self.registry.all_instances(): - for ch in dev.info.channels: + for ch in dev.info.signals: if not ch.enabled or not ch.writable: continue - label = f"{dev.info.device_id} / {ch.name} ({ch.channel_id})" - self._target_cb.addItem(label, userData=(dev.info.device_id, ch.channel_id)) + label = f"{dev.info.device_id} / {ch.name} ({ch.signal_id})" + self._target_cb.addItem(label, userData=(dev.info.device_id, ch.signal_id)) hw_form.addRow("Target Signal:", self._target_cb) self._active_low_chk = QCheckBox("Active-low output (ON sends LOW)") diff --git a/ui/control_panel.py b/ui/control_panel.py index dafeabf..f008853 100644 --- a/ui/control_panel.py +++ b/ui/control_panel.py @@ -160,7 +160,7 @@ class ControlWidget(QFrame): if dev is None: stale = True else: - ch = dev.get_channel(self.channel_id) + ch = dev.get_signal(self.channel_id) if ch is None or not ch.enabled: stale = True elif dev.status not in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED): @@ -225,8 +225,7 @@ class ControlWidget(QFrame): Write value to the linked device channel. Routing: - DigitalIODevice → write_channel() → ArduinoLayer.digital_write() → W:Dxx:val - AnalogInputDevice (arduino backend) → write_channel() → ArduinoLayer.digital_write() + ArduinoDevice (do* signal) → write_channel() → ArduinoLayer.digital_write() → W:Dxx:val Any device with no write support → logs a warning, emits signal only The actual write is dispatched to a background worker (_enqueue_write) @@ -238,7 +237,7 @@ class ControlWidget(QFrame): if self.registry and self.device_id and self.channel_id: dev = self.registry.get_instance(self.device_id) if dev: - ch = dev.get_channel(self.channel_id) + ch = dev.get_signal(self.channel_id) if ch is not None and not ch.enabled: print(f"[Control] write_channel({self.channel_id}, {value}) skipped on " f"{self.device_id} — channel is disabled") diff --git a/ui/device_panel.py b/ui/device_panel.py deleted file mode 100644 index 932f536..0000000 --- a/ui/device_panel.py +++ /dev/null @@ -1,148 +0,0 @@ -""" -ui/device_panel.py — Left sidebar: device list with status indicators. -""" - -from PyQt6.QtWidgets import ( - QWidget, QVBoxLayout, QHBoxLayout, QLabel, - QPushButton, QScrollArea, QFrame, QSizePolicy, -) -from PyQt6.QtCore import Qt, pyqtSignal, QTimer -from devices.base_device import DeviceStatus -from devices.device_registry import DeviceRegistry -from core.acquisition import AcquisitionEngine - - -_STATUS_STYLE = { - DeviceStatus.CONNECTED: "color:#22c55e;", - DeviceStatus.SIMULATED: "color:#3b82f6;", - DeviceStatus.DISCONNECTED: "color:#475569;", - DeviceStatus.CONNECTING: "color:#f59e0b;", - DeviceStatus.ERROR: "color:#ef4444;", -} -_STATUS_LABEL = { - DeviceStatus.CONNECTED: "CONNECTED", - DeviceStatus.SIMULATED: "SIMULATED", - DeviceStatus.DISCONNECTED: "OFFLINE", - DeviceStatus.CONNECTING: "CONNECTING", - DeviceStatus.ERROR: "ERROR", -} - - -class DeviceCard(QFrame): - config_requested = pyqtSignal(str) - - def __init__(self, device): - super().__init__() - self.device = device - self.setObjectName("deviceCard") - self._build() - - def _build(self): - layout = QVBoxLayout(self) - layout.setContentsMargins(10, 10, 10, 10) - layout.setSpacing(5) - - # Header row: icon + name + status dot - hdr = QHBoxLayout() - icon_lbl = QLabel(self.device.info.icon) - icon_lbl.setObjectName("deviceIcon") - hdr.addWidget(icon_lbl) - - name_lbl = QLabel(self.device.info.name) - name_lbl.setObjectName("deviceName") - hdr.addWidget(name_lbl, 1) - - self._dot = QLabel("●") - self._dot.setStyleSheet(_STATUS_STYLE.get(self.device.status, "color:#475569;") + " font-size:10px;") - hdr.addWidget(self._dot) - layout.addLayout(hdr) - - # Device ID - id_lbl = QLabel(self.device.info.device_id) - id_lbl.setObjectName("deviceSub") - layout.addWidget(id_lbl) - - # Status text - self._status_lbl = QLabel(_STATUS_LABEL.get(self.device.status, "UNKNOWN")) - style = _STATUS_STYLE.get(self.device.status, "color:#475569;") - self._status_lbl.setStyleSheet( - style + " font-family:'IBM Plex Mono',monospace; font-size:10px; font-weight:700; letter-spacing:1px;" - ) - layout.addWidget(self._status_lbl) - - # Channel count - n = sum(1 for c in self.device.info.channels if c.enabled) - ch_lbl = QLabel(f"{n} ch · {self.device.info.device_type}") - ch_lbl.setObjectName("deviceChannelCount") - layout.addWidget(ch_lbl) - - # Config button - cfg = QPushButton("Configure") - cfg.setObjectName("configButton") - cfg.clicked.connect(lambda: self.config_requested.emit(self.device.info.device_id)) - layout.addWidget(cfg) - - def refresh(self): - dot_style = _STATUS_STYLE.get(self.device.status, "color:#475569;") + " font-size:10px;" - label_style = _STATUS_STYLE.get(self.device.status, "color:#475569;") + \ - " font-family:'IBM Plex Mono',monospace; font-size:10px; font-weight:700; letter-spacing:1px;" - self._dot.setStyleSheet(dot_style) - self._status_lbl.setText(_STATUS_LABEL.get(self.device.status, "UNKNOWN")) - self._status_lbl.setStyleSheet(label_style) - - -class DevicePanel(QWidget): - config_requested = pyqtSignal(str) - - def __init__(self, registry: DeviceRegistry, engine: AcquisitionEngine): - super().__init__() - self.registry = registry - self.engine = engine - self._cards = {} - self._build() - - # Refresh status every 2 s - self._timer = QTimer(self) - self._timer.setInterval(2000) - self._timer.timeout.connect(self._refresh_status) - self._timer.start() - - def _build(self): - layout = QVBoxLayout(self) - layout.setContentsMargins(0, 0, 0, 0) - layout.setSpacing(0) - - hdr = QLabel(" DEVICES") - hdr.setObjectName("panelHeader") - hdr.setMinimumHeight(28) - layout.addWidget(hdr) - - scroll = QScrollArea() - scroll.setObjectName("deviceScroll") - scroll.setWidgetResizable(True) - scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff) - - self._container = QWidget() - self._inner = QVBoxLayout(self._container) - self._inner.setContentsMargins(8, 8, 8, 8) - self._inner.setSpacing(8) - self._inner.addStretch() - - scroll.setWidget(self._container) - layout.addWidget(scroll) - - def refresh(self): - for card in self._cards.values(): - self._inner.removeWidget(card) - card.deleteLater() - self._cards.clear() - - for dev in self.registry.all_instances(): - card = DeviceCard(dev) - card.config_requested.connect(self.config_requested.emit) - self._inner.insertWidget(self._inner.count() - 1, card) - self._cards[dev.info.device_id] = card - - def _refresh_status(self): - for dev_id, card in self._cards.items(): - card.refresh() diff --git a/ui/devices_window.py b/ui/devices_window.py deleted file mode 100644 index 0548330..0000000 --- a/ui/devices_window.py +++ /dev/null @@ -1,224 +0,0 @@ -""" -ui/devices_window.py - -Floating "Devices" settings window. -Opens from the toolbar ⚙ DEVICES button; can stay open alongside the main window. -Contains: device status cards, Add Device, and per-device Configure. -""" - -from PyQt6.QtWidgets import ( - QWidget, QVBoxLayout, QHBoxLayout, QLabel, - QPushButton, QScrollArea, QFrame, QSizePolicy, - QDialog, -) -from PyQt6.QtCore import Qt, pyqtSignal, QTimer -from PyQt6.QtGui import QCloseEvent - -from devices.base_device import DeviceStatus -from devices.device_registry import DeviceRegistry -from core.acquisition import AcquisitionEngine - - -_STATUS_STYLE = { - DeviceStatus.CONNECTED: "color:#22c55e;", - DeviceStatus.SIMULATED: "color:#3b82f6;", - DeviceStatus.DISCONNECTED: "color:#475569;", - DeviceStatus.CONNECTING: "color:#f59e0b;", - DeviceStatus.ERROR: "color:#ef4444;", -} -_STATUS_LABEL = { - DeviceStatus.CONNECTED: "CONNECTED", - DeviceStatus.SIMULATED: "SIMULATED", - DeviceStatus.DISCONNECTED: "OFFLINE", - DeviceStatus.CONNECTING: "CONNECTING", - DeviceStatus.ERROR: "ERROR", -} - - -class DeviceCard(QFrame): - config_requested = pyqtSignal(str) - - def __init__(self, device): - super().__init__() - self.device = device - self.setObjectName("deviceCard") - self._build() - - def _build(self): - layout = QVBoxLayout(self) - layout.setContentsMargins(10, 10, 10, 10) - layout.setSpacing(5) - - # Header: icon + name + status dot - hdr = QHBoxLayout() - icon_lbl = QLabel(self.device.info.icon) - icon_lbl.setObjectName("deviceIcon") - hdr.addWidget(icon_lbl) - - name_lbl = QLabel(self.device.info.name) - name_lbl.setObjectName("deviceName") - hdr.addWidget(name_lbl, 1) - - self._dot = QLabel("●") - self._dot.setStyleSheet( - _STATUS_STYLE.get(self.device.status, "color:#475569;") + " font-size:10px;" - ) - hdr.addWidget(self._dot) - layout.addLayout(hdr) - - # ID - id_lbl = QLabel(self.device.info.device_id) - id_lbl.setObjectName("deviceSub") - layout.addWidget(id_lbl) - - # Status text - self._status_lbl = QLabel(_STATUS_LABEL.get(self.device.status, "UNKNOWN")) - self._apply_status_style() - layout.addWidget(self._status_lbl) - - # Channel count · type - n = sum(1 for c in self.device.info.channels if c.enabled) - ch_lbl = QLabel(f"{n} ch · {self.device.info.device_type}") - ch_lbl.setObjectName("deviceChannelCount") - layout.addWidget(ch_lbl) - - # Configure button - cfg = QPushButton("Configure") - cfg.setObjectName("configButton") - cfg.clicked.connect(lambda: self.config_requested.emit(self.device.info.device_id)) - layout.addWidget(cfg) - - def refresh(self): - dot_style = _STATUS_STYLE.get(self.device.status, "color:#475569;") + " font-size:10px;" - self._dot.setStyleSheet(dot_style) - self._status_lbl.setText(_STATUS_LABEL.get(self.device.status, "UNKNOWN")) - self._apply_status_style() - - def _apply_status_style(self): - s = _STATUS_STYLE.get(self.device.status, "color:#475569;") - self._status_lbl.setStyleSheet( - s + " font-family:'IBM Plex Mono',monospace;" - " font-size:10px; font-weight:700; letter-spacing:1px;" - ) - - -class DevicesWindow(QWidget): - """ - Standalone floating window for device management. - - Signals - ------- - config_requested(device_id) — user clicked Configure on a card - device_added() — a new device was added via the dialog - closed() — window was closed (so toolbar btn can untoggle) - """ - - config_requested = pyqtSignal(str) - device_added = pyqtSignal() - closed = pyqtSignal() - - def __init__(self, registry: DeviceRegistry, engine: AcquisitionEngine, parent=None): - super().__init__( - parent, - # Acts like a real standalone window but stays on top of parent - Qt.WindowType.Window | - Qt.WindowType.Tool, - ) - self.registry = registry - self.engine = engine - self._cards: dict = {} - - self.setWindowTitle("Devices") - self.setMinimumSize(300, 480) - self.resize(320, 600) - self._build() - self.refresh() - - self._timer = QTimer(self) - self._timer.setInterval(2000) - self._timer.timeout.connect(self._refresh_status) - self._timer.start() - - # ── Build ───────────────────────────────────────────────────────────── - - def _build(self): - root = QVBoxLayout(self) - root.setContentsMargins(0, 0, 0, 0) - root.setSpacing(0) - - # ── Title bar row ────────────────────────────────────────────── - title_bar = QWidget() - title_bar.setObjectName("devWindowTitleBar") - title_bar.setFixedHeight(40) - tb_lay = QHBoxLayout(title_bar) - tb_lay.setContentsMargins(12, 0, 8, 0) - - title_lbl = QLabel("⚙ DEVICES") - title_lbl.setObjectName("devWindowTitle") - tb_lay.addWidget(title_lbl, 1) - - self._add_btn = QPushButton("＋ Add") - self._add_btn.setObjectName("addDeviceButton") - self._add_btn.clicked.connect(self._on_add) - tb_lay.addWidget(self._add_btn) - - root.addWidget(title_bar) - - # ── Divider ──────────────────────────────────────────────────── - div = QFrame() - div.setFrameShape(QFrame.Shape.HLine) - div.setObjectName("devWindowDivider") - root.addWidget(div) - - # ── Scrollable device cards ──────────────────────────────────── - scroll = QScrollArea() - scroll.setObjectName("deviceScroll") - scroll.setWidgetResizable(True) - scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff) - - self._container = QWidget() - self._inner = QVBoxLayout(self._container) - self._inner.setContentsMargins(10, 10, 10, 10) - self._inner.setSpacing(10) - self._inner.addStretch() - - scroll.setWidget(self._container) - root.addWidget(scroll) - - # ── Public API ──────────────────────────────────────────────────────── - - def refresh(self): - """Rebuild all device cards from current registry state.""" - for card in self._cards.values(): - self._inner.removeWidget(card) - card.deleteLater() - self._cards.clear() - - for dev in self.registry.all_instances(): - card = DeviceCard(dev) - card.config_requested.connect(self.config_requested.emit) - self._inner.insertWidget(self._inner.count() - 1, card) - self._cards[dev.info.device_id] = card - - # ── Slots ──────────────────────────────────────────────────────────── - - def _on_add(self): - from ui.add_device_dialog import AddDeviceDialog - from ui.config_dialog import DeviceConfigDialog - dlg = AddDeviceDialog(self.registry, self) - if dlg.exec(): - dev = dlg.created_device - if dev: - dev.connect() - self.registry.add_instance(dev) - self.engine.add_device(dev) - self.refresh() - self.device_added.emit() - - def _refresh_status(self): - for card in self._cards.values(): - card.refresh() - - def closeEvent(self, event: QCloseEvent): - self.closed.emit() - event.accept() diff --git a/ui/main_window.py b/ui/main_window.py index b632535..0e21b2e 100644 --- a/ui/main_window.py +++ b/ui/main_window.py @@ -93,9 +93,6 @@ class _DockTitleBar(QWidget): lay.addWidget(close_btn) -from devices.arduino_device import ArduinoDevice -from devices.nidaqmx_device import NidaqmxDevice -from devices.serial_device import SerialDevice from devices.device_registry import DeviceRegistry from core.acquisition import AcquisitionEngine from core.signal_processor import SignalProcessor @@ -148,24 +145,11 @@ class MainWindow(QMainWindow): self._wheel_blocker = _WheelBlocker(self) QApplication.instance().installEventFilter(self._wheel_blocker) - self._init_demo_devices() self._build_ui() self._connect_signals() self._init_plugins() self._apply_settings_on_startup() - # ── Demo ────────────────────────────────────────────────────────────── - - def _init_demo_devices(self): - for dev in [ - NidaqmxDevice(device_id="ni_0", num_analog=4, num_di=2, num_do=4, simulate=True), - ArduinoDevice(device_id="ard_0", num_analog=4, num_di=2, num_do=4, simulate=True), - SerialDevice (device_id="ser_0", num_channels=3, simulate=True), - ]: - dev.connect() - self.registry.add_instance(dev) - self.engine.add_device(dev) - # ── UI ──────────────────────────────────────────────────────────────── def _build_ui(self): @@ -276,7 +260,6 @@ class MainWindow(QMainWindow): # ── Control Panel dock ──────────────────────────────────────────── self._ctrl = ControlPanel(self.registry, processor=self.processor, engine=self.engine) self._ctrl.setMinimumWidth(200); self._ctrl.setMaximumWidth(340) - self._ctrl._add_demo_widgets() self._ctrl_dock = QDockWidget("Control Panel", self) self._ctrl_dock.setObjectName("controlPanelDock") @@ -561,6 +544,7 @@ class MainWindow(QMainWindow): self._win_config.channel_unit_changed.connect(self._on_channel_unit_changed) self._win_config.pipeline_changed.connect(lambda: None) self._win_config.derived_changed.connect(self._on_derived_changed) + self._win_config.channel_renamed.connect(self._on_channel_renamed) self._win_config.layout_applied.connect(self._chart.apply_layout) self._win_config.tabs.setCurrentIndex(tab) self._show_win(self._win_config, "right") @@ -762,6 +746,13 @@ class MainWindow(QMainWindow): if self._win_config: self._win_config.refresh_plot(self._chart._cfg) + def _on_channel_renamed(self, old_dev: str, old_id: str, new_dev: str, new_id: str): + """A Signal or Channel's id changed — rewrite every plot trace, + control target, and filter pipeline that still points at the old + (device_id, channel_id) pair so it doesn't silently stop updating.""" + from core.reference_registry import propagate_rename + propagate_rename(self, (old_dev, old_id), (new_dev, new_id)) + # ── Run / Log ───────────────────────────────────────────────────────── def _toggle_run(self, c: bool): diff --git a/ui/namespace_panel.py b/ui/namespace_panel.py index 44750c6..9616db7 100644 --- a/ui/namespace_panel.py +++ b/ui/namespace_panel.py @@ -107,9 +107,9 @@ class NamespacePanel(QWidget): def _desired_keys(self): keys = [] for dev in self.registry.all_instances(): - for ch in dev.info.channels: + for ch in dev.info.signals: if ch.enabled: - keys.append(("signal", dev.info.device_id, ch.channel_id)) + keys.append(("signal", dev.info.device_id, ch.signal_id)) for dc in self.processor.get_derived(): keys.append(("channel", dc.channel_id)) for name in self.processor.list_vars(): @@ -131,7 +131,7 @@ class NamespacePanel(QWidget): if kind == "signal": _, dev_id, ch_id = key dev = self.registry.get_instance(dev_id) - ch = dev.get_channel(ch_id) if dev else None + ch = dev.get_signal(ch_id) if dev else None name = ch.name if ch else ch_id is_wr = bool(ch and ch.writable) self._set_row(row, "Signal", ch_id, name, "WRITE" if is_wr else "READ", diff --git a/ui/plot_builder.py b/ui/plot_builder.py deleted file mode 100644 index 7445bda..0000000 --- a/ui/plot_builder.py +++ /dev/null @@ -1,510 +0,0 @@ -""" -ui/plot_builder.py - -Plot Builder — floating window for operators to: - • Create named subplot rows - • Drag channels (physical + derived) into any row - • Name each axis, set Y range, grid, color - • Reorder rows with ▲▼ arrows - • Toggle per-trace visibility - • Export / Import config as JSON - -Emits config_applied(ChartConfig) to rebuild the strip chart. -""" - -from __future__ import annotations -import json -from copy import deepcopy -from dataclasses import dataclass, field, asdict -from typing import List, Optional, Tuple - -from PyQt6.QtWidgets import ( - QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, - QScrollArea, QFrame, QLineEdit, QDoubleSpinBox, - QCheckBox, QComboBox, QColorDialog, QToolButton, - QSizePolicy, QFileDialog, QMessageBox, QSpinBox, -) -from PyQt6.QtCore import Qt, pyqtSignal -from PyQt6.QtGui import QColor, QFont, QCloseEvent - -from devices.device_registry import DeviceRegistry -from core.signal_processor import SignalProcessor - - -# ══════════════════════════════════════════════════════════════════════════════ -# Data model (same as before, kept self-contained for clarity) -# ══════════════════════════════════════════════════════════════════════════════ - -@dataclass -class TraceSpec: - device_id: str - channel_id: str - label: str = "" - color: str = "#00d4ff" - width: float = 1.8 - visible: bool = True - style: str = "solid" # solid | dash | dot - -@dataclass -class PlotSpec: - title: str = "Plot" - y_label: str = "" - y_auto: bool = True - y_min: float = -10.0 - y_max: float = 10.0 - grid: bool = True - height: int = 1 # relative weight in splitter - traces: List[TraceSpec] = field(default_factory=list) - -@dataclass -class LayoutConfig: - plots: List[PlotSpec] = field(default_factory=list) - time_window_s: float = 30.0 - link_x: bool = True - show_legend: bool = True - - def to_json(self) -> str: - return json.dumps(asdict(self), indent=2) - - @staticmethod - def from_json(s: str) -> "LayoutConfig": - d = json.loads(s) - plots = [] - for p in d.get("plots", []): - traces = [TraceSpec(**t) for t in p.pop("traces", [])] - plots.append(PlotSpec(**p, traces=traces)) - d.pop("plots") - return LayoutConfig(**d, plots=plots) - - -def build_default_layout(registry: DeviceRegistry, - processor: SignalProcessor) -> LayoutConfig: - cfg = LayoutConfig() - for dev in registry.all_instances(): - enabled = [ch for ch in dev.info.channels if ch.enabled] - if not enabled: continue - p = PlotSpec(title=dev.info.name, - y_label=enabled[0].unit if enabled else "") - for ch in enabled: - p.traces.append(TraceSpec( - device_id=dev.info.device_id, - channel_id=ch.channel_id, - label=ch.name, color=ch.color, - )) - if p.traces: cfg.plots.append(p) - # Add derived channels - for dc in processor.get_derived(): - p = PlotSpec(title=dc.name, y_label=dc.unit) - p.traces.append(TraceSpec( - device_id="derived", channel_id=dc.channel_id, - label=dc.name, color=dc.color, - )) - cfg.plots.append(p) - return cfg - - -# ══════════════════════════════════════════════════════════════════════════════ -# Color swatch button -# ══════════════════════════════════════════════════════════════════════════════ - -class ColorSwatch(QPushButton): - color_changed = pyqtSignal(str) - - def __init__(self, color: str = "#00d4ff"): - super().__init__() - self._color = color - self.setFixedSize(26, 20) - self._paint() - self.clicked.connect(self._pick) - - def _paint(self): - self.setStyleSheet( - f"QPushButton{{background:{self._color};border:1px solid #2a3558;" - f"border-radius:3px;}}QPushButton:hover{{border-color:#3b82f6;}}") - - def _pick(self): - c = QColorDialog.getColor(QColor(self._color), self) - if c.isValid(): - self._color = c.name(); self._paint() - self.color_changed.emit(self._color) - - @property - def color(self): return self._color - @color.setter - def color(self, v): - self._color = v; self._paint() - - -# ══════════════════════════════════════════════════════════════════════════════ -# Trace row inside a plot block -# ══════════════════════════════════════════════════════════════════════════════ - -class TraceRow(QFrame): - removed = pyqtSignal(object) - changed = pyqtSignal() - - def __init__(self, trace: TraceSpec): - super().__init__() - self.trace = trace - self.setObjectName("traceRow") - self._build() - - def _build(self): - lay = QHBoxLayout(self) - lay.setContentsMargins(6, 3, 6, 3); lay.setSpacing(6) - - self._vis = QCheckBox(); self._vis.setChecked(self.trace.visible) - self._vis.setToolTip("Visible"); self._vis.toggled.connect(self._on_vis) - lay.addWidget(self._vis) - - self._color = ColorSwatch(self.trace.color) - self._color.color_changed.connect(self._on_color) - lay.addWidget(self._color) - - src = QLabel(f"{self.trace.device_id} / {self.trace.channel_id}") - src.setObjectName("traceSource"); src.setMinimumWidth(130) - lay.addWidget(src) - - self._lbl = QLineEdit(self.trace.label) - self._lbl.setObjectName("traceLabel"); self._lbl.setPlaceholderText("Label…") - self._lbl.textChanged.connect(lambda t: setattr(self.trace, "label", t)) - lay.addWidget(self._lbl, 1) - - self._style = QComboBox(); self._style.setObjectName("traceStyleCb") - self._style.addItems(["solid", "dash", "dot"]) - self._style.setCurrentText(self.trace.style); self._style.setFixedWidth(64) - self._style.currentTextChanged.connect(lambda s: setattr(self.trace, "style", s)) - lay.addWidget(self._style) - - self._w = QDoubleSpinBox(); self._w.setObjectName("traceWidthSpin") - self._w.setRange(0.5, 6.0); self._w.setSingleStep(0.5) - self._w.setValue(self.trace.width); self._w.setFixedWidth(56) - self._w.valueChanged.connect(lambda v: setattr(self.trace, "width", v)) - lay.addWidget(self._w) - - rm = QToolButton(); rm.setText("✕"); rm.setObjectName("traceRemoveBtn") - rm.setFixedSize(22, 22); rm.clicked.connect(lambda: self.removed.emit(self)) - lay.addWidget(rm) - - def _on_vis(self, v): self.trace.visible = v; self.changed.emit() - def _on_color(self, c): self.trace.color = c; self.changed.emit() - - -# ══════════════════════════════════════════════════════════════════════════════ -# Plot block — one collapsible row per PlotSpec -# ══════════════════════════════════════════════════════════════════════════════ - -class PlotBlock(QFrame): - removed = pyqtSignal(object) - changed = pyqtSignal() - move_up = pyqtSignal(object) - move_dn = pyqtSignal(object) - - def __init__(self, spec: PlotSpec, registry: DeviceRegistry, - processor: SignalProcessor): - super().__init__() - self.spec = spec - self.registry = registry - self.processor = processor - self._rows: List[TraceRow] = [] - self.setObjectName("plotBlock") - self._build() - - def _build(self): - outer = QVBoxLayout(self); outer.setContentsMargins(0,0,0,0); outer.setSpacing(0) - - # ── Header ──────────────────────────────────────────────────── - hdr = QWidget(); hdr.setObjectName("plotBlockHeader"); hdr.setFixedHeight(32) - hl = QHBoxLayout(hdr); hl.setContentsMargins(8,0,6,0); hl.setSpacing(4) - - self._title = QLineEdit(self.spec.title) - self._title.setObjectName("plotBlockTitle") - self._title.textChanged.connect(lambda t: (setattr(self.spec, "title", t), self.changed.emit())) - hl.addWidget(self._title, 1) - - for txt, sig in [("▲", self.move_up), ("▼", self.move_dn)]: - b = QToolButton(); b.setText(txt); b.setObjectName("plotMoveBtn") - b.setFixedSize(22,22); b.clicked.connect(lambda _, s=sig: s.emit(self)) - hl.addWidget(b) - - rm = QToolButton(); rm.setText("✕"); rm.setObjectName("plotRemoveBtn") - rm.setFixedSize(22,22); rm.clicked.connect(lambda: self.removed.emit(self)) - hl.addWidget(rm) - outer.addWidget(hdr) - - # ── Axis / range settings ───────────────────────────────────── - sett = QWidget(); sett.setObjectName("plotBlockSettings") - sl = QHBoxLayout(sett); sl.setContentsMargins(8,6,8,6); sl.setSpacing(8) - - sl.addWidget(QLabel("Y Axis:")) - self._ylabel = QLineEdit(self.spec.y_label) - self._ylabel.setObjectName("plotYLabelEdit"); self._ylabel.setMaximumWidth(130) - self._ylabel.setPlaceholderText("e.g. Voltage (V)") - self._ylabel.textChanged.connect(lambda t: setattr(self.spec, "y_label", t)) - sl.addWidget(self._ylabel) - - self._auto = QCheckBox("Auto Y"); self._auto.setChecked(self.spec.y_auto) - self._auto.toggled.connect(self._on_auto) - sl.addWidget(self._auto) - - sl.addWidget(QLabel("Min:")) - self._ymin = QDoubleSpinBox(); self._ymin.setRange(-1e9,1e9) - self._ymin.setValue(self.spec.y_min); self._ymin.setFixedWidth(72) - self._ymin.setEnabled(not self.spec.y_auto) - self._ymin.valueChanged.connect(lambda v: setattr(self.spec, "y_min", v)) - sl.addWidget(self._ymin) - - sl.addWidget(QLabel("Max:")) - self._ymax = QDoubleSpinBox(); self._ymax.setRange(-1e9,1e9) - self._ymax.setValue(self.spec.y_max); self._ymax.setFixedWidth(72) - self._ymax.setEnabled(not self.spec.y_auto) - self._ymax.valueChanged.connect(lambda v: setattr(self.spec, "y_max", v)) - sl.addWidget(self._ymax) - - self._grid = QCheckBox("Grid"); self._grid.setChecked(self.spec.grid) - self._grid.toggled.connect(lambda v: setattr(self.spec, "grid", v)) - sl.addWidget(self._grid) - - sl.addWidget(QLabel("H:")) - self._height_spin = QSpinBox(); self._height_spin.setRange(1,10) - self._height_spin.setValue(self.spec.height); self._height_spin.setFixedWidth(44) - self._height_spin.setToolTip("Relative row height") - self._height_spin.valueChanged.connect(lambda v: setattr(self.spec, "height", v)) - sl.addWidget(self._height_spin) - sl.addStretch() - outer.addWidget(sett) - - # ── Trace rows ──────────────────────────────────────────────── - traces_w = QWidget(); traces_w.setObjectName("plotBlockTraces") - self._tlay = QVBoxLayout(traces_w) - self._tlay.setContentsMargins(4,2,4,4); self._tlay.setSpacing(2) - - for tr in self.spec.traces: - self._add_trace_row(tr) - - # Channel picker + add button - pick_row = QHBoxLayout() - self._picker = self._make_picker() - pick_row.addWidget(self._picker, 1) - add_btn = QPushButton("＋ Add Channel") - add_btn.setObjectName("addTraceBtn") - add_btn.clicked.connect(self._on_add_trace) - pick_row.addWidget(add_btn) - self._tlay.addLayout(pick_row) - outer.addWidget(traces_w) - - def _make_picker(self) -> QComboBox: - cb = QComboBox(); cb.setObjectName("channelPickerCb") - cb.setPlaceholderText("Select channel…") - for dev in self.registry.all_instances(): - for ch in dev.info.channels: - if not ch.enabled: - continue - cb.addItem(f"{ch.name} ({dev.info.device_id}/{ch.channel_id})", - userData=(dev.info.device_id, ch.channel_id, ch.name, ch.color)) - for dc in self.processor.get_derived(): - if not dc.enabled: - continue - cb.addItem(f"{dc.name} ([derived]/{dc.channel_id})", - userData=("derived", dc.channel_id, dc.name, dc.color)) - return cb - - def _add_trace_row(self, trace: TraceSpec): - row = TraceRow(trace) - row.removed.connect(self._remove_trace) - row.changed.connect(self.changed) - self._rows.append(row) - self._tlay.insertWidget(self._tlay.count() - 1, row) - - def _remove_trace(self, row: TraceRow): - if row.trace in self.spec.traces: self.spec.traces.remove(row.trace) - self._rows.remove(row) - self._tlay.removeWidget(row); row.deleteLater() - self.changed.emit() - - def _on_add_trace(self): - data = self._picker.currentData() - if not data: return - dev_id, ch_id, ch_name, ch_color = data - tr = TraceSpec(device_id=dev_id, channel_id=ch_id, - label=ch_name, color=ch_color) - self.spec.traces.append(tr) - self._add_trace_row(tr) - self.changed.emit() - - def _on_auto(self, v: bool): - self.spec.y_auto = v - self._ymin.setEnabled(not v); self._ymax.setEnabled(not v) - - -# ══════════════════════════════════════════════════════════════════════════════ -# PlotBuilderWindow -# ══════════════════════════════════════════════════════════════════════════════ - -class PlotBuilderWindow(QWidget): - """ - Floating Plot Builder. Emits config_applied(LayoutConfig) on Apply. - """ - - config_applied = pyqtSignal(object) # LayoutConfig - closed = pyqtSignal() - - def __init__(self, registry: DeviceRegistry, - processor: SignalProcessor, - current: Optional[LayoutConfig] = None, - parent=None): - super().__init__(parent, Qt.WindowType.Window | Qt.WindowType.Tool) - self.registry = registry - self.processor = processor - self.cfg = deepcopy(current) if current \ - else build_default_layout(registry, processor) - - self.setWindowTitle("Plot Builder") - self.setMinimumSize(700, 780) - self.resize(800, 780) - self._blocks: List[PlotBlock] = [] - self._build() - self._populate() - - def _build(self): - root = QVBoxLayout(self); root.setContentsMargins(0,0,0,0); root.setSpacing(0) - - # ── Top bar ────────────────────────────────────────────────── - top = QWidget(); top.setObjectName("cfgTopBar"); top.setFixedHeight(44) - tl = QHBoxLayout(top); tl.setContentsMargins(12,0,12,0) - title = QLabel("📐 PLOT BUILDER"); title.setObjectName("cfgTopBarTitle") - tl.addWidget(title, 1) - self._live = QCheckBox("Live preview"); self._live.setObjectName("cfgLiveChk") - tl.addWidget(self._live) - root.addWidget(top) - - # ── Global settings ────────────────────────────────────────── - glob = QWidget(); glob.setObjectName("cfgGlobalBar") - gl = QHBoxLayout(glob); gl.setContentsMargins(12,6,12,6); gl.setSpacing(12) - gl.addWidget(QLabel("Time window:")) - self._win = QDoubleSpinBox(); self._win.setRange(1,3600) - self._win.setSuffix(" s"); self._win.setValue(self.cfg.time_window_s) - self._win.setObjectName("cfgGlobalSpin") - self._win.valueChanged.connect(self._global_changed) - gl.addWidget(self._win) - self._linkx = QCheckBox("Link X"); self._linkx.setChecked(self.cfg.link_x) - self._linkx.setObjectName("cfgLiveChk"); self._linkx.toggled.connect(self._global_changed) - gl.addWidget(self._linkx) - self._legend = QCheckBox("Legend"); self._legend.setChecked(self.cfg.show_legend) - self._legend.setObjectName("cfgLiveChk"); self._legend.toggled.connect(self._global_changed) - gl.addWidget(self._legend) - gl.addStretch() - add_row_btn = QPushButton("＋ Add Plot Row"); add_row_btn.setObjectName("addDeviceButton") - add_row_btn.clicked.connect(self._add_plot) - gl.addWidget(add_row_btn) - root.addWidget(glob) - - div = QFrame(); div.setFrameShape(QFrame.Shape.HLine); div.setObjectName("devWindowDivider") - root.addWidget(div) - - # ── Scrollable plot blocks ─────────────────────────────────── - scroll = QScrollArea(); scroll.setWidgetResizable(True); scroll.setObjectName("deviceScroll") - self._container = QWidget() - self._blay = QVBoxLayout(self._container) - self._blay.setContentsMargins(10,10,10,10); self._blay.setSpacing(12) - self._blay.addStretch() - scroll.setWidget(self._container) - root.addWidget(scroll, 1) - - # ── Bottom bar ─────────────────────────────────────────────── - btm = QWidget(); btm.setObjectName("cfgBottomBar") - bl = QHBoxLayout(btm); bl.setContentsMargins(12,8,12,8) - for lbl, fn in [("⟳ Rebuild from Devices", self._reset), - ("↓ Export JSON", self._export), - ("↑ Import JSON", self._import)]: - b = QPushButton(lbl); b.setObjectName("configButton"); b.clicked.connect(fn) - bl.addWidget(b) - bl.addStretch() - ap = QPushButton("✓ Apply"); ap.setObjectName("applyButton") - ap.clicked.connect(self._apply) - bl.addWidget(ap) - root.addWidget(btm) - - def _populate(self): - for b in self._blocks: - self._blay.removeWidget(b); b.deleteLater() - self._blocks.clear() - for spec in self.cfg.plots: - self._insert_block(spec) - - def _insert_block(self, spec: PlotSpec): - blk = PlotBlock(spec, self.registry, self.processor) - blk.removed.connect(self._remove_block) - blk.changed.connect(self._maybe_live) - blk.move_up.connect(self._move_up) - blk.move_dn.connect(self._move_dn) - self._blocks.append(blk) - self._blay.insertWidget(self._blay.count() - 1, blk) - - def _add_plot(self): - spec = PlotSpec(title=f"Plot {len(self.cfg.plots)+1}") - self.cfg.plots.append(spec) - self._insert_block(spec) - self._maybe_live() - - def _remove_block(self, blk: PlotBlock): - if blk.spec in self.cfg.plots: self.cfg.plots.remove(blk.spec) - self._blocks.remove(blk) - self._blay.removeWidget(blk); blk.deleteLater() - self._maybe_live() - - def _move_up(self, blk: PlotBlock): - i = self._blocks.index(blk) - if i == 0: return - self.cfg.plots.insert(i-1, self.cfg.plots.pop(i)) - self._blocks.insert(i-1, self._blocks.pop(i)) - self._blay.removeWidget(blk); self._blay.insertWidget(i-1, blk) - self._maybe_live() - - def _move_dn(self, blk: PlotBlock): - i = self._blocks.index(blk) - if i >= len(self._blocks)-1: return - self.cfg.plots.insert(i+1, self.cfg.plots.pop(i)) - self._blocks.insert(i+1, self._blocks.pop(i)) - self._blay.removeWidget(blk); self._blay.insertWidget(i+1, blk) - self._maybe_live() - - def _global_changed(self): - self.cfg.time_window_s = self._win.value() - self.cfg.link_x = self._linkx.isChecked() - self.cfg.show_legend = self._legend.isChecked() - self._maybe_live() - - def _maybe_live(self): - if self._live.isChecked(): - self.config_applied.emit(deepcopy(self.cfg)) - - def _apply(self): - self.config_applied.emit(deepcopy(self.cfg)) - - def _reset(self): - self.cfg = build_default_layout(self.registry, self.processor) - self._win.setValue(self.cfg.time_window_s) - self._populate(); self._maybe_live() - - def _export(self): - path, _ = QFileDialog.getSaveFileName(self, "Export Layout", "layout.json", "JSON (*.json)") - if path: - open(path, "w").write(self.cfg.to_json()) - - def _import(self): - path, _ = QFileDialog.getOpenFileName(self, "Import Layout", "", "JSON (*.json)") - if path: - try: - self.cfg = LayoutConfig.from_json(open(path).read()) - self._win.setValue(self.cfg.time_window_s) - self._populate(); self._maybe_live() - except Exception as e: - QMessageBox.critical(self, "Import failed", str(e)) - - def refresh_channels(self): - """Call after devices/derived channels change.""" - self._populate() - - def closeEvent(self, event: QCloseEvent): - self.closed.emit(); event.accept() diff --git a/ui/plot_config.py b/ui/plot_config.py deleted file mode 100644 index c6ecaa4..0000000 --- a/ui/plot_config.py +++ /dev/null @@ -1,631 +0,0 @@ -""" -ui/plot_config.py - -Plot Configuration Builder — floating window. - -Lets the user define: - • How many subplot rows to show - • Which channels go in each row (any mix of devices/channels) - • Per-channel: label override, color, line width, visibility - • Per-plot: Y-axis label, Y min/max (or auto), grid on/off - • Global: time window, sample decimation, background color - -The resulting config is stored as a list of PlotConfig dataclasses -and emitted via the `config_applied` signal so StripChartWidget can -rebuild itself on demand. -""" - -from __future__ import annotations -import json -from copy import deepcopy -from dataclasses import dataclass, field, asdict -from typing import List, Optional - -from PyQt6.QtWidgets import ( - QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, - QScrollArea, QFrame, QLineEdit, QDoubleSpinBox, QSpinBox, - QCheckBox, QComboBox, QColorDialog, QGroupBox, QFormLayout, - QSizePolicy, QSplitter, QListWidget, QListWidgetItem, - QAbstractItemView, QToolButton, QSlider, -) -from PyQt6.QtCore import Qt, pyqtSignal, QSize -from PyQt6.QtGui import QColor, QIcon, QCloseEvent - -from devices.device_registry import DeviceRegistry - - -# ══════════════════════════════════════════════════════════════════════════════ -# Data model -# ══════════════════════════════════════════════════════════════════════════════ - -@dataclass -class ChannelTrace: - device_id: str - channel_id: str - label: str = "" # "" → use channel name - color: str = "#00d4ff" - width: float = 1.8 - visible: bool = True - style: str = "solid" # "solid" | "dash" | "dot" - - -@dataclass -class PlotConfig: - title: str = "Plot" - y_label: str = "" - y_auto: bool = True - y_min: float = -10.0 - y_max: float = 10.0 - grid: bool = True - traces: List[ChannelTrace] = field(default_factory=list) - - -@dataclass -class ChartConfig: - plots: List[PlotConfig] = field(default_factory=list) - time_window_s: float = 30.0 - bg_color: str = "#0b0e13" - link_x: bool = True - show_legend: bool = True - - def to_json(self) -> str: - return json.dumps(asdict(self), indent=2) - - @staticmethod - def from_json(s: str) -> "ChartConfig": - d = json.loads(s) - plots = [] - for p in d.get("plots", []): - traces = [ChannelTrace(**t) for t in p.pop("traces", [])] - plots.append(PlotConfig(**p, traces=traces)) - d["plots"] = plots - return ChartConfig(**{k: v for k, v in d.items() if k != "plots"}, - plots=plots) - - -def default_config(registry: DeviceRegistry) -> ChartConfig: - """Build a sensible default config from whatever devices are registered.""" - cfg = ChartConfig() - for dev in registry.all_instances(): - if not dev.info.channels: - continue - p = PlotConfig( - title=dev.info.name, - y_label=dev.info.channels[0].unit if dev.info.channels else "", - ) - for ch in dev.info.channels: - if ch.enabled: - p.traces.append(ChannelTrace( - device_id=dev.info.device_id, - channel_id=ch.channel_id, - label=ch.name, - color=ch.color, - )) - if p.traces: - cfg.plots.append(p) - return cfg - - -# ══════════════════════════════════════════════════════════════════════════════ -# Colour swatch button -# ══════════════════════════════════════════════════════════════════════════════ - -class ColorButton(QPushButton): - color_changed = pyqtSignal(str) - - def __init__(self, color: str = "#00d4ff"): - super().__init__() - self._color = color - self.setFixedSize(28, 22) - self._apply() - self.clicked.connect(self._pick) - - def _apply(self): - self.setStyleSheet( - f"QPushButton {{ background:{self._color}; border:1px solid #2a3558;" - f" border-radius:3px; }} " - f"QPushButton:hover {{ border-color:#3b82f6; }}" - ) - - def _pick(self): - c = QColorDialog.getColor(QColor(self._color), self, "Pick colour") - if c.isValid(): - self._color = c.name() - self._apply() - self.color_changed.emit(self._color) - - @property - def color(self): return self._color - - @color.setter - def color(self, v: str): - self._color = v - self._apply() - - -# ══════════════════════════════════════════════════════════════════════════════ -# Trace row widget (one row per ChannelTrace inside a plot) -# ══════════════════════════════════════════════════════════════════════════════ - -class TraceRow(QFrame): - removed = pyqtSignal(object) # self - changed = pyqtSignal() - - def __init__(self, trace: ChannelTrace, registry: DeviceRegistry): - super().__init__() - self.trace = trace - self.registry = registry - self.setObjectName("traceRow") - self._build() - - def _build(self): - lay = QHBoxLayout(self) - lay.setContentsMargins(6, 4, 6, 4) - lay.setSpacing(6) - - # Visibility checkbox - self._vis = QCheckBox() - self._vis.setChecked(self.trace.visible) - self._vis.setToolTip("Show/hide trace") - self._vis.toggled.connect(self._on_vis) - lay.addWidget(self._vis) - - # Color swatch - self._color_btn = ColorButton(self.trace.color) - self._color_btn.color_changed.connect(self._on_color) - lay.addWidget(self._color_btn) - - # Source label dev_id / ch_id - src = QLabel(f"{self.trace.device_id} / {self.trace.channel_id}") - src.setObjectName("traceSource") - src.setMinimumWidth(120) - lay.addWidget(src) - - # Label override - self._label_edit = QLineEdit(self.trace.label) - self._label_edit.setPlaceholderText("Label…") - self._label_edit.setObjectName("traceLabel") - self._label_edit.textChanged.connect(self._on_label) - lay.addWidget(self._label_edit, 1) - - # Line style - self._style_cb = QComboBox() - self._style_cb.addItems(["solid", "dash", "dot"]) - self._style_cb.setCurrentText(self.trace.style) - self._style_cb.setObjectName("traceStyleCb") - self._style_cb.setFixedWidth(64) - self._style_cb.currentTextChanged.connect(self._on_style) - lay.addWidget(self._style_cb) - - # Width - self._width_spin = QDoubleSpinBox() - self._width_spin.setRange(0.5, 6.0) - self._width_spin.setSingleStep(0.5) - self._width_spin.setValue(self.trace.width) - self._width_spin.setFixedWidth(56) - self._width_spin.setObjectName("traceWidthSpin") - self._width_spin.valueChanged.connect(self._on_width) - lay.addWidget(self._width_spin) - - # Remove button - rm = QToolButton() - rm.setText("✕") - rm.setObjectName("traceRemoveBtn") - rm.setFixedSize(22, 22) - rm.clicked.connect(lambda: self.removed.emit(self)) - lay.addWidget(rm) - - def _on_vis(self, v): self.trace.visible = v; self.changed.emit() - def _on_color(self, c): self.trace.color = c; self.changed.emit() - def _on_label(self, t): self.trace.label = t; self.changed.emit() - def _on_style(self, s): self.trace.style = s; self.changed.emit() - def _on_width(self, w): self.trace.width = w; self.changed.emit() - - -# ══════════════════════════════════════════════════════════════════════════════ -# Plot block widget (one collapsible section per PlotConfig) -# ══════════════════════════════════════════════════════════════════════════════ - -class PlotBlock(QFrame): - removed = pyqtSignal(object) # self - changed = pyqtSignal() - move_up = pyqtSignal(object) - move_dn = pyqtSignal(object) - - def __init__(self, plot_cfg: PlotConfig, registry: DeviceRegistry, index: int): - super().__init__() - self.plot_cfg = plot_cfg - self.registry = registry - self.index = index - self.setObjectName("plotBlock") - self._trace_rows: list[TraceRow] = [] - self._build() - - def _build(self): - outer = QVBoxLayout(self) - outer.setContentsMargins(0, 0, 0, 0) - outer.setSpacing(0) - - # ── Header bar ────────────────────────────────────────────── - hdr = QWidget(); hdr.setObjectName("plotBlockHeader"); hdr.setFixedHeight(32) - hdr_lay = QHBoxLayout(hdr); hdr_lay.setContentsMargins(8, 0, 6, 0) - - self._title_edit = QLineEdit(self.plot_cfg.title) - self._title_edit.setObjectName("plotBlockTitle") - self._title_edit.textChanged.connect(self._on_title) - hdr_lay.addWidget(self._title_edit, 1) - - up_btn = QToolButton(); up_btn.setText("▲"); up_btn.setObjectName("plotMoveBtn") - up_btn.setFixedSize(22, 22); up_btn.clicked.connect(lambda: self.move_up.emit(self)) - dn_btn = QToolButton(); dn_btn.setText("▼"); dn_btn.setObjectName("plotMoveBtn") - dn_btn.setFixedSize(22, 22); dn_btn.clicked.connect(lambda: self.move_dn.emit(self)) - rm_btn = QToolButton(); rm_btn.setText("✕"); rm_btn.setObjectName("plotRemoveBtn") - rm_btn.setFixedSize(22, 22); rm_btn.clicked.connect(lambda: self.removed.emit(self)) - - for b in (up_btn, dn_btn, rm_btn): hdr_lay.addWidget(b) - outer.addWidget(hdr) - - # ── Settings row ──────────────────────────────────────────── - settings = QWidget(); settings.setObjectName("plotBlockSettings") - s_lay = QHBoxLayout(settings); s_lay.setContentsMargins(8, 6, 8, 6) - - s_lay.addWidget(QLabel("Y Label:")) - self._ylabel_edit = QLineEdit(self.plot_cfg.y_label) - self._ylabel_edit.setPlaceholderText("e.g. Voltage (V)") - self._ylabel_edit.setObjectName("plotYLabelEdit") - self._ylabel_edit.textChanged.connect(self._on_ylabel) - s_lay.addWidget(self._ylabel_edit) - - self._auto_chk = QCheckBox("Auto Y") - self._auto_chk.setChecked(self.plot_cfg.y_auto) - self._auto_chk.toggled.connect(self._on_auto) - s_lay.addWidget(self._auto_chk) - - s_lay.addWidget(QLabel("Min:")) - self._ymin = QDoubleSpinBox() - self._ymin.setRange(-1e9, 1e9); self._ymin.setValue(self.plot_cfg.y_min) - self._ymin.setFixedWidth(72); self._ymin.setEnabled(not self.plot_cfg.y_auto) - self._ymin.valueChanged.connect(self._on_ymin) - s_lay.addWidget(self._ymin) - - s_lay.addWidget(QLabel("Max:")) - self._ymax = QDoubleSpinBox() - self._ymax.setRange(-1e9, 1e9); self._ymax.setValue(self.plot_cfg.y_max) - self._ymax.setFixedWidth(72); self._ymax.setEnabled(not self.plot_cfg.y_auto) - self._ymax.valueChanged.connect(self._on_ymax) - s_lay.addWidget(self._ymax) - - self._grid_chk = QCheckBox("Grid") - self._grid_chk.setChecked(self.plot_cfg.grid) - self._grid_chk.toggled.connect(self._on_grid) - s_lay.addWidget(self._grid_chk) - - s_lay.addStretch() - outer.addWidget(settings) - - # ── Traces container ───────────────────────────────────────── - self._traces_widget = QWidget() - self._traces_widget.setObjectName("plotBlockTraces") - self._traces_layout = QVBoxLayout(self._traces_widget) - self._traces_layout.setContentsMargins(4, 2, 4, 4) - self._traces_layout.setSpacing(2) - - for tr in self.plot_cfg.traces: - self._add_trace_row(tr) - - # ── Add trace button ───────────────────────────────────────── - add_row = QHBoxLayout() - self._ch_picker = self._build_channel_picker() - add_row.addWidget(self._ch_picker, 1) - add_tr_btn = QPushButton("＋ Add Trace") - add_tr_btn.setObjectName("addTraceBtn") - add_tr_btn.clicked.connect(self._on_add_trace) - add_row.addWidget(add_tr_btn) - self._traces_layout.addLayout(add_row) - - outer.addWidget(self._traces_widget) - - # ── Helpers ────────────────────────────────────────────────────────── - - def _build_channel_picker(self) -> QComboBox: - cb = QComboBox() - cb.setObjectName("channelPickerCb") - cb.setPlaceholderText("Select channel…") - for dev in self.registry.all_instances(): - for ch in dev.info.channels: - if not ch.enabled: - continue - label = f"{ch.name} ({dev.info.device_id}/{ch.channel_id})" - cb.addItem(label, userData=(dev.info.device_id, ch.channel_id, - ch.name, ch.color)) - return cb - - def _add_trace_row(self, trace: ChannelTrace): - row = TraceRow(trace, self.registry) - row.removed.connect(self._remove_trace_row) - row.changed.connect(self.changed) - self._trace_rows.append(row) - # Insert before the add-row (last item) - idx = self._traces_layout.count() - 1 - self._traces_layout.insertWidget(idx, row) - - def _remove_trace_row(self, row: TraceRow): - self.plot_cfg.traces.remove(row.trace) - self._trace_rows.remove(row) - self._traces_layout.removeWidget(row) - row.deleteLater() - self.changed.emit() - - def _on_add_trace(self): - data = self._ch_picker.currentData() - if data is None: - return - dev_id, ch_id, ch_name, ch_color = data - tr = ChannelTrace(device_id=dev_id, channel_id=ch_id, - label=ch_name, color=ch_color) - self.plot_cfg.traces.append(tr) - self._add_trace_row(tr) - self.changed.emit() - - # ── Config slots ───────────────────────────────────────────────────── - - def _on_title(self, t): self.plot_cfg.title = t; self.changed.emit() - def _on_ylabel(self, t): self.plot_cfg.y_label = t; self.changed.emit() - def _on_auto(self, v): - self.plot_cfg.y_auto = v - self._ymin.setEnabled(not v); self._ymax.setEnabled(not v) - self.changed.emit() - def _on_ymin(self, v): self.plot_cfg.y_min = v; self.changed.emit() - def _on_ymax(self, v): self.plot_cfg.y_max = v; self.changed.emit() - def _on_grid(self, v): self.plot_cfg.grid = v; self.changed.emit() - - -# ══════════════════════════════════════════════════════════════════════════════ -# PlotConfigWindow — the main floating builder -# ══════════════════════════════════════════════════════════════════════════════ - -class PlotConfigWindow(QWidget): - """ - Floating plot configuration builder. - - Emits `config_applied(ChartConfig)` when the user clicks Apply or - when live-preview is enabled and any setting changes. - """ - - config_applied = pyqtSignal(object) # ChartConfig - closed = pyqtSignal() - - def __init__(self, registry: DeviceRegistry, - current_config: Optional[ChartConfig] = None, - parent=None): - super().__init__( - parent, - Qt.WindowType.Window | Qt.WindowType.Tool, - ) - self.registry = registry - self.cfg = deepcopy(current_config) if current_config \ - else default_config(registry) - - self.setWindowTitle("Plot Configuration") - self.setMinimumSize(680, 560) - self.resize(760, 640) - self._blocks: list[PlotBlock] = [] - self._build() - self._populate() - - # ── Build shell ─────────────────────────────────────────────────────── - - def _build(self): - root = QVBoxLayout(self) - root.setContentsMargins(0, 0, 0, 0) - root.setSpacing(0) - - # ── Top bar ────────────────────────────────────────────────── - topbar = QWidget(); topbar.setObjectName("cfgTopBar"); topbar.setFixedHeight(44) - tb_lay = QHBoxLayout(topbar); tb_lay.setContentsMargins(12, 0, 12, 0) - - title = QLabel("📈 PLOT CONFIGURATION") - title.setObjectName("cfgTopBarTitle") - tb_lay.addWidget(title, 1) - - self._live_chk = QCheckBox("Live preview") - self._live_chk.setChecked(False) - self._live_chk.setObjectName("cfgLiveChk") - tb_lay.addWidget(self._live_chk) - - root.addWidget(topbar) - - # ── Global settings strip ──────────────────────────────────── - glob = QWidget(); glob.setObjectName("cfgGlobalBar") - g_lay = QHBoxLayout(glob); g_lay.setContentsMargins(12, 6, 12, 6) - - g_lay.addWidget(QLabel("Time window:")) - self._win_spin = QDoubleSpinBox() - self._win_spin.setRange(1, 3600); self._win_spin.setSuffix(" s") - self._win_spin.setValue(self.cfg.time_window_s) - self._win_spin.setObjectName("cfgGlobalSpin") - self._win_spin.valueChanged.connect(self._on_global_change) - g_lay.addWidget(self._win_spin) - g_lay.addSpacing(16) - - self._linkx_chk = QCheckBox("Link X axes") - self._linkx_chk.setChecked(self.cfg.link_x) - self._linkx_chk.setObjectName("cfgLiveChk") - self._linkx_chk.toggled.connect(self._on_global_change) - g_lay.addWidget(self._linkx_chk) - - self._legend_chk = QCheckBox("Show legend") - self._legend_chk.setChecked(self.cfg.show_legend) - self._legend_chk.setObjectName("cfgLiveChk") - self._legend_chk.toggled.connect(self._on_global_change) - g_lay.addWidget(self._legend_chk) - - g_lay.addStretch() - - add_plot_btn = QPushButton("＋ Add Plot Row") - add_plot_btn.setObjectName("addDeviceButton") - add_plot_btn.clicked.connect(self._add_plot) - g_lay.addWidget(add_plot_btn) - - root.addWidget(glob) - - # divider - div = QFrame(); div.setFrameShape(QFrame.Shape.HLine) - div.setObjectName("devWindowDivider"); root.addWidget(div) - - # ── Scrollable plot blocks ─────────────────────────────────── - scroll = QScrollArea() - scroll.setWidgetResizable(True) - scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff) - scroll.setObjectName("deviceScroll") - - self._container = QWidget() - self._blocks_lay = QVBoxLayout(self._container) - self._blocks_lay.setContentsMargins(10, 10, 10, 10) - self._blocks_lay.setSpacing(12) - self._blocks_lay.addStretch() - - scroll.setWidget(self._container) - root.addWidget(scroll, 1) - - # ── Bottom action bar ──────────────────────────────────────── - btm = QWidget(); btm.setObjectName("cfgBottomBar") - b_lay = QHBoxLayout(btm); b_lay.setContentsMargins(12, 8, 12, 8) - - self._reset_btn = QPushButton("⟳ Reset to Default") - self._reset_btn.setObjectName("configButton") - self._reset_btn.clicked.connect(self._reset_defaults) - b_lay.addWidget(self._reset_btn) - - self._export_btn = QPushButton("↓ Export JSON") - self._export_btn.setObjectName("configButton") - self._export_btn.clicked.connect(self._export_json) - b_lay.addWidget(self._export_btn) - - self._import_btn = QPushButton("↑ Import JSON") - self._import_btn.setObjectName("configButton") - self._import_btn.clicked.connect(self._import_json) - b_lay.addWidget(self._import_btn) - - b_lay.addStretch() - - self._apply_btn = QPushButton("✓ Apply") - self._apply_btn.setObjectName("applyButton") - self._apply_btn.clicked.connect(self._apply) - b_lay.addWidget(self._apply_btn) - - root.addWidget(btm) - - # ── Populate from config ─────────────────────────────────────────────── - - def _populate(self): - for blk in self._blocks: - self._blocks_lay.removeWidget(blk) - blk.deleteLater() - self._blocks.clear() - - for i, p in enumerate(self.cfg.plots): - self._insert_block(p, i) - - def _insert_block(self, plot_cfg: PlotConfig, index: int): - blk = PlotBlock(plot_cfg, self.registry, index) - blk.removed.connect(self._remove_block) - blk.changed.connect(self._on_block_changed) - blk.move_up.connect(self._move_block_up) - blk.move_dn.connect(self._move_block_dn) - self._blocks.append(blk) - self._blocks_lay.insertWidget(self._blocks_lay.count() - 1, blk) - - # ── Block management ────────────────────────────────────────────────── - - def _add_plot(self): - p = PlotConfig(title=f"Plot {len(self.cfg.plots) + 1}") - self.cfg.plots.append(p) - self._insert_block(p, len(self._blocks)) - self._maybe_live() - - def _remove_block(self, blk: PlotBlock): - if blk.plot_cfg in self.cfg.plots: - self.cfg.plots.remove(blk.plot_cfg) - self._blocks.remove(blk) - self._blocks_lay.removeWidget(blk) - blk.deleteLater() - self._maybe_live() - - def _move_block_up(self, blk: PlotBlock): - i = self._blocks.index(blk) - if i == 0: return - self.cfg.plots.insert(i - 1, self.cfg.plots.pop(i)) - self._blocks.insert(i - 1, self._blocks.pop(i)) - self._blocks_lay.removeWidget(blk) - self._blocks_lay.insertWidget(i - 1, blk) - self._maybe_live() - - def _move_block_dn(self, blk: PlotBlock): - i = self._blocks.index(blk) - if i >= len(self._blocks) - 1: return - self.cfg.plots.insert(i + 1, self.cfg.plots.pop(i)) - self._blocks.insert(i + 1, self._blocks.pop(i)) - self._blocks_lay.removeWidget(blk) - self._blocks_lay.insertWidget(i + 1, blk) - self._maybe_live() - - # ── Global settings ─────────────────────────────────────────────────── - - def _on_global_change(self): - self.cfg.time_window_s = self._win_spin.value() - self.cfg.link_x = self._linkx_chk.isChecked() - self.cfg.show_legend = self._legend_chk.isChecked() - self._maybe_live() - - def _on_block_changed(self): - self._maybe_live() - - def _maybe_live(self): - if self._live_chk.isChecked(): - self.config_applied.emit(deepcopy(self.cfg)) - - # ── Actions ─────────────────────────────────────────────────────────── - - def _apply(self): - self.config_applied.emit(deepcopy(self.cfg)) - - def _reset_defaults(self): - self.cfg = default_config(self.registry) - self._win_spin.setValue(self.cfg.time_window_s) - self._populate() - self._maybe_live() - - def _export_json(self): - from PyQt6.QtWidgets import QFileDialog - path, _ = QFileDialog.getSaveFileName( - self, "Export Plot Config", "plot_config.json", - "JSON Files (*.json)" - ) - if path: - with open(path, "w") as f: - f.write(self.cfg.to_json()) - - def _import_json(self): - from PyQt6.QtWidgets import QFileDialog - path, _ = QFileDialog.getOpenFileName( - self, "Import Plot Config", "", "JSON Files (*.json)" - ) - if path: - try: - with open(path) as f: - self.cfg = ChartConfig.from_json(f.read()) - self._win_spin.setValue(self.cfg.time_window_s) - self._populate() - self._maybe_live() - except Exception as e: - from PyQt6.QtWidgets import QMessageBox - QMessageBox.critical(self, "Import failed", str(e)) - - def closeEvent(self, event: QCloseEvent): - self.closed.emit() - event.accept() - - def update_registry(self): - """Call after devices are added/removed to refresh channel pickers.""" - self._populate() diff --git a/ui/strip_chart.py b/ui/strip_chart.py index 6c644be..bdffb91 100644 --- a/ui/strip_chart.py +++ b/ui/strip_chart.py @@ -54,8 +54,8 @@ _STYLES = { def _windowed(times_dq, values_dq, seconds: float): """Last `seconds` of a (times, values) deque pair, walking in from the - newest sample only — same trick as ChannelBuffer.window(), needed here - for derived-channel buffers which aren't wrapped in ChannelBuffer.""" + newest sample only — same trick as SignalBuffer.window(), needed here + for derived-channel buffers which aren't wrapped in SignalBuffer.""" if not times_dq: return [], [] cutoff = times_dq[-1] - seconds @@ -210,7 +210,7 @@ class StripChartWidget(QWidget): return dc.unit return "" dev = self.registry.get_instance(device_id) - ch = dev.get_channel(channel_id) if dev else None + ch = dev.get_signal(channel_id) if dev else None return ch.unit if ch else "" @pyqtSlot(str, str, str) @@ -246,7 +246,7 @@ class StripChartWidget(QWidget): ts_l, vs_l = _windowed(bufs[0], bufs[1], window) if not ts_l: return else: - buf = self.engine.get_buffer(device_id, channel_id) + buf = self.engine.get_signal_buffer(device_id, channel_id) if buf is None or len(buf) < 2: return ts_l, vs_l = buf.window(window) if not ts_l: return @@ -273,7 +273,7 @@ class StripChartWidget(QWidget): mn = min(len(x_arr), len(y_arr)) e["curve"].setData(x_arr[-mn:], y_arr[-mn:]) else: - xbuf = self.engine.get_buffer(x_dev, x_ch) + xbuf = self.engine.get_signal_buffer(x_dev, x_ch) if xbuf and len(xbuf) >= 2: _, xvs = xbuf.all() x_arr = np.array(xvs, dtype=np.float64) diff --git a/ui/windows/channels_window.py b/ui/windows/channels_window.py index 87d7840..f581851 100644 --- a/ui/windows/channels_window.py +++ b/ui/windows/channels_window.py @@ -15,7 +15,7 @@ Classes: PipelineTab — splitter content (physical + virtual sections) ChannelPipelineBlock — collapsible filter+math stack for one physical channel ChannelPickerDialog — checkbox dialog for selecting channels to add - DerivedBlock — editor for one virtual/derived channel + ChannelBlock — editor for one virtual/derived channel FilterRow — single filter stage row inside a pipeline block SignalsListTab — unused; kept for profile-load compatibility """ @@ -35,7 +35,7 @@ from PyQt6.QtGui import QCloseEvent, QFont, QFontMetrics from devices.device_registry import DeviceRegistry from core.signal_processor import ( - SignalProcessor, ChannelPipeline, DerivedChannel, + SignalProcessor, ChannelPipeline, Channel, FilterBase, FILTER_CLASSES, ) from ui.windows.plot_window import _DragHandle, _DragReorderContainer @@ -60,8 +60,8 @@ def _channel_display_name(registry: DeviceRegistry, else: for dev in registry.all_instances(): if dev.info.device_id == dev_id: - for ch in dev.info.channels: - if ch.channel_id == ch_id: + for ch in dev.info.signals: + if ch.signal_id == ch_id: return ch.name if ch.name else ch_id return ch_id @@ -72,16 +72,16 @@ def _channel_combo(registry: DeviceRegistry, show_unit: bool = False) -> QComboBox: cb = QComboBox(); cb.setObjectName("channelPickerCb") for dev in registry.all_instances(): - for ch in dev.info.channels: + for ch in dev.info.signals: if not ch.enabled: continue - label = f"{ch.name} ({dev.info.device_id}/{ch.channel_id})" + label = f"Signal: {ch.name} ({dev.info.device_id}/{ch.signal_id})" if show_unit and ch.unit: label += f" [{ch.unit}]" - cb.addItem(label, userData=(dev.info.device_id, ch.channel_id)) + cb.addItem(label, userData=(dev.info.device_id, ch.signal_id)) if include_derived and processor: for dc in processor.get_derived(): - label = f"[derived] {dc.channel_id} ({dc.name})" + label = f"Channel: {dc.channel_id} ({dc.name})" if show_unit and dc.unit: label += f" [{dc.unit}]" cb.addItem(label, userData=("derived", dc.channel_id)) @@ -110,17 +110,17 @@ class ChannelPickerDialog(QDialog): any_available = False for dev in registry.all_instances(): - for ch in dev.info.channels: + for ch in dev.info.signals: if not ch.enabled: continue - if (dev.info.device_id, ch.channel_id) in already_shown: + if (dev.info.device_id, ch.signal_id) in already_shown: continue any_available = True - label = f"{ch.name} ({dev.info.device_id}/{ch.channel_id})" + label = f"{ch.name} ({dev.info.device_id}/{ch.signal_id})" if ch.unit: label += f" [{ch.unit}]" chk = QCheckBox(label) - self._checks.append((chk, dev.info.device_id, ch.channel_id)) + self._checks.append((chk, dev.info.device_id, ch.signal_id)) cl.addWidget(chk) if not any_available: @@ -237,7 +237,7 @@ class ChannelPipelineBlock(QFrame): self._unit = unit outer = QVBoxLayout(self); outer.setContentsMargins(0, 0, 0, 0); outer.setSpacing(0) - # Header (omitted when embedded in another block, e.g. DerivedBlock's + # Header (omitted when embedded in another block, e.g. ChannelBlock's # Filters section — the parent already provides a header/toggle) if self._show_header: hdr = QWidget(); hdr.setObjectName("plotBlockHeader"); hdr.setFixedHeight(32) @@ -405,6 +405,7 @@ class ChannelPipelineBlock(QFrame): class PipelineTab(QWidget): changed = pyqtSignal() derived_changed = pyqtSignal() + channel_renamed = pyqtSignal(str, str, str, str) # old_dev, old_id, new_dev, new_id def __init__(self, registry: DeviceRegistry, processor: SignalProcessor): super().__init__() @@ -412,7 +413,7 @@ class PipelineTab(QWidget): self.processor = processor self._shown: Set[Tuple[str, str]] = set() self._blocks: dict[Tuple[str, str], ChannelPipelineBlock] = {} - self._derived_blocks: List[DerivedBlock] = [] + self._derived_blocks: List[ChannelBlock] = [] self._color_idx = 0 self._build() @@ -494,9 +495,9 @@ class PipelineTab(QWidget): QTimer.singleShot(0, _init_split) for dev in self.registry.all_instances(): - for ch in dev.info.channels: + for ch in dev.info.signals: if ch.enabled: - self._add_block(dev.info.device_id, ch.channel_id, ch.name, ch.unit) + self._add_block(dev.info.device_id, ch.signal_id, ch.name, ch.unit) # ── Physical channel block management ────────────────────────────────────── @@ -516,7 +517,7 @@ class PipelineTab(QWidget): if key not in self._shown: dev = self.registry.get_instance(dev_id) if dev: - ch = dev.get_channel(ch_id) + ch = dev.get_signal(ch_id) if ch: self._add_block(dev_id, ch_id, ch.name, ch.unit) else: @@ -541,7 +542,7 @@ class PipelineTab(QWidget): if dev is None: self.on_device_removed(dev_id) return - valid = {ch.channel_id for ch in dev.info.channels} + valid = {ch.signal_id for ch in dev.info.signals} keys = [(d, c) for (d, c) in list(self._shown) if d == dev_id and c not in valid] for key in keys: @@ -560,7 +561,7 @@ class PipelineTab(QWidget): else: dev = self.registry.get_instance(dev_id) if dev: - ch = dev.get_channel(ch_id) + ch = dev.get_signal(ch_id) if ch and ch.enabled: self._add_block(dev_id, ch_id, ch_name, ch.unit) self._refresh_src_cb() @@ -580,13 +581,13 @@ class PipelineTab(QWidget): self._src_cb.clear() self._src_cb.insertItem(0, "Signal: none (empty channel)", userData=None) for dev in self.registry.all_instances(): - for ch in dev.info.channels: + for ch in dev.info.signals: if not ch.enabled: continue - label = f"{ch.name} ({dev.info.device_id}/{ch.channel_id})" - self._src_cb.addItem(label, userData=(dev.info.device_id, ch.channel_id)) + label = f"Signal: {ch.name} ({dev.info.device_id}/{ch.signal_id})" + self._src_cb.addItem(label, userData=(dev.info.device_id, ch.signal_id)) for dc in self.processor.get_derived(): - label = f"[derived] {dc.channel_id} ({dc.name})" + label = f"Channel: {dc.channel_id} ({dc.name})" self._src_cb.addItem(label, userData=("derived", dc.channel_id)) if current: for i in range(self._src_cb.count()): @@ -607,7 +608,7 @@ class PipelineTab(QWidget): dev = self.registry.get_instance(dev_id) if not dev: continue - ch = dev.get_channel(ch_id) + ch = dev.get_signal(ch_id) if not ch: continue self._add_block(dev_id, ch_id, ch.name, ch.unit) @@ -635,23 +636,27 @@ class PipelineTab(QWidget): src_name = f"Channel {len(self._derived_blocks) + 1}" name = _dedupe_name(src_name, existing_names) channel_id = _slug_id(name, existing_ids) - dc = DerivedChannel(channel_id=channel_id, name=name, + dc = Channel(channel_id=channel_id, name=name, kind="expression", color=color) blk = self._make_derived_block(dc) self._virt_inner.insertWidget(self._virt_inner.count() - 1, blk) if src: blk._add_src(src) - def _make_derived_block(self, dc: DerivedChannel) -> DerivedBlock: - blk = DerivedBlock(dc, self.registry, self.processor) + def _make_derived_block(self, dc: Channel) -> ChannelBlock: + blk = ChannelBlock(dc, self.registry, self.processor) blk.removed.connect(self._rm_virtual) blk.applied.connect(lambda _: self.derived_changed.emit()) - blk.id_renamed.connect(self.refresh_derived) + blk.id_renamed.connect(self._on_block_id_renamed) blk._drag_handle.set_drag_source(self, blk) self._derived_blocks.append(blk) return blk - def _rm_virtual(self, blk: DerivedBlock): + def _on_block_id_renamed(self, old_id: str, new_id: str): + self.refresh_derived() + self.channel_renamed.emit("derived", old_id, "derived", new_id) + + def _rm_virtual(self, blk: ChannelBlock): self.processor.remove_derived(blk.dc.channel_id) self._derived_blocks.remove(blk) self._virt_inner.removeWidget(blk); blk.deleteLater() @@ -659,7 +664,7 @@ class PipelineTab(QWidget): # ── Drag-to-reorder — see _DragReorderContainer in ui/windows/plot_window.py - def _index_of(self, block: DerivedBlock) -> int: + def _index_of(self, block: ChannelBlock) -> int: return self._derived_blocks.index(block) def _drop_index(self, y: float) -> int: @@ -708,7 +713,7 @@ class PipelineTab(QWidget): self.derived_changed.emit() def refresh_derived(self): - """Rebuild DerivedBlock widgets from processor state (called after profile load).""" + """Rebuild ChannelBlock widgets from processor state (called after profile load).""" for blk in self._derived_blocks: self._virt_inner.removeWidget(blk); blk.deleteLater() self._derived_blocks.clear() @@ -723,7 +728,7 @@ class PipelineTab(QWidget): # ══════════════════════════════════════════════════════════════════════════════ -# Derived editor for one DerivedChannel +# Derived editor for one Channel # ══════════════════════════════════════════════════════════════════════════════ _BUILTIN = ["none", "scale_offset", "derivative", "second_derivative", "power", "rms", "difference", "sum"] @@ -806,13 +811,17 @@ def _sub_xrefs(code: str, names: list) -> str: return re.sub(r'\bx\[(\d+)\]', repl, code) -class DerivedBlock(QFrame): +class ChannelBlock(QFrame): removed = pyqtSignal(object) - applied = pyqtSignal(object) # DerivedChannel - id_renamed = pyqtSignal() # this Channel's ID changed — other blocks' - # source pickers may now be stale - - def __init__(self, dc: DerivedChannel, registry: DeviceRegistry, + applied = pyqtSignal(object) # Channel + id_renamed = pyqtSignal(str, str) # old_id, new_id — this Channel's ID + # changed; other blocks' source + # pickers, and everything outside + # this processor referencing it + # (plot traces, controls), may now + # be stale — see propagate_rename() + + def __init__(self, dc: Channel, registry: DeviceRegistry, processor: SignalProcessor): super().__init__() self.dc = dc; self.registry = registry; self.processor = processor @@ -1192,12 +1201,8 @@ class DerivedBlock(QFrame): self._st.setStyleSheet("color:#ef4444;") return renamed = channel_id != self._orig_id + old_id = self._orig_id if renamed: - # Other channels sourced from this one under the old ID would - # otherwise keep pointing at a reference that no longer exists — - # fix their sources up front, carry the ring buffer + latest - # value over so the rename doesn't reset their plotted history. - self.processor.rename_derived(self._orig_id, channel_id) self._orig_id = channel_id sources = [cb.currentData() for cb in self._src_combos if cb.currentData()] if self.dc.kind in ("derivative", "second_derivative"): @@ -1215,11 +1220,11 @@ class DerivedBlock(QFrame): elif self.dc.kind in ("function", "custom_script"): self.dc.script = self._code.toPlainText() err = self.processor.add_derived(self.dc) - # The rename above already committed regardless of whether this - # channel's own expression/script compiles — other blocks' source - # pickers need to refresh either way. + # Emitted regardless of whether this channel's own expression/script + # compiles — propagate_rename() (via id_renamed) still needs to run + # so other channels' sources / plot traces / controls don't dangle. if renamed: - self.id_renamed.emit() + self.id_renamed.emit(old_id, channel_id) if err: self._st.setText(f"⚠ {err[:100]}") self._st.setStyleSheet("color:#ef4444;") @@ -1261,12 +1266,12 @@ class SignalsListTab(QWidget): if it.widget(): it.widget().deleteLater() for dev in self.registry.all_instances(): - for ch in dev.info.channels: + for ch in dev.info.signals: dev_id_cap = dev.info.device_id - ch_id_cap = ch.channel_id + ch_id_cap = ch.signal_id self._lay.addWidget(self._make_row( color = ch.color, - name = f"{ch.channel_id} — {ch.name}", + name = f"{ch.signal_id} — {ch.name}", unit = ch.unit, source = dev_id_cap, enabled = ch.enabled, diff --git a/ui/windows/config_window.py b/ui/windows/config_window.py index 604b6ed..62789f1 100644 --- a/ui/windows/config_window.py +++ b/ui/windows/config_window.py @@ -28,9 +28,9 @@ from PyQt6.QtGui import QCloseEvent, QColor, QShowEvent from devices.device_registry import DeviceRegistry from core.acquisition import AcquisitionEngine -from core.signal_processor import SignalProcessor, DerivedChannel +from core.signal_processor import SignalProcessor, Channel -from ui.windows.devices_window import DeviceRow, ChannelsTab +from ui.windows.devices_window import DeviceRow, SignalsTab from ui.windows.channels_window import PipelineTab, _slug_id, _dedupe_name from ui.windows.plot_window import ( PlotWindow, LayoutConfig, LayoutCanvas, PaneBlock, @@ -43,8 +43,8 @@ from ui.namespace_panel import NamespacePanel # Enhanced signals table — adds Dir badge and pipeline ⚙ button # ══════════════════════════════════════════════════════════════════════════════ -class _SignalsTab(ChannelsTab): - """ChannelsTab with a Dir (READ/WRITE) badge. Operations (filters/calibration/math) +class _SignalsTab(SignalsTab): + """SignalsTab with a Dir (READ/WRITE) badge. Operations (filters/calibration/math) are configured per Channel in the Channels pane, not here.""" _C_DEVICE = 0 @@ -58,6 +58,7 @@ class _SignalsTab(ChannelsTab): _C_MKCH = 8 # "+ Channel" button channel_created = pyqtSignal() + channel_renamed = pyqtSignal(str, str, str, str) # old_dev, old_id, new_dev, new_id def __init__(self, registry: DeviceRegistry, processor: SignalProcessor): self._processor = processor @@ -91,7 +92,7 @@ class _SignalsTab(ChannelsTab): self._table.setRowCount(0) row = 0 for dev in self.registry.all_instances(): - for ch in dev.info.channels: + for ch in dev.info.signals: self._table.insertRow(row) dev_item = QTableWidgetItem(f"{dev.info.icon} {dev.info.device_id}") @@ -99,8 +100,8 @@ class _SignalsTab(ChannelsTab): dev_item.setForeground(QColor("#64748b")) self._table.setItem(row, self._C_DEVICE, dev_item) - ch_item = QTableWidgetItem(ch.channel_id) - ch_item.setFlags(ch_item.flags() & ~Qt.ItemFlag.ItemIsEditable) + ch_item = QTableWidgetItem(ch.signal_id) + ch_item.setData(Qt.ItemDataRole.UserRole, (dev.info.device_id, ch.signal_id)) ch_item.setForeground(QColor(ch.color)) self._table.setItem(row, self._C_CH_ID, ch_item) @@ -122,7 +123,7 @@ class _SignalsTab(ChannelsTab): chk.setEnabled(False) else: dev_id_cap = dev.info.device_id - ch_id_cap = ch.channel_id + ch_id_cap = ch.signal_id chk.toggled.connect( lambda checked, d=dev_id_cap, c=ch_id_cap: self._on_enabled(d, c, checked) @@ -131,7 +132,7 @@ class _SignalsTab(ChannelsTab): self._table.setCellWidget(row, self._C_ENABLED, chk_w) name_item = QTableWidgetItem(ch.name) - name_item.setData(Qt.ItemDataRole.UserRole, (dev.info.device_id, ch.channel_id)) + name_item.setData(Qt.ItemDataRole.UserRole, (dev.info.device_id, ch.signal_id)) self._table.setItem(row, self._C_NAME, name_item) self._table.setItem(row, self._C_UNIT, QTableWidgetItem(ch.unit)) @@ -148,7 +149,7 @@ class _SignalsTab(ChannelsTab): mkch_btn.setFixedSize(24, 22) mkch_btn.setToolTip("Create a Channel from this Signal") d_cap, c_cap, n_cap, u_cap, col_cap = ( - dev.info.device_id, ch.channel_id, ch.name, ch.unit, ch.color + dev.info.device_id, ch.signal_id, ch.name, ch.unit, ch.color ) mkch_btn.clicked.connect( lambda _=False, d=d_cap, c=c_cap, n=n_cap, u=u_cap, col=col_cap: @@ -161,6 +162,49 @@ class _SignalsTab(ChannelsTab): self._table.blockSignals(False) + def _on_cell_changed(self, row: int, col: int): + if col == self._C_CH_ID: + self._on_id_edited(row) + return + super()._on_cell_changed(row, col) + + def _on_id_edited(self, row: int): + """Signal IDs are stable/backend-facing by default (see + devices/base_device.py::SignalConfig) but user-editable here, same + as a derived Channel's ID — renaming propagates through + reference_registry.propagate_rename() (wired via channel_renamed) + so plot traces / controls / pipelines pointing at the old id follow + it instead of going dangling (the bug that motivated this feature: + an Arduino analog pin remap used to change the id outright).""" + id_item = self._table.item(row, self._C_CH_ID) + if id_item is None: + return + ref = id_item.data(Qt.ItemDataRole.UserRole) + if ref is None: + return + dev_id, old_id = ref + new_id = id_item.text().strip() + if not new_id or new_id == old_id: + self._table.blockSignals(True) + id_item.setText(old_id) + self._table.blockSignals(False) + return + dev = self.registry.get_instance(dev_id) + if not dev or any(s.signal_id == new_id for s in dev.info.signals if s.signal_id != old_id): + self._table.blockSignals(True) + id_item.setText(old_id) + self._table.blockSignals(False) + return + ch = dev.get_signal(old_id) + if not ch: + return + ch.signal_id = new_id + id_item.setData(Qt.ItemDataRole.UserRole, (dev_id, new_id)) + name_item = self._table.item(row, self._C_NAME) + if name_item is not None: + name_item.setData(Qt.ItemDataRole.UserRole, (dev_id, new_id)) + self.channel_renamed.emit(dev_id, old_id, dev_id, new_id) + def _on_create_channel(self, dev_id: str, ch_id: str, ch_name: str, unit: str, color: str): """Create a passthrough Channel (kind="none") from this Signal, named after it — the "1 Signal -> 1 Channel" quick path. No-ops if @@ -173,7 +217,7 @@ class _SignalsTab(ChannelsTab): existing_ids = {d.channel_id for d in existing} name = _dedupe_name(ch_name, existing_names) channel_id = _slug_id(name, existing_ids) - dc = DerivedChannel(channel_id=channel_id, name=name, kind="none", + dc = Channel(channel_id=channel_id, name=name, kind="none", sources=[(dev_id, ch_id)], source_names=[ch_name], unit=unit, color=color or "#00d4ff") err = self._processor.add_derived(dc) @@ -275,6 +319,7 @@ class ConfigWindow(QWidget): channel_unit_changed = pyqtSignal(str, str, str) pipeline_changed = pyqtSignal() derived_changed = pyqtSignal() + channel_renamed = pyqtSignal(str, str, str, str) # old_dev, old_id, new_dev, new_id layout_applied = pyqtSignal(object) closed = pyqtSignal() @@ -360,6 +405,7 @@ class ConfigWindow(QWidget): self._signals_tab.channel_visibility_changed.connect(self._on_sig_visibility) self._signals_tab.channel_name_changed.connect(self._on_sig_name) self._signals_tab.channel_unit_changed.connect(self._on_sig_unit) + self._signals_tab.channel_renamed.connect(self.channel_renamed) sig_l.addWidget(self._signals_tab, 1) splitter.addWidget(sig_w) @@ -468,6 +514,7 @@ class ConfigWindow(QWidget): self._pipeline_tab = PipelineTab(self.registry, self.processor) self._pipeline_tab.changed.connect(self.pipeline_changed) self._pipeline_tab.derived_changed.connect(self.derived_changed) + self._pipeline_tab.channel_renamed.connect(self.channel_renamed) self._pipeline_tab.refresh_derived() # load derived channels already in processor # (e.g. window built fresh after profile load) inner_splitters = self._pipeline_tab.findChildren(QSplitter) diff --git a/ui/windows/devices_window.py b/ui/windows/devices_window.py index ae88c6b..5d47681 100644 --- a/ui/windows/devices_window.py +++ b/ui/windows/devices_window.py @@ -5,7 +5,7 @@ DEVICES window — toolbar section 2. Tabs: Physical Devices — add / remove / connect devices; shows status cards - Channels — per-device channel list: rename, set unit, enable/disable + Signals — per-device signal list: rename, set unit, enable/disable """ from PyQt6.QtWidgets import ( @@ -84,7 +84,7 @@ class DeviceRow(QFrame): self._status_lbl = QLabel(_STATUS_TEXT.get(self.device.status, "?")) self._status_lbl.setObjectName("deviceSub") self._set_status_style() - n = sum(1 for c in self.device.info.channels if c.enabled) + n = sum(1 for c in self.device.info.signals if c.enabled) ch_lbl = QLabel(f"{n} ch") ch_lbl.setObjectName("deviceChannelCount") stat_col.addWidget(self._status_lbl) @@ -129,7 +129,7 @@ class DeviceRow(QFrame): ) -class ChannelsTab(QWidget): +class SignalsTab(QWidget): """Shows all channels across all devices in an editable table.""" channel_visibility_changed = pyqtSignal(str, str, bool) # dev_id, ch_id, enabled @@ -174,12 +174,12 @@ class ChannelsTab(QWidget): self._table.setRowCount(0) row = 0 for dev in self.registry.all_instances(): - for ch in dev.info.channels: + for ch in dev.info.signals: self._table.insertRow(row) # Device + signal ID folded into one non-editable column — # the separate "Signal ID" column was removed as redundant. - dev_item = QTableWidgetItem(f"{dev.info.icon} {dev.info.device_id} / {ch.channel_id}") + dev_item = QTableWidgetItem(f"{dev.info.icon} {dev.info.device_id} / {ch.signal_id}") dev_item.setFlags(dev_item.flags() & ~Qt.ItemFlag.ItemIsEditable) dev_item.setForeground(QColor("#64748b")) self._table.setItem(row, self._C_DEVICE, dev_item) @@ -192,7 +192,7 @@ class ChannelsTab(QWidget): chk = QCheckBox() chk.setChecked(ch.enabled) dev_id_cap = dev.info.device_id - ch_id_cap = ch.channel_id + ch_id_cap = ch.signal_id chk.toggled.connect( lambda checked, d=dev_id_cap, c=ch_id_cap: self._on_enabled(d, c, checked) ) @@ -200,7 +200,7 @@ class ChannelsTab(QWidget): self._table.setCellWidget(row, self._C_ENABLED, chk_container) name_item = QTableWidgetItem(ch.name) - name_item.setData(Qt.ItemDataRole.UserRole, (dev.info.device_id, ch.channel_id)) + name_item.setData(Qt.ItemDataRole.UserRole, (dev.info.device_id, ch.signal_id)) self._table.setItem(row, self._C_NAME, name_item) unit_item = QTableWidgetItem(ch.unit) @@ -218,7 +218,7 @@ class ChannelsTab(QWidget): def _on_enabled(self, dev_id: str, ch_id: str, checked: bool): dev = self.registry.get_instance(dev_id) if dev: - ch = dev.get_channel(ch_id) + ch = dev.get_signal(ch_id) if ch: ch.enabled = checked self.channel_visibility_changed.emit(dev_id, ch_id, checked) @@ -234,7 +234,7 @@ class ChannelsTab(QWidget): dev = self.registry.get_instance(dev_id) if not dev: return - ch = dev.get_channel(ch_id) + ch = dev.get_signal(ch_id) if not ch: return if col == self._C_NAME: @@ -318,7 +318,7 @@ class DevicesWindow(QWidget): tabs.addTab(dev_tab, " Physical Devices ") # Tab 2: Channels - self._ch_tab = ChannelsTab(self.registry) + self._ch_tab = SignalsTab(self.registry) self._ch_tab.channel_visibility_changed.connect(self.channel_visibility_changed) self._ch_tab.channel_name_changed.connect(self.channel_name_changed) self._ch_tab.channel_unit_changed.connect(self.channel_unit_changed) diff --git a/ui/windows/plot_window.py b/ui/windows/plot_window.py index e810365..c0aea4e 100644 --- a/ui/windows/plot_window.py +++ b/ui/windows/plot_window.py @@ -341,18 +341,18 @@ class TraceChannelPickerDialog(QDialog): any_available = False for dev in registry.all_instances(): - for ch in dev.info.channels: + for ch in dev.info.signals: if not ch.enabled: continue any_available = True - chk = QCheckBox(f"{ch.name} ({dev.info.device_id}/{ch.channel_id})") - self._checks.append((chk, dev.info.device_id, ch.channel_id, ch.name, ch.color)) + chk = QCheckBox(f"Signal: {ch.name} ({dev.info.device_id}/{ch.signal_id})") + self._checks.append((chk, dev.info.device_id, ch.signal_id, ch.name, ch.color)) cl.addWidget(chk) for dc in processor.get_derived(): if not dc.enabled: continue any_available = True - chk = QCheckBox(f"{dc.name} ([virtual]/{dc.channel_id})") + chk = QCheckBox(f"Channel: {dc.name} ([virtual]/{dc.channel_id})") self._checks.append((chk, "derived", dc.channel_id, dc.name, dc.color)) cl.addWidget(chk) @@ -498,15 +498,15 @@ class PaneBlock(QFrame): self._x_cb.setMinimumWidth(150) self._x_cb.addItem("⏱ Time (elapsed s)", userData="time") for dev in self.registry.all_instances(): - for ch in dev.info.channels: + for ch in dev.info.signals: if not ch.enabled: continue - self._x_cb.addItem(f"{ch.name} ({dev.info.device_id}/{ch.channel_id})", - userData=f"{dev.info.device_id}/{ch.channel_id}") + self._x_cb.addItem(f"Signal: {ch.name} ({dev.info.device_id}/{ch.signal_id})", + userData=f"{dev.info.device_id}/{ch.signal_id}") for dc in self.processor.get_derived(): if not dc.enabled: continue - self._x_cb.addItem(f"{dc.name} ([virtual]/{dc.channel_id})", + self._x_cb.addItem(f"Channel: {dc.name} ([virtual]/{dc.channel_id})", userData=f"derived/{dc.channel_id}") for i in range(self._x_cb.count()): if self._x_cb.itemData(i) == self.spec.x_source: |