Source code for ewoksxrdct.tasks.create_nxxrdct

import numpy as np

from ewokscore import Task
from ewokscore.model import BaseInputModel
from ewokscore.model import BaseOutputModel
from nxxrdct import NXxrdct
from pydantic import Field




class CreateNxxrdctInputModel(BaseInputModel):
    title: str = Field(..., description="Title of experiment.")
    nx_path: str = Field(
        ...,
        description="Full path to the NX file.",
    )
    beam_incident_energy: float = Field(
        ..., description="Incident energy of the beam in kiloelectron volts (keV)."
    )
    sample_name: str = Field(..., description="Name of the sample.")
    sample_rotation_angle: np.ndarray = Field(
        ..., description="Rotation angle of the sample (one per frame)."
    )
    detector_data: np.ndarray = Field(..., description="Data from the detector.")
    detector_count_time: float = Field(
        ..., description="Time to record a single frame."
    )





class CreateNxxrdctOutputModel(BaseOutputModel):
    nx_path: str = Field(
        ...,
        description="Full path to the NX file.",
    )





class CreateNxxrdct(
    Task, input_model=CreateNxxrdctInputModel, output_model=CreateNxxrdctOutputModel
):


    def run(self):
        """Execute the task to create an NX file from the given inputs."""
        self.save_NX_file()
        self.outputs.nx_path = self.inputs.nx_path




    def save_NX_file(self):
        """Create the NX file and save it using the given inputs."""
        nx = NXxrdct()
        nx.title = self.inputs.title
        nx.beam.incident_energy = self.inputs.beam_incident_energy
        nx.sample.name = self.inputs.sample_name
        nx.sample.rotation_angle = self.inputs.sample_rotation_angle
        nx.instrument.detector.data = self.inputs.detector_data
        nx.instrument.detector.count_time = self.inputs.detector_count_time

        nx.save(file_path=self.inputs.nx_path, data_path="entry")