diff --git a/.gitignore b/.gitignore index 14a680a..8a082ed 100644 --- a/.gitignore +++ b/.gitignore @@ -11,3 +11,4 @@ venv *venv* .tox/ dist/ +build/ diff --git a/src/s2python/ppbc/__init__.py b/src/s2python/ppbc/__init__.py new file mode 100644 index 0000000..1e0b4d3 --- /dev/null +++ b/src/s2python/ppbc/__init__.py @@ -0,0 +1,12 @@ +from s2python.ppbc.ppbc_schedule_instruction import PPBCScheduleInstruction +from s2python.ppbc.ppbc_end_interruption_instruction import ( + PPBCEndInterruptionInstruction, +) +from s2python.ppbc.ppbc_power_profile_definition import PPBCPowerProfileDefinition +from s2python.ppbc.ppbc_power_sequence_container import PPBCPowerSequenceContainer +from s2python.ppbc.ppbc_power_sequence import PPBCPowerSequence +from s2python.ppbc.ppbc_power_profile_status import PPBCPowerProfileStatus +from s2python.ppbc.ppbc_power_sequence_container_status import ( + PPBCPowerSequenceContainerStatus, +) +from s2python.ppbc.ppbc_power_sequence_element import PPBCPowerSequenceElement diff --git a/src/s2python/ppbc/ppbc_end_interruption_instruction.py b/src/s2python/ppbc/ppbc_end_interruption_instruction.py new file mode 100644 index 0000000..d53c527 --- /dev/null +++ b/src/s2python/ppbc/ppbc_end_interruption_instruction.py @@ -0,0 +1,32 @@ +import uuid + +from s2python.generated.gen_s2 import ( + PPBCEndInterruptionInstruction as GenPPBCEndInterruptionInstruction, +) + +from s2python.validate_values_mixin import ( + S2Message, + catch_and_convert_exceptions, +) + + +@catch_and_convert_exceptions +class PPBCEndInterruptionInstruction( + GenPPBCEndInterruptionInstruction, S2Message["PPBCEndInterruptionInstruction"] +): + model_config = GenPPBCEndInterruptionInstruction.model_config + model_config["validate_assignment"] = True + + id: uuid.UUID = GenPPBCEndInterruptionInstruction.model_fields["id"] # type: ignore[assignment] + power_profile_id: uuid.UUID = GenPPBCEndInterruptionInstruction.model_fields[ + "power_profile_id" + ] # type: ignore[assignment] + sequence_container_id: uuid.UUID = GenPPBCEndInterruptionInstruction.model_fields[ + "sequence_container_id" + ] # type: ignore[assignment] + power_sequence_id: uuid.UUID = GenPPBCEndInterruptionInstruction.model_fields[ + "power_sequence_id" + ] # type: ignore[assignment] + abnormal_condition: bool = GenPPBCEndInterruptionInstruction.model_fields[ + "abnormal_condition" + ] # type: ignore[assignment] diff --git a/src/s2python/ppbc/ppbc_power_profile_definition.py b/src/s2python/ppbc/ppbc_power_profile_definition.py new file mode 100644 index 0000000..cc4ba6a --- /dev/null +++ b/src/s2python/ppbc/ppbc_power_profile_definition.py @@ -0,0 +1,27 @@ +from typing import List +import uuid + +from s2python.generated.gen_s2 import ( + PPBCPowerProfileDefinition as GenPPBCPowerProfileDefinition, +) + +from s2python.validate_values_mixin import ( + S2Message, + catch_and_convert_exceptions, +) + +from s2python.ppbc.ppbc_power_sequence_container import PPBCPowerSequenceContainer + + +@catch_and_convert_exceptions +class PPBCPowerProfileDefinition( + GenPPBCPowerProfileDefinition, S2Message["PPBCPowerProfileDefinition"] +): + model_config = GenPPBCPowerProfileDefinition.model_config + model_config["validate_assignment"] = True + + message_id: uuid.UUID = GenPPBCPowerProfileDefinition.model_fields["message_id"] # type: ignore[assignment] + id: uuid.UUID = GenPPBCPowerProfileDefinition.model_fields["id"] # type: ignore[assignment] + power_sequences_containers: List[PPBCPowerSequenceContainer] = ( + GenPPBCPowerProfileDefinition.model_fields["power_sequences_containers"] # type: ignore[assignment] + ) diff --git a/src/s2python/ppbc/ppbc_power_profile_status.py b/src/s2python/ppbc/ppbc_power_profile_status.py new file mode 100644 index 0000000..d44f7e2 --- /dev/null +++ b/src/s2python/ppbc/ppbc_power_profile_status.py @@ -0,0 +1,26 @@ +from typing import List + +from s2python.generated.gen_s2 import ( + PPBCPowerProfileStatus as GenPPBCPowerProfileStatus, +) + +from s2python.validate_values_mixin import ( + S2Message, + catch_and_convert_exceptions, +) + +from s2python.ppbc.ppbc_power_sequence_container_status import ( + PPBCPowerSequenceContainerStatus, +) + + +@catch_and_convert_exceptions +class PPBCPowerProfileStatus( + GenPPBCPowerProfileStatus, S2Message["PPBCPowerProfileStatus"] +): + model_config = GenPPBCPowerProfileStatus.model_config + model_config["validate_assignment"] = True + + sequence_container_status: List[PPBCPowerSequenceContainerStatus] = ( + GenPPBCPowerProfileStatus.model_fields["sequence_container_status"] # type: ignore[assignment] + ) diff --git a/src/s2python/ppbc/ppbc_power_sequence.py b/src/s2python/ppbc/ppbc_power_sequence.py new file mode 100644 index 0000000..95e2758 --- /dev/null +++ b/src/s2python/ppbc/ppbc_power_sequence.py @@ -0,0 +1,30 @@ +from typing import List +import uuid + +from s2python.generated.gen_s2 import ( + PPBCPowerSequence as GenPPBCPowerSequence, +) + +from s2python.validate_values_mixin import ( + S2Message, + catch_and_convert_exceptions, +) + +from s2python.ppbc.ppbc_power_sequence_element import PPBCPowerSequenceElement +from s2python.common import Duration + + +@catch_and_convert_exceptions +class PPBCPowerSequence(GenPPBCPowerSequence, S2Message["PPBCPowerSequence"]): + model_config = GenPPBCPowerSequence.model_config + model_config["validate_assignment"] = True + + id: uuid.UUID = GenPPBCPowerSequence.model_fields["id"] # type: ignore[assignment] + elements: List[PPBCPowerSequenceElement] = GenPPBCPowerSequence.model_fields[ + "elements" + ] # type: ignore[assignment] + is_interruptible: bool = GenPPBCPowerSequence.model_fields["is_interruptible"] # type: ignore[assignment] + max_pause_before: Duration = GenPPBCPowerSequence.model_fields["max_pause_before"] # type: ignore[assignment] + abnormal_condition_only: bool = GenPPBCPowerSequence.model_fields[ + "abnormal_condition_only" + ] # type: ignore[assignment] diff --git a/src/s2python/ppbc/ppbc_power_sequence_container.py b/src/s2python/ppbc/ppbc_power_sequence_container.py new file mode 100644 index 0000000..3a11163 --- /dev/null +++ b/src/s2python/ppbc/ppbc_power_sequence_container.py @@ -0,0 +1,27 @@ +from typing import List +import uuid + + +from s2python.generated.gen_s2 import ( + PPBCPowerSequenceContainer as GenPPBCPowerSequenceContainer, +) + +from s2python.validate_values_mixin import ( + S2Message, + catch_and_convert_exceptions, +) + +from s2python.ppbc.ppbc_power_sequence import PPBCPowerSequence + + +@catch_and_convert_exceptions +class PPBCPowerSequenceContainer( + GenPPBCPowerSequenceContainer, S2Message["PPBCPowerSequenceContainer"] +): + model_config = GenPPBCPowerSequenceContainer.model_config + model_config["validate_assignment"] = True + + id: uuid.UUID = GenPPBCPowerSequenceContainer.model_fields["id"] # type: ignore[assignment] + power_sequences: List[PPBCPowerSequence] = ( + GenPPBCPowerSequenceContainer.model_fields["power_sequences"] # type: ignore[assignment] + ) diff --git a/src/s2python/ppbc/ppbc_power_sequence_container_status.py b/src/s2python/ppbc/ppbc_power_sequence_container_status.py new file mode 100644 index 0000000..624e4d6 --- /dev/null +++ b/src/s2python/ppbc/ppbc_power_sequence_container_status.py @@ -0,0 +1,32 @@ +import uuid +from typing import Union + +from s2python.generated.gen_s2 import ( + PPBCPowerSequenceContainerStatus as GenPPBCPowerSequenceContainerStatus, +) + +from s2python.validate_values_mixin import ( + S2Message, + catch_and_convert_exceptions, +) + + +@catch_and_convert_exceptions +class PPBCPowerSequenceContainerStatus( + GenPPBCPowerSequenceContainerStatus, S2Message["PPBCPowerSequenceContainerStatus"] +): + model_config = GenPPBCPowerSequenceContainerStatus.model_config + model_config["validate_assignment"] = True + + power_profile_id: uuid.UUID = GenPPBCPowerSequenceContainerStatus.model_fields[ + "power_profile_id" # type: ignore[assignment] + ] + sequence_container_id: uuid.UUID = GenPPBCPowerSequenceContainerStatus.model_fields[ + "sequence_container_id" # type: ignore[assignment] + ] + selected_sequence_id: Union[uuid.UUID, None] = ( + GenPPBCPowerSequenceContainerStatus.model_fields["selected_sequence_id"] # type: ignore[assignment] + ) + progress: Union[uuid.UUID, None] = GenPPBCPowerSequenceContainerStatus.model_fields[ + "progress" # type: ignore[assignment] + ] diff --git a/src/s2python/ppbc/ppbc_power_sequence_element.py b/src/s2python/ppbc/ppbc_power_sequence_element.py new file mode 100644 index 0000000..1372063 --- /dev/null +++ b/src/s2python/ppbc/ppbc_power_sequence_element.py @@ -0,0 +1,25 @@ +from typing import List + +from s2python.generated.gen_s2 import ( + PPBCPowerSequenceElement as GenPPBCPowerSequenceElement, +) + +from s2python.validate_values_mixin import ( + S2Message, + catch_and_convert_exceptions, +) + +from s2python.common import Duration, PowerForecastValue + + +@catch_and_convert_exceptions +class PPBCPowerSequenceElement( + GenPPBCPowerSequenceElement, S2Message["PPBCPowerSequenceElement"] +): + model_config = GenPPBCPowerSequenceElement.model_config + model_config["validate_assignment"] = True + + duration: Duration = GenPPBCPowerSequenceElement.model_fields["duration"] # type: ignore[assignment] + power_values: List[PowerForecastValue] = GenPPBCPowerSequenceElement.model_fields[ + "power_values" + ] # type: ignore[assignment] diff --git a/src/s2python/ppbc/ppbc_schedule_instruction.py b/src/s2python/ppbc/ppbc_schedule_instruction.py new file mode 100644 index 0000000..c794b78 --- /dev/null +++ b/src/s2python/ppbc/ppbc_schedule_instruction.py @@ -0,0 +1,33 @@ +import uuid + +from s2python.generated.gen_s2 import ( + PPBCScheduleInstruction as GenPPBCScheduleInstruction, +) +from s2python.validate_values_mixin import ( + catch_and_convert_exceptions, + S2Message, +) + + +@catch_and_convert_exceptions +class PPBCScheduleInstruction( + GenPPBCScheduleInstruction, S2Message["PPBCScheduleInstruction"] +): + model_config = GenPPBCScheduleInstruction.model_config + model_config["validate_assignment"] = True + + id: uuid.UUID = GenPPBCScheduleInstruction.model_fields["id"] # type: ignore[assignment] + + power_profile_id: uuid.UUID = GenPPBCScheduleInstruction.model_fields[ + "power_profile_id" + ] # type: ignore[assignment] + + message_id: uuid.UUID = GenPPBCScheduleInstruction.model_fields["message_id"] # type: ignore[assignment] + + sequence_container_id: uuid.UUID = GenPPBCScheduleInstruction.model_fields[ + "sequence_container_id" + ] # type: ignore[assignment] + + power_sequence_id: uuid.UUID = GenPPBCScheduleInstruction.model_fields[ + "power_sequence_id" + ] # type: ignore[assignment] diff --git a/src/s2python/ppbc/ppbc_start_interruption_instruction.py b/src/s2python/ppbc/ppbc_start_interruption_instruction.py new file mode 100644 index 0000000..f6d25ff --- /dev/null +++ b/src/s2python/ppbc/ppbc_start_interruption_instruction.py @@ -0,0 +1,32 @@ +import uuid + +from s2python.generated.gen_s2 import ( + PPBCStartInterruptionInstruction as GenPPBCStartInterruptionInstruction, +) + +from s2python.validate_values_mixin import ( + S2Message, + catch_and_convert_exceptions, +) + + +@catch_and_convert_exceptions +class PPBCStartInterruptionInstruction( + GenPPBCStartInterruptionInstruction, S2Message["PPBCStartInterruptionInstruction"] +): + model_config = GenPPBCStartInterruptionInstruction.model_config + model_config["validate_assignment"] = True + + id: uuid.UUID = GenPPBCStartInterruptionInstruction.model_fields["id"] # type: ignore[assignment] + power_profile_id: uuid.UUID = GenPPBCStartInterruptionInstruction.model_fields[ + "power_profile_id" + ] # type: ignore[assignment] + sequence_container_id: uuid.UUID = GenPPBCStartInterruptionInstruction.model_fields[ + "sequence_container_id" + ] # type: ignore[assignment] + power_sequence_id: uuid.UUID = GenPPBCStartInterruptionInstruction.model_fields[ + "power_sequence_id" + ] # type: ignore[assignment] + abnormal_condition: bool = GenPPBCStartInterruptionInstruction.model_fields[ + "abnormal_condition" + ] # type: ignore[assignment] diff --git a/src/s2python/s2_control_type.py b/src/s2python/s2_control_type.py index f9a4545..43f3b8e 100644 --- a/src/s2python/s2_control_type.py +++ b/src/s2python/s2_control_type.py @@ -3,6 +3,7 @@ from s2python.common import ControlType as ProtocolControlType from s2python.frbc import FRBCInstruction +from s2python.ppbc import PPBCScheduleInstruction from s2python.validate_values_mixin import S2Message if typing.TYPE_CHECKING: @@ -36,10 +37,33 @@ def handle_instruction( ) -> None: ... @abc.abstractmethod - def activate(self, conn: "S2Connection") -> None: ... + def activate(self, conn: "S2Connection") -> None: + """Overwrite with the actual dctivation logic of your Resource Manager for this particular control type.""" @abc.abstractmethod - def deactivate(self, conn: "S2Connection") -> None: ... + def deactivate(self, conn: "S2Connection") -> None: + """Overwrite with the actual deactivation logic of your Resource Manager for this particular control type.""" + + +class PPBCControlType(S2ControlType): + def get_protocol_control_type(self) -> ProtocolControlType: + return ProtocolControlType.POWER_PROFILE_BASED_CONTROL + + def register_handlers(self, handlers: "MessageHandlers") -> None: + handlers.register_handler(PPBCScheduleInstruction, self.handle_instruction) + + @abc.abstractmethod + def handle_instruction( + self, conn: "S2Connection", msg: S2Message, send_okay: typing.Callable[[], None] + ) -> None: ... + + @abc.abstractmethod + def activate(self, conn: "S2Connection") -> None: + """Overwrite with the actual dctivation logic of your Resource Manager for this particular control type.""" + + @abc.abstractmethod + def deactivate(self, conn: "S2Connection") -> None: + """Overwrite with the actual deactivation logic of your Resource Manager for this particular control type.""" class NoControlControlType(S2ControlType): diff --git a/src/s2python/s2_parser.py b/src/s2python/s2_parser.py index 906a286..e1a5c43 100644 --- a/src/s2python/s2_parser.py +++ b/src/s2python/s2_parser.py @@ -24,6 +24,8 @@ FRBCTimerStatus, FRBCUsageForecast, ) +from s2python.ppbc import PPBCScheduleInstruction + from s2python.validate_values_mixin import S2Message from s2python.s2_validation_error import S2ValidationError @@ -44,6 +46,7 @@ "FRBC.SystemDescription": FRBCSystemDescription, "FRBC.TimerStatus": FRBCTimerStatus, "FRBC.UsageForecast": FRBCUsageForecast, + "PPBC.ScheduleInstruction": PPBCScheduleInstruction, "Handshake": Handshake, "HandshakeResponse": HandshakeResponse, "InstructionStatusUpdate": InstructionStatusUpdate, @@ -86,7 +89,9 @@ def parse_as_any_message(unparsed_message: Union[dict, str, bytes]) -> S2Message return TYPE_TO_MESSAGE_CLASS[message_type].model_validate(message_json) @staticmethod - def parse_as_message(unparsed_message: Union[dict, str, bytes], as_message: Type[M]) -> M: + def parse_as_message( + unparsed_message: Union[dict, str, bytes], as_message: Type[M] + ) -> M: """Parse the message to a specific S2 python message. :param unparsed_message: The message as a JSON-formatted string or as a JSON-parsed dictionary. @@ -98,7 +103,9 @@ def parse_as_message(unparsed_message: Union[dict, str, bytes], as_message: Type return as_message.from_dict(message_json) @staticmethod - def parse_message_type(unparsed_message: Union[dict, str, bytes]) -> Optional[S2MessageType]: + def parse_message_type( + unparsed_message: Union[dict, str, bytes], + ) -> Optional[S2MessageType]: """Parse only the message type from the unparsed message. This is useful to call before `parse_as_message` to retrieve the message type and allows for strictly-typed