Source code for ewoksxrdct.tasks.read_fscan2d

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 ReadFscan2d(
    Task,
    input_model=ReadInputModel,
    output_model=ReadOutputModel,
):
    """This task assumes the bliss master file follows ESRF's beamline
    conventions for fscan2d scans. The master file must contain:

    - `measurement/{translation_motor_name}`: 1D translation positions
    - `measurement/{rotation_motor_name}`: 1D rotation angles
    - `instrument/fscan_parameters/fast_motor`: name of the fast motor
    - `instrument/fscan_parameters/slow_motor`: name of the slow motor
    - `instrument/fscan_parameters/fast_start_pos`: scalar, fast motor start position
    - `instrument/fscan_parameters/fast_step_size`: scalar, fast motor step size
    - `instrument/fscan_parameters/fast_npoints`: scalar, number of fast motor points
    - `instrument/fscan_parameters/slow_start_pos`: scalar, slow motor start position
    - `instrument/fscan_parameters/slow_step_size`: scalar, slow motor step size
    - `instrument/fscan_parameters/slow_npoints`: scalar, number of slow motor points
    """



    def run(self) -> None:
        """Execute the task to read the raw motor and azimuthally integrated
        data for fscan2d scans 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
            )[()]

            fast_motor_name = load_hdf5_path(
                f, f"{scan}/instrument/fscan_parameters/fast_motor", file
            )[()].decode()
            slow_motor_name = load_hdf5_path(
                f, f"{scan}/instrument/fscan_parameters/slow_motor", file
            )[()].decode()
            fast_start = load_hdf5_path(
                f, f"{scan}/instrument/fscan_parameters/fast_start_pos", file
            )[()].item()
            fast_step = load_hdf5_path(
                f, f"{scan}/instrument/fscan_parameters/fast_step_size", file
            )[()].item()
            fast_npoints = load_hdf5_path(
                f, f"{scan}/instrument/fscan_parameters/fast_npoints", file
            )[()].item()
            slow_start = load_hdf5_path(
                f, f"{scan}/instrument/fscan_parameters/slow_start_pos", file
            )[()].item()
            slow_step = load_hdf5_path(
                f, f"{scan}/instrument/fscan_parameters/slow_step_size", file
            )[()].item()
            slow_npoints = load_hdf5_path(
                f, f"{scan}/instrument/fscan_parameters/slow_npoints", file
            )[()].item()

        if (
            fast_motor_name == self.inputs.rotation_motor_name
            and slow_motor_name == self.inputs.translation_motor_name
        ):
            rotation_is_fast = True
        elif (
            fast_motor_name == self.inputs.translation_motor_name
            and slow_motor_name == self.inputs.rotation_motor_name
        ):
            rotation_is_fast = False
        else:
            raise ValueError(
                f"Either 'rotation_motor_name' ({self.inputs.rotation_motor_name!r}) or"
                f" 'translation_motor_name' ({self.inputs.translation_motor_name!r}) "
                f"does not match the fast/slow motors recorded for this scan "
                f"({fast_motor_name!r}/{slow_motor_name!r})."
            )

        validate_motor_counts(fast_npoints * slow_npoints, 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(fast_npoints * slow_npoints, intensity)

        # The fast motor records the center of each step (see fscan's
        # MusstCenterCalc, same convention as aeroystepscan's rotation motor).
        # The slow motor is a standard step-scan master and records the exact
        # commanded target position.
        fast_grid_params = (
            fast_start + fast_step / 2,
            fast_start + fast_npoints * fast_step - fast_step / 2,
            fast_npoints,
        )
        slow_grid_params = (
            slow_start,
            slow_start + (slow_npoints - 1) * slow_step,
            slow_npoints,
        )

        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
        if rotation_is_fast:
            self.outputs.rotation_grid_params = fast_grid_params
            self.outputs.translation_grid_params = slow_grid_params
        else:
            self.outputs.rotation_grid_params = slow_grid_params
            self.outputs.translation_grid_params = fast_grid_params