Source code for ewoksfluo.tasks.input.concat_bliss

from typing import Dict
from typing import List

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

from ...io.blissconcat import concatenate_bliss_scans




class Inputs(BaseInputModel):
    bliss_scan_uris: List[str] = Field(
        description="List of Bliss scan URI's to concatenate.",
        examples=[
            [
                "/data/dataset1.h5::/2.1",
                "/data/dataset1.h5::/4.1",
                "/data/dataset2.h5::/100.1",
                "/data/dataset2.h5::/101.1",
            ]
        ],
    )
    bliss_scan_uri: str = Field(
        description="Concatenated Bliss scan URI to save.",
        examples=[
            "/result/concatenated.h5",
            "/result/concatenated.h5::/2.1",
        ],
    )
    virtual_axes: Dict[str, str] = Field(
        default_factory=dict,
        description="Virtual positioners which are typically the sum of scalar and moving motor positions.",
        examples=[{"sy": "<nsy>+<nspy>", "sz": "<nsz>+<nspz>"}],
    )
    axes_units: Dict[str, str] = Field(
        default_factory=dict,
        description="Axes units to be used when missing.",
        examples=[{"nsy": "mm", "nsz": "mm", "nspy": "um", "nspz": "um"}],
    )





class Outputs(BaseOutputModel):
    bliss_scan_uri: str = Field(
        description="Bliss scan URI.", examples=["/result/concatenated.h5::/2.1"]
    )





class ConcatBliss(Task, input_model=Inputs, output_model=Outputs):
    """Concatenate Bliss scans."""



    def run(self):
        self.outputs.bliss_scan_uri = concatenate_bliss_scans(
            self.inputs.bliss_scan_uris,
            self.inputs.bliss_scan_uri,
            virtual_axes=self.inputs.virtual_axes,
            axes_units=self.inputs.axes_units,
        )