diff --git a/superxas_bec/devices/timepix/timepix.py b/superxas_bec/devices/timepix/timepix.py index 60a9867..bf678df 100644 --- a/superxas_bec/devices/timepix/timepix.py +++ b/superxas_bec/devices/timepix/timepix.py @@ -5,11 +5,13 @@ of the backend is stored in the timepix_fly_client module. This is combined with interface in EPICS, which is implemented via the 'ASItpxCam' class. """ +from __future__ import annotations + import enum import threading import time import traceback -from typing import Any, Literal +from typing import TYPE_CHECKING, Any, Literal import numpy as np from bec_lib.logger import bec_logger @@ -29,18 +31,33 @@ from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface impor PixelMap, ) +if TYPE_CHECKING: # pragma: no cover + from ophyd import Device + class AndStatus(StatusBase): """Custom AndStatus for TimePix detector.""" def __init__( self, - left: StatusBase | DeviceStatus | list[StatusBase | DeviceStatus], - right: StatusBase | DeviceStatus | list[StatusBase | DeviceStatus], + left: StatusBase | DeviceStatus | list[StatusBase | DeviceStatus] | None, + name: str | Device | None = None, + right: StatusBase | DeviceStatus | list[StatusBase | DeviceStatus] | None = None, **kwargs, ): self.left = left if isinstance(left, list) else [left] - self.right = right if isinstance(right, list) else [right] + if right is not None: + self.right = right if isinstance(right, list) else [right] + else: + self.right = [] + self.all_statuses = self.left + self.right + if name is None: + name = "unname_status" + elif isinstance(name, Device): + name = name.name + else: + name = name + self.name = name super().__init__(**kwargs) self._trace_attributes["left"] = [st._trace_attributes for st in self.left] self._trace_attributes["right"] = [st._trace_attributes for st in self.right] @@ -50,39 +67,19 @@ class AndStatus(StatusBase): if self._externally_initiated_completion: return - all_statuses = self.left + self.right - for st in all_statuses: + for st in self.all_statuses: with st._lock: if st.done and not st.success: self.set_exception(st.exception()) return - if all(st.done for st in all_statuses) and all(st.success for st in all_statuses): + if all(st.done for st in self.all_statuses) and all( + st.success for st in self.all_statuses + ): self.set_finished() - # with self.left._lock: - # with self.right._lock: - # l_success = self.left.success - # r_success = self.right.success - # l_done = self.left.done - # r_done = self.right.done - # # At least one is done. - # # If it failed, do not wait for the second one. - # if (not l_success) and l_done: - # self.set_exception(self.left.exception()) - # elif (not r_success) and r_done: - # self.set_exception(self.right.exception()) - - # elif l_success and r_success and l_done and r_done: - # # Both are done, successfully. - # self.set_finished() - # # Else one is done, successfully, and we wait for #2, - # # when this function will be called again. - - for st in self.left + self.right: + for st in self.all_statuses: st.add_callback(inner) - # self.left.add_callback(inner) - # self.right.add_callback(inner) def __repr__(self): return "({self.left!r} & {self.right!r})".format(self=self) @@ -100,6 +97,70 @@ class AndStatus(StatusBase): return False + def set_exception(self, exc): + super().set_exception(exc) + for st in self.all_statuses: + with st._lock: + st.set_exception( + RuntimeError(f"AndStatus exception on high-level status, caused by {exc}") + ) + + def _run_callbacks(self): + """ + Set the Event and run the callbacks. + """ + if self.timeout is None: + timeout = None + else: + timeout = self.timeout + self.settle_time + if not self._settled_event.wait(timeout): + # We have timed out. It's possible that set_finished() has already + # been called but we got here before the settle_time timer expired. + # And it's possible that in this space be between the above + # statement timing out grabbing the lock just below, + # set_exception(exc) has been called. Both of these possibilties + # are accounted for. + self.log.warning("%r has timed out", self) + with self._externally_initiated_completion_lock: + # Set the exception and mark the Status as done, unless + # set_exception(exc) was called externally before we grabbed + # the lock. + if self._exception is None: + exc = TimeoutError( + f"Status with name {self.name} failed to complete in specified timeout of {self.timeout + self.settle_time}." + ) + self._exception = exc + # Mark this as "settled". + try: + self._settled() + except Exception: + # No alternative but to log this. We can't supersede set_exception, + # and we have to continue and run the callbacks. + self.log.exception("%r encountered error during _settled()", self) + # Now we know whether or not we have succeed or failed, either by + # timeout above or by set_exception(exc), so we can set the Event that + # will mark this Status as done. + with self._lock: + self._event.set() + if self._exception is not None: + try: + self._handle_failure() + except Exception: + self.log.exception("%r encountered an error during _handle_failure()", self) + # The callbacks have access to self, from which they can distinguish + # success or failure. + for cb in self._callbacks: + try: + cb(self) + except Exception: + self.log.exception( + "An error was raised on a background thread while " + "running the callback %r(%r).", + cb, + self, + ) + self._callbacks.clear() + logger = bec_logger.logger @@ -246,7 +307,7 @@ class Timepix(PSIDeviceBase, TimePixControl): self._n_energy_points = 3 self._troistep = 1 self._troin = 5000 - self._pv_timeout = 3 + self._pv_timeout = 5 self._readout_time = 2.1e-3 # 2.1ms readout time to ensure readout is >2ms, required from ASI serval server.. self.r_lock = threading.RLock() # Lock to access the message buffer safely super().__init__( @@ -282,7 +343,12 @@ class Timepix(PSIDeviceBase, TimePixControl): end_frame (dict): The EndFrame. Dictionary representation of detailed structure described in model .timepix_fly_client.timepix_fly_interface.TimepixEndFrame """ - n_energy_points = start_frame["NumEnergyPoints"] + n_energy_points = start_frame.get("NumEnergyPoints", None) + if n_energy_points is None: + logger.error( + f"NumEnergyPoints not found in start_frame: {start_frame}. Have we received the correct frame?" + ) + return # TODO What should we do here if n_energy_points and troin do not match with the expected values? if n_energy_points != self._n_energy_points: logger.error( @@ -425,7 +491,7 @@ class Timepix(PSIDeviceBase, TimePixControl): f"Exposure time {exp_time} must be greater than readout time {self._readout_time}." ) num_images = self.scan_info.msg.scan_parameters.get("frames_per_trigger", 1) - logger.info(f"Setting exposure time to {exp_time} and number of images to {num_images}") + logger.debug(f"Setting exposure time to {exp_time} and number of images to {num_images}") self.cam.acquire_time.set(exp_time - self._readout_time).wait(timeout=self._pv_timeout) self.cam.acquire_period.set(exp_time).wait(timeout=self._pv_timeout) @@ -439,13 +505,13 @@ class Timepix(PSIDeviceBase, TimePixControl): TRoiN=self.troin, output_uri=f"tcp:{self.backend.hostname}:{self.backend.socket_port}", ) - logger.info(f"Current TimePixFly configuration: {other_config}") + logger.debug(f"Current TimePixFly configuration: {other_config}") pixel_map = self.pixel_map self.backend.on_stage(other_config=other_config, pixel_map=pixel_map) # Fetch the backend socket info net_add = self.backend.timepix_fly_client.get_net_addresses() - logger.info(f"Using net_add for timepix_fly backend {net_add}") + logger.debug(f"Using net_add for timepix_fly backend {net_add}") self.cam.raw_file_template.set("").wait(timeout=self._pv_timeout) self.cam.raw_file_path.set(f"tcp://connect@{net_add.address}").wait( timeout=self._pv_timeout @@ -457,62 +523,77 @@ class Timepix(PSIDeviceBase, TimePixControl): def on_pre_scan(self) -> StatusBase: """Called right before the scan starts on all devices automatically.""" - status_acquire = CompareStatus(self.cam.acquire_busy, ACQUIRESTATUS.DONE) - status_detector_state = CompareStatus(self.cam.detector_state, DETECTORSTATE.IDLE) - status_detector = AndStatus(status_acquire, status_detector_state) - self.cancel_on_stop(status_detector) - return status_detector + # TODO check detector error state and raise if we see an error in the detector status + status_acquire = CompareStatus(self.cam.acquire_busy, ACQUIRESTATUS.DONE, timeout=3) + self.cancel_on_stop(status_acquire) + return status_acquire + + def _trigger_callback(self, status: DeviceStatus) -> None: + """Trigger callback to start the acquisition.""" + if status.done and status.success: + status.device.cam.acquire.put(1) + return + logger.error( + f"Status callback from TimePixFly backend trigger failed. Exception {status.exception()}" + ) def on_trigger(self) -> DeviceStatus | StatusBase | None: """Called when the device is triggered.""" - def trigger_callback(status: DeviceStatus): - """Trigger callback to start the acquisition.""" - if status.done: - logger.info(f"Calling acquire on detector.") - status.device.cam.acquire.put(1) - logger.info( - f"Status callback from backend trigger. done {status.done}, success {status.success} and exception {status._exception}" + # First we make sure that the backend reach 'config' state. This needs to happend before each trigger. + status_backend_config = DeviceStatus(self) + self.cancel_on_stop(status_backend_config) + self.backend.timepix_fly_client.add_status_callback( + status=status_backend_config, + success=[TimePixFlyStatus.CONFIG], + error=[TimePixFlyStatus.EXCEPT, TimePixFlyStatus.SHUTDOWN], + ) + try: + status_backend_config.wait(timeout=5) + except TimeoutError: + # pylint:disable=raise-missing-from + raise TimeoutError( + f"TimePixFly backend of device {self.name} failed to reach 'config' state in trigger" ) - # Detector will be ready to start, as either pre_scan or the status_camera from a previous - # trigger will ensure that the detector is in ACQUIRESTATUS.DONE state. - status_backend = DeviceStatus(self) - # Prepare the camera status that resolves when the camera is finished acquiring + # Prepare status objects for coordination of actions status_camera = TransitionStatus( self.cam.acquire_busy, [ACQUIRESTATUS.DONE, ACQUIRESTATUS.ACQUIRING, ACQUIRESTATUS.DONE] ) - # Prepare the backend, attach the status to the state of the backend - status_backend = self.backend.on_trigger(status=status_backend) + status_backend_on_trigger = DeviceStatus(self) + status_backend_on_trigger.add_callback(self.trigger_callback) + status_backend_collect_started = DeviceStatus(self) - status_collect_backend = DeviceStatus(self, timeout=10) + # Add Collect callback self.backend.timepix_fly_client.add_status_callback( - status=status_collect_backend, + status_backend_collect_started, success=[TimePixFlyStatus.COLLECT], error=[TimePixFlyStatus.EXCEPT, TimePixFlyStatus.SHUTDOWN, TimePixFlyStatus.CONFIG], ) - # Add callback that starts the acquisition on the detector - status_backend.add_callback(trigger_callback) - status = AndStatus(status_backend, status_camera) - st = AndStatus(status, status_collect_backend) - self.cancel_on_stop(st) - # NOTE, the callback to sent the data will always be called from the backend - # as it is attached via self.backend.add_callback() in on_connected. + # Start on trigger on backend + status_backend_on_trigger = self.backend.on_trigger(status=status_backend_on_trigger) + + status = AndStatus( + [status_camera, status_backend_on_trigger, status_backend_collect_started], + name=f"{self.name}_trigger_status", + ) + self.cancel_on_stop(status) return status def on_complete(self) -> DeviceStatus | StatusBase | None: """Called to inquire if a device has completed a scans.""" # Detector Control status_detector = CompareStatus(self.cam.acquire_busy, ACQUIRESTATUS.DONE) - self.cancel_on_stop(status_detector) # TimepixFFly status_backend = DeviceStatus(self) - self.cancel_on_stop(status_backend) # Add callback to the backend complete handling status_backend = self.backend.on_complete(status=status_backend) # Combine the statuses - complete_status = AndStatus(status_backend, status_detector) + complete_status = AndStatus( + [status_backend, status_detector], name=f"{self.name}_complete_status" + ) + self.cancel_on_stop(complete_status) return complete_status def on_kickoff(self) -> DeviceStatus | StatusBase | None: @@ -527,6 +608,7 @@ class Timepix(PSIDeviceBase, TimePixControl): def on_destroy(self): """Cleanup method to stop the device and clean up resources.""" + self.backend.on_stop() self.backend.on_destroy() diff --git a/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_backend.py b/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_backend.py index 80f1c8f..67c6fa2 100644 --- a/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_backend.py +++ b/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_backend.py @@ -9,7 +9,6 @@ hooks for all the relevant ophyd interface, 'on_stage', from __future__ import annotations -import time import atexit import json import signal @@ -114,8 +113,6 @@ class TimepixFlyBackend: other_config (OtherConfigModel): The configuration for the Timepix Fly detector. pixel_map (PixelMap): The pixel map for the Timepix Fly detector. """ - # Ensure that the message buffer is empty, should never contain data from previous scan - # But to be sure, it should definitely be resetted before starting a new scan status = StatusBase() self.cancel_on_stop(status) self.timepix_fly_client.add_status_callback( @@ -123,10 +120,6 @@ class TimepixFlyBackend: success=[TimePixFlyStatus.CONFIG], error=[TimePixFlyStatus.EXCEPT, TimePixFlyStatus.SHUTDOWN], ) - # if other_config.output_uri != f"tcp:{self.hostname}:{self.socket_port}": - # other_config.output_uri = f"tcp:{self.hostname}:{self.socket_port}" - # logger.info(f"Setting output URI to {other_config.output_uri}.") - # Make sure backend is in config state try: status.wait(timeout=5.0) except Exception: @@ -165,10 +158,10 @@ class TimepixFlyBackend: Returns: StatusBase | DeviceStatus: The status object that will be updated with the operation's result """ - # TODO add check that backend is in CONFIG! - time.sleep(0.05) + # TODO, could be removed as it's checkd from the top level! if status is None: status = StatusBase() + self.cancel_on_stop(status) self.timepix_fly_client.add_status_callback( status, success=[TimePixFlyStatus.AWAIT_CONNECTION], @@ -192,6 +185,7 @@ class TimepixFlyBackend: """ if status is None: status = StatusBase() + self.cancel_on_stop(status) self.timepix_fly_client.add_status_callback( status, success=[TimePixFlyStatus.CONFIG], @@ -237,9 +231,12 @@ class TimepixFlyBackend: def stop_all_status_objects(self): """Stop all status objects that are currently running.""" for status in self._status_objects: - if not status.done: - status.set_exception(RuntimeError("Operation cancelled by stop command.")) - logger.info(f"Cancelled status object: {status}") + with status._lock: + if not status.done: + status.set_exception( + RuntimeError("Stop called on device, all status objects cancelled.") + ) + logger.info(f"Cancelled status object: {status}") self._status_objects.clear() def add_callback(self, callback: callable, kwd: dict | None = None) -> str: @@ -369,6 +366,7 @@ class TimepixFlyBackend: logger.error(f"Error accepting connection: {content}") continue logger.debug(f"Connection accepted from {addr} for timepix_fly backend.") + # Clear the message buffer before entering the loop. if self.__msg_buffer: logger.warning(f"Found messages in msg_buffer: {self.__msg_buffer}") self.__msg_buffer.clear() diff --git a/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_client.py b/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_client.py index dcc7d3f..6137093 100644 --- a/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_client.py +++ b/superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_client.py @@ -344,7 +344,7 @@ class TimepixFlyClient: Start the TimePix detector by sending a GET request to the start endpoint. This method is a wrapper around the REST API call to start the detector. """ - logger.info(f"Start send from pixfly client") + logger.debug(f"Start called from client") self._get(get_cmd="?start=true") self._started = True @@ -479,7 +479,6 @@ class TimepixFlyClient: raise ValueError( f"Value must be an instance of OtherConfigModel. Received {type(other_config)}, {other_config}." ) - # logger.info(f"Value send via rest from set_other_config {other_config.model_dump()}") self._put(put_cmd="other-config", value=other_config.model_dump(), put_response_model=None) def get_net_addresses(self) -> NetAddresses: