Feature/dei 225 filter extremes rule

decoimpact/business/entities/rules/combine_results_rule.py

@@ -19,7 +19,7 @@
 from decoimpact.business.entities.rules.i_multi_array_based_rule import (
     IMultiArrayBasedRule,
 )
-from decoimpact.business.entities.rules.multi_array_operation_type import (
+from decoimpact.business.entities.rules.options.multi_array_operation_type import (
     MultiArrayOperationType,
 )
 from decoimpact.business.entities.rules.rule_base import RuleBase

decoimpact/business/entities/rules/depth_average_rule.py

@@ -54,11 +54,14 @@ def execute(
         interface_suffix = _get_layer_suffix(self._layer_type, logger)
 
         bed_level_values = _extract_variable_based_on_suffix(
-            value_arrays, BED_LEVEL_SUFFIX)
+            value_arrays, BED_LEVEL_SUFFIX
+        )
         depths_interfaces = _extract_variable_based_on_suffix(
-            value_arrays, interface_suffix)
+            value_arrays, interface_suffix
+        )
         water_level_values = _extract_variable_based_on_suffix(
-            value_arrays, WATER_LEVEL_SUFFIX)
+            value_arrays, WATER_LEVEL_SUFFIX
+        )
 
         # Get the dimension names for the interfaces and for the layers
         dim_interfaces_name = list(depths_interfaces.dims)[0]
@@ -118,9 +121,8 @@ def execute(
 
 
 def _extract_variable_based_on_suffix(
-        value_arrays: Dict[str, _xr.DataArray],
-        suffix: str
-        ) -> List:
+    value_arrays: Dict[str, _xr.DataArray], suffix: str
+) -> List:
     """Extract the values from the XArray dataset based on the name
     suffixes by matching the name, irrespective of the dummy name prefix.
 
@@ -134,10 +136,7 @@ def _extract_variable_based_on_suffix(
     return [value_arrays[name] for name in value_arrays if suffix in name][0]
 
 
-def _get_layer_suffix(
-        layer_type: str,
-        logger: ILogger
-        ):
+def _get_layer_suffix(layer_type: str, logger: ILogger):
     """Get the interface suffix depending on whether the model is a sigma or z
     layer model. Give error if the interface suffix cannot be determined.
 
@@ -151,7 +150,9 @@ def _get_layer_suffix(
         return INTERFACES_SIGMA_SUFFIX
     if layer_type.lower() == "z":
         return INTERFACES_Z_SUFFIX
-    logger.log_error(f"Layer type {layer_type} unknown. Allowed layer "
-                     "type: z or sigma. Interface "
-                     "variable could not be determined.")
+    logger.log_error(
+        f"Layer type {layer_type} unknown. Allowed layer "
+        "type: z or sigma. Interface "
+        "variable could not be determined."
+    )
     return "_unknown"

decoimpact/business/entities/rules/filter_extremes_rule.py

@@ -0,0 +1,137 @@
+# This file is part of D-EcoImpact
+# Copyright (C) 2022-2024 Stichting Deltares
+# This program is free software distributed under the
+# GNU Affero General Public License version 3.0
+# A copy of the GNU Affero General Public License can be found at
+# https://github.com/Deltares/D-EcoImpact/blob/main/LICENSE.md
+"""
+Module for FilterExtremesRule class
+
+Classes:
+    FilterExtremesRule
+"""
+
+from typing import List
+
+import xarray as _xr
+import scipy as _sc
+import numpy as _np
+
+from decoimpact.business.entities.rules.i_array_based_rule import IArrayBasedRule
+from decoimpact.business.entities.rules.options.options_filter_extreme_rule import (
+    ExtremeTypeOptions,
+)
+from decoimpact.business.entities.rules.rule_base import RuleBase
+from decoimpact.business.entities.rules.time_operation_settings import (
+    TimeOperationSettings,
+)
+from decoimpact.business.utils.data_array_utils import get_time_dimension_name
+from decoimpact.crosscutting.i_logger import ILogger
+from decoimpact.data.dictionary_utils import get_dict_element
+
+
+class FilterExtremesRule(RuleBase, IArrayBasedRule):
+    """Implementation for the filter extremes rule"""
+
+    # pylint: disable=too-many-arguments
+    def __init__(
+        self,
+        name: str,
+        input_variable_names: List[str],
+        extreme_type: ExtremeTypeOptions,
+        distance: int,
+        time_scale: str,
+        mask: bool,
+    ):
+        super().__init__(name, input_variable_names)
+        self._settings = TimeOperationSettings(
+            {"second": "s", "hour": "h", "day": "D", "month": "M", "year": "Y"}
+        )
+        self._extreme_type: ExtremeTypeOptions = extreme_type
+        self._distance = distance
+        self._settings.time_scale = time_scale
+        self._mask = mask
+
+    @property
+    def settings(self):
+        """Time operation settings"""
+        return self._settings
+
+    @property
+    def extreme_type(self) -> ExtremeTypeOptions:
+        """Type of extremes (peaks or troughs)"""
+        return self._extreme_type
+
+    @property
+    def distance(self) -> int:
+        """Minimal distance between peaks"""
+        return self._distance
+
+    @property
+    def mask(self) -> bool:
+        """Return either directly the values of the filtered array or a
+        True/False array"""
+        return self._mask
+
+    def validate(self, logger: ILogger) -> bool:
+        """Validates if the rule is valid
+
+        Returns:
+            bool: wether the rule is valid
+        """
+        return self.settings.validate(self.name, logger)
+
+    def execute(self, value_array: _xr.DataArray, logger: ILogger) -> _xr.DataArray:
+        """
+        Retrieve the extremes
+        extreme_type: Either retrieve the values at the peaks or troughs
+        mask: If False return the values at the peaks, otherwise return a
+        1 at the extreme locations.
+
+        Args:
+            value_array (DataArray): Values to filter at extremes
+
+        Returns:
+            DataArray: Filtered DataArray with only the extremes remaining
+            at all other times the values are set to NaN
+        """
+
+        time_scale = get_dict_element(
+            self.settings.time_scale, self.settings.time_scale_mapping
+        )
+
+        time_dim_name = get_time_dimension_name(value_array, logger)
+        time = value_array.time.values
+        timestep = (time[-1] - time[0]) / len(time)
+        width_time = _np.timedelta64(self.distance, time_scale)
+        distance = width_time / timestep
+
+        results = _xr.apply_ufunc(
+            self._process_peaks,
+            value_array,
+            input_core_dims=[[time_dim_name]],
+            output_core_dims=[[time_dim_name]],
+            vectorize=True,
+            kwargs={
+                "distance": distance,
+                "mask": self.mask,
+                "extreme_type": self.extreme_type,
+            },
+        )
+
+        results = results.transpose(*value_array.dims)
+        return results
+
+    def _process_peaks(
+        self, arr: _xr.DataArray, distance: float, mask: bool, extreme_type: str
+    ):
+        factor = 1
+        if extreme_type == "troughs":
+            factor = -1
+        peaks, _ = _sc.signal.find_peaks(factor * arr, distance=distance)
+        values = arr[peaks]
+        if mask:
+            values = True
+        new_arr = _np.full_like(arr, _np.nan, dtype=float)
+        new_arr[peaks] = values
+        return new_arr

decoimpact/business/entities/rules/options/options_filter_extreme_rule.py

@@ -0,0 +1,20 @@
+# This file is part of D-EcoImpact
+# Copyright (C) 2022-2024 Stichting Deltares
+# This program is free software distributed under the
+# GNU Affero General Public License version 3.0
+# A copy of the GNU Affero General Public License can be found at
+# https://github.com/Deltares/D-EcoImpact/blob/main/LICENSE.md
+"""
+Module for ExtremeTypeOptions Class
+
+Classes:
+    ExtremeTypeOptions
+"""
+from enum import StrEnum, auto
+
+
+class ExtremeTypeOptions(StrEnum):
+    """Classify the extreme type options."""
+
+    PEAKS = auto()
+    TROUGHS = auto()

decoimpact/business/workflow/model_builder.py

@@ -20,10 +20,11 @@
 from decoimpact.business.entities.rules.classification_rule import ClassificationRule
 from decoimpact.business.entities.rules.combine_results_rule import CombineResultsRule
 from decoimpact.business.entities.rules.depth_average_rule import DepthAverageRule
+from decoimpact.business.entities.rules.filter_extremes_rule import FilterExtremesRule
 from decoimpact.business.entities.rules.formula_rule import FormulaRule
 from decoimpact.business.entities.rules.i_rule import IRule
 from decoimpact.business.entities.rules.layer_filter_rule import LayerFilterRule
-from decoimpact.business.entities.rules.multi_array_operation_type import (
+from decoimpact.business.entities.rules.options.multi_array_operation_type import (
     MultiArrayOperationType,
 )
 from decoimpact.business.entities.rules.multiply_rule import MultiplyRule
@@ -41,6 +42,7 @@
 from decoimpact.data.api.i_combine_results_rule_data import ICombineResultsRuleData
 from decoimpact.data.api.i_data_access_layer import IDataAccessLayer
 from decoimpact.data.api.i_depth_average_rule_data import IDepthAverageRuleData
+from decoimpact.data.api.i_filter_extremes_rule_data import IFilterExtremesRuleData
 from decoimpact.data.api.i_formula_rule_data import IFormulaRuleData
 from decoimpact.data.api.i_layer_filter_rule_data import ILayerFilterRuleData
 from decoimpact.data.api.i_model_data import IModelData
@@ -109,6 +111,15 @@ def _create_rule(rule_data: IRuleData) -> IRule:
                 rule_data.input_variables,
                 rule_data.layer_type,
             )
+        elif isinstance(rule_data, IFilterExtremesRuleData):
+            rule = FilterExtremesRule(
+                rule_data.name,
+                rule_data.input_variables,
+                rule_data.extreme_type,
+                rule_data.distance,
+                rule_data.time_scale,
+                rule_data.mask,
+            )
         elif isinstance(rule_data, ILayerFilterRuleData):
             rule = LayerFilterRule(
                 rule_data.name,

decoimpact/data/api/i_filter_extremes_rule_data.py

@@ -0,0 +1,45 @@
+# This file is part of D-EcoImpact
+# Copyright (C) 2022-2024 Stichting Deltares
+# This program is free software distributed under the
+# GNU Affero General Public License version 3.0
+# A copy of the GNU Affero General Public License can be found at
+# https://github.com/Deltares/D-EcoImpact/blob/main/LICENSE.md
+"""Module for IFilterExtremesRuleData interface
+
+Interfaces:
+    IFilterExtremesRuleData
+"""
+
+from abc import ABC, abstractmethod
+from typing import List
+
+from decoimpact.data.api.i_rule_data import IRuleData
+
+
+class IFilterExtremesRuleData(IRuleData, ABC):
+    """Data for a filter extremes rule"""
+
+    @property
+    @abstractmethod
+    def input_variables(self) -> List[str]:
+        """List with input variable name"""
+
+    @property
+    @abstractmethod
+    def extreme_type(self) -> str:
+        """Type of extremes [peaks or throughs]"""
+
+    @property
+    @abstractmethod
+    def distance(self) -> int:
+        """Property for the distance between peaks"""
+
+    @property
+    @abstractmethod
+    def time_scale(self) -> str:
+        """Property for the timescale of the distance between peaks"""
+
+    @property
+    @abstractmethod
+    def mask(self) -> bool:
+        """Property for mask"""

decoimpact/data/entities/depth_average_rule_data.py

@@ -5,10 +5,10 @@
 # A copy of the GNU Affero General Public License can be found at
 # https://github.com/Deltares/D-EcoImpact/blob/main/LICENSE.md
 """
-Module for (multiple) ClassificationRule class
+Module for (multiple) DepthAverageRule class
 
 Classes:
-    (multiple) ClassificationRuleData
+    (multiple) DepthAverageRuleData
 
 """
 

decoimpact/data/entities/filter_extremes_rule_data.py

@@ -0,0 +1,64 @@
+# This file is part of D-EcoImpact
+# Copyright (C) 2022-2024 Stichting Deltares
+# This program is free software distributed under the
+# GNU Affero General Public License version 3.0
+# A copy of the GNU Affero General Public License can be found at
+# https://github.com/Deltares/D-EcoImpact/blob/main/LICENSE.md
+"""
+Module for FilterExtremesRuleData class
+
+Classes:
+    FilterExtremesRuleData
+
+"""
+
+from typing import List
+
+from decoimpact.data.api.i_filter_extremes_rule_data import IFilterExtremesRuleData
+from decoimpact.data.entities.rule_data import RuleData
+
+
+class FilterExtremesRuleData(IFilterExtremesRuleData, RuleData):
+    """Class for storing data related to filter extremes rule"""
+
+    # pylint: disable=too-many-arguments
+    def __init__(
+        self,
+        name: str,
+        input_variables: List[str],
+        extreme_type: str,
+        distance: int,
+        time_scale: str,
+        mask: bool,
+    ):
+        super().__init__(name)
+        self._input_variables = input_variables
+        self._extreme_type = extreme_type
+        self._distance = distance
+        self._time_scale = time_scale
+        self._mask = mask
+
+    @property
+    def input_variables(self) -> List[str]:
+        """List with input variables"""
+        return self._input_variables
+
+    @property
+    def extreme_type(self) -> str:
+        """Property for the extremes type"""
+        return self._extreme_type
+
+    @property
+    def distance(self) -> int:
+        """Property for the distance between peaks"""
+        return self._distance
+
+    @property
+    def time_scale(self) -> str:
+        """Property for the timescale of the distance between peaks"""
+        return self._time_scale
+
+    @property
+    def mask(self) -> bool:
+        """Property for mask"""
+        return self._mask