diff --git a/ophyd_devices/sim/sim_camera.py b/ophyd_devices/sim/sim_camera.py index 51690ca..f44bd03 100644 --- a/ophyd_devices/sim/sim_camera.py +++ b/ophyd_devices/sim/sim_camera.py @@ -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()) diff --git a/ophyd_devices/utils/bec_roi_signals/ad_roi_signal.py b/ophyd_devices/utils/bec_roi_signals/ad_roi_signal.py index 8662b3b..37e7bfa 100644 --- a/ophyd_devices/utils/bec_roi_signals/ad_roi_signal.py +++ b/ophyd_devices/utils/bec_roi_signals/ad_roi_signal.py @@ -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()) diff --git a/ophyd_devices/utils/bec_roi_signals/roi_processing.py b/ophyd_devices/utils/bec_roi_signals/roi_processing.py index a2f046e..ba19471 100644 --- a/ophyd_devices/utils/bec_roi_signals/roi_processing.py +++ b/ophyd_devices/utils/bec_roi_signals/roi_processing.py @@ -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." diff --git a/ophyd_devices/utils/bec_roi_signals/signals.py b/ophyd_devices/utils/bec_roi_signals/signals.py index 6b8d436..61d217c 100644 --- a/ophyd_devices/utils/bec_roi_signals/signals.py +++ b/ophyd_devices/utils/bec_roi_signals/signals.py @@ -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."