combine_results_rule

Module for CombineResultsRule Class

!!! classes CombineResultsRule

CombineResultsRule (RuleBase, IMultiArrayBasedRule)

Implementation for the combine results rule

Source code in rules/combine_results_rule.py

class CombineResultsRule(RuleBase, IMultiArrayBasedRule):
    """Implementation for the combine results rule"""

    def __init__(
        self,
        name: str,
        input_variable_names: List[str],
        operation_type: MultiArrayOperationType,
        ignore_nan: bool = False,
    ):
        super().__init__(name, input_variable_names)
        self._operation_type: MultiArrayOperationType = operation_type
        self._ignore_nan = ignore_nan
        self._operations = self._create_operations()

    @property
    def operation_type(self) -> MultiArrayOperationType:
        """Name of the rule"""
        return self._operation_type

    @property
    def ignore_nan(self) -> bool:
        """Indicates if NaN values should be ignored in the calculations"""
        return self._ignore_nan

    def validate(self, logger: ILogger) -> bool:
        if self._operation_type not in self._operations:

            message = (
                f"Operation type {self._operation_type} is currently" " not supported."
            )

            logger.log_error(message)
            return False

        if len(self._input_variable_names) < 2:
            logger.log_error("Minimum of two input variables required.")
            return False

        return True

    def execute(
        self, value_arrays: Dict[str, _xr.DataArray], logger: ILogger
    ) -> _xr.DataArray:
        """Calculate simple statistical operations with two or more input arrays
        Args:
            input_arrays (DataArray): array list containing the variables
        Returns:
            DataArray: Input arrays
        """
        if len(value_arrays) != len(self._input_variable_names):
            raise ValueError("Not all expected arrays where provided.")

        np_arrays = [a_array.to_numpy() for a_array in value_arrays.values()]
        if not self._check_dimensions(np_arrays):
            raise ValueError("The arrays must have the same dimensions.")

        operation_to_use = self._operations[self._operation_type]

        first_value_array = next(iter(value_arrays.values()))

        result_variable = _xr.DataArray(
            data=operation_to_use(np_arrays),
            dims=first_value_array.dims,
            attrs=first_value_array.attrs,
        )

        return result_variable

    def _create_operations(self) -> dict[MultiArrayOperationType, Callable]:
        if self.ignore_nan:
            return {
                MultiArrayOperationType.MULTIPLY: lambda npa: _np.prod(npa, axis=0),
                MultiArrayOperationType.MIN: lambda npa: _np.nanmin(npa, axis=0),
                MultiArrayOperationType.MAX: lambda npa: _np.nanmax(npa, axis=0),
                MultiArrayOperationType.AVERAGE: lambda npa: _np.nanmean(npa, axis=0),
                MultiArrayOperationType.MEDIAN: lambda npa: _np.nanmedian(npa, axis=0),
                MultiArrayOperationType.ADD: lambda npa: _np.nansum(npa, axis=0),
                MultiArrayOperationType.SUBTRACT: lambda npa: _np.subtract(
                    npa[0], _np.nansum(npa[1:], axis=0)
                ),
            }
        # and if ignore_nan is False:
        return {
            MultiArrayOperationType.MULTIPLY: lambda npa: _np.prod(npa, axis=0),
            MultiArrayOperationType.MIN: lambda npa: _np.min(npa, axis=0),
            MultiArrayOperationType.MAX: lambda npa: _np.max(npa, axis=0),
            MultiArrayOperationType.AVERAGE: lambda npa: _np.average(npa, axis=0),
            MultiArrayOperationType.MEDIAN: lambda npa: _np.median(npa, axis=0),
            MultiArrayOperationType.ADD: lambda npa: _np.sum(npa, axis=0),
            MultiArrayOperationType.SUBTRACT: lambda npa: _np.subtract(
                npa[0], _np.sum(npa[1:], axis=0)
            ),
        }

    def _check_dimensions(self, np_arrays: List[_np.ndarray]) -> bool:
        """Brief check if all the arrays to be combined have the
           same size/dimension/length
        Args:
            np_arrays: List of numpy arrays
        Returns:
            Boolean: True of False
        """

        expected_dimensions = np_arrays[0].ndim
        for a_array in np_arrays[1:]:
            if expected_dimensions != _np.ndim(a_array):
                return False

        expected_shape = np_arrays[0].shape
        for a_array in np_arrays[1:]:
            if expected_shape != a_array.shape:
                return False
        return True

ignore_nan: bool property readonly

Indicates if NaN values should be ignored in the calculations

operation_type: MultiArrayOperationType property readonly

Name of the rule

execute(self, value_arrays, logger)

Calculate simple statistical operations with two or more input arrays

Parameters:

Name	Type	Description	Default
input_arrays	DataArray	array list containing the variables	required

Returns:

Type	Description
DataArray	Input arrays

Source code in rules/combine_results_rule.py

def execute(
    self, value_arrays: Dict[str, _xr.DataArray], logger: ILogger
) -> _xr.DataArray:
    """Calculate simple statistical operations with two or more input arrays
    Args:
        input_arrays (DataArray): array list containing the variables
    Returns:
        DataArray: Input arrays
    """
    if len(value_arrays) != len(self._input_variable_names):
        raise ValueError("Not all expected arrays where provided.")

    np_arrays = [a_array.to_numpy() for a_array in value_arrays.values()]
    if not self._check_dimensions(np_arrays):
        raise ValueError("The arrays must have the same dimensions.")

    operation_to_use = self._operations[self._operation_type]

    first_value_array = next(iter(value_arrays.values()))

    result_variable = _xr.DataArray(
        data=operation_to_use(np_arrays),
        dims=first_value_array.dims,
        attrs=first_value_array.attrs,
    )

    return result_variable

validate(self, logger)

Validates if the rule is valid

Returns:

Type	Description
bool	wether the rule is valid

Source code in rules/combine_results_rule.py

def validate(self, logger: ILogger) -> bool:
    if self._operation_type not in self._operations:

        message = (
            f"Operation type {self._operation_type} is currently" " not supported."
        )

        logger.log_error(message)
        return False

    if len(self._input_variable_names) < 2:
        logger.log_error("Minimum of two input variables required.")
        return False

    return True