""" core/profile.py Lab profile — saves and loads the complete operator configuration: • 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, Tuple # ── Helpers ─────────────────────────────────────────────────────────────────── def _safe(d: dict, key: str, default=None): return d.get(key, default) # ══════════════════════════════════════════════════════════════════════════════ # Legacy (pre-2.0) schema migration # ══════════════════════════════════════════════════════════════════════════════ 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". """ 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 # ══════════════════════════════════════════════════════════════════════════════ # Profile data model # ══════════════════════════════════════════════════════════════════════════════ @dataclass class Profile: name: str = "Untitled Profile" version: str = "2.0" devices: List[Dict] = field(default_factory=list) controls: List[Dict] = field(default_factory=list) signals: List[Dict] = field(default_factory=list) pipelines: 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) # plugin_state = {plugin_id: plugin.get_save_state()} plugins_enabled: List[str] = field(default_factory=list) # plugins_enabled = [plugin_id, ...] — which plugins were on when saved plugins_manifest: List[Dict] = field(default_factory=list) # plugins_manifest = [{plugin_id, name, version, description, requires, source_url}] script_vars: Dict[str, Any] = field(default_factory=dict) # script_vars — shared vars dict accessible as `vars` in all expressions def to_json(self) -> str: return json.dumps(asdict(self), indent=2) @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", "2.0"), devices = d.get("devices", []), controls = d.get("controls", []), signals = d.get("signals", []), pipelines = d.get("pipelines", []), channels = d.get("channels", []), plot = d.get("plot"), settings = d.get("settings", {}), plugin_state = d.get("plugin_state", {}), plugins_enabled = d.get("plugins_enabled", []), plugins_manifest = d.get("plugins_manifest", []), script_vars = d.get("script_vars", {}), ) def save(self, path: str): os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) with open(path, "w") as f: f.write(self.to_json()) @staticmethod def load(path: str) -> "Profile": with open(path) as f: return Profile.from_json(f.read()) # ══════════════════════════════════════════════════════════════════════════════ # ProfileManager — serialises / deserialises the live app state # ══════════════════════════════════════════════════════════════════════════════ class ProfileManager: """ Converts between a Profile dict and the live application objects. Usage: # Save profile = ProfileManager.capture(registry, processor, chart_cfg, control_specs, settings) profile.save("my_lab.labui") # Load profile = Profile.load("my_lab.labui") ProfileManager.apply(profile, registry, processor, ...) """ # Plugin device factories registered at runtime (device_type → class) _extra_factories: dict = {} @staticmethod def register_device_factory(device_type: str, factory): """Register a plugin device class so it can be restored from profiles.""" ProfileManager._extra_factories[device_type] = factory @staticmethod def unregister_device_factory(device_type: str): ProfileManager._extra_factories.pop(device_type, None) # ── Capture ────────────────────────────────────────────────────────────── @staticmethod def capture( registry, processor, plot_cfg, control_specs: List, settings: dict, profile_name: str = "Profile", plugin_manager=None, ) -> Profile: p = Profile(name=profile_name) # Devices for dev in registry.all_instances(): if hasattr(dev, "get_save_config"): p.devices.append(dev.get_save_config()) # Controls p.controls = [spec.to_dict() for spec in control_specs] # Signal overrides for dev in registry.all_instances(): for ch in dev.info.signals: p.signals.append({ "device_id": dev.info.device_id, "signal_id": ch.signal_id, "name": ch.name, "unit": ch.unit, "enabled": ch.enabled, "color": ch.color, }) # Signal pipelines for key, pipeline in processor._pipelines.items(): p.pipelines.append({ "device_id": pipeline.device_id, "channel_id": pipeline.channel_id, "enabled": pipeline.enabled, "filters": [f.to_dict() for f in pipeline.filters], }) # Channels (derived) for dc in processor.get_derived(): p.channels.append({ "channel_id": dc.channel_id, "name": dc.name, "unit": dc.unit, "color": dc.color, "kind": dc.kind, "sources": [list(s) for s in dc.sources], "source_names": list(dc.source_names), "expression": dc.expression, "script": dc.script, "params": {k: v for k, v in dc.params.items() if not k.startswith("_")}, "enabled": dc.enabled, }) # Plot layout if plot_cfg is not None: try: from ui.windows.plot_window import LayoutConfig if hasattr(plot_cfg, "to_json"): p.plot = json.loads(plot_cfg.to_json()) except Exception: pass p.settings = dict(settings) # Shared script variables with processor._lock: p.script_vars = dict(processor._script_vars) # Plugin state + enabled list + manifest snapshot if plugin_manager is not None: p.plugins_enabled = plugin_manager.get_enabled_ids() for pid in p.plugins_enabled: m = plugin_manager.get_manifest(pid) if m: p.plugins_manifest.append({ "plugin_id": m.plugin_id, "name": m.name, "version": m.version, "description": m.description, "requires": m.requires, "source_url": m.source_url, }) for plugin in plugin_manager.get_loaded(): try: state = plugin.get_save_state() if state: p.plugin_state[plugin.plugin_id] = state except Exception: pass return p # ── Apply ───────────────────────────────────────────────────────────────── @staticmethod def apply( profile: Profile, registry, processor, control_panel, settings_ref: dict, engine=None, plugin_manager=None, ): """Apply a loaded profile to the live application state.""" from ui.control_editor import ControlSpec from core.signal_processor import ( ChannelPipeline, Channel, filter_from_dict, ) # ── Devices ────────────────────────────────────────────────────── _DEVICE_FACTORIES = dict(ProfileManager._extra_factories) try: from devices.arduino_device import ArduinoDevice _DEVICE_FACTORIES["arduino"] = ArduinoDevice except Exception: pass try: from devices.nidaqmx_device import NidaqmxDevice _DEVICE_FACTORIES["nidaqmx"] = NidaqmxDevice except Exception: pass try: from devices.serial_device import SerialDevice _DEVICE_FACTORIES["serial"] = SerialDevice except Exception: pass # Clear all existing devices — profile defines the complete device set for dev in list(registry.all_instances()): try: dev.disconnect() except Exception: pass registry.remove_instance(dev.info.device_id) if engine is not None: engine.remove_device(dev.info.device_id) # Clear signal pipelines and derived channels — profile defines the # complete set; otherwise the previous profile's leftovers stick around. processor._pipelines.clear() for dc in list(processor.get_derived()): processor.remove_derived(dc.channel_id) for dev_cfg in profile.devices: dev_type = dev_cfg.get("device_type") dev_id = dev_cfg.get("device_id") if not dev_id: continue factory = _DEVICE_FACTORIES.get(dev_type) if not factory: print(f"[Profile] Unknown device type: {dev_type}") continue try: # "name" is a display label, not a constructor arg — every device # factory builds its own default name internally, so apply it # after construction instead of passing it through. # "port_desc"/"port_serial" are identity metadata for port # re-detection below, not constructor args either. custom_name = dev_cfg.get("name") port_desc = dev_cfg.get("port_desc", "") port_serial = dev_cfg.get("port_serial", "") kwargs = {k: v for k, v in dev_cfg.items() if k not in ("device_type", "name", "port_desc", "port_serial")} # Re-map the saved port to whatever it enumerates as on *this* # machine — COM/tty numbers aren't stable across machines (or # even across replugs on some OSes). Only if the saved port # isn't already present verbatim (same-machine reload — nothing # to fix) and we have something to match against. if not kwargs.get("simulate", True) and "port" in kwargs and (port_desc or port_serial): from api_layers.arduino_layer import ArduinoLayer current_ports = {dev for dev, _ in ArduinoLayer.list_ports()} if kwargs["port"] not in current_ports: found = ArduinoLayer.find_port_by_identity(port_desc, port_serial) if found: print(f"[Profile] '{dev_id}': port {kwargs['port']!r} not found, " f"re-mapped to {found!r} by device identity") kwargs["port"] = found else: print(f"[Profile] '{dev_id}': port {kwargs['port']!r} not found and " f"no matching device on this machine — will show ERROR " f"until reconfigured") dev = factory(**kwargs) if custom_name: dev.info.name = custom_name registry.add_instance(dev) dev.connect() if engine is not None: engine.add_device(dev) except Exception as e: print(f"[Profile] Could not restore device {dev_id}: {e}") # ── Signal overrides ───────────────────────────────────────────── for ch_data in profile.signals: dev = registry.get_instance(ch_data["device_id"]) if not dev: continue ch = dev.get_signal(ch_data["signal_id"]) if not ch: print(f"[Profile] '{ch_data['device_id']}': signal '{ch_data['signal_id']}' " f"no longer exists — saved override for '{ch_data.get('name', '?')}' dropped") continue ch.name = ch_data.get("name", ch.name) ch.unit = ch_data.get("unit", ch.unit) ch.enabled = ch_data.get("enabled", ch.enabled) if ch_data.get("color"): ch.color = ch_data["color"] # ── Signal pipelines ───────────────────────────────────────────── for pd in profile.pipelines: filters = [filter_from_dict(fd) for fd in pd.get("filters", [])] pipeline = ChannelPipeline( device_id=pd["device_id"], channel_id=pd["channel_id"], filters=filters, enabled=pd.get("enabled", True), ) processor.set_pipeline(pipeline) # ── Channels (derived) ──────────────────────────────────────────── for dd in profile.channels: sources = [tuple(s) for s in dd.get("sources", [])] dc = Channel( channel_id=dd["channel_id"], name=dd.get("name", dd["channel_id"]), unit=dd.get("unit", ""), color=dd.get("color", "#f72585"), kind=dd.get("kind", "expression"), sources=sources, source_names=dd.get("source_names", []), expression=dd.get("expression", ""), script=dd.get("script", ""), params=dd.get("params", {}), enabled=dd.get("enabled", True), ) processor.add_derived(dc) # ── Controls ───────────────────────────────────────────────────── specs = [ControlSpec.from_dict(d) for d in profile.controls] control_panel.load_specs(specs) # ── Script vars ────────────────────────────────────────────────── if profile.script_vars: with processor._lock: processor._script_vars.update(profile.script_vars) # ── Settings ───────────────────────────────────────────────────── settings_ref.update(profile.settings) # ── Plot layout ─────────────────────────────────────────────────── plot_cfg = None if profile.plot: try: from ui.windows.plot_window import LayoutConfig plot_cfg = LayoutConfig.from_json(json.dumps(profile.plot)) except Exception as e: print(f"[Profile] Could not restore plot layout: {e}") # Plugin state if plugin_manager is not None and profile.plugin_state: for plugin_id, state in profile.plugin_state.items(): plugin = plugin_manager.get_plugin(plugin_id) if plugin is not None: try: plugin.apply_save_state(state) except Exception as e: print(f"[Profile] Plugin '{plugin_id}' state restore error: {e}") return plot_cfg # caller applies this to the chart