Source code for ewoksfluo.tasks.input.pick_scan

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

from ..nexus_utils import wait_scan_finished




class Inputs(BaseInputModel):
    filename: str = Field(
        description="Bliss dataset HDF5 file name.", examples=["/data/dataset.h5"]
    )
    scan_number: int = Field(description="Scan number.", examples=[1, 2])





class Outputs(BaseOutputModel):
    bliss_scan_uri: str = Field(
        description="Bliss scan URI.", examples=["/data/dataset.h5::/1.1"]
    )





class PickScan(Task, input_model=Inputs, output_model=Outputs):
    """Select a single Bliss scan."""



    def run(self):
        filename = self.inputs.filename
        scan_number = self.inputs.scan_number
        bliss_scan_uri = f"{filename}::/{scan_number}.1"
        wait_scan_finished(bliss_scan_uri)
        self.outputs.bliss_scan_uri = bliss_scan_uri