Source code for ewoksxrdct.tasks.read_aeroystepscan

import os
from pathlib import Path

from ewokscore import Task

from .utils.hdf5 import load_hdf5_path
from .utils.hdf5 import open_hdf5_file
from .utils.read_models import ReadInputModel
from .utils.read_models import ReadOutputModel
from .utils.validate import validate_frame_count
from .utils.validate import validate_motor_counts




class ReadAeroystepscan(
    Task,
    input_model=ReadInputModel,
    output_model=ReadOutputModel,
):
    """This task assumes the bliss master file follows ESRF's beamline
    conventions for aeroystepscans. The master file must contain:

    - `measurement/{translation_motor_name}`: 1D translation positions
    - `measurement/{rotation_motor_name}`: 1D rotation angles
    - `instrument/fscan_parameters/x_range`: scalar, total rotation range
    - `instrument/fscan_parameters/x_step_size`: scalar, rotation step size
    - `instrument/fscan_parameters/x_start_pos`: scalar, rotation start position
    - `instrument/fscan_parameters/y_start_pos`: scalar, translation start position
    - `instrument/fscan_parameters/y_step_size_arr`: scalar, translation step size
    - `instrument/fscan_parameters/y_npoints_arr`: scalar, number of translation points

    """



    def run(self) -> None:
        """Execute the task to read the raw motor and azimuthally integrated
        data for aeroystepscans (ESRF-only) and derive the grid info needed
        for regridding."""

        bliss_master = self.inputs.bliss_master_path
        with open_hdf5_file(bliss_master) as f:
            file = Path(bliss_master).name
            scan = f"{self.inputs.scan_number}.1"

            translations = load_hdf5_path(
                f, f"{scan}/measurement/{self.inputs.translation_motor_name}", file
            )[()]
            rotations = load_hdf5_path(
                f, f"{scan}/measurement/{self.inputs.rotation_motor_name}", file
            )[()]
            x_range = load_hdf5_path(
                f, f"{scan}/instrument/fscan_parameters/x_range", file
            )[()].item()
            x_step = load_hdf5_path(
                f, f"{scan}/instrument/fscan_parameters/x_step_size", file
            )[()].item()
            x_start = load_hdf5_path(
                f, f"{scan}/instrument/fscan_parameters/x_start_pos", file
            )[()].item()
            y_start = load_hdf5_path(
                f, f"{scan}/instrument/fscan_parameters/y_start_pos", file
            )[()].item()
            y_step = load_hdf5_path(
                f, f"{scan}/instrument/fscan_parameters/y_step_size_arr", file
            )[()].item()
            y_points = load_hdf5_path(
                f, f"{scan}/instrument/fscan_parameters/y_npoints_arr", file
            )[()].item()

        N_trans = y_points
        # x_range is already a span (last trigger minus first trigger), not
        # an absolute end position, so x_start does not factor into the
        # point count -- see fscan's AeroScanXYSMaster.validate().
        N_rot = int(x_range / x_step)

        validate_motor_counts(N_trans * N_rot, translations, rotations)

        ai_filename = self.inputs.integration_output_path
        with open_hdf5_file(ai_filename) as f:
            ai_file = Path(ai_filename).name
            radial_axis = load_hdf5_path(
                f, self.inputs.integration_radial_axis_path, ai_file
            )[()]
            intensity = load_hdf5_path(
                f, self.inputs.integration_intensity_path, ai_file
            )[()]

        # Check the data sizing will be consistent in the final NX file.
        validate_frame_count(N_trans * N_rot, intensity)

        self.outputs.rotation_angles = rotations
        self.outputs.translation_values = translations
        self.outputs.integration_intensity_values = intensity
        self.outputs.integration_radial_axis = os.path.basename(
            self.inputs.integration_radial_axis_path
        )
        self.outputs.integration_radial_axis_values = radial_axis
        # The rotation motor records the center of each angular step (see
        # fscan's MusstCenterCalc), not its start.
        self.outputs.rotation_grid_params = (
            x_start + x_step / 2,
            x_start + x_range - x_step / 2,
            N_rot,
        )
        self.outputs.translation_grid_params = (
            y_start,
            y_start + ((y_points - 1) * y_step),
            N_trans,
        )