From 3705d16e5152d83a8000f9dda05d0d6fbb24e5cf Mon Sep 17 00:00:00 2001 From: appel_c Date: Wed, 17 Jun 2026 11:11:05 +0200 Subject: [PATCH] feat(bec_roi_signals): initial implementation of ROI signals and processing --- .../base_classes/psi_device_base.py | 14 ++ ophyd_devices/sim/sim_camera.py | 80 ++++++++ .../utils/bec_roi_signals/__init__.py | 1 + .../utils/bec_roi_signals/ad_roi_signal.py | 39 ++++ .../utils/bec_roi_signals/roi_processing.py | 189 ++++++++++++++++++ .../utils/bec_roi_signals/signals.py | 187 +++++++++++++++++ 6 files changed, 510 insertions(+) create mode 100644 ophyd_devices/utils/bec_roi_signals/__init__.py create mode 100644 ophyd_devices/utils/bec_roi_signals/ad_roi_signal.py create mode 100644 ophyd_devices/utils/bec_roi_signals/roi_processing.py create mode 100644 ophyd_devices/utils/bec_roi_signals/signals.py diff --git a/ophyd_devices/interfaces/base_classes/psi_device_base.py b/ophyd_devices/interfaces/base_classes/psi_device_base.py index 9d0bdf7..aa34c7f 100644 --- a/ophyd_devices/interfaces/base_classes/psi_device_base.py +++ b/ophyd_devices/interfaces/base_classes/psi_device_base.py @@ -73,6 +73,20 @@ class PSIDeviceBase(Device): self.scan_info = scan_info self.on_init() + def wait_for_connection(self, *args, **kwargs): + """Wait for the device to be connected. + + This method is called from BEC after the device is added to the device manager. + It waits for the device to be connected and then calls the on_connected method. + """ + for walk in self.walk_components(): + if ( + hasattr(walk.item.cls, "REQUIRES_WAIT_FOR_CONNECTION") + and walk.item.cls.REQUIRES_WAIT_FOR_CONNECTION is True + ): + getattr(self, walk.dotted_name).wait_for_connection(*args, **kwargs) + super().wait_for_connection(*args, **kwargs) + ######################################## # Additional Properties and Attributes # ######################################## diff --git a/ophyd_devices/sim/sim_camera.py b/ophyd_devices/sim/sim_camera.py index 83b0e77..51690ca 100644 --- a/ophyd_devices/sim/sim_camera.py +++ b/ophyd_devices/sim/sim_camera.py @@ -9,10 +9,78 @@ from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase from ophyd_devices.sim.sim_data import SimulatedDataCamera from ophyd_devices.sim.sim_signals import ReadOnlySignal, SetableSignal from ophyd_devices.sim.sim_utils import H5Writer +from ophyd_devices.utils.bec_roi_signals import LITERAL_ROI_PROCESSING_CONFIG, ROIProcessing from ophyd_devices.utils.bec_signals import FileEventSignal, PreviewSignal logger = bec_logger.logger +SIM_CONFIG: LITERAL_ROI_PROCESSING_CONFIG = { + "basic_statistics": {"scalar_outputs": ["sum", "mean", "min", "max"], "waveform_outputs": []} +} + + +class SimCameraROIProcessing(ROIProcessing): + """ROI processing signal for simulated camera setups.""" + + def get_scalar_outputs(self) -> list[str]: + return SIM_CONFIG["basic_statistics"]["scalar_outputs"] + + def get_waveform_outputs(self) -> list[str]: + return SIM_CONFIG["basic_statistics"]["waveform_outputs"] + + def get_available_analysis_operations(self) -> list[str]: + return list(SIM_CONFIG.keys()) + + def wait_for_connection(self, *args, **kwargs): + super().wait_for_connection(*args, **kwargs) + self.parent.image.subscribe( + self._on_image_update, event_type=self.parent.image.SUB_VALUE, run=False + ) + + def _on_image_update(self, value, **kwargs): + """Callback for image updates.""" + if not self.active.get(): + return # Do nothing if not enabled + + if not isinstance(value, np.ndarray): + logger.warning(f"Received non-numpy array value: {value}") + return # Do nothing if not numpy array + if "basic_statistics" not in self.selected_operations.get(): + return + roi_image = self._extract_roi(value) + if roi_image.size == 0: + logger.warning(f"ROI {self.roi_name.get()} is outside image bounds or has zero size.") + return + self.result_scalar.put(self._compute_basic_statistics(roi_image)) + + def _extract_roi(self, image: np.ndarray) -> np.ndarray: + """Return the configured rectangular ROI clipped to the image bounds.""" + x = int(self.x.get()) + y = int(self.y.get()) + width = int(self.width.get()) + height = int(self.height.get()) + if width <= 0 or height <= 0: + return image[0:0, 0:0] + + x0 = max(0, x) + y0 = max(0, y) + x1 = min(image.shape[1], x + width) + y1 = min(image.shape[0], y + height) + if x1 <= x0 or y1 <= y0: + return image[0:0, 0:0] + return image[y0:y1, x0:x1] + + def _compute_basic_statistics(self, image: np.ndarray) -> dict[str, dict[str, float]]: + """Compute basic statistics for the given image.""" + timestamp = self.parent.image.timestamp + values = { + "sum": {"value": float(np.sum(image)), "timestamp": timestamp}, + "mean": {"value": float(np.mean(image)), "timestamp": timestamp}, + "min": {"value": float(np.min(image)), "timestamp": timestamp}, + "max": {"value": float(np.max(image)), "timestamp": timestamp}, + } + return {name: values[name] for name in self.get_scalar_outputs() if name in values} + class SimCameraControl(Device): """SimCamera Control layer""" @@ -74,6 +142,18 @@ class SimCamera(PSIDeviceBase, SimCameraControl): """ + roi_processing = Cpt( + SimCameraROIProcessing, + name="roi_processing", + active=True, + x=25, + y=25, + width=50, + height=50, + selected_operations=["basic_statistics"], + kind=Kind.normal, + ) + def __init__(self, name: str, scan_info=None, device_manager=None, **kwargs): super().__init__(name=name, scan_info=scan_info, device_manager=device_manager, **kwargs) self.file_path = None diff --git a/ophyd_devices/utils/bec_roi_signals/__init__.py b/ophyd_devices/utils/bec_roi_signals/__init__.py new file mode 100644 index 0000000..8082d63 --- /dev/null +++ b/ophyd_devices/utils/bec_roi_signals/__init__.py @@ -0,0 +1 @@ +from .roi_processing import LITERAL_ROI_PROCESSING_CONFIG, ROIProcessing diff --git a/ophyd_devices/utils/bec_roi_signals/ad_roi_signal.py b/ophyd_devices/utils/bec_roi_signals/ad_roi_signal.py new file mode 100644 index 0000000..8662b3b --- /dev/null +++ b/ophyd_devices/utils/bec_roi_signals/ad_roi_signal.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +from ophyd import Component as Cpt +from ophyd import Device + +from ophyd_devices.devices.areadetector.plugins import ROIPlugin, StatsPlugin +from ophyd_devices.utils.bec_roi_signals.roi_processing import ( + LITERAL_ROI_PROCESSING_CONFIG, + ROIProcessing, +) + +NDPLUGIN_STATS_CONFIG: LITERAL_ROI_PROCESSING_CONFIG = { + "basic_statistics": { + "scalar_outputs": ["sum", "mean", "min", "max", "sigma"], + "waveform_outputs": [], + } +} + + +class ADROIProcessing(ROIProcessing): + """ROI processing signal for AD detector setups at PSI.""" + + roi1 = Cpt(ROIPlugin, prefix="ROI1:", kind="normal") + stats1 = Cpt(StatsPlugin, prefix="STATS1:", kind="normal") + + def get_scalar_outputs(self) -> list[str]: + return NDPLUGIN_STATS_CONFIG["basic_statistics"]["scalar_outputs"] + + def get_waveform_outputs(self) -> list[str]: + return NDPLUGIN_STATS_CONFIG["basic_statistics"]["waveform_outputs"] + + def get_available_analysis_operations(self) -> list[str]: + return list(NDPLUGIN_STATS_CONFIG.keys()) + + +class MyDevice(Device): + """Example device with ROI processing.""" + + roi_processing = Cpt(ADROIProcessing, prefix="ROI1:", kind="normal") diff --git a/ophyd_devices/utils/bec_roi_signals/roi_processing.py b/ophyd_devices/utils/bec_roi_signals/roi_processing.py new file mode 100644 index 0000000..a2f046e --- /dev/null +++ b/ophyd_devices/utils/bec_roi_signals/roi_processing.py @@ -0,0 +1,189 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, Literal + +from bec_lib.endpoints import MessageEndpoints +from bec_lib.logger import bec_logger +from bec_lib.messages import ROIAnalysisConfig, ROIAvailableAnalysisMessage, ROIConfigurationMessage +from bec_lib.redis_connector import RedisConnector +from ophyd import Component as Cpt +from ophyd import Device, Signal, SignalRO +from ophyd.device import required_for_connection + +from ophyd_devices.utils.bec_roi_signals.signals import ( + AvailableOperationsSignal, + ConfigUpdateReceivedSignal, + SelectedOperationSignal, +) +from ophyd_devices.utils.bec_signals import DynamicSignal + +if TYPE_CHECKING: + from bec_lib.connector import MessageObject + +logger = bec_logger.logger + +LITERAL_ROI_PROCESSING_CONFIG = dict[ + str, dict[Literal["scalar_outputs", "waveform_outputs"], list[str]] +] + + +class ROIProcessing(Device, ABC): + """Abstract base class for ROI processing signals""" + + REQUIRES_WAIT_FOR_CONNECTION = True + + result_scalar = Cpt( + DynamicSignal, + name="result_scalar", + max_size=1000, + signals=[], + ndim=0, + async_update={"type": "add", "max_shape": [None]}, + ) + result_waveform = Cpt( + DynamicSignal, + name="result_waveform", + max_size=1000, + signals=[], + ndim=1, + async_update={"type": "add", "max_shape": [None, None]}, + ) + + active = Cpt(Signal, value=False, kind="config") + roi_name = Cpt(Signal, value="roi", kind="config") + x = Cpt(Signal, value=0, kind="config") + y = Cpt(Signal, value=0, kind="config") + width = Cpt(Signal, value=10, kind="config") + height = Cpt(Signal, value=10, kind="config") + selected_operations = Cpt(SelectedOperationSignal, value=[], kind="config") + available_operations = Cpt(AvailableOperationsSignal, kind="config") + update_received = Cpt(ConfigUpdateReceivedSignal, kind="omitted") + + def __init__(self, *args, **kwargs): + kwargs, signal_kwargs = self._get_kwargs_for_signals(kwargs) + super().__init__(*args, **kwargs) + self._signal_kwargs = signal_kwargs + self._connector: RedisConnector | None = None + # Setup the DynamicSignal outputs based on the available operations and outputs + self._prepare_result_signals() + + def _prepare_result_signals(self): + """Prepare the result signals based on the available operations and outputs.""" + self.result_scalar.signals = self.result_scalar._unify_signals(self.get_scalar_outputs()) + self.result_waveform.signals = self.result_waveform._unify_signals( + self.get_waveform_outputs() + ) + + def _get_kwargs_for_signals(self, kwargs) -> tuple[dict, dict[str, dict]]: + """Get the kwargs for the signals of this device.""" + ret_kwargs = {} + for k in ["active", "roi_name", "x", "y", "width", "height", "selected_operations"]: + if k in kwargs: + ret_kwargs[k] = kwargs.pop(k) + return kwargs, ret_kwargs + + @required_for_connection + def wait_for_connection(self, all_signals=False, timeout=...): + """Wait for the ROI processing signal to be connected to Redis.""" + self._connector: RedisConnector | None = ( + self.root.device_manager.connector if hasattr(self.root, "device_manager") else None + ) + if self._connector is None: + raise RuntimeError( + f"Signal {self.name} is not connected to Redis, please provide a Redis Connector during the initialization." + ) + super().wait_for_connection(all_signals, timeout) + # Set initial values for the configuration signals from the kwargs provided during initialization + for signal_name, value in self._signal_kwargs.items(): + signal = getattr(self, signal_name) + signal.put(value) + + # Connect to relevant endpoint for ROI configuration updates + self.update_received.subscribe( + self._on_config_update, event_type=self.update_received.SUB_VALUE, run=False + ) + self._connector.register( + MessageEndpoints.roi_config(device=self.root.name, signal=self.endpoint_signal_name), + cb=self.receive_roi_configuration_message, + ) + self.publish_available_analysis() + + def destroy(self): + """Clean up the ROI processing signal and unregister from Redis.""" + if self._connector is not None: + self._connector.unregister( + MessageEndpoints.roi_config( + device=self.root.name, signal=self.endpoint_signal_name + ), + cb=self.receive_roi_configuration_message, + ) + super().destroy() + + @abstractmethod + def get_scalar_outputs(self) -> list[str]: + """Return the scalar output signal for the ROI processing signal.""" + + @abstractmethod + def get_waveform_outputs(self) -> list[str]: + """Return the waveform output signal for the ROI processing signal.""" + + @abstractmethod + def get_available_analysis_operations(self) -> list[str]: + """Return the available analysis operations for the ROI processing signal.""" + + @property + def endpoint_signal_name(self) -> str: + """Return the component signal name used for BEC ROI endpoints.""" + return getattr(self, "dotted_name", None) or getattr(self, "attr_name", None) or self.name + + def compile_roi_analysis_config(self) -> ROIAnalysisConfig: + """Compile the ROI analysis configuration from the available operations and outputs.""" + return ROIAnalysisConfig( + available_operations=self.get_available_analysis_operations(), + waveform_results=self.get_waveform_outputs(), + scalar_results=self.get_scalar_outputs(), + ) + + def publish_available_analysis(self) -> None: + """Publish the available ROI analysis operations and result signals.""" + if self._connector is None: + raise RuntimeError( + f"Signal {self.name} is not connected to Redis and cannot publish ROI analysis." + ) + message = ROIAvailableAnalysisMessage( + device=self.root.name, + signal=self.endpoint_signal_name, + config=self.compile_roi_analysis_config(), + ) + self._connector.set_and_publish( + MessageEndpoints.available_roi_analysis( + device=self.root.name, signal=self.endpoint_signal_name + ), + message, + ) + + def receive_roi_configuration_message(self, message: MessageObject): + """Receive a ROI configuration message and update the ROI processing signal accordingly.""" + message: ROIConfigurationMessage = message.value + if message.device != self.root.name or message.signal != self.endpoint_signal_name: + logger.warning( + f"Ignoring ROI configuration for {message.device}.{message.signal}; " + f"this ROIProcessing is {self.root.name}.{self.endpoint_signal_name}." + ) + return + self.update_received.put(message) + + def _on_config_update(self, value, **kwargs): + """Callback for when a new ROI configuration message is received.""" + if not isinstance(value, ROIConfigurationMessage): + raise TypeError( + f"Expected ROIConfigurationMessage, got {type(value).__name__} instead." + ) + self.active.put(value.active) + self.roi_name.put(value.name) + self.x.put(value.roi_config.x) + self.y.put(value.roi_config.y) + self.width.put(value.roi_config.width) + self.height.put(value.roi_config.height) + self.selected_operations.put(value.selected_operations) diff --git a/ophyd_devices/utils/bec_roi_signals/signals.py b/ophyd_devices/utils/bec_roi_signals/signals.py new file mode 100644 index 0000000..6b8d436 --- /dev/null +++ b/ophyd_devices/utils/bec_roi_signals/signals.py @@ -0,0 +1,187 @@ +from __future__ import annotations + +import threading +from typing import TYPE_CHECKING + +from bec_lib.endpoints import MessageEndpoints +from bec_lib.logger import bec_logger +from bec_lib.messages import ROIConfigurationMessage, ScanStatusMessage +from ophyd import Signal, SignalRO +from ophyd.status import Status + +if TYPE_CHECKING: + from bec_lib.redis_connector import MessageObject, RedisConnector + +logger = bec_logger.logger + + +class SelectedOperationSignal(Signal): + """ + Signal for selecting ROI analysis operations. + + The parent class must implement ``get_available_analysis_operations()``. + """ + + def check_value(self, value: list[str], **kwargs): + """Check that all selected operations are available.""" + available_operations = self.parent.get_available_analysis_operations() + if not all(v in available_operations for v in value): + raise ValueError( + f"Invalid operation(s): {value}. Must be one of {available_operations}." + ) + return value + + +class AvailableOperationsSignal(SignalRO): + """ + Signal for providing the available ROI analysis operations. + + The parent class must implement ``get_available_analysis_operations()``. + """ + + def get(self, **kwargs): + """Return the list of available operations.""" + return self.parent.get_available_analysis_operations() + + +class ConfigUpdateReceivedSignal(Signal): + """Signal to publish ROI configuration updates, optionally coalescing them while blocked.""" + + def __init__(self, name, parent=None, **kwargs): + super().__init__(name=name, parent=parent, **kwargs) + self._update_received = False + self._connector: RedisConnector | None = None + self._metadata["connected"] = False + self._next_config = None + self._next_config_kwargs = None + self._updates_blocked = False + self._update_lock = threading.RLock() + + @property + def updates_blocked(self) -> bool: + """Return whether incoming updates are currently held back.""" + return self._updates_blocked + + @property + def has_pending_update(self) -> bool: + """Return whether a blocked update is waiting to be published.""" + return self._update_received + + @property + def next_config(self): + """Return the latest blocked update without publishing it.""" + return self._next_config + + def destroy(self): + """Clean up the signal and unregister from Redis.""" + if self._connector is not None: + self._connector.unregister( + MessageEndpoints.scan_status(), self._handle_scan_status_update + ) + super().destroy() + + @property + def connected(self) -> bool: + """Return whether the signal is connected to Redis.""" + self.wait_for_connection() + return self._metadata.get("connected", False) + + def wait_for_connection(self, *args, **kwargs): + self._connector: RedisConnector | None = ( + self.root.device_manager.connector if hasattr(self.root, "device_manager") else None + ) + if self._connector is None: + raise RuntimeError( + f"Signal {self.name} is not connected to Redis, please provide a Redis Connector during the initialization." + ) + self._metadata["connected"] = True + self._connector.register(MessageEndpoints.scan_status(), cb=self._handle_scan_status_update) + + def _handle_scan_status_update(self, message: MessageObject): + msg: ScanStatusMessage = message.value + if msg.status == "open": + self.block_updates() + elif msg.status != "paused": + self.unblock_updates() + + def block_updates(self) -> None: + """Hold back incoming updates until :meth:`unblock_updates` is called.""" + with self._update_lock: + self._updates_blocked = True + + def unblock_updates(self): + """Allow updates again and publish the latest update received while blocked.""" + with self._update_lock: + self._updates_blocked = False + if not self._update_received: + return None + + value = self._next_config + kwargs = self._next_config_kwargs or {} + self._next_config = None + self._next_config_kwargs = None + self._update_received = False + + try: + super().put(value, **kwargs) + except Exception: + self._next_config = value + self._next_config_kwargs = dict( + timestamp=None, force=False, metadata=None, timeout=None + ) + self._update_received = True + raise + return value + + def put( + self, + value: ROIConfigurationMessage, + *, + timestamp=None, + force=False, + metadata=None, + timeout=None, + **kwargs, + ): + """Publish ``value`` immediately, or stage it as the next update while blocked.""" + if self._metadata.get("connected") is not True: + raise RuntimeError( + f"Signal {self.name} is not connected to Redis and cannot publish updates." + ) + with self._update_lock: + self.check_value(value) + if value.block_while_scanning is False or not self._updates_blocked: + super().put( + value, + timestamp=timestamp, + force=force, + metadata=metadata, + timeout=timeout, + **kwargs, + ) + return + + self._next_config = value + self._next_config_kwargs = dict( + timestamp=timestamp, force=force, metadata=metadata, timeout=timeout, **kwargs + ) + self._update_received = True + + def set(self, value, *, timeout=None, settle_time=None, **kwargs): + """Set the signal value, staging it if updates are currently blocked.""" + status = Status(self, timeout=timeout, settle_time=settle_time or 0) + try: + self.put(value, **kwargs) + except Exception as exc: + status.set_exception(exc) + else: + status.set_finished() + return status + + def check_value(self, value, **kwargs): + """Check that the value is a ROI configuration message.""" + if not isinstance(value, ROIConfigurationMessage): + raise ValueError( + f"Invalid value: {value}. Must be an instance of ROIConfigurationMessage." + ) + super().check_value(value, **kwargs)