mirror of
https://github.com/bec-project/ophyd_devices.git
synced 2026-07-28 18:12:56 +02:00
feat(bec_roi_signals): initial implementation of ROI signals and processing
This commit is contained in:
@@ -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 #
|
||||
########################################
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
from .roi_processing import LITERAL_ROI_PROCESSING_CONFIG, ROIProcessing
|
||||
@@ -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")
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user