From 23ef9c1c6f776bb0aaee82d149cf45a696d62f3a Mon Sep 17 00:00:00 2001 From: x09lb Date: Fri, 14 Aug 2026 16:04:19 +0200 Subject: [PATCH] Pressure monitor strip tool widget --- .../widgets/pressure_monitor/__init__.py | 0 .../pressure_monitor/pressure_monitor.py | 1207 +++++++++++++++++ .../pressure_monitor.pyproject | 1 + .../pressure_monitor_plugin.py | 62 + .../register_pressure_monitor.py | 18 + 5 files changed, 1288 insertions(+) create mode 100644 xil_bec/bec_widgets/widgets/pressure_monitor/__init__.py create mode 100644 xil_bec/bec_widgets/widgets/pressure_monitor/pressure_monitor.py create mode 100644 xil_bec/bec_widgets/widgets/pressure_monitor/pressure_monitor.pyproject create mode 100644 xil_bec/bec_widgets/widgets/pressure_monitor/pressure_monitor_plugin.py create mode 100644 xil_bec/bec_widgets/widgets/pressure_monitor/register_pressure_monitor.py diff --git a/xil_bec/bec_widgets/widgets/pressure_monitor/__init__.py b/xil_bec/bec_widgets/widgets/pressure_monitor/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/xil_bec/bec_widgets/widgets/pressure_monitor/pressure_monitor.py b/xil_bec/bec_widgets/widgets/pressure_monitor/pressure_monitor.py new file mode 100644 index 0000000..fff83a6 --- /dev/null +++ b/xil_bec/bec_widgets/widgets/pressure_monitor/pressure_monitor.py @@ -0,0 +1,1207 @@ +"""Rolling time-window monitor for BEC signals (e.g. pressure gauges). + +The widget subscribes directly to the ``device_readback`` endpoint of the selected +devices, keeps the samples in a bounded in-memory ring buffer and shows them as +pressure-vs-time curves in a moving time window. Nothing is written to disk while +running; the data currently held by the plot can be exported to CSV or copied to +the clipboard on demand. + +Typical use:: + + from .bec_widgets.widgets.pressure_monitor.pressure_monitor import ( + PressureMonitor, + ) + + w = PressureMonitor(device_tags=["pressure"], time_span=300.0) + w.select_devices(["gauge_mc", "gauge_oh"]) + w.show() +""" + +from __future__ import annotations + +import csv +import io +import math +import time +from datetime import datetime +from typing import TYPE_CHECKING, Literal + +import numpy as np +import pyqtgraph as pg +from bec_lib.callback_handler import EventType +from bec_lib.device import Device as BECDevice +from bec_lib.device import ReadoutPriority +from bec_lib.device import Signal as BECSignal +from bec_lib.endpoints import MessageEndpoints +from bec_lib.logger import bec_logger +from pydantic import Field +from qtpy.QtCore import Qt, QTimer +from qtpy.QtCore import Signal as QSignal +from qtpy.QtGui import QGuiApplication +from qtpy.QtWidgets import ( + QCheckBox, + QDoubleSpinBox, + QFileDialog, + QHBoxLayout, + QLabel, + QLineEdit, + QListWidget, + QListWidgetItem, + QPushButton, + QSizePolicy, + QSplitter, + QVBoxLayout, + QWidget, +) + +from bec_widgets.utils.bec_connector import ConnectionConfig +from bec_widgets.utils.bec_widget import BECWidget +from bec_widgets.utils.colors import Colors +from bec_widgets.utils.error_popups import SafeProperty, SafeSlot +from bec_widgets.widgets.plots.plot_base import PlotBase + +if TYPE_CHECKING: # pragma: no cover + from bec_lib.client import BECClient + +logger = bec_logger.logger + +# Anything above this is almost certainly a broken timestamp source rather than data. +MAX_POINTS_PER_CURVE = 500_000 + + +# -------------------------------------------------------------------------------------- +# Data storage +# -------------------------------------------------------------------------------------- +class TimeSeriesBuffer: + """Growable, trimmable (timestamp, value) buffer backed by two numpy arrays. + + The buffer is append-only at the head and trimmed at the tail, which keeps the + memory bounded by the configured time span without reallocating on every sample. + """ + + __slots__ = ("_t", "_v", "_n", "_max_points") + + def __init__(self, capacity: int = 4096, max_points: int = MAX_POINTS_PER_CURVE): + capacity = max(8, int(capacity)) + self._t = np.empty(capacity, dtype=np.float64) + self._v = np.empty(capacity, dtype=np.float64) + self._n = 0 + self._max_points = int(max_points) + + def __len__(self) -> int: + return self._n + + def append(self, timestamp: float, value: float) -> None: + """Append a single sample.""" + if self._n == self._t.size: + self._grow() + self._t[self._n] = timestamp + self._v[self._n] = value + self._n += 1 + if self._n > self._max_points: + self._drop_head(self._n - self._max_points) + + def _grow(self) -> None: + new_capacity = min(max(8, self._t.size * 2), self._max_points + 1) + if new_capacity <= self._t.size: + # Hard cap reached, make room by dropping the oldest 10%. + self._drop_head(max(1, self._n // 10)) + return + new_t = np.empty(new_capacity, dtype=np.float64) + new_v = np.empty(new_capacity, dtype=np.float64) + new_t[: self._n] = self._t[: self._n] + new_v[: self._n] = self._v[: self._n] + self._t = new_t + self._v = new_v + + def _drop_head(self, count: int) -> None: + count = min(count, self._n) + if count <= 0: + return + keep = self._n - count + self._t[:keep] = self._t[count : self._n] + self._v[:keep] = self._v[count : self._n] + self._n = keep + + def trim(self, t_min: float) -> None: + """Drop all samples older than ``t_min``.""" + if self._n == 0: + return + cut = int(np.searchsorted(self._t[: self._n], t_min, side="left")) + self._drop_head(cut) + + def data(self) -> tuple[np.ndarray, np.ndarray]: + """Return copies of the current timestamps and values.""" + return self._t[: self._n].copy(), self._v[: self._n].copy() + + def window(self, t_min: float, t_max: float) -> tuple[np.ndarray, np.ndarray]: + """Return copies of the samples with ``t_min <= t <= t_max``.""" + if self._n == 0: + return np.empty(0), np.empty(0) + lo = int(np.searchsorted(self._t[: self._n], t_min, side="left")) + hi = int(np.searchsorted(self._t[: self._n], t_max, side="right")) + return self._t[lo:hi].copy(), self._v[lo:hi].copy() + + def clear(self) -> None: + self._n = 0 + + +class MonitoredSignal: + """Bookkeeping for one monitored device signal and its curve.""" + + def __init__( + self, + device: str, + signal_key: str, + candidate_keys: set[str], + label: str, + units: str, + source: str, + curve: pg.PlotDataItem, + ): + self.device = device + self.signal_key = signal_key + self.candidate_keys = candidate_keys + self.label = label + self.units = units + self.source = source + self.curve = curve + self.buffer = TimeSeriesBuffer() + + +# -------------------------------------------------------------------------------------- +# Plot +# -------------------------------------------------------------------------------------- +class RollingSignalPlotConfig(ConnectionConfig): + """Serializable configuration of :class:`RollingSignalPlot`.""" + + devices: list[str] = Field(default_factory=list, description="Monitored device names.") + time_span: float = Field(300.0, description="Displayed time window in seconds.") + x_mode: str = Field("relative", description="'relative' (s before now) or 'absolute' (clock).") + timestamp_source: str = Field( + "local", description="'local' (arrival time) or 'device' (timestamp in the readback)." + ) + color_palette: str = Field("plasma", description="Colormap used to assign curve colors.") + + +class RollingSignalPlot(PlotBase): + """A PlotBase showing live device readbacks in a moving time window. + + The plot owns one ring buffer per monitored device. Samples arrive through the + BEC dispatcher (``device_readback`` endpoint) and are appended as they come in; + redrawing is decoupled from the update rate by a refresh timer, so a fast gauge + cannot stall the GUI. + """ + + PLUGIN = False + RPC = False + ICON_NAME = "monitoring" + + USER_ACCESS = [ + *PlotBase.USER_ACCESS, + "add_device", + "remove_device", + "set_devices", + "devices", + "time_span", + "time_span.setter", + "paused", + "paused.setter", + "refresh_interval", + "refresh_interval.setter", + "auto_follow", + "auto_follow.setter", + "x_mode", + "x_mode.setter", + "timestamp_source", + "timestamp_source.setter", + "y_quantity", + "y_quantity.setter", + "clear_data", + "export_csv", + "displayed_data_as_text", + ] + + devices_changed = QSignal(list) + + def __init__( + self, + parent: QWidget | None = None, + config: RollingSignalPlotConfig | dict | None = None, + client: "BECClient | None" = None, + gui_id: str | None = None, + time_span: float = 300.0, + popups: bool = True, + **kwargs, + ): + if config is None: + config = RollingSignalPlotConfig(widget_class=self.__class__.__name__) + elif isinstance(config, dict): + config = RollingSignalPlotConfig(**config) + self.config: RollingSignalPlotConfig = config + + self._entries: dict[str, MonitoredSignal] = {} + self._connected_devices: list[str] = [] + self._paused = False + self._auto_follow = True + self._time_span = float(time_span) + self._refresh_interval = 200 # ms + self._latest_device_timestamp = 0.0 + self._date_axis: pg.DateAxisItem | None = None + self._default_bottom_axis = None + + super().__init__( + parent=parent, config=config, client=client, gui_id=gui_id, popups=popups, **kwargs + ) + + self.config.time_span = self._time_span + self._default_bottom_axis = self.plot_item.getAxis("bottom") + + self._y_quantity = "pressure" + self.plot_item.setLabel("bottom", "time before now", units="s") + # Log decades read much better without pyqtgraph's automatic SI prefix. + self.plot_item.getAxis("left").enableAutoSIPrefix(False) + self._update_y_label() + self.plot_item.enableAutoRange(x=False) + self.plot_item.enableAutoRange(y=True) + self.plot_item.showGrid(x=True, y=True, alpha=0.2) + + self._render_timer = QTimer(self) + self._render_timer.setTimerType(Qt.TimerType.CoarseTimer) + self._render_timer.timeout.connect(self._redraw) + self._render_timer.start(self._refresh_interval) + + # ---------------------------------------------------------------- device handling + @property + def devices(self) -> list[str]: + """Names of the currently monitored devices, in display order.""" + return list(self._entries) + + def set_devices(self, devices: list[str]) -> None: + """Replace the set of monitored devices. + + Devices that are already monitored keep their buffered data. + """ + devices = [str(d) for d in devices] + for name in [d for d in self._entries if d not in devices]: + self.remove_device(name, _resubscribe=False) + for name in devices: + if name not in self._entries: + self.add_device(name, _resubscribe=False) + # Re-order the internal dict so colors and legend follow the requested order. + self._entries = {name: self._entries[name] for name in devices if name in self._entries} + self._refresh_subscriptions() + self._assign_colors() + self._update_y_label() + self.devices_changed.emit(self.devices) + + def add_device(self, device: str, signal: str | None = None, _resubscribe: bool = True) -> None: + """Start monitoring ``device``. + + Args: + device: Name of the BEC device. + signal: Optional explicit signal key inside the readback message. Defaults + to the device name for signal-type devices (``EpicsSignalRO`` and + friends) or to the first hinted signal for composite devices. + _resubscribe: Internal flag, leave at the default. + """ + if device in self._entries: + return + if device not in self.dev: + logger.warning(f"PressureMonitor: device {device} is unknown to BEC, skipping.") + return + + signal_key, candidates = self._resolve_signal_key(device, signal) + units, source = self._signal_metadata(device, signal_key) + label = f"{device} [{units}]" if units else device + + curve = pg.PlotDataItem(name=label) + curve.setClipToView(True) + curve.setDownsampling(auto=True, method="peak") + self.plot_item.addItem(curve) + + self._entries[device] = MonitoredSignal( + device=device, + signal_key=signal_key, + candidate_keys=candidates, + label=label, + units=units, + source=source, + curve=curve, + ) + self._seed_from_cache(self._entries[device]) + if _resubscribe: + self._refresh_subscriptions() + self._assign_colors() + self._update_y_label() + self.devices_changed.emit(self.devices) + + def remove_device(self, device: str, _resubscribe: bool = True) -> None: + """Stop monitoring ``device`` and drop its buffered data.""" + entry = self._entries.pop(device, None) + if entry is None: + return + try: + self.plot_item.removeItem(entry.curve) + except Exception: # pragma: no cover - defensive, item may already be gone + logger.warning(f"PressureMonitor: could not remove curve for {device}.") + entry.buffer.clear() + if _resubscribe: + self._refresh_subscriptions() + self._assign_colors() + self._update_y_label() + self.devices_changed.emit(self.devices) + + def _resolve_signal_key(self, device: str, signal: str | None) -> tuple[str, set[str]]: + """Return the readback key to read and a set of acceptable alternatives.""" + obj = self.dev.get(device) + candidates: set[str] = {device} + + info = getattr(obj, "_info", {}) or {} + signals_info = info.get("signals", {}) or {} + hinted: list[str] = [] + for sig_name, sig_info in signals_info.items(): + obj_name = sig_info.get("obj_name") or f"{device}_{sig_name}" + candidates.add(obj_name) + if sig_info.get("kind_str") == "hinted": + hinted.append(obj_name) + + if signal: + key = signal if signal in candidates else f"{device}_{signal}" + candidates.add(signal) + candidates.add(key) + return key, candidates + + if isinstance(obj, BECSignal): + # ophyd Signals report themselves under their own name. + return device, candidates + if hinted: + return hinted[0], candidates + return device, candidates + + def _signal_metadata(self, device: str, signal_key: str) -> tuple[str, str]: + """Best-effort lookup of engineering units and EPICS source for a signal.""" + obj = self.dev.get(device) + units, source = "", "" + for getter in ("describe", "describe_configuration"): + try: + described = getattr(obj, getter)() or {} + except Exception: + continue + entry = described.get(signal_key) or described.get(device) or {} + if isinstance(entry, dict): + units = units or entry.get("units") or entry.get("egu") or "" + source = source or entry.get("source") or "" + if not units: + try: + units = getattr(obj, "egu", "") or "" + except Exception: + units = "" + return str(units), str(source) + + def _seed_from_cache(self, entry: MonitoredSignal) -> None: + """Prime a freshly added curve with the last cached reading, if available.""" + obj = self.dev.get(entry.device) + try: + reading = obj.read(cached=True) or {} + except Exception: + return + signals = reading if isinstance(reading, dict) else {} + for key in (entry.signal_key, *entry.candidate_keys): + payload = signals.get(key) + if isinstance(payload, dict) and payload.get("value") is not None: + entry.signal_key = key + self._append_sample(entry, payload) + return + + # ------------------------------------------------------------------ subscriptions + def _refresh_subscriptions(self) -> None: + """Re-subscribe the readback slot to exactly the monitored devices.""" + wanted = self.devices + if wanted == self._connected_devices: + return + if self._connected_devices: + self.bec_dispatcher.disconnect_slot( + self.on_device_readback, + [MessageEndpoints.device_readback(name) for name in self._connected_devices], + ) + if wanted: + self.bec_dispatcher.connect_slot( + self.on_device_readback, + [MessageEndpoints.device_readback(name) for name in wanted], + ) + self._connected_devices = wanted + self.config.devices = wanted + + @SafeSlot(dict, dict) + def on_device_readback(self, msg: dict, metadata: dict) -> None: + """Append incoming readback samples to the matching buffer.""" + if self._paused: + return + signals = msg.get("signals") or {} + if not signals: + return + for entry in self._entries.values(): + payload = signals.get(entry.signal_key) + if payload is None: + # The device may publish under a different key than we guessed. + for key in entry.candidate_keys: + if key in signals: + entry.signal_key = key + payload = signals[key] + break + if isinstance(payload, dict): + self._append_sample(entry, payload) + + def _append_sample(self, entry: MonitoredSignal, payload: dict) -> None: + value = payload.get("value") + if value is None or isinstance(value, (str, bytes, list, tuple, dict, np.ndarray)): + return + try: + value = float(value) + except (TypeError, ValueError): + return + if not math.isfinite(value): + return + + device_ts = payload.get("timestamp") + try: + device_ts = float(device_ts) + except (TypeError, ValueError): + device_ts = 0.0 + if math.isfinite(device_ts) and device_ts > 0: + self._latest_device_timestamp = max(self._latest_device_timestamp, device_ts) + + if self.config.timestamp_source == "device" and device_ts > 0: + timestamp = device_ts + else: + timestamp = time.time() + + entry.buffer.append(timestamp, value) + + # ------------------------------------------------------------------------ drawing + def _reference_time(self) -> float: + """The right edge of the moving window.""" + if self.config.timestamp_source == "device" and self._latest_device_timestamp > 0: + return self._latest_device_timestamp + return time.time() + + def _redraw(self) -> None: + """Trim the buffers to the window and push the data to the curves.""" + now = self._reference_time() + t_min = now - self._time_span + for entry in self._entries.values(): + # Keep a little margin so a curve does not visibly snap at the left edge. + entry.buffer.trim(t_min - 0.05 * self._time_span) + t, v = entry.buffer.data() + if t.size == 0: + entry.curve.setData([], []) + continue + x = t if self.config.x_mode == "absolute" else t - now + entry.curve.setData(x, v) + + if not self._auto_follow: + return + if self.config.x_mode == "absolute": + self.plot_item.setXRange(t_min, now, padding=0) + else: + self.plot_item.setXRange(-self._time_span, 0, padding=0) + + def _assign_colors(self) -> None: + """Give every curve a distinguishable color from the configured palette.""" + entries = list(self._entries.values()) + if not entries: + return + try: + colors = Colors.golden_angle_color( + colormap=self.config.color_palette, num=max(10, len(entries)), format="HEX" + ) + except Exception: # pragma: no cover - fall back to pyqtgraph defaults + colors = [pg.intColor(i, hues=max(9, len(entries))).name() for i in range(len(entries))] + for i, entry in enumerate(entries): + pen = pg.mkPen(color=colors[i % len(colors)], width=2) + entry.curve.setPen(pen) + + def _update_y_label(self) -> None: + """Show the engineering units on the y axis when all curves agree on them.""" + units = {entry.units for entry in self._entries.values() if entry.units} + label = self._y_quantity + if len(units) == 1: + label = f"{self._y_quantity} [{units.pop()}]" + self.plot_item.setLabel("left", label, units=None) + + @SafeProperty(str, doc="Quantity name shown on the y axis, e.g. 'pressure'.") + def y_quantity(self) -> str: + return self._y_quantity + + @y_quantity.setter + def y_quantity(self, value: str) -> None: + self._y_quantity = str(value) + self._update_y_label() + + def apply_theme(self, theme: str) -> None: + super().apply_theme(theme) + self._assign_colors() + + # --------------------------------------------------------------------- properties + @SafeProperty(float, doc="Displayed time window in seconds.") + def time_span(self) -> float: + return self._time_span + + @time_span.setter + def time_span(self, value: float) -> None: + value = float(value) + if value <= 0: + return + self._time_span = value + self.config.time_span = value + self._redraw() + + @SafeProperty(bool, doc="Freeze acquisition; the plot keeps showing the buffered data.") + def paused(self) -> bool: + return self._paused + + @paused.setter + def paused(self, value: bool) -> None: + self._paused = bool(value) + + @SafeProperty(bool, doc="Keep the x range locked to the moving window.") + def auto_follow(self) -> bool: + return self._auto_follow + + @auto_follow.setter + def auto_follow(self, value: bool) -> None: + self._auto_follow = bool(value) + if self._auto_follow: + self._redraw() + + @SafeProperty(int, doc="Redraw interval in milliseconds.") + def refresh_interval(self) -> int: + return self._refresh_interval + + @refresh_interval.setter + def refresh_interval(self, value: int) -> None: + value = max(20, int(value)) + self._refresh_interval = value + self._render_timer.start(value) + + @SafeProperty(str, doc="'relative' shows seconds before now, 'absolute' shows wall-clock time.") + def x_mode(self) -> str: + return self.config.x_mode + + @x_mode.setter + def x_mode(self, value: str) -> None: + value = "absolute" if str(value).lower().startswith("abs") else "relative" + if value == self.config.x_mode: + return + self.config.x_mode = value + if value == "absolute": + if self._date_axis is None: + self._date_axis = pg.DateAxisItem(orientation="bottom") + self.plot_item.setAxisItems({"bottom": self._date_axis}) + self.plot_item.setLabel("bottom", "time", units=None) + else: + if self._default_bottom_axis is not None: + self.plot_item.setAxisItems({"bottom": self._default_bottom_axis}) + self.plot_item.setLabel("bottom", "time before now", units="s") + self._redraw() + + @SafeProperty(str, doc="'local' uses arrival time, 'device' uses the readback timestamp.") + def timestamp_source(self) -> str: + return self.config.timestamp_source + + @timestamp_source.setter + def timestamp_source(self, value: str) -> None: + value = "device" if str(value).lower().startswith("dev") else "local" + if value == self.config.timestamp_source: + return + self.config.timestamp_source = value + self.clear_data() + + # ------------------------------------------------------------------------- export + @SafeSlot() + def clear_data(self) -> None: + """Drop all buffered samples.""" + for entry in self._entries.values(): + entry.buffer.clear() + entry.curve.setData([], []) + self._latest_device_timestamp = 0.0 + + def displayed_data(self) -> dict[str, dict]: + """Return the data currently inside the time window. + + Returns: + Mapping ``device -> {"timestamps": ndarray, "values": ndarray, + "units": str, "source": str, "signal": str}``. + """ + now = self._reference_time() + t_min = now - self._time_span + out: dict[str, dict] = {} + for name, entry in self._entries.items(): + t, v = entry.buffer.window(t_min, now) + out[name] = { + "timestamps": t, + "values": v, + "units": entry.units, + "source": entry.source, + "signal": entry.signal_key, + } + return out + + def displayed_data_as_text( + self, + layout: Literal["wide", "long"] = "wide", + delimiter: str = ",", + include_header_comments: bool = True, + ) -> str: + """Serialize the currently displayed data to delimited text. + + Args: + layout: ``"wide"`` writes one column per device on a union time axis + (missing samples are left empty), ``"long"`` writes one row per sample. + delimiter: Field separator, ``","`` for CSV and ``"\\t"`` for the clipboard. + include_header_comments: Prepend ``#`` comment lines with export metadata. + """ + data = self.displayed_data() + stream = io.StringIO() + writer = csv.writer(stream, delimiter=delimiter, lineterminator="\n") + + if include_header_comments: + now = self._reference_time() + stream.write(f"# BEC rolling signal monitor export\n") + stream.write(f"# exported_at: {datetime.now().astimezone().isoformat()}\n") + stream.write(f"# time_span_s: {self._time_span}\n") + stream.write(f"# timestamp_source: {self.config.timestamp_source}\n") + stream.write(f"# window_end_epoch: {now:.6f}\n") + for name, entry in data.items(): + stream.write( + f"# device: {name}; signal: {entry['signal']}; " + f"units: {entry['units'] or 'n/a'}; source: {entry['source'] or 'n/a'}; " + f"points: {entry['timestamps'].size}\n" + ) + + if layout == "long": + writer.writerow(["device", "timestamp", "iso_time", "value", "units"]) + for name, entry in data.items(): + for t, v in zip(entry["timestamps"], entry["values"]): + writer.writerow( + [ + name, + f"{t:.6f}", + datetime.fromtimestamp(t).astimezone().isoformat(), + f"{v!r}", + entry["units"], + ] + ) + return stream.getvalue() + + names = list(data) + arrays = [data[name]["timestamps"] for name in names if data[name]["timestamps"].size] + if not arrays: + writer.writerow(["timestamp", "iso_time", *names]) + return stream.getvalue() + + axis = np.unique(np.concatenate(arrays)) + columns: list[np.ndarray] = [] + for name in names: + column = np.full(axis.size, np.nan) + t = data[name]["timestamps"] + if t.size: + idx = np.searchsorted(axis, t) + column[idx] = data[name]["values"] + columns.append(column) + + writer.writerow(["timestamp", "iso_time", *names]) + for row, t in enumerate(axis): + cells = ["" if np.isnan(col[row]) else repr(float(col[row])) for col in columns] + writer.writerow([f"{t:.6f}", datetime.fromtimestamp(t).astimezone().isoformat(), *cells]) + return stream.getvalue() + + def export_csv( + self, + path: str, + layout: Literal["wide", "long"] = "wide", + include_header_comments: bool = True, + ) -> str: + """Write the currently displayed data to ``path`` as CSV and return the path.""" + text = self.displayed_data_as_text( + layout=layout, delimiter=",", include_header_comments=include_header_comments + ) + with open(path, "w", encoding="utf-8", newline="") as file: + file.write(text) + logger.info(f"PressureMonitor: exported displayed data to {path}") + return path + + # ------------------------------------------------------------------------ cleanup + def cleanup(self) -> None: + self._render_timer.stop() + for name in list(self._entries): + self.remove_device(name, _resubscribe=False) + self._connected_devices = [] + super().cleanup() + + +# -------------------------------------------------------------------------------------- +# Composite widget: device selection + controls + plot +# -------------------------------------------------------------------------------------- +class PressureMonitorConfig(ConnectionConfig): + """Serializable configuration of :class:`PressureMonitor`.""" + + device_tags: list[str] = Field( + default_factory=list, description="Only list devices carrying one of these BEC deviceTags." + ) + device_classes: list[str] = Field( + default_factory=list, description="Only list devices of these ophyd classes." + ) + readout_priorities: list[str] = Field( + default_factory=list, description="Only list devices with these readout priorities." + ) + signals_only: bool = Field( + True, description="Restrict the list to signal-type devices (EpicsSignalRO and friends)." + ) + + +class PressureMonitor(BECWidget, QWidget): + """Select devices, choose a time span, watch them scroll by, export what you see. + + Args: + parent: Optional parent widget. + client: Optional BEC client. + config: :class:`PressureMonitorConfig` or dict. + gui_id: Optional GUI identifier. + devices: Devices to monitor right away. + time_span: Initial width of the moving window in seconds. + device_tags: Restrict the selectable devices to these BEC device tags. + device_classes: Restrict the selectable devices to these ophyd class names. + signals_only: Only list signal-type devices. Defaults to True. + """ + + ICON_NAME = "readiness_score" + PLUGIN = True + RPC = True + + USER_ACCESS = [ + *BECWidget.USER_ACCESS, + "plot", + "select_devices", + "selected_devices", + "available_devices", + "time_span", + "time_span.setter", + "paused", + "paused.setter", + "log_y", + "log_y.setter", + "device_tags", + "device_tags.setter", + "device_classes", + "device_classes.setter", + "signals_only", + "signals_only.setter", + "refresh_device_list", + "clear_data", + "export_csv", + "copy_to_clipboard", + ] + + selection_changed = QSignal(list) + # Emitted from the (non-Qt) BEC callback thread, delivered to the GUI thread. + device_config_updated = QSignal() + + def __init__( + self, + parent: QWidget | None = None, + client: "BECClient | None" = None, + config: PressureMonitorConfig | dict | None = None, + gui_id: str | None = None, + devices: list[str] | None = None, + time_span: float = 300.0, + device_tags: list[str] | None = None, + device_classes: list[str] | None = None, + signals_only: bool = True, + **kwargs, + ): + if config is None: + config = PressureMonitorConfig(widget_class=self.__class__.__name__) + elif isinstance(config, dict): + config = PressureMonitorConfig(**config) + self.config: PressureMonitorConfig = config + if device_tags is not None: + self.config.device_tags = list(device_tags) + if device_classes is not None: + self.config.device_classes = list(device_classes) + self.config.signals_only = bool(signals_only) + + super().__init__(parent=parent, client=client, config=config, gui_id=gui_id, **kwargs) + self.get_bec_shortcuts() + + self._updating_list = False + self._device_update_callback_id: int | None = None + self._init_ui(time_span=time_span) + self.refresh_device_list() + + # The BEC callback fires on a worker thread; hop to the GUI thread via a signal. + self.device_config_updated.connect(self.refresh_device_list) + try: + self._device_update_callback_id = self.client.callbacks.register( + EventType.DEVICE_UPDATE, self._on_device_config_update + ) + except Exception: # pragma: no cover - older bec_lib without this event type + logger.debug("PressureMonitor: device_update callback not available.") + + if devices: + self.select_devices(devices) + + # ------------------------------------------------------------------------- layout + def _init_ui(self, time_span: float) -> None: + outer = QVBoxLayout(self) + outer.setContentsMargins(4, 4, 4, 4) + outer.setSpacing(4) + + # --- control bar ------------------------------------------------------- + controls = QHBoxLayout() + controls.setSpacing(6) + + controls.addWidget(QLabel("Time span:")) + self.span_spinbox = QDoubleSpinBox() + self.span_spinbox.setRange(1.0, 86_400.0) + self.span_spinbox.setDecimals(1) + self.span_spinbox.setSingleStep(10.0) + self.span_spinbox.setSuffix(" s") + self.span_spinbox.setValue(float(time_span)) + self.span_spinbox.setToolTip("Width of the moving time window") + controls.addWidget(self.span_spinbox) + + self.follow_checkbox = QCheckBox("Follow") + self.follow_checkbox.setChecked(True) + self.follow_checkbox.setToolTip("Keep the x range locked to the moving window") + controls.addWidget(self.follow_checkbox) + + self.log_checkbox = QCheckBox("Log y") + self.log_checkbox.setToolTip("Logarithmic pressure axis") + controls.addWidget(self.log_checkbox) + + self.pause_button = QPushButton("Pause") + self.pause_button.setCheckable(True) + self.pause_button.setToolTip("Stop appending new samples; the buffer is kept") + controls.addWidget(self.pause_button) + + self.clear_button = QPushButton("Clear") + self.clear_button.setToolTip("Discard all buffered samples") + controls.addWidget(self.clear_button) + + controls.addStretch(1) + + self.copy_button = QPushButton("Copy") + self.copy_button.setToolTip("Copy the displayed data to the clipboard (tab separated)") + controls.addWidget(self.copy_button) + + self.export_button = QPushButton("Export CSV…") + self.export_button.setToolTip("Write the data currently displayed to a CSV file") + controls.addWidget(self.export_button) + + outer.addLayout(controls) + + # --- splitter: device selection | plot --------------------------------- + self.splitter = QSplitter(Qt.Orientation.Horizontal, self) + + selection_panel = QWidget(self.splitter) + selection_layout = QVBoxLayout(selection_panel) + selection_layout.setContentsMargins(0, 0, 0, 0) + selection_layout.setSpacing(3) + + self.filter_edit = QLineEdit() + self.filter_edit.setPlaceholderText("Filter devices…") + self.filter_edit.setClearButtonEnabled(True) + selection_layout.addWidget(self.filter_edit) + + self.device_list = QListWidget() + self.device_list.setSelectionMode(QListWidget.SelectionMode.NoSelection) + self.device_list.setToolTip("Tick the devices to plot") + selection_layout.addWidget(self.device_list) + + list_buttons = QHBoxLayout() + self.none_button = QPushButton("None") + self.none_button.setToolTip("Uncheck all devices") + self.refresh_button = QPushButton("Refresh") + self.refresh_button.setToolTip("Reload the device list from BEC") + list_buttons.addWidget(self.none_button) + list_buttons.addWidget(self.refresh_button) + selection_layout.addLayout(list_buttons) + + selection_panel.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Expanding) + selection_panel.setMaximumWidth(280) + + self.plot = RollingSignalPlot(parent=self.splitter, client=self.client, time_span=time_span) + + self.splitter.addWidget(selection_panel) + self.splitter.addWidget(self.plot) + self.splitter.setStretchFactor(0, 0) + self.splitter.setStretchFactor(1, 1) + outer.addWidget(self.splitter) + + # --- wiring ------------------------------------------------------------ + self.span_spinbox.valueChanged.connect(self._on_span_changed) + self.follow_checkbox.toggled.connect(self._on_follow_toggled) + self.log_checkbox.toggled.connect(self._on_log_toggled) + self.pause_button.toggled.connect(self._on_pause_toggled) + self.clear_button.clicked.connect(self.clear_data) + self.copy_button.clicked.connect(self.copy_to_clipboard) + self.export_button.clicked.connect(self._on_export_clicked) + self.filter_edit.textChanged.connect(self._apply_list_filter) + self.none_button.clicked.connect(self._on_select_none) + self.refresh_button.clicked.connect(self.refresh_device_list) + self.device_list.itemChanged.connect(self._on_item_changed) + + # ------------------------------------------------------------------ device list + @staticmethod + def _device_tags_of(obj) -> set[str]: + """Read the BEC device tags, tolerating the different bec_lib accessors.""" + for accessor in ("get_device_tags", "device_tags"): + try: + tags = getattr(obj, accessor) + except Exception: + continue + if callable(tags): + try: + tags = tags() + except Exception: + continue + if tags: + return {str(tag) for tag in tags} + try: + return {str(tag) for tag in (obj._config.get("deviceTags") or [])} + except Exception: + return set() + + def _device_matches_filters(self, name: str) -> bool: + obj = self.dev.get(name) + if obj is None: + return False + if self.config.signals_only and not isinstance(obj, BECSignal): + return False + if not self.config.signals_only and not isinstance(obj, (BECSignal, BECDevice)): + return False + if self.config.device_classes: + class_name = getattr(obj, "_class_name", type(obj).__name__) + if class_name not in self.config.device_classes: + return False + if self.config.device_tags: + if not self._device_tags_of(obj).intersection(self.config.device_tags): + return False + if self.config.readout_priorities: + try: + priority = ReadoutPriority(obj.readout_priority).value + except Exception: + priority = None + if priority not in self.config.readout_priorities: + return False + return True + + @SafeSlot() + def refresh_device_list(self) -> None: + """Reload the selectable devices from the BEC device manager.""" + selected = set(self.selected_devices) + names = sorted(name for name in self.dev if self._device_matches_filters(name)) + + self._updating_list = True + try: + self.device_list.clear() + for name in names: + item = QListWidgetItem(name) + item.setFlags(item.flags() | Qt.ItemFlag.ItemIsUserCheckable) + item.setCheckState( + Qt.CheckState.Checked if name in selected else Qt.CheckState.Unchecked + ) + obj = self.dev.get(name) + class_name = getattr(obj, "_class_name", type(obj).__name__) + item.setToolTip(f"{name} ({class_name})") + self.device_list.addItem(item) + finally: + self._updating_list = False + self._apply_list_filter(self.filter_edit.text()) + # Drop monitored devices that disappeared from the config. + for name in self.plot.devices: + if name not in names: + self.plot.remove_device(name) + + @property + def available_devices(self) -> list[str]: + """Device names currently offered in the selection list.""" + return [self.device_list.item(i).text() for i in range(self.device_list.count())] + + @SafeSlot(str) + def _apply_list_filter(self, text: str) -> None: + needle = (text or "").strip().lower() + for i in range(self.device_list.count()): + item = self.device_list.item(i) + item.setHidden(bool(needle) and needle not in item.text().lower()) + + @SafeSlot(object) + def _on_item_changed(self, _item) -> None: + if self._updating_list: + return + self.select_devices(self._checked_devices()) + + def _checked_devices(self) -> list[str]: + out = [] + for i in range(self.device_list.count()): + item = self.device_list.item(i) + if item.checkState() == Qt.CheckState.Checked: + out.append(item.text()) + return out + + @SafeSlot() + def _on_select_none(self) -> None: + self.select_devices([]) + + def _on_device_config_update(self, *_args, **_kwargs) -> None: # pragma: no cover + """Called by BEC on a worker thread whenever the device config changes.""" + self.device_config_updated.emit() + + # -------------------------------------------------------------------- selection + @property + def selected_devices(self) -> list[str]: + """Devices currently plotted.""" + return self.plot.devices + + def select_devices(self, devices: list[str]) -> None: + """Plot exactly ``devices`` (checkboxes are synchronized).""" + devices = [d for d in devices] + self.plot.set_devices(devices) + wanted = set(self.plot.devices) + + self._updating_list = True + try: + for i in range(self.device_list.count()): + item = self.device_list.item(i) + state = ( + Qt.CheckState.Checked + if item.text() in wanted + else Qt.CheckState.Unchecked + ) + if item.checkState() != state: + item.setCheckState(state) + finally: + self._updating_list = False + self.selection_changed.emit(self.plot.devices) + + # -------------------------------------------------------------------- properties + @SafeProperty(float, doc="Displayed time window in seconds.") + def time_span(self) -> float: + return self.plot.time_span + + @time_span.setter + def time_span(self, value: float) -> None: + self.plot.time_span = float(value) + if self.span_spinbox.value() != float(value): + self.span_spinbox.setValue(float(value)) + + @SafeProperty(bool, doc="Freeze acquisition.") + def paused(self) -> bool: + return self.plot.paused + + @paused.setter + def paused(self, value: bool) -> None: + self.pause_button.setChecked(bool(value)) + + @SafeProperty(bool, doc="Logarithmic pressure axis.") + def log_y(self) -> bool: + return self.plot.y_log + + @log_y.setter + def log_y(self, value: bool) -> None: + self.log_checkbox.setChecked(bool(value)) + + @SafeProperty("QStringList", doc="Only list devices carrying one of these BEC device tags.") + def device_tags(self) -> list[str]: + return list(self.config.device_tags) + + @device_tags.setter + def device_tags(self, value: list[str]) -> None: + self.config.device_tags = [str(v) for v in (value or [])] + self.refresh_device_list() + + @SafeProperty("QStringList", doc="Only list devices of these ophyd classes.") + def device_classes(self) -> list[str]: + return list(self.config.device_classes) + + @device_classes.setter + def device_classes(self, value: list[str]) -> None: + self.config.device_classes = [str(v) for v in (value or [])] + self.refresh_device_list() + + @SafeProperty(bool, doc="Restrict the list to signal-type devices.") + def signals_only(self) -> bool: + return self.config.signals_only + + @signals_only.setter + def signals_only(self, value: bool) -> None: + self.config.signals_only = bool(value) + self.refresh_device_list() + + # ------------------------------------------------------------------------- slots + @SafeSlot(float) + def _on_span_changed(self, value: float) -> None: + self.plot.time_span = float(value) + + @SafeSlot(bool) + def _on_follow_toggled(self, checked: bool) -> None: + self.plot.auto_follow = checked + + @SafeSlot(bool) + def _on_log_toggled(self, checked: bool) -> None: + self.plot.y_log = checked + + @SafeSlot(bool) + def _on_pause_toggled(self, checked: bool) -> None: + self.plot.paused = checked + self.pause_button.setText("Resume" if checked else "Pause") + + @SafeSlot() + def clear_data(self) -> None: + """Discard all buffered samples.""" + self.plot.clear_data() + + @SafeSlot() + def copy_to_clipboard(self) -> None: + """Copy the currently displayed data to the clipboard as tab separated text.""" + text = self.plot.displayed_data_as_text(layout="wide", delimiter="\t") + clipboard = QGuiApplication.clipboard() + if clipboard is not None: + clipboard.setText(text) + + @SafeSlot() + def _on_export_clicked(self) -> None: + default_name = f"pressure_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" + path, _ = QFileDialog.getSaveFileName( + self, "Export displayed data", default_name, "CSV files (*.csv);;All files (*)" + ) + if path: + self.plot.export_csv(path) + + def export_csv(self, path: str, layout: Literal["wide", "long"] = "wide") -> str: + """Write the data currently displayed to ``path`` as CSV.""" + return self.plot.export_csv(path, layout=layout) + + # ----------------------------------------------------------------------- cleanup + def cleanup(self) -> None: + if self._device_update_callback_id is not None: + try: + self.client.callbacks.remove(self._device_update_callback_id) + except Exception: # pragma: no cover + logger.debug("PressureMonitor: could not remove device_update callback.") + self._device_update_callback_id = None + super().cleanup() + + +if __name__ == "__main__": # pragma: no cover + import sys + + from qtpy.QtWidgets import QApplication + + app = QApplication(sys.argv) + widget = PressureMonitor(time_span=300.0) + widget.resize(1100, 600) + widget.show() + sys.exit(app.exec_()) diff --git a/xil_bec/bec_widgets/widgets/pressure_monitor/pressure_monitor.pyproject b/xil_bec/bec_widgets/widgets/pressure_monitor/pressure_monitor.pyproject new file mode 100644 index 0000000..96cb305 --- /dev/null +++ b/xil_bec/bec_widgets/widgets/pressure_monitor/pressure_monitor.pyproject @@ -0,0 +1 @@ +{'files': ['pressure_monitor.py']} diff --git a/xil_bec/bec_widgets/widgets/pressure_monitor/pressure_monitor_plugin.py b/xil_bec/bec_widgets/widgets/pressure_monitor/pressure_monitor_plugin.py new file mode 100644 index 0000000..9cac852 --- /dev/null +++ b/xil_bec/bec_widgets/widgets/pressure_monitor/pressure_monitor_plugin.py @@ -0,0 +1,62 @@ +# Copyright (C) 2022 The Qt Company Ltd. +# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause + +from qtpy.QtDesigner import QDesignerCustomWidgetInterface +from qtpy.QtWidgets import QWidget + +from bec_widgets.utils.bec_designer import designer_material_icon + +# NOTE: Qt Designer imports this file as a top-level script, so the import below must be +# absolute. Replace `my_beamline_plugin` with the package name of your BEC plugin repo. +from xil_bec.bec_widgets.widgets.pressure_monitor.pressure_monitor import ( + PressureMonitor, +) + +DOM_XML = """ + + + + +""" + + +class PressureMonitorPlugin(QDesignerCustomWidgetInterface): # pragma: no cover + def __init__(self): + super().__init__() + self._form_editor = None + + def createWidget(self, parent): + if parent is None: + return QWidget() + t = PressureMonitor(parent) + return t + + def domXml(self): + return DOM_XML + + def group(self): + return "BEC Plots" + + def icon(self): + return designer_material_icon(PressureMonitor.ICON_NAME) + + def includeFile(self): + return "pressure_monitor" + + def initialize(self, form_editor): + self._form_editor = form_editor + + def isContainer(self): + return False + + def isInitialized(self): + return self._form_editor is not None + + def name(self): + return "PressureMonitor" + + def toolTip(self): + return "Rolling time-window monitor for pressure gauges and other BEC signals" + + def whatsThis(self): + return self.toolTip() diff --git a/xil_bec/bec_widgets/widgets/pressure_monitor/register_pressure_monitor.py b/xil_bec/bec_widgets/widgets/pressure_monitor/register_pressure_monitor.py new file mode 100644 index 0000000..450e401 --- /dev/null +++ b/xil_bec/bec_widgets/widgets/pressure_monitor/register_pressure_monitor.py @@ -0,0 +1,18 @@ +def main(): # pragma: no cover + from qtpy import PYSIDE6 + + if not PYSIDE6: + print("PYSIDE6 is not available in the environment. Cannot patch designer.") + return + from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection + + # NOTE: replace `my_beamline_plugin` with the package name of your BEC plugin repo. + from xil_bec.bec_widgets.widgets.pressure_monitor.pressure_monitor_plugin import ( + PressureMonitorPlugin, + ) + + QPyDesignerCustomWidgetCollection.addCustomWidget(PressureMonitorPlugin()) + + +if __name__ == "__main__": # pragma: no cover + main()