Source code for ewoksfluo.tasks.input.spec_to_bliss

from typing import List
from typing import Union

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

from ...io.convert import spec_to_bliss




class Inputs(BaseInputModel):
    input_filename: str = Field(
        description="SPEC master file name.", examples=["/data/specfile.dat"]
    )
    output_filename: str = Field(
        description="Bliss dataset HDF5 file name.",
        examples=["/results/dataset.h5"],
    )
    scan_numbers: Union[int, List[int], None] = Field(
        default=None, description="Scan numbers.", examples=[1, [1, 2, 3]]
    )
    subscan_numbers: Union[int, List[int], None] = Field(
        default=None, description="Subscan numbers.", examples=[1, [1, 2]]
    )





class Outputs(BaseOutputModel):
    output_filename: str = Field(
        description="Bliss dataset HDF5 file name.",
        examples=["/results/dataset.h5"],
    )





class SpecToBliss(Task, input_model=Inputs, output_model=Outputs):
    """Convert SPEC master file to BLISS dataset HDF5 file."""



    def run(self):
        spec_filename = self.inputs.input_filename
        bliss_filename = self.inputs.output_filename
        spec_to_bliss(
            spec_filename,
            bliss_filename,
            scans=self.inputs.scan_numbers,
            subscans=self.inputs.subscan_numbers,
            mode="a",
        )
        self.outputs.output_filename = bliss_filename