diff --git a/src/s2python/common/power_forecast_element.py b/src/s2python/common/power_forecast_element.py index 7d84a73..38fab23 100644 --- a/src/s2python/common/power_forecast_element.py +++ b/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 diff --git a/src/s2python/common/power_measurement.py b/src/s2python/common/power_measurement.py index afd15cf..669ac08 100644 --- a/src/s2python/common/power_measurement.py +++ b/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 diff --git a/src/s2python/pebc/pebc_allowed_limit_range.py b/src/s2python/pebc/pebc_allowed_limit_range.py index 1df2c32..81e82b6 100644 --- a/src/s2python/pebc/pebc_allowed_limit_range.py +++ b/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 diff --git a/src/s2python/pebc/pebc_power_constraints.py b/src/s2python/pebc/pebc_power_constraints.py index 59c721d..71ab1bc 100644 --- a/src/s2python/pebc/pebc_power_constraints.py +++ b/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 diff --git a/tests/unit/common/power_forecast_element_test.py b/tests/unit/common/power_forecast_element_test.py index 4f68f45..ec2505d 100644 --- a/tests/unit/common/power_forecast_element_test.py +++ b/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)), + ) \ No newline at end of file diff --git a/tests/unit/common/power_measurement_test.py b/tests/unit/common/power_measurement_test.py index a8c555c..66106dd 100644 --- a/tests/unit/common/power_measurement_test.py +++ b/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)) + ), + ) diff --git a/tests/unit/pebc/pebc_allowed_limit_range_test.py b/tests/unit/pebc/pebc_allowed_limit_range_test.py new file mode 100644 index 0000000..c27964e --- /dev/null +++ b/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) diff --git a/tests/unit/pebc/pebc_power_constraints_test.py b/tests/unit/pebc/pebc_power_constraints_test.py new file mode 100644 index 0000000..20c739d --- /dev/null +++ b/tests/unit/pebc/pebc_power_constraints_test.py @@ -0,0 +1,248 @@ +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 PEBCPowerConstraintsTest(TestCase): + def test__from_json__happy_path(self): + # Arrange + json_str = """ +{ + "message_type": "PEBC.PowerConstraints", + "message_id": "5165cd7f-04bd-4c78-8fdd-b504cb0013a3", + "id": "7fe73aa9-1ce0-41e1-9a5f-ea7795687e5e", + "valid_from": "2025-05-12T12:00:00.000000Z", + "valid_until": "2025-05-12T13:00:00.000000Z", + "consequence_type": "VANISH", + "allowed_limit_ranges": [ + { + "commodity_quantity": "ELECTRIC.POWER.L1", + "limit_type": "UPPER_LIMIT", + "range_boundary": { + "start_of_range": 0.0, + "end_of_range": 0.0 + }, + "abnormal_condition_only": false + }, + { + "commodity_quantity": "ELECTRIC.POWER.L1", + "limit_type": "LOWER_LIMIT", + "range_boundary": { + "start_of_range": 0.0, + "end_of_range": -2000.0 + }, + "abnormal_condition_only": false + } + ] +} + """ + + # Act + pebc_power_constraints: PEBCPowerConstraints = PEBCPowerConstraints.from_json( + json_str + ) + + self.assertEqual( + pebc_power_constraints.id, + uuid.UUID("7fe73aa9-1ce0-41e1-9a5f-ea7795687e5e"), + ) + self.assertEqual( + pebc_power_constraints.message_id, + uuid.UUID("5165cd7f-04bd-4c78-8fdd-b504cb0013a3"), + ) + self.assertEqual(pebc_power_constraints.message_type, "PEBC.PowerConstraints") + self.assertEqual( + pebc_power_constraints.consequence_type, + PEBCPowerEnvelopeConsequenceType.VANISH, + ) + + self.assertEqual( + pebc_power_constraints.valid_from, + datetime( + year=2025, + month=5, + day=12, + hour=12, + minute=0, + second=0, + tzinfo=offset(offset=timedelta(seconds=0.0)), + ), + ) + + self.assertEqual( + pebc_power_constraints.valid_until, + datetime( + year=2025, + month=5, + day=12, + hour=13, + minute=0, + second=0, + tzinfo=offset(offset=timedelta(seconds=0.0)), + ), + ) + + self.assertEqual(len(pebc_power_constraints.allowed_limit_ranges), 2) + + def test__to_json__happy_path(self): + # Arrange + pebc_power_constraints = PEBCPowerConstraints( + message_type="PEBC.PowerConstraints", + message_id=uuid.UUID("5165cd7f-04bd-4c78-8fdd-b504cb0013a3"), + id=uuid.UUID("7fe73aa9-1ce0-41e1-9a5f-ea7795687e5e"), + valid_from=datetime( + year=2025, + month=5, + day=12, + hour=12, + minute=0, + second=0, + tzinfo=offset(offset=timedelta(seconds=0.0)), + ), + valid_until=datetime( + year=2025, + month=5, + day=12, + hour=13, + minute=0, + second=0, + tzinfo=offset(offset=timedelta(seconds=0.0)), + ), + consequence_type=PEBCPowerEnvelopeConsequenceType.VANISH, + allowed_limit_ranges=[ + PEBCAllowedLimitRange( + commodity_quantity=CommodityQuantity.ELECTRIC_POWER_L1, + limit_type=PEBCPowerEnvelopeLimitType.UPPER_LIMIT, + range_boundary=NumberRange(start_of_range=0.0, end_of_range=0.0), + abnormal_condition_only=False, + ), + PEBCAllowedLimitRange( + commodity_quantity=CommodityQuantity.ELECTRIC_POWER_L1, + limit_type=PEBCPowerEnvelopeLimitType.LOWER_LIMIT, + range_boundary=NumberRange( + start_of_range=0.0, end_of_range=-2000.0 + ), + abnormal_condition_only=False, + ), + ], + ) + + # Act + json_str = pebc_power_constraints.to_json() + + # Assert + expected_json = { + "message_type": "PEBC.PowerConstraints", + "message_id": "5165cd7f-04bd-4c78-8fdd-b504cb0013a3", + "id": "7fe73aa9-1ce0-41e1-9a5f-ea7795687e5e", + "valid_from": "2025-05-12T12:00:00Z", + "valid_until": "2025-05-12T13:00:00Z", + "consequence_type": "VANISH", + "allowed_limit_ranges": [ + { + "commodity_quantity": "ELECTRIC.POWER.L1", + "limit_type": "UPPER_LIMIT", + "range_boundary": {"start_of_range": 0.0, "end_of_range": 0.0}, + "abnormal_condition_only": False, + }, + { + "commodity_quantity": "ELECTRIC.POWER.L1", + "limit_type": "LOWER_LIMIT", + "range_boundary": {"start_of_range": 0.0, "end_of_range": -2000.0}, + "abnormal_condition_only": False, + }, + ], + } + self.assertEqual(json.loads(json_str), expected_json) + + def test__from_json__missing_upper_limit(self): + # Arrange + json_str = """ +{ + "message_type": "PEBC.PowerConstraints", + "message_id": "5165cd7f-04bd-4c78-8fdd-b504cb0013a3", + "id": "7fe73aa9-1ce0-41e1-9a5f-ea7795687e5e", + "valid_from": "2025-05-12T12:00:00.000000Z", + "valid_until": "2025-05-12T13:00:00.000000Z", + "consequence_type": "VANISH", + "allowed_limit_ranges": [ + { + "commodity_quantity": "ELECTRIC.POWER.L1", + "limit_type": "LOWER_LIMIT", + "range_boundary": { + "start_of_range": 0.0, + "end_of_range": -2000.0 + }, + "abnormal_condition_only": false + } + ] +} + """ + with self.assertRaises(S2ValidationError) as context: + PEBCPowerConstraints.from_json(json_str) + + def test__from_json__missing_lower_limit(self): + # Arrange + json_str = """ +{ + "message_type": "PEBC.PowerConstraints", + "message_id": "5165cd7f-04bd-4c78-8fdd-b504cb0013a3", + "id": "7fe73aa9-1ce0-41e1-9a5f-ea7795687e5e", + "valid_from": "2025-05-12T12:00:00.000000Z", + "valid_until": "2025-05-12T13:00:00.000000Z", + "consequence_type": "VANISH", + "allowed_limit_ranges": [ + { + "commodity_quantity": "ELECTRIC.POWER.L1", + "limit_type": "UPPER_LIMIT", + "range_boundary": { + "start_of_range": 0.0, + "end_of_range": 0.0 + }, + "abnormal_condition_only": false + } + ] +} + """ + with self.assertRaises(S2ValidationError) as context: + PEBCPowerConstraints.from_json(json_str) + + def test__from_json__valid_until_before_valid_from(self): + # Arrange + json_str = """ +{ + "message_type": "PEBC.PowerConstraints", + "message_id": "5165cd7f-04bd-4c78-8fdd-b504cb0013a3", + "id": "7fe73aa9-1ce0-41e1-9a5f-ea7795687e5e", + "valid_from": "2025-05-12T13:00:00.000000Z", + "valid_until": "2025-05-12T12:00:00.000000Z", + "consequence_type": "VANISH", + "allowed_limit_ranges": [ + { + "commodity_quantity": "ELECTRIC.POWER.L1", + "limit_type": "UPPER_LIMIT", + "range_boundary": { + "start_of_range": 0.0, + "end_of_range": 0.0 + }, + "abnormal_condition_only": false + }, + { + "commodity_quantity": "ELECTRIC.POWER.L1", + "limit_type": "LOWER_LIMIT", + "range_boundary": { + "start_of_range": 0.0, + "end_of_range": -2000.0 + }, + "abnormal_condition_only": false + } + ] +} + """ + with self.assertRaises(S2ValidationError) as context: + PEBCPowerConstraints.from_json(json_str)