WIP with Fede's scans

This commit is contained in:
gac-x05la
2025-02-17 17:46:52 +01:00
parent e8b3aedb10
commit f6fecfdc3f
4 changed files with 162 additions and 46 deletions

View File

@@ -36,6 +36,10 @@ class AerotechDriveDataCollectionMixin(CustomDeviceMixin):
# NOTE: Scans don't have to fully configure the device
if "ddc_trigger" in scanargs:
d["ddc_trigger"] = scanargs["ddc_trigger"]
if "ddc_source0" in scanargs:
d["ddc_source0"] = scanargs["ddc_source0"]
if "ddc_source1" in scanargs:
d["ddc_source1"] = scanargs["ddc_source1"]
if "ddc_num_points" in scanargs and scanargs["ddc_num_points"] is not None:
d["num_points_total"] = scanargs["ddc_num_points"]
@@ -64,8 +68,7 @@ class AerotechDriveDataCollectionMixin(CustomDeviceMixin):
# Stage the data collection if not in internally launced mode
# NOTE: Scripted scans start acquiring from the scrits
if self.parent.scaninfo.scan_type not in ("script", "scripted"):
self.parent.bluestage()
self.parent.bluestage()
def on_unstage(self):
"""Standard bluesky unstage"""
@@ -115,7 +118,7 @@ class aa1AxisDriveDataCollection(PSIDeviceBase):
_buffer1 = Component(EpicsSignalRO, "BUFFER1", auto_monitor=True, kind=Kind.normal)
custom_prepare_cls = AerotechDriveDataCollectionMixin
USER_ACCESS = ["configure", "reset"]
USER_ACCESS = ["configure", "reset", "collect"]
def configure(self, d: dict = None) -> tuple:
"""Configure data capture
@@ -144,6 +147,7 @@ class aa1AxisDriveDataCollection(PSIDeviceBase):
def bluestage(self) -> None:
"""Bluesky-style stage"""
self._switch.set("ResetRB", settle_time=0.1).wait()
self._switch.set("Start", settle_time=0.2).wait()
def blueunstage(self):

View File

@@ -1,2 +1,2 @@
from .tutorial_fly_scan import AcquireDark, AcquireWhite, AcquireRefs, TutorialFlyScanContLine
from .tutorial_fly_scan import AcquireDark, AcquireWhite, AcquireRefs, AcquireProjections, TutorialFlyScanContLine
from .tomcat_scans import TomcatSnapNStep, TomcatSimpleSequence

View File

@@ -30,6 +30,10 @@ class AcquireDark(Acquire):
Predefined acquisition mode (default= 'default')
file_path : str, optional
File path for standard daq
ddc_trigger : int, optional
Drive Data Capture Trigger
ddc_source0 : int, optional
Drive Data capture Input0
Returns:
ScanReport
@@ -53,10 +57,10 @@ class AcquireDark(Acquire):
class AcquireWhite(Acquire):
scan_name = "acquire_white"
required_kwargs = ["exp_burst", "sample_position_out", "sample_angle_out"]
required_kwargs = ["exp_burst", "sample_position_out", "sample_angle_out", "motor"]
gui_config = {"Acquisition parameters": ["exp_burst"]}
def __init__(self, exp_burst: int, sample_position_out: float, sample_angle_out: float, **kwargs):
def __init__(self, exp_burst: int, sample_position_out: float, sample_angle_out: float, motor: DeviceBase, **kwargs):
"""
Acquire flat field images. This scan is used to acquire flat field images. The flat field image is an image taken
with the shutter open but the sample out of the beam. Flat field images are used to correct the data images for
@@ -69,6 +73,8 @@ class AcquireWhite(Acquire):
Position to move the sample stage to position the sample out of beam and take flat field images
sample_angle_out : float
Angular position where to take the flat field images
motor : DeviceBase
Motor to be moved to move the sample out of beam
exp_time : float, optional
Exposure time [ms]. If not specified, the currently configured value on the camera will be used
exp_period : float, optional
@@ -81,6 +87,10 @@ class AcquireWhite(Acquire):
Predefined acquisition mode (default= 'default')
file_path : str, optional
File path for standard daq
ddc_trigger : int, optional
Drive Data Capture Trigger
ddc_source0 : int, optional
Drive Data capture Input0
Returns:
ScanReport
@@ -93,15 +103,16 @@ class AcquireWhite(Acquire):
self.burst_at_each_point = 1
self.sample_position_out = sample_position_out
self.sample_angle_out = sample_angle_out
self.motor_sample = motor
self.scan_motors = ["eyex", "eyez", "es1_roty"] # change to the correct shutter device
self.scan_motors = ["eyex", self.motor_sample, "es1_roty"] # change to the correct shutter device
self.dark_shutter_pos_out = 1 ### change with a variable
self.dark_shutter_pos_in = 0 ### change with a variable
def scan_core(self):
# open the shutter and move the sample stage to the out position
self.scan_motors = ["eyez", "es1_roty"] # change to the correct shutter device
self.scan_motors = [self.motor_sample, "es1_roty"] # change to the correct shutter device
yield from self._move_scan_motors_and_wait([self.sample_position_out, self.sample_angle_out])
self.scan_motors = ["eyex"] # change to the correct shutter device
yield from self._move_scan_motors_and_wait([self.dark_shutter_pos_out])
@@ -111,26 +122,31 @@ class AcquireWhite(Acquire):
# TODO add closing of fast shutter
yield from self._move_scan_motors_and_wait([self.dark_shutter_pos_in])
class AcquireProjectins(Acquire):
class AcquireProjections(AsyncFlyScanBase):
scan_name = "acquire_projections"
required_kwargs = ["exp_burst", "sample_position_in", "start_position", "angular_range"]
required_kwargs = ["motor", "exp_burst", "sample_position_in", "start_angle", "angular_range"]
gui_config = {"Acquisition parameters": ["exp_burst"]}
def __init__(self,
motor: DeviceBase,
exp_burst: int,
sample_position_in: float,
start_position: float,
start_angle: float,
angular_range: float,
**kwargs):
"""
Acquire projection images.
Args:
motor :
motor : DeviceBase
Motor to move continuously from start to stop position
exp_burst : int
Number of flat field images to acquire (no default)
sample_position_in : float
Position to move the sample stage to position the sample in the beam
start_position : float
start_angle : float
Angular start position for the scan
angular_range : float
Angular range
@@ -146,36 +162,81 @@ class AcquireProjectins(Acquire):
Predefined acquisition mode (default= 'default')
file_path : str, optional
File path for standard daq
ddc_trigger : int, optional
Drive Data Capture Trigger
ddc_source0 : int, optional
Drive Data capture Input0
Returns:
ScanReport
Examples:
>>> scans.acquire_white(5, 20)
>>> scans.acquire_projections()
"""
super().__init__(**kwargs)
self.motor = motor
self.burst_at_each_point = 1
self.sample_position_in = sample_position_in
self.start_position = start_position
self.start_angle = start_angle
self.angular_range = angular_range
self.scan_motors = ["eyex", "eyez", "es1_roty"] # change to the correct shutter device
#self.scan_motors = ["eyex", "eyez", "es1_roty"] # change to the correct shutter device
self.dark_shutter_pos_out = 1 ### change with a variable
self.dark_shutter_pos_in = 0 ### change with a variable
def prepare_positions(self):
self.positions = np.array([[self.start_angle], [self.start_angle+self.angular_range]])
self.num_pos = None
yield from self._set_position_offset()
def scan_core(self):
# open the shutter and move the sample stage to the out position
self.scan_motors = ["eyez", "es1_roty"] # change to the correct shutter device
yield from self._move_scan_motors_and_wait([self.sample_position_out, self.sample_angle_out])
# move to in position and go to start position
self.scan_motors = ["eyez", self.motor]
yield from self._move_scan_motors_and_wait([self.sample_position_in, self.positions[0][0]])
# open the shutter
self.scan_motors = ["eyex"] # change to the correct shutter device
yield from self._move_scan_motors_and_wait([self.dark_shutter_pos_out])
# TODO add opening of fast shutter
yield from super().scan_core()
# start the flyer
# flyer_request = yield from self.stubs.set_with_response(
# device=self.motor, value=self.positions[1][0]
# )
flyer_request = yield from self.stubs.set(
device=self.motor, value=self.positions[1][0], wait=True
)
self.connector.send_client_info(
"Starting the scan", show_asap=True, rid=self.metadata.get("RID")
)
# TODO add closing of fast shutter
yield from self._move_scan_motors_and_wait([self.dark_shutter_pos_in])
# send a trigger
# yield from self.stubs.trigger(group="trigger", point_id=self.point_id)
yield from self.stubs.trigger(wait=False)
while True:
# read the data
# yield from self.stubs.read_and_wait(
# group="primary", wait_group="readout_primary", point_id=self.point_id
# )
yield from self.stubs.read(
device=self.motor, point_id=self.point_id, wait=True
)
time.sleep(1)
if self.stubs.request_is_completed(flyer_request):
# stop the scan if the motor has reached the stop position
break
# increase the point id
self.point_id += 1
def finalize(self):
yield from super().finalize()
self.num_pos = self.point_id + 1
class AcquireRefs(Acquire):
@@ -192,14 +253,6 @@ class AcquireRefs(Acquire):
sample_position_out: float = 5000,
file_prefix_dark: str = 'tmp_dark',
file_prefix_white: str = 'tmp_white',
exp_time: float = 0,
exp_period: float = 0,
image_width: int = 2016,
image_height: int = 2016,
acq_mode: str = 'default',
file_path: str = 'tmp',
nr_writers: int = 2,
base_path: str = 'tmp',
**kwargs
):
"""
@@ -247,14 +300,6 @@ class AcquireRefs(Acquire):
self.num_flats = num_flats
self.file_prefix_dark = file_prefix_dark
self.file_prefix_white = file_prefix_white
self.exp_time = exp_time
self.exp_period = exp_period
self.image_width = image_width
self.image_height = image_height
self.acq_mode = acq_mode
self.file_path = file_path
self.nr_writers = nr_writers
self.base_path = base_path
def scan_core(self):
@@ -270,8 +315,9 @@ class AcquireRefs(Acquire):
exp_burst=self.num_darks,
file_prefix=self.file_prefix_dark,
device_manager=self.device_manager,
metadata=self.metadata
metadata=self.metadata,
)
yield from darks.scan_core()
self.point_id = darks.point_id

View File

@@ -16,6 +16,7 @@ class Measurement:
self.nimages_white = 100
self.start_angle = 0
self.angular_range = 180
self.sample_angle_out = 0
self.sample_position_in = 0
self.sample_position_out = 1
@@ -268,13 +269,14 @@ class Measurement:
else:
print("Roiy: " + str(self.roiy))
print("Start angle: " + str(self.start_angle))
print("Angular range: " + str(self.angular_range))
print("Sample angle out: " + str(self.sample_angle_out))
print("Sample position in: " + str(self.sample_position_in))
print("Sample position out: " + str(self.sample_position_out))
def acquire_darks(self,nimages_dark=None, exposure_time=None, exposure_period=None,
roix=None, roiy=None, acq_mode=None):
roix=None, roiy=None, acq_mode=None, **kwargs):
"""
Acquire a set of dark images with shutters closed.
@@ -315,17 +317,19 @@ class Measurement:
print("Handing over to 'scans.acquire_dark")
scans.acquire_dark(exp_burst=self.nimages_dark, exp_time=self.exposure_time, exp_period=self.exposure_period, image_width=self.roix,
image_height=self.roiy, acq_mode=acq_mode, file_path=self.file_path, nr_writers=2, base_path=self.base_path,
file_prefix=self.file_prefix)
file_prefix=self.file_prefix, ddc_trigger=4, ddc_source0=1, **kwargs)
def acquire_whites(self,nimages_white=None, sample_angle_out=None, sample_position_out=None,
def acquire_whites(self,motor="eyez", nimages_white=None, sample_angle_out=None, sample_position_out=None,
exposure_time=None, exposure_period=None,
roix=None, roiy=None, acq_mode=None):
roix=None, roiy=None, acq_mode=None, **kwargs):
"""
Acquire a set of whites images with shutters open and sample out of beam.
Parameters
----------
motor : DeviceBase
Motor to be moved to move the sample out of beam
nimages_whites : int, optional
Number of white images to acquire (no default)
sample_angle_out : float, optional
@@ -348,6 +352,7 @@ class Measurement:
m.acquire_whites(nimages_whites=100, exposure_time=5)
"""
self.motor_sample = motor
if nimages_white != None:
self.nimages_white = nimages_white
if sample_angle_out != None:
@@ -367,16 +372,76 @@ class Measurement:
### TODO: camera reset
print("Handing over to 'scans.acquire_whites")
scans.acquire_white(exp_burst=self.nimages_white, sample_angle_out=self.sample_angle_out, sample_position_out= self.sample_position_out,
scans.acquire_white(motor=self.motor_sample, exp_burst=self.nimages_white, sample_angle_out=self.sample_angle_out, sample_position_out= self.sample_position_out,
exp_time=self.exposure_time, exp_period=self.exposure_period, image_width=self.roix,
image_height=self.roiy, acq_mode=acq_mode, file_path=self.file_path, nr_writers=2, base_path=self.base_path,
file_prefix=self.file_prefix)
file_prefix=self.file_prefix, ddc_trigger=4, ddc_source0=1, **kwargs)
def acquire_projections(self, nimages=None, sample_position_in=None,
start_angle=None, angular_range=None,
exposure_time=None, exposure_period=None,
roix=None, roiy=None, acq_mode=None, **kwargs):
"""
Acquire a set of whites images with shutters open and sample out of beam.
Parameters
----------
nimages : int, optional
Number of projection images to acquire (no default)
sample_position_in : float, optional
Sample stage X position for sample in the beam [um]
start_angle : float, optional
Starting angular position [deg]
angular_range : float, optional
Angular range [deg]
exposure_time : float, optional
Exposure time [ms]. If not specified, the currently configured value on the camera will be used
exposure_period : float, optional
Exposure period [ms]
roix : int, optional
ROI size in the x-direction [pixels]
roiy : int, optional
ROI size in the y-direction [pixels]
acq_mode : str, optional
Predefined acquisition mode (default=None)
Example:
--------
m.acquire_projections(nimages_projections=100, exposure_time=5)
"""
if nimages != None:
self.nimages = nimages
if sample_position_in != None:
self.sample_position_in = sample_position_in
if start_angle != None:
self.start_angle = start_angle
if angular_range != None:
self.angular_range = angular_range
if exposure_time != None:
self.exposure_time = exposure_time
if exposure_period != None:
self.exposure_period = exposure_period
if roix != None:
self.roix = roix
if roiy != None:
self.roiy = roiy
self.build_filename(acquisition_type='data')
### TODO: camera reset
print("Handing over to 'scans.acquire_projections")
scans.acquire_projections(motor="es1_roty", exp_burst=self.nimages, sample_position_in= self.sample_position_in,
start_angle = self.start_angle, angular_range = self.angular_range,
exp_time=self.exposure_time, exp_period=self.exposure_period, image_width=self.roix,
image_height=self.roiy, acq_mode=acq_mode, file_path=self.file_path, nr_writers=2,
base_path=self.base_path,file_prefix=self.file_prefix, ddc_trigger=4, ddc_source0=1, **kwargs)
def acquire_refs(self,nimages_dark=None, nimages_white=None, sample_angle_out=None,
sample_position_in=None, sample_position_out=None,
exposure_time=None, exposure_period=None,
roix=None, roiy=None, acq_mode=None):
roix=None, roiy=None, acq_mode=None, **kwargs):
"""
Acquire reference images (darks + whites) and return to beam position.
@@ -441,4 +506,5 @@ class Measurement:
sample_position_in=self.sample_position_in, sample_position_out=self.sample_position_out,
exp_time=self.exposure_time, exp_period=self.exposure_period, image_width=self.roix,
image_height=self.roiy, acq_mode='default', file_path=self.file_path, nr_writers=2, base_path=self.base_path,
file_prefix_dark=file_prefix_dark, file_prefix_white=file_prefix_white)
file_prefix_dark=file_prefix_dark, file_prefix_white=file_prefix_white,
ddc_trigger=4, ddc_source0=1, **kwargs)