Source code for ewoksfluo.tasks.sum_detectors.sum_results

import logging
from typing import List

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

from ..math import format_expression_template
from ..models import FluoInputs
from .sum_utils import detector_weight_iterator
from .sum_utils import save_summed_xrf_results

_logger = logging.getLogger(__name__)




class Inputs(FluoInputs):
    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="Output root URI.",
        examples=[
            "/results/dataset.h5",
            "/results/dataset.h5::/1.1",
            "/results/dataset.h5::/1.1/sum",
        ],
    )

    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: str = Field(
        default="1./<instrument/{}/live_time>",
        description="Expression to evaluate and multiply each spectrum with before summing. "
        "Templates marked with '<...>' are relative to `bliss_scan_uri`.",
    )





class Outputs(BaseOutputModel):
    bliss_scan_uri: str = Field(
        description="Bliss scan URI.", examples=["/data/dataset.h5::/1.1"]
    )
    output_root_uri: str = Field(
        description="Output NXentry URI.", examples=["/results/dataset.h5::/1.1"]
    )
    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
        }
        self.outputs.xrf_results_uri = save_summed_xrf_results(
            self.inputs.xrf_results_uris,
            detector_weights,
            self.inputs.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