Added Validation to PEBC Message Classes + Tests of the validation

src/s2python/common/power_forecast_element.py

@@ -1,6 +1,12 @@
 from typing import List
+from typing_extensions import Self
 
-from s2python.generated.gen_s2 import PowerForecastElement as GenPowerForecastElement
+from pydantic import model_validator
+
+from s2python.generated.gen_s2 import (
+    CommodityQuantity,
+    PowerForecastElement as GenPowerForecastElement,
+)
 from s2python.validate_values_mixin import (
     catch_and_convert_exceptions,
     S2MessageComponent,
@@ -18,3 +24,19 @@ class PowerForecastElement(GenPowerForecastElement, S2MessageComponent):
     power_values: List[PowerForecastValue] = (  # type: ignore[reportIncompatibleVariableOverride]
         GenPowerForecastElement.model_fields["power_values"]  # type: ignore[assignment]
     )
+
+    @model_validator(mode="after")
+    def validate_values_at_most_one_per_commodity_quantity(self) -> Self:
+        """Validates the power measurement values to check that there is at most 1 PowerValue per CommodityQuantity."""
+
+        has_value: dict[CommodityQuantity, bool] = {}
+
+        for value in self.power_values:
+            if has_value.get(value.commodity_quantity, False):
+                raise ValueError(
+                    self,
+                    f"There must be at most 1 PowerForecastValue per CommodityQuantity",
+                )
+            has_value[value.commodity_quantity] = True
+
+        return self

src/s2python/common/power_measurement.py

@@ -1,8 +1,13 @@
 from typing import List
+from typing_extensions import Self
 import uuid
 
+from pydantic import model_validator
 from s2python.common.power_value import PowerValue
-from s2python.generated.gen_s2 import PowerMeasurement as GenPowerMeasurement
+from s2python.generated.gen_s2 import (
+    PowerMeasurement as GenPowerMeasurement,
+    CommodityQuantity,
+)
 from s2python.validate_values_mixin import (
     catch_and_convert_exceptions,
     S2MessageComponent,
@@ -16,3 +21,20 @@ class PowerMeasurement(GenPowerMeasurement, S2MessageComponent):
 
     message_id: uuid.UUID = GenPowerMeasurement.model_fields["message_id"]  # type: ignore[assignment,reportIncompatibleVariableOverride]
     values: List[PowerValue] = GenPowerMeasurement.model_fields["values"]  # type: ignore[assignment,reportIncompatibleVariableOverride]
+
+    @model_validator(mode="after")
+    def validate_values_at_most_one_per_commodity_quantity(self) -> Self:
+        """Validates the power measurement values to check that there is at most 1 PowerValue per CommodityQuantity."""
+
+        has_value: dict[CommodityQuantity, bool] = {}
+
+        for value in self.values:
+            if has_value.get(value.commodity_quantity, False):
+                raise ValueError(
+                    self,
+                    f"The measured PowerValues must contain at most one item per CommodityQuantity.",
+                )
+
+            has_value[value.commodity_quantity] = True
+
+        return self

src/s2python/pebc/pebc_allowed_limit_range.py

@@ -1,3 +1,5 @@
+from typing_extensions import Self
+from pydantic import model_validator
 from s2python.generated.gen_s2 import (
     PEBCAllowedLimitRange as GenPEBCAllowedLimitRange,
     PEBCPowerEnvelopeLimitType as GenPEBCPowerEnvelopeLimitType,
@@ -24,3 +26,18 @@ class PEBCAllowedLimitRange(GenPEBCAllowedLimitRange, S2MessageComponent):
     abnormal_condition_only: bool = [
         GenPEBCAllowedLimitRange.model_fields["abnormal_condition_only"]  # type: ignore[assignment,reportIncompatibleVariableOverride]
     ]
+
+    @model_validator(mode="after")
+    def validate_range_boundary(self) -> Self:
+        # According to the specification "There shall be at least one PEBC.AllowedLimitRange for the UPPER_LIMIT
+        # and at least one AllowedLimitRange for the LOWER_LIMIT." However for something that produces energy
+        # end_of_range=-2000 and start_of_range=0 is valid. Therefore absolute value used here.
+        # TODO: Check that this is the correct interpretation of the wording
+        if abs(self.range_boundary.start_of_range) > abs(
+            self.range_boundary.end_of_range
+        ):
+            raise ValueError(
+                self,
+                f"The start of the range must shall be smaller or equal than the end of the range.",
+            )
+        return self

src/s2python/pebc/pebc_power_constraints.py

@@ -1,9 +1,14 @@
 import uuid
-from typing import List
+from typing import List, Dict, Tuple
+from typing_extensions import Self
 
+from pydantic import model_validator
+
+from s2python.common import CommodityQuantity, NumberRange
 from s2python.generated.gen_s2 import (
     PEBCPowerConstraints as GenPEBCPowerConstraints,
     PEBCPowerEnvelopeConsequenceType as GenPEBCPowerEnvelopeConsequenceType,
+    PEBCPowerEnvelopeLimitType,
 )
 from s2python.pebc.pebc_allowed_limit_range import PEBCAllowedLimitRange
 from s2python.validate_values_mixin import (
@@ -25,3 +30,48 @@ class PEBCPowerConstraints(GenPEBCPowerConstraints, S2MessageComponent):
     allowed_limit_ranges: List[PEBCAllowedLimitRange] = GenPEBCPowerConstraints.model_fields[  # type: ignore[reportIncompatibleVariableOverride]
         "allowed_limit_ranges"
     ]  # type: ignore[assignment]
+
+    @model_validator(mode="after")
+    def validate_has_one_upper_one_lower_limit_range(self) -> Self:
+
+        commodity_type_ranges: Dict[CommodityQuantity, Tuple[bool, bool]] = {}
+
+        for limit_range in self.allowed_limit_ranges:
+            current: Tuple[bool, bool] = commodity_type_ranges.get(
+                limit_range.commodity_quantity, (False, False)
+            )
+
+            if limit_range.limit_type == PEBCPowerEnvelopeLimitType.UPPER_LIMIT:
+                current = (
+                    True,
+                    current[1],
+                )
+
+            if limit_range.limit_type == PEBCPowerEnvelopeLimitType.LOWER_LIMIT:
+                current = (
+                    current[0],
+                    True,
+                )
+
+            commodity_type_ranges[limit_range.commodity_quantity] = current
+
+        valid = True
+
+        for upper, lower in commodity_type_ranges.values():
+            valid = valid and upper and lower
+
+        if not (valid):
+            raise ValueError(
+                self,
+                f"There shall be at least one PEBC.AllowedLimitRange for the UPPER_LIMIT and at least one AllowedLimitRange for the LOWER_LIMIT.",
+            )
+
+        return self
+
+    @model_validator(mode="after")
+    def validate_valid_until_after_valid_from(self) -> Self:
+        if self.valid_until is not None and self.valid_until < self.valid_from:
+            raise ValueError(
+                self, f"valid_until cannot be set to a value that is before valid_from."
+            )
+        return self

tests/unit/common/power_forecast_element_test.py

@@ -1,6 +1,10 @@
+import datetime
 import json
 from datetime import timedelta
 from unittest import TestCase
+import uuid
+
+from s2python.s2_validation_error import S2ValidationError
 
 from s2python.common import (
     PowerForecastElement,
@@ -59,3 +63,22 @@ def test__to_json__happy_path(self):
             ],
         }
         self.assertEqual(json.loads(json_str), expected_json)
+
+    def test__init__multiple_power_forecast_values_for_commodity_quantity(self):
+
+        # Arrange / Act / Assert
+        with self.assertRaises(S2ValidationError):
+            PowerForecastElement(
+                power_values=[
+                    PowerForecastValue(  # pyright: ignore[reportCallIssue]
+                        commodity_quantity=CommodityQuantity.NATURAL_GAS_FLOW_RATE,
+                        value_expected=500.2,
+                    ),
+
+                    PowerForecastValue(  # pyright: ignore[reportCallIssue]
+                        commodity_quantity=CommodityQuantity.NATURAL_GAS_FLOW_RATE,
+                        value_expected=500.2,
+                    )
+                ],
+                duration=Duration.from_timedelta(timedelta(seconds=4)),
+            )

tests/unit/common/power_measurement_test.py

@@ -3,6 +3,8 @@
 import uuid
 from unittest import TestCase
 
+from s2python.s2_validation_error import S2ValidationError
+
 from s2python.common import PowerMeasurement, PowerValue, CommodityQuantity
 
 
@@ -62,3 +64,34 @@ def test__to_json__happy_path(self):
             "measurement_timestamp": "2023-08-03T12:48:42+01:00",
         }
         self.assertEqual(json.loads(json_str), expected_json)
+
+    def test__init__no_power_measurement_values(self):
+
+        # Arrange / Act / Assert
+        with self.assertRaises(S2ValidationError):
+            PowerMeasurement(
+                values=[],
+                message_id=uuid.UUID("2bdec96b-be3b-4ba9-afa0-c4a0632cced8"),
+                measurement_timestamp=datetime(
+                    2023, 8, 3, 12, 48, 42, tzinfo=offset(timedelta(hours=1))
+                ),
+            )
+
+    def test__init__multiple_power_measurement_values_for_commodity_quantity(self):
+
+        # Arrange / Act / Assert
+        with self.assertRaises(S2ValidationError):
+            PowerMeasurement(
+                values=[
+                    PowerValue(
+                        commodity_quantity=CommodityQuantity.OIL_FLOW_RATE, value=42.42
+                    ),
+                    PowerValue(
+                        commodity_quantity=CommodityQuantity.OIL_FLOW_RATE, value=42.42
+                    ),
+                ],
+                message_id=uuid.UUID("2bdec96b-be3b-4ba9-afa0-c4a0632cced8"),
+                measurement_timestamp=datetime(
+                    2023, 8, 3, 12, 48, 42, tzinfo=offset(timedelta(hours=1))
+                ),
+            )

tests/unit/pebc/pebc_allowed_limit_range_test.py

@@ -0,0 +1,87 @@
+from datetime import timedelta, datetime, timezone as offset
+import json
+from unittest import TestCase
+import uuid
+
+from s2python.common import *
+from s2python.pebc import *
+from s2python.s2_validation_error import S2ValidationError
+
+
+class PEBCAllowedLimitRangeTest(TestCase):
+    def test__from_json__happy_path(self):
+        # Arrange
+        json_str = """
+{
+    "commodity_quantity": "ELECTRIC.POWER.L1",
+    "limit_type": "UPPER_LIMIT",
+    "range_boundary": {
+        "start_of_range": 0.0,
+        "end_of_range": 1000.0
+    },
+    "abnormal_condition_only": false
+}
+        """
+
+        # Act
+        allowed_limit_range = PEBCAllowedLimitRange.from_json(json_str)
+
+        # Assert
+        self.assertEqual(
+            allowed_limit_range.commodity_quantity,
+            CommodityQuantity.ELECTRIC_POWER_L1,
+        )
+        self.assertEqual(
+            allowed_limit_range.limit_type,
+            PEBCPowerEnvelopeLimitType.UPPER_LIMIT,
+        )
+        self.assertEqual(
+            allowed_limit_range.range_boundary.start_of_range, 0.0
+        )
+        self.assertEqual(
+            allowed_limit_range.range_boundary.end_of_range, 1000.0
+        )
+        self.assertEqual(
+            allowed_limit_range.abnormal_condition_only, False
+        )
+
+    def test__to_json__happy_path(self):
+        # Arrange
+        allowed_limit_range = PEBCAllowedLimitRange(
+            commodity_quantity=CommodityQuantity.ELECTRIC_POWER_L1,
+            limit_type=PEBCPowerEnvelopeLimitType.UPPER_LIMIT,
+            range_boundary=NumberRange(
+                start_of_range=0.0, end_of_range=1000.0
+            ),
+            abnormal_condition_only=False,
+        )
+
+        # Act
+        json_str = allowed_limit_range.to_json()
+
+        # Assert
+        expected_json = {
+            "commodity_quantity": "ELECTRIC.POWER.L1",
+            "limit_type": "UPPER_LIMIT",
+            "range_boundary": {"start_of_range": 0.0, "end_of_range": 1000.0},
+            "abnormal_condition_only": False,
+        }
+        self.assertEqual(json.loads(json_str), expected_json)
+
+    def test__from_json__invalid_range_boundary(self):
+        # Arrange
+        json_str = """
+{
+    "commodity_quantity": "ELECTRIC.POWER.L1",
+    "limit_type": "UPPER_LIMIT",
+    "range_boundary": {
+        "start_of_range": 1000.0,
+        "end_of_range": 0.0
+    },
+    "abnormal_condition_only": false
+}
+        """
+
+        # Act & Assert
+        with self.assertRaises(S2ValidationError) as context:
+            PEBCAllowedLimitRange.from_json(json_str)