Extracted common functionality to paring helper time_operation_parsing

decoimpact/data/parsers/parser_rolling_statistics_rule.py

@@ -10,16 +10,16 @@
 Classes:
     ParserRollingStatisticsRule
 """
-from typing import Any, Dict, Tuple
+from typing import Any, Dict
 
 from decoimpact.crosscutting.i_logger import ILogger
 from decoimpact.data.api.i_rule_data import IRuleData
-from decoimpact.data.api.time_operation_type import TimeOperationType
 from decoimpact.data.dictionary_utils import get_dict_element
 from decoimpact.data.entities.rolling_statistics_rule_data import (
     RollingStatisticsRuleData,
 )
 from decoimpact.data.parsers.i_parser_rule_base import IParserRuleBase
+from decoimpact.data.parsers.time_operation_parsing import parse_operation_values
 
 
 class ParserRollingStatisticsRule(IParserRuleBase):
@@ -39,11 +39,11 @@ def parse_dict(self, dictionary: Dict[str, Any], logger: ILogger) -> IRuleData:
             RuleBase: Rule based on the provided data
         """
         # get elements
-        name = get_dict_element("name", dictionary)
-        input_variable_name = get_dict_element("input_variable", dictionary)
-        operation = get_dict_element("operation", dictionary)
-        time_scale = get_dict_element("time_scale", dictionary)
-        period = get_dict_element("period", dictionary)
+        name: str = get_dict_element("name", dictionary)
+        input_variable_name: str = get_dict_element("input_variable", dictionary)
+        operation: str = get_dict_element("operation", dictionary)
+        time_scale: str = get_dict_element("time_scale", dictionary)
+        period: float = get_dict_element("period", dictionary)
         description = get_dict_element("description", dictionary, False)
         output_variable_name = get_dict_element("output_variable", dictionary)
 
@@ -52,7 +52,7 @@ def parse_dict(self, dictionary: Dict[str, Any], logger: ILogger) -> IRuleData:
                       a float or integer value. Received: {period}"
             raise ValueError(message)
 
-        operation_value, operation_parameter = self._get_operation_values(operation)
+        operation_value, operation_parameter = parse_operation_values(operation)
 
         rule_data = RollingStatisticsRuleData(
             name, operation_value, input_variable_name, period
@@ -64,38 +64,3 @@ def parse_dict(self, dictionary: Dict[str, Any], logger: ILogger) -> IRuleData:
         rule_data.description = description
 
         return rule_data
-
-    def _get_operation_values(
-        self, operation_str: str
-    ) -> Tuple[TimeOperationType, float]:
-        # if operation contains percentile,
-        # extract percentile value as operation_parameter from operation:
-        if str(operation_str)[:10] == "PERCENTILE":
-            try:
-                operation_parameter = float(str(operation_str)[11:-1])
-            except ValueError as exc:
-                message = (
-                    "Operation percentile is missing valid value like 'percentile(10)'"
-                )
-                raise ValueError(message) from exc
-
-            # test if operation_parameter is within expected limits:
-            if (
-                operation_parameter is None
-                or operation_parameter < 0
-                or operation_parameter > 100
-            ):
-                message = "Operation percentile should be a number between 0 and 100."
-                raise ValueError(message)
-            return TimeOperationType.PERCENTILE, operation_parameter
-
-        # validate operation
-        match_operation = [o for o in TimeOperationType if o.name == operation_str]
-        operation_value = next(iter(match_operation), None)
-
-        if not operation_value:
-            message = f"Operation is not of a predefined type. Should be in: \
-                      {[o.name for o in TimeOperationType]}. Received: {operation_str}"
-            raise ValueError(message)
-
-        return operation_value, 0

decoimpact/data/parsers/parser_time_aggregation_rule.py

@@ -14,10 +14,10 @@
 
 from decoimpact.crosscutting.i_logger import ILogger
 from decoimpact.data.api.i_rule_data import IRuleData
-from decoimpact.data.api.time_operation_type import TimeOperationType
 from decoimpact.data.dictionary_utils import get_dict_element
 from decoimpact.data.entities.time_aggregation_rule_data import TimeAggregationRuleData
 from decoimpact.data.parsers.i_parser_rule_base import IParserRuleBase
+from decoimpact.data.parsers.time_operation_parsing import parse_operation_values
 
 
 class ParserTimeAggregationRule(IParserRuleBase):
@@ -44,39 +44,7 @@ def parse_dict(self, dictionary: Dict[str, Any], logger: ILogger) -> IRuleData:
         time_scale: str = get_dict_element("time_scale", dictionary)
         output_variable_name: str = get_dict_element("output_variable", dictionary)
 
-        operation_parameter = 0
-
-        # if operation contains percentile,
-        # extract percentile value as operation_parameter from operation:
-        if str(operation)[:10] == "PERCENTILE":
-            try:
-                operation_parameter = float(str(operation)[11:-1])
-            except ValueError as exc:
-                message = (
-                    "Operation percentile is missing valid value like 'percentile(10)'"
-                )
-                raise ValueError(message) from exc
-            operation = "PERCENTILE"
-
-        # validate operation
-        match_operation = [o for o in TimeOperationType if o.name == operation]
-        operation_value = next(iter(match_operation), None)
-
-        # validate operation_value (percentile(n); n = operation_value)
-        if not operation_value:
-            message = f"Operation is not of a predefined type. Should be in: \
-                      {[o.name for o in TimeOperationType]}. Received: {operation}"
-            raise ValueError(message)
-
-        # test if operation_parameter is within expected limits:
-        if operation_value == TimeOperationType.PERCENTILE:
-            if (
-                operation_parameter is None
-                or operation_parameter < 0
-                or operation_parameter > 100
-            ):
-                message = "Operation percentile should be a number between 0 and 100."
-                raise ValueError(message)
+        operation_value, operation_parameter = parse_operation_values(operation)
 
         rule_data = TimeAggregationRuleData(name, operation_value, input_variable_name)
 

decoimpact/data/parsers/time_operation_parsing.py

@@ -0,0 +1,47 @@
+# 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 ParserTimeAggregationRule class
+
+Classes:
+    ParserTimeAggregationRule
+"""
+from typing import Tuple
+
+from decoimpact.data.api.time_operation_type import TimeOperationType
+
+
+def parse_operation_values(operation_str: str) -> Tuple[TimeOperationType, float]:
+    # if operation contains percentile,
+    # extract percentile value as operation_parameter from operation:
+    if str.startswith(operation_str, "PERCENTILE"):
+        try:
+            operation_parameter = float(str(operation_str)[11:-1])
+        except ValueError as exc:
+            message = (
+                "Operation percentile is missing valid value like 'percentile(10)'"
+            )
+            raise ValueError(message) from exc
+
+        # test if operation_parameter is within expected limits:
+        if operation_parameter < 0 or operation_parameter > 100:
+            message = "Operation percentile should be a number between 0 and 100."
+            raise ValueError(message)
+        return TimeOperationType.PERCENTILE, operation_parameter
+
+    # validate operation
+    match_operation = [o for o in TimeOperationType if o.name == operation_str]
+    operation_value = next(iter(match_operation), None)
+
+    if not operation_value:
+        message = (
+            f"Operation '{operation_str}' is not of a predefined type. Should be in:"
+            + f"{[o.name for o in TimeOperationType]}."
+        )
+        raise ValueError(message)
+
+    return operation_value, 0

tests/data/parsers/test_parser_rolling_statistics_rule.py

@@ -80,8 +80,10 @@ def test_parse_operation_type():
     exception_raised = exc_info.value
 
     # Assert
-    expected_message = f"Operation is not of a predefined type. Should be in: \
-                      {[o.name for o in TimeOperationType]}. Received: Minimum"
+    expected_message = (
+        "Operation 'Minimum' is not of a predefined type. Should be in:"
+        + f"{[o.name for o in TimeOperationType]}."
+    )
     assert exception_raised.args[0] == expected_message
 
 

tests/data/parsers/test_parser_time_aggregation_rule.py

@@ -103,8 +103,10 @@ def test_parse_operation_type():
     exception_raised = exc_info.value
 
     # Assert
-    expected_message = f"Operation is not of a predefined type. Should be in: \
-                      {[o.name for o in TimeOperationType]}. Received: Minimum"
+    expected_message = (
+        "Operation 'Minimum' is not of a predefined type. Should be in:"
+        + f"{[o.name for o in TimeOperationType]}."
+    )
     assert exception_raised.args[0] == expected_message