Feature/dei 225 filter extremes rule

decoimpact/business/entities/rules/depth_average_rule.py

@@ -47,11 +47,14 @@ def execute(
         variables = next(iter(value_arrays.values()))
 
         bed_level_values = self._extract_variable_based_on_suffix(
-            value_arrays, BED_LEVEL_SUFFIX)
+            value_arrays, BED_LEVEL_SUFFIX
+        )
         depths_interfaces = self._extract_variable_based_on_suffix(
-            value_arrays, INTERFACES_Z_SUFFIX)
+            value_arrays, INTERFACES_Z_SUFFIX
+        )
         water_level_values = self._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]
@@ -110,10 +113,8 @@ def execute(
         )
 
     def _extract_variable_based_on_suffix(
-            self,
-            value_arrays: Dict[str, _xr.DataArray],
-            suffix: str
-            ) -> List:
+        self, 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.
 

decoimpact/business/entities/rules/filter_extremes_rule.py

@@ -0,0 +1,128 @@
+# 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.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"""
+
+    def __init__(
+        self,
+        name: str,
+        input_variable_names: List[str],
+        extreme_type: str,
+        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 = 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) -> str:
+        """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 peak or through values
+        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)
+        old_dr = _xr.DataArray(value_array)
+        time = old_dr.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,
+            old_dr,
+            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(*old_dr.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/workflow/model_builder.py

@@ -20,6 +20,7 @@
 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
@@ -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
@@ -108,6 +110,15 @@ def _create_rule(rule_data: IRuleData) -> IRule:
                 rule_data.name,
                 rule_data.input_variables,
             )
+        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,63 @@
+# 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"""
+
+    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

decoimpact/data/parsers/parser_filter_extremes_rule.py

@@ -0,0 +1,56 @@
+# 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 ParserFilterExtremesRule class
+
+Classes:
+    ParserFilterExtremesRule
+"""
+from typing import Any, Dict, List
+
+from decoimpact.crosscutting.i_logger import ILogger
+from decoimpact.data.api.i_rule_data import IRuleData
+from decoimpact.data.dictionary_utils import get_dict_element
+from decoimpact.data.entities.filter_extremes_rule_data import FilterExtremesRuleData
+from decoimpact.data.parsers.i_parser_rule_base import IParserRuleBase
+
+
+class ParserFilterExtremesRule(IParserRuleBase):
+    """Class for creating a ParserFilterExtremesRule"""
+
+    @property
+    def rule_type_name(self) -> str:
+        """Type name for the rule"""
+        return "filter_extremes_rule"
+
+    def parse_dict(self, dictionary: Dict[str, Any], logger: ILogger) -> IRuleData:
+        """Parses the provided dictionary to a IRuleData
+        Args:
+            dictionary (Dict[str, Any]): Dictionary holding the values
+                                         for making the rule
+        Returns:
+            RuleBase: Rule based on the provided data
+        """
+        name: str = get_dict_element("name", dictionary)
+        input_variable_names: List[str] = [
+            get_dict_element("input_variable", dictionary)
+        ]
+        output_variable_name: str = get_dict_element("output_variable", dictionary)
+        description: str = get_dict_element("description", dictionary, False) or ""
+        extreme_type: str = get_dict_element("extreme_type", dictionary)
+        distance: int = get_dict_element("distance", dictionary) or 0
+        time_scale: str = get_dict_element("time_scale", dictionary) or "D"
+
+        mask: bool = get_dict_element("mask", dictionary) or False
+        rule_data = FilterExtremesRuleData(
+            name, input_variable_names, extreme_type, distance, time_scale, mask
+        )
+
+        rule_data.output_variable = output_variable_name
+        rule_data.description = description
+
+        return rule_data

decoimpact/data/parsers/rule_parsers.py

@@ -16,6 +16,7 @@
 from decoimpact.data.parsers.parser_axis_filter_rule import ParserAxisFilterRule
 from decoimpact.data.parsers.parser_classification_rule import ParserClassificationRule
 from decoimpact.data.parsers.parser_combine_results_rule import ParserCombineResultsRule
+from decoimpact.data.parsers.parser_filter_extremes_rule import ParserFilterExtremesRule
 from decoimpact.data.parsers.parser_formula_rule import ParserFormulaRule
 from decoimpact.data.parsers.parser_layer_filter_rule import ParserLayerFilterRule
 from decoimpact.data.parsers.parser_multiply_rule import ParserMultiplyRule
@@ -43,3 +44,4 @@ def rule_parsers() -> Iterator[IParserRuleBase]:
     yield ParserClassificationRule()
     yield ParserAxisFilterRule()
     yield ParserDepthAverageRule()
+    yield ParserFilterExtremesRule()