Source code for ewoksid16a.tasks.fluo.helpers

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Wed Mar 25 13:05:03 2026

@author: blissadm
"""

from ewokscore import Task
from silx.io import h5py_utils
from silx.io.url import DataUrl

DEFAULTS = {
    "output_root_group": None,
    "fscan_parameters": "instrument/fscan_parameters",
    "instrument_positioners_template": "measurement/{}",
    "horizontal_motors": ["sy", "spy", "spyp"],
}




class PositionersFromFscan(
    Task,
    input_names=[
        "bliss_scan_uri",
        "output_root_uri",
        "xrf_results_uri",
    ],
    optional_input_names=[
        "output_root_group",
        "fscan_parameters",
        "horizontal_motors",
        "instrument_positioners_template",
    ],
    output_names=[
        "xrf_results_uri",
        "bliss_scan_uri",
        "output_root_uri",
        "output_root_group",
        "positioners",
    ],
):



    def run(self):
        params = {
            **DEFAULTS,
            **self.get_input_values(),
        }
        bliss_scan_uri = DataUrl(params["bliss_scan_uri"])
        fscan_parameters = params["fscan_parameters"]
        ipt = params["instrument_positioners_template"]
        horz_motors = params["horizontal_motors"]

        with h5py_utils.open_item(bliss_scan_uri.file_path(), bliss_scan_uri.data_path()) as scan:  # type: ignore[reportGeneralTypeIssues]

            fp = scan[fscan_parameters]

            fastmot = fp["fast_motor"][()].decode()
            slowmot = fp["slow_motor"][()].decode()

            if ipt.format(fastmot) not in scan:
                if ipt.format(f"axis_{fastmot}") in scan:
                    fastmot = f"axis_{fastmot}"
                else:
                    raise RuntimeError(
                        f"Can't find {fastmot} nor axis_{fastmot} in positioners!"
                    )

            if ipt.format(slowmot) not in scan:
                if ipt.format(f"axis_{slowmot}") in scan:
                    slowmot = f"axis_{slowmot}"
                else:
                    raise RuntimeError(
                        f"Can't find {slowmot} nor axis_{slowmot} in positioners!"
                    )

        print(f"{fastmot=} {slowmot=}")

        if slowmot in horz_motors:
            fastmot, slowmot = slowmot, fastmot

        self.outputs.positioners = [slowmot, fastmot]

        self.outputs.bliss_scan_uri = params["bliss_scan_uri"]
        self.outputs.output_root_uri = params["output_root_uri"]
        self.outputs.xrf_results_uri = params["xrf_results_uri"]
        self.outputs.output_root_group = params["output_root_group"]