bec/superxas_bec
0
0
Fork 0
Code Issues 2 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

wip review timepix logic for status objects

Browse Source
This commit is contained in:
appel_c
2025-09-11 17:11:55 +02:00
parent c1b15d270a
commit a9a9b2e58f
3 changed files with 157 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
superxas_bec/devices/timepix/timepix.py
+146 -64
View File
@@ -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()
superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_backend.py
+10 -12
View File
@@ -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()
superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_client.py
+1 -2
View File
@@ -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: