Source code for ewoksfluo.tasks.sum_detectors.sum_results

import logging
from typing import List
from typing import Optional

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

from ...io import hdf5
from ...io import output_uri
from ..math import format_expression_template
from .sum_utils import detector_weight_iterator
from .sum_utils import save_summed_xrf_results

_logger = logging.getLogger(__name__)




class Inputs(BaseInputModel):
    bliss_scan_uri: str = Field(
        description="Bliss scan URI.", examples=["/data/dataset.h5::/1.1"]
    )
    detector_names: List[str] = Field(
        description="NXdetector group name.",
        min_length=1,
        examples=[["fx_nano_det0", "fx_nano_det1"]],
    )
    output_root_uri: str = Field(
        description="Target HDF5 file URI with optional data path.",
        examples=[
            "/results/dataset.h5",
            "/results/dataset.h5::/1.1",
            "/results/dataset.h5::/1.1/sum",
        ],
    )
    output_root_group: Optional[str] = Field(
        default=None, description="Optional group underneath ``output_root_uri``."
    )
    xrf_results_uris: List[str] = Field(
        description="Output NXprocess results URIs.",
        examples=[
            [
                "/results/dataset.h5::/1.1/fit/mca1/results",
                "/results/dataset.h5::/1.1/fit/mca2/results",
            ]
        ],
    )
    detector_normalization_template: Optional[str] = Field(
        default="1./<instrument/{}/live_time>",
        description="Expression to evaluate and multiply each spectrum with before summing. "
        "HDF5 URI's <...> are relative to ``bliss_scan_uris``.",
    )





class Outputs(BaseOutputModel):
    bliss_scan_uri: str = Field(
        description="Bliss scan URI.", examples=["/data/dataset.h5::/1.1"]
    )
    output_root_uri: str = Field(
        description="Original output root URI received as input.",
        examples=["/results/dataset.h5::/1.1"],
    )
    output_root_group: Optional[str] = Field(
        default=None, description="Original output root group received as input."
    )
    xrf_results_uri: str = Field(
        description="Output NXprocess results URI.",
        examples=["/results/dataset.h5::/1.1/sum/results"],
    )





class SumXrfResults(Task, input_model=Inputs, output_model=Outputs):
    """Add single-scan XRF results of multiple detectors."""



    def run(self) -> None:
        weight_expressions = [
            format_expression_template(
                self.inputs.detector_normalization_template, name
            )
            for name in self.inputs.detector_names
        ]
        for weight_expression, detector_name in zip(
            weight_expressions, self.inputs.detector_names
        ):
            _logger.info(
                "Detector %r weights for summing XRF results: %s",
                detector_name,
                weight_expression,
            )
        detector_weights = detector_weight_iterator(
            self.inputs.bliss_scan_uri, weight_expressions
        )

        process_config = {
            "detector_normalization_template": self.inputs.detector_normalization_template
        }
        _, scan_h5path = hdf5.split_h5uri(self.inputs.bliss_scan_uri)
        output_root_uri = output_uri.compose_full_output_uri(
            self.inputs.output_root_uri,
            default_output_data_path=scan_h5path,
            extra_data_paths=(self.inputs.output_root_group, "sum"),
        )

        self.outputs.xrf_results_uri = save_summed_xrf_results(
            self.inputs.xrf_results_uris,
            detector_weights,
            output_root_uri,
            process_config,
        )
        self.outputs.bliss_scan_uri = self.inputs.bliss_scan_uri
        self.outputs.output_root_uri = self.inputs.output_root_uri
        self.outputs.output_root_group = self.inputs.output_root_group