bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2026-07-29 10:32:57 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: cleanup

Browse Source
This commit is contained in:
appel_c
2026-07-20 15:48:27 +02:00
parent 3705d16e51
commit 0f85e87a50
4 changed files with 104 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ophyd_devices/sim/sim_camera.py
+8 -2
View File
@@ -23,10 +23,16 @@ class SimCameraROIProcessing(ROIProcessing):
"""ROI processing signal for simulated camera setups."""
def get_scalar_outputs(self) -> list[str]:
return SIM_CONFIG["basic_statistics"]["scalar_outputs"]
scalar_outpus = []
for v in SIM_CONFIG.values():
scalar_outpus.extend(v["scalar_outputs"])
return scalar_outpus
def get_waveform_outputs(self) -> list[str]:
return SIM_CONFIG["basic_statistics"]["waveform_outputs"]
waveform_outputs = []
for v in SIM_CONFIG.values():
waveform_outputs.extend(v["waveform_outputs"])
return waveform_outputs
def get_available_analysis_operations(self) -> list[str]:
return list(SIM_CONFIG.keys())
ophyd_devices/utils/bec_roi_signals/ad_roi_signal.py
+8 -2
View File
@@ -24,10 +24,16 @@ class ADROIProcessing(ROIProcessing):
stats1 = Cpt(StatsPlugin, prefix="STATS1:", kind="normal")
def get_scalar_outputs(self) -> list[str]:
return NDPLUGIN_STATS_CONFIG["basic_statistics"]["scalar_outputs"]
scalar_outpus = []
for v in NDPLUGIN_STATS_CONFIG.values():
scalar_outpus.extend(v["scalar_outputs"])
return scalar_outpus
def get_waveform_outputs(self) -> list[str]:
return NDPLUGIN_STATS_CONFIG["basic_statistics"]["waveform_outputs"]
waveform_outputs = []
for v in NDPLUGIN_STATS_CONFIG.values():
waveform_outputs.extend(v["waveform_outputs"])
return waveform_outputs
def get_available_analysis_operations(self) -> list[str]:
return list(NDPLUGIN_STATS_CONFIG.keys())
ophyd_devices/utils/bec_roi_signals/roi_processing.py
+70 -45
View File
@@ -8,8 +8,7 @@ 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 import Device, Kind, Signal
from ophyd_devices.utils.bec_roi_signals.signals import (
AvailableOperationsSignal,
@@ -40,6 +39,7 @@ class ROIProcessing(Device, ABC):
signals=[],
ndim=0,
async_update={"type": "add", "max_shape": [None]},
data_type="processed",
)
result_waveform = Cpt(
DynamicSignal,
@@ -48,44 +48,59 @@ class ROIProcessing(Device, ABC):
signals=[],
ndim=1,
async_update={"type": "add", "max_shape": [None, None]},
data_type="processed",
)
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")
active = Cpt(Signal, value=False, kind=Kind.normal)
roi_name = Cpt(Signal, value="roi", kind=Kind.normal)
x = Cpt(Signal, value=0, kind=Kind.normal)
y = Cpt(Signal, value=0, kind=Kind.normal)
width = Cpt(Signal, value=0, kind=Kind.normal)
height = Cpt(Signal, value=0, kind=Kind.normal)
selected_operations = Cpt(SelectedOperationSignal, value=[], kind=Kind.normal)
available_operations = Cpt(AvailableOperationsSignal, kind=Kind.config)
update_received = Cpt(ConfigUpdateReceivedSignal, kind=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()
self._update_result_signal_names()
def _prepare_result_signals(self):
"""Prepare the result signals based on the available operations and outputs."""
def _update_result_signal_names(self):
"""
Update the signal names for the DynamicSignals "result_scalar" and "result_waveform" based on
the outputs defined in the subclass. This ensures that the signals.
Subclasses must specify every possible output signal in the get_scalar_outputs and get_waveform_outputs methods.
"""
# pylint: disable=protected-access
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."""
"""
Utility method to extract kwargs for the signals defined in this class.
This allows to set initial values for the signals during initialization of the ROIProcessing 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."""
"""
Wait for the ROI processing signal to be connected to Redis. The root device must ensure
that wait_for_connection is called when connecting the device. Otherwise, the ROIProcessing
component will not work properly.
The root device also needs a device_manager and a RedisConnector available.
Please use PSIDeviceBase as a base class for your root device to ensure this is the case.
-> ophyd_devices.interfaces.base_classes.psi_device_base.PSIDeviceBase
"""
self._connector: RedisConnector | None = (
self.root.device_manager.connector if hasattr(self.root, "device_manager") else None
)
@@ -94,7 +109,7 @@ class ROIProcessing(Device, ABC):
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
# Setup initial values for the signal as provided in the kwargs during initialization
for signal_name, value in self._signal_kwargs.items():
signal = getattr(self, signal_name)
signal.put(value)
@@ -104,38 +119,40 @@ class ROIProcessing(Device, ABC):
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,
MessageEndpoints.roi_config(device=self.root.name, signal=self.dotted_name),
cb=self._receive_roi_configuration_message,
)
self.publish_available_analysis()
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,
MessageEndpoints.roi_config(device=self.root.name, signal=self.dotted_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."""
"""
Returns a list of strings with all possible scalar output signals for the ROI processing signal.
All possible outputs must be listed, although not all of them have to be there at the same time.
"""
@abstractmethod
def get_waveform_outputs(self) -> list[str]:
"""Return the waveform output signal for the ROI processing signal."""
"""
Returns a list of strings with all possible waveform output signals for the ROI processing signal.
All possible outputs must be listed, although not all of them have to be there at the same time.
"""
@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
"""
Returns a list of strings with all possible analysis operations for the ROI processing signal.
All possible operations must be listed, although not all of them have to be active at the same time.
"""
def compile_roi_analysis_config(self) -> ROIAnalysisConfig:
"""Compile the ROI analysis configuration from the available operations and outputs."""
@@ -145,7 +162,7 @@ class ROIProcessing(Device, ABC):
scalar_results=self.get_scalar_outputs(),
)
def publish_available_analysis(self) -> None:
def _publish_available_analysis(self) -> None:
"""Publish the available ROI analysis operations and result signals."""
if self._connector is None:
raise RuntimeError(
@@ -153,29 +170,37 @@ class ROIProcessing(Device, ABC):
)
message = ROIAvailableAnalysisMessage(
device=self.root.name,
signal=self.endpoint_signal_name,
signal=self.dotted_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
),
MessageEndpoints.available_roi_analysis(device=self.root.name, signal=self.dotted_name),
message,
)
def receive_roi_configuration_message(self, message: MessageObject):
"""Receive a ROI configuration message and update the ROI processing signal accordingly."""
def _receive_roi_configuration_message(self, message: MessageObject):
"""
Receive a ROI configuration message and update the ROI processing signal accordingly.
Args:
message (MessageObject): The received message object containing the ROI configuration.
"""
message: ROIConfigurationMessage = message.value
if message.device != self.root.name or message.signal != self.endpoint_signal_name:
if message.device != self.root.name or message.signal != self.dotted_name:
logger.warning(
f"Ignoring ROI configuration for {message.device}.{message.signal}; "
f"this ROIProcessing is {self.root.name}.{self.endpoint_signal_name}."
f"this ROIProcessing is {self.root.name}.{self.dotted_name}."
)
return
self.update_received.put(message)
def _on_config_update(self, value, **kwargs):
"""Callback for when a new ROI configuration message is received."""
def _on_config_update(self, value: ROIConfigurationMessage, **kwargs):
"""
Callback for when a new ROI configuration message is received.
Args:
value (ROIConfigurationMessage): The received ROI configuration message.
"""
if not isinstance(value, ROIConfigurationMessage):
raise TypeError(
f"Expected ROIConfigurationMessage, got {type(value).__name__} instead."
ophyd_devices/utils/bec_roi_signals/signals.py
+18 -4
View File
@@ -45,7 +45,10 @@ class AvailableOperationsSignal(SignalRO):
class ConfigUpdateReceivedSignal(Signal):
"""Signal to publish ROI configuration updates, optionally coalescing them while blocked."""
"""
Signal to publish ROI configuration updates. Based on the updated configuration,
updates will be published right away or blocked until released.
"""
def __init__(self, name, parent=None, **kwargs):
super().__init__(name=name, parent=parent, **kwargs)
@@ -74,16 +77,20 @@ class ConfigUpdateReceivedSignal(Signal):
def destroy(self):
"""Clean up the signal and unregister from Redis."""
if self._connector is not None:
if self._connector is not None and self._metadata.get("connected") is True:
self._connector.unregister(
MessageEndpoints.scan_status(), self._handle_scan_status_update
)
self._metadata["connected"] = False
super().destroy()
@property
def connected(self) -> bool:
"""Return whether the signal is connected to Redis."""
self.wait_for_connection()
if self._destroyed:
return False
if self._metadata.get("connected") is not True:
self.wait_for_connection()
return self._metadata.get("connected", False)
def wait_for_connection(self, *args, **kwargs):
@@ -143,7 +150,14 @@ class ConfigUpdateReceivedSignal(Signal):
timeout=None,
**kwargs,
):
"""Publish ``value`` immediately, or stage it as the next update while blocked."""
"""
Put a new ROI configuration message to the signal. If updates are currently blocked,
the update will be cached and published once `unblock_updates` is called. Any subsequent
updates will overwrite the cached update until updates are unblocked.
Args:
value (ROIConfigurationMessage): The new ROI configuration message to put.
"""
if self._metadata.get("connected") is not True:
raise RuntimeError(
f"Signal {self.name} is not connected to Redis and cannot publish updates."