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

add script for online testing

Browse Source
This commit is contained in:
appel_c
2025-09-11 17:11:55 +02:00
parent 931f71a2c9
commit 08f6aeb889
2 changed files with 118 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
superxas_bec/devices/timepix/timepix.py
+116 -9
View File
@@ -5,6 +5,7 @@ is implemented via EPICS IOC.
"""
import enum
from time import time
from bec_lib.logger import bec_logger
from ophyd import ADBase
@@ -13,6 +14,7 @@ from ophyd import DeviceStatus, StatusBase
from ophyd_devices import AndStatus, CompareStatus, TransitionStatus
from ophyd_devices.devices.areadetector.cam import ASItpxCam
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
from typeguard import typechecked
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_backend import TimepixFlyBackend
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface import (
@@ -138,8 +140,56 @@ class Timepix(PSIDeviceBase, TimePixControl):
self.backend = TimepixFlyBackend(
backend_rest_url=backend_rest_url, hostname=hostname, socket_port=socket_port
)
self._pixel_map = PixelMap(
chips=[
[{"i": 256 ^ 2 - 1, "p": [0, 1], "f": [0.5, 0.5]}],
[{"i": 255 * 256, "p": [0, 1], "f": [0.5, 0.5]}],
[{"i": 255, "p": [1, 2], "f": [0.5, 0.5]}],
[{"i": 0, "p": [1, 2], "f": [0.5, 0.5]}],
]
)
self._troistep = 1
self._troin = 5000
@property
def pixel_map(self) -> PixelMap:
"""Get the current pixel map of the TimePix detector."""
return self._pixel_map
@pixel_map.setter
@typechecked
def pixel_map(self, value: PixelMap):
self._pixel_map = value
@property
def troistep(self) -> int:
"""Get the current ROI step size."""
return self._troistep
@troistep.setter
@typechecked
def troistep(self, value: int):
"""Set the ROI step size."""
if value <= 0:
raise ValueError("ROI step size must be a positive integer.")
self._troistep = value
@property
def troin(self) -> int:
"""Get the current ROI number of pixels."""
return self._troin
@troin.setter
@typechecked
def troin(self, value: int):
"""Set the ROI number of pixels."""
if value <= 0:
raise ValueError("ROI number of pixels must be a positive integer.")
self._troin = value
######################################################################
### Beamline specific methods for the TimePix Detector integration ###
######################################################################
def on_init(self) -> None:
"""
Called when the device is initialized.
@@ -173,19 +223,28 @@ class Timepix(PSIDeviceBase, TimePixControl):
Information about the upcoming scan can be accessed from the scan_info (self.scan_info.msg) object.
"""
# currently hardcode acquire time.. -> discuss logic here
# TODO discuss if the acquire period should be equivalent to the exposure time,
# and acquire time always 2ms shorter than the acquire period.
# num_images is currently hardcoded, is this basically a burst frame at each point of a step scan?
self.cam.acquire_time.set(0.998).wait(timeout=self._pv_timeout)
self.cam.acquire_period.set(1.0).wait(timeout=self._pv_timeout)
self.cam.num_images.set(10).wait(timeout=self._pv_timeout)
self.cam.raw_enable.set(1).wait(timeout=self._pv_timeout)
# TODO: can we inspect this from the backend instead!!
self.cam.raw_file_template.set("").wait(timeout=self._pv_timeout)
self.cam.raw_file_path.set(f"tcp://connect@127.0.0.1:8451").wait(timeout=self._pv_timeout)
# -------------------------
# Prepare TimePixFly
other_config = self.scan_info.msg.scan_parameters.get("other_config", None)
pixel_map = self.scan_info.msg.scan_parameters.get("pixel_map", None)
if other_config or pixel_map is None:
raise RuntimeError(
f"Scan info does not contain 'other_config' or 'pixel_map' for device {self.name}."
)
# TODO RoiStep and TRoinN are those config parameters across many scans,
# or parameter that are configured similar to energypoints --> pixel_map?
other_config = OtherConfigModel(
TRoiStep=self.troistep,
TRoiN=self.troin,
output_uri=f"tcp://{self.backend.hostname}:{self.backend.socket_port}",
)
pixel_map = self.pixel_map
self.backend.on_stage(other_config=other_config, pixel_map=pixel_map)
def on_unstage(self) -> None:
@@ -214,7 +273,7 @@ class Timepix(PSIDeviceBase, TimePixControl):
status_backend.add_callback(trigger_callback)
# Trigger the backend to start, then the callback will trigger the acquisition on the detector.
# This adds a status callback from the backend, which will resolve the status once the backend is finished.
self.backend.on_trigger(status=status_backend)
status_backend = self.backend.on_trigger(status=status_backend)
status_camera = TransitionStatus(
self.cam.acquire_busy, [ACQUIRESTATUS.ACQUIRING, ACQUIRESTATUS.DONE]
@@ -234,7 +293,7 @@ class Timepix(PSIDeviceBase, TimePixControl):
status_backend = DeviceStatus(self)
self.cancel_on_stop(status_backend)
# Add callback to the backend complete handling
self.backend.on_complete(status=status_backend)
status_backend = self.backend.on_complete(status=status_backend)
# Combine the statuses
complete_status = AndStatus(status_backend, status_detector)
return complete_status
@@ -256,7 +315,55 @@ class Timepix(PSIDeviceBase, TimePixControl):
# pylint: disable=protected-access
if __name__ == "__main__": # pragma: no cover
pass
timepix = Timepix(
name="timepix",
prefix="X10DA-ES-TPX1:",
backend_rest_url="P4-0017.psi.ch:8452",
hostname="x10da-bec-001.psi.ch",
)
try:
timepix.wait_for_connection(all_signals=True, timeout=10)
timepix.on_connected()
print("Timepix connected and initialized.")
timepix.stage()
print("Timepix staged.")
timepix.pre_scan()
print("Timepix pre-scan completed.")
msgs = []
for ii in range(5):
print(f"Starting scan {ii + 1}...; sleeping for 1s before each trigger before start")
time.sleep(1)
status = timepix.trigger()
print("Timepix trigger sent.")
start_time = time.time()
while not status.done:
try:
status.wait(timeout=1)
except Exception as exc:
print(f" Trigger status not done yet after ({time.time() - start_time:.2f}s)")
if time.time() - start_time > 30:
print("Breaking loop manually after 30 seconds of waiting.")
print(f"Received number of messsages: {len(timepix.backend.msg_buffer)}")
msgs.append(timepix.backend.msg_buffer)
timepix.backend.reset_message_buffer()
print(f"Resettet message buffer after scan {ii + 1}.")
status = timepix.complete()
print("Waiting for timepix to complete.")
status.wait(timeout=10)
print("Timepix scan completed.")
for ii, msg in enumerate(msgs):
print(f"Received {len(msg)} messages for trigger {ii}.")
timepix.unstage()
print("Timepix unstaged.")
except Exception as e:
print(f"An error occurred: {e}")
logger.error(f"An error occurred: {e}")
finally:
timepix.destroy()
print("Timepix destroyed.")
# Create a Timepix object
# timepix = Timepix(name="TimePixDetector", prefix="")
superxas_bec/devices/timepix/timepix_fly_client/timepix_fly_backend.py
+2
View File
@@ -332,6 +332,8 @@ if __name__ == "__main__": # pragma: no cover
timepix.on_stage(other_config=config, pixel_map=pixel_map)
print("TimepixFlyBackend staged with configuration and pixel map.")
for ii in range(10):
print(f"Starting scan {ii + 1}...; sleeping for 1 before start")
time.sleep(1)
status = timepix.on_trigger()
print("TimepixFlyBackend pre-scan started.")
status.wait(timeout=10)