Source code for est.core.process.roi

from est.core.types import XASObject
from est.core.types import Spectra
from .process import Process
import logging
from typing import Union

_logger = logging.getLogger(__name__)




def xas_roi(xas_obj, **optional_inputs):
    """
    apply roi on the XASObject.spectra

    :param xas_obj: object containing the configuration and spectra to process
    :type: Union[:class:`.XASObject`, dict]
    :return: spectra dict
    :rtype: :class:`.XASObject`
    """
    process = ROIProcess(inputs={"xas_obj": xas_obj, **optional_inputs})
    process.run()
    return process.get_output_value("xas_obj", None)



class _ROI:
    def __init__(self, origin, size):
        self.origin = origin
        self.size = size

    @property
    def origin(self) -> tuple:
        return self.__origin

    @origin.setter
    def origin(self, origin: Union[list, tuple]):
        self.__origin = int(origin[0]), int(origin[1])

    @property
    def size(self) -> tuple:
        return self.__size

    @size.setter
    def size(self, size: Union[list, tuple]):
        self.__size = int(size[0]), int(size[1])

    def to_dict(self) -> dict:
        return {"origin": self.origin, "size": self.size}

    @staticmethod
    def from_dict(ddict: dict):
        return _ROI(origin=ddict["origin"], size=ddict["size"])

    @staticmethod
    def from_silx_def(silx_roi):
        origin = silx_roi.getOrigin()
        size = silx_roi.getSize()
        return _ROI(origin=origin, size=size)




class ROIProcess(
    Process,
    name="roi",
    input_names=["xas_obj", "roi_size", "roi_origin"],
    output_names=["xas_obj"],
):
    def _apply_roi(self, xas_obj, roi):
        _logger.warning("applying roi")
        assert isinstance(xas_obj, XASObject)
        assert isinstance(roi, _ROI)
        assert type(roi.origin[0]) is int
        assert type(roi.origin[1]) is int
        assert type(roi.size[0]) is int
        assert type(roi.size[1]) is int
        ymin, ymax = roi.origin[1], roi.origin[1] + roi.size[1]
        xmin, xmax = roi.origin[0], roi.origin[0] + roi.size[0]
        # clip roi
        ymin = max(ymin, 0)
        ymax = min(ymax, xas_obj.spectra.shape[0])
        xmin = max(xmin, 0)
        xmax = min(xmax, xas_obj.spectra.shape[1])

        assert type(ymin) is int
        assert type(ymax) is int
        assert type(xmin) is int
        assert type(xmax) is int
        # need to copy it. In orangecontrib for example for now we use the same object.
        # so if this one is modify the 'trigger / reprocessing' will be confusing to users.
        xas_obj = xas_obj.copy()
        data = xas_obj.spectra.data  # .reshape(xas_obj.spectra.shape)
        spectra = data[ymin:ymax, xmin:xmax]
        xas_obj.spectra = Spectra(
            energy=xas_obj.energy,
            spectra=spectra,
        )
        return xas_obj



    def run(self):
        """

        :param xas_obj: object containing the configuration and spectra to process
        :type: Union[:class:`.XASObject`, dict]
        :return: spectra dict
        :rtype: :class:`.XASObject`
        """
        # preprocessing on the roi
        if 0 in self.inputs.roi_size:
            raise ValueError(
                f"Unabme to apply a roi of size 0. Requested ROI size is {self.inputs.roi_size}"
            )

        # handle -1 value
        roi_size = list(self.inputs.roi_size)
        roi_origin = list(self.inputs.roi_origin)
        xas_obj = self.getXasObject(self.inputs.xas_obj)
        if roi_size[0] == -1:
            roi_origin[0] = 0
            roi_size[0] = xas_obj.spectra.shape[1]
        if roi_size[1] == -1:
            roi_origin[1] = 0
            roi_size[1] = xas_obj.spectra.shape[0]
        # define the ROI
        self.roi = _ROI(
            origin=roi_origin,
            size=roi_size,
        )

        # processing
        xas_obj = self.inputs.xas_obj
        _xas_obj = self.getXasObject(xas_obj=xas_obj)

        _xas_obj = self._apply_roi(xas_obj=_xas_obj, roi=self.roi)
        self.outputs.xas_obj = _xas_obj  # .to_dict()




    @staticmethod
    def program_name():
        return "roi"