DDBC skeleton class implementation

src/s2python/ddbc/__init__.py

@@ -0,0 +1,12 @@
+from s2python.ddbc.ddbc_actuator_description import DDBCActuatorDescription
+from s2python.ddbc.ddbc_operation_mode import DDBCOperationMode
+from s2python.ddbc.ddbc_instruction import DDBCInstruction
+from s2python.ddbc.ddbc_actuator_status import DDBCActuatorStatus
+from s2python.ddbc.ddbc_average_demand_rate_forecast_element import (
+    DDBCAverageDemandRateForecastElement,
+)
+from s2python.ddbc.ddbc_average_demand_rate_forecast import (
+    DDBCAverageDemandRateForecast,
+)
+from s2python.ddbc.ddbc_system_description import DDBCSystemDescription
+from s2python.ddbc.ddbc_timer_status import DDBCTimerStatus

src/s2python/ddbc/ddbc_actuator_description.py

@@ -0,0 +1,30 @@
+from typing import List
+import uuid
+
+from s2python.generated.gen_s2 import (
+    DDBCActuatorDescription as GenDDBCActuatorDescription,
+)
+from s2python.generated.gen_s2 import Commodity
+from s2python.ddbc.ddbc_operation_mode import DDBCOperationMode
+
+from s2python.common.timer import Timer
+
+from s2python.validate_values_mixin import (
+    catch_and_convert_exceptions,
+    S2MessageComponent,
+)
+
+
+@catch_and_convert_exceptions
+class DDBCActuatorDescription(GenDDBCActuatorDescription, S2MessageComponent["DDBCActuatorDescription"]):
+    model_config = GenDDBCActuatorDescription.model_config
+    model_config["validate_assignment"] = True
+
+    id: uuid.UUID = GenDDBCActuatorDescription.model_fields["id"]  # type: ignore[assignment]
+    supported_commodites: List[Commodity] = GenDDBCActuatorDescription.model_fields[
+        "supported_commodites"
+    ]  # type: ignore[assignment]
+    timers: List[Timer] = GenDDBCActuatorDescription.model_fields["timers"]  # type: ignore[assignment]
+    operation_modes: List[DDBCOperationMode] = GenDDBCActuatorDescription.model_fields[
+        "operation_modes"
+    ]  # type: ignore[assignment]

src/s2python/ddbc/ddbc_actuator_status.py

@@ -0,0 +1,22 @@
+import uuid
+
+from s2python.generated.gen_s2 import DDBCActuatorStatus as GenDDBCActuatorStatus
+from s2python.validate_values_mixin import (
+    catch_and_convert_exceptions,
+    S2MessageComponent,
+)
+
+
+@catch_and_convert_exceptions
+class DDBCActuatorStatus(GenDDBCActuatorStatus, S2MessageComponent["DDBCActuatorStatus"]):
+    model_config = GenDDBCActuatorStatus.model_config
+    model_config["validate_assignment"] = True
+
+    message_id: uuid.UUID = GenDDBCActuatorStatus.model_fields["message_id"]  # type: ignore[assignment]
+    actuator_id: uuid.UUID = GenDDBCActuatorStatus.model_fields["actuator_id"]  # type: ignore[assignment]
+    active_operation_mode_id: uuid.UUID = GenDDBCActuatorStatus.model_fields[
+        "active_operation_mode_id"
+    ]  # type: ignore[assignment]
+    operation_mode_factor: float = GenDDBCActuatorStatus.model_fields[
+        "operation_mode_factor"
+    ]  # type: ignore[assignment]

src/s2python/ddbc/ddbc_average_demand_rate_forecast.py

@@ -0,0 +1,30 @@
+from typing import List
+import uuid
+
+from s2python.generated.gen_s2 import (
+    DDBCAverageDemandRateForecast as GenDDBCAverageDemandRateForecast,
+)
+from s2python.ddbc.ddbc_average_demand_rate_forecast_element import (
+    DDBCAverageDemandRateForecastElement,
+)
+
+from s2python.validate_values_mixin import (
+    catch_and_convert_exceptions,
+    S2MessageComponent,
+)
+
+
+@catch_and_convert_exceptions
+class DDBCAverageDemandRateForecast(
+    GenDDBCAverageDemandRateForecast,
+    S2MessageComponent["DDBCAverageDemandRateForecast"],
+):
+    model_config = GenDDBCAverageDemandRateForecast.model_config
+    model_config["validate_assignment"] = True
+
+    message_id: uuid.UUID = GenDDBCAverageDemandRateForecast.model_fields["message_id"]  # type: ignore[assignment]
+    elements: List[
+        DDBCAverageDemandRateForecastElement
+    ] = GenDDBCAverageDemandRateForecast.model_fields[
+        "elements"
+    ]  # type: ignore[assignment]

src/s2python/ddbc/ddbc_average_demand_rate_forecast_element.py

@@ -0,0 +1,21 @@
+from s2python.generated.gen_s2 import Duration
+
+from s2python.generated.gen_s2 import (
+    DDBCAverageDemandRateForecastElement as GenDDBCAverageDemandRateForecastElement,
+)
+
+from s2python.validate_values_mixin import catch_and_convert_exceptions, S2MessageComponent
+
+
+@catch_and_convert_exceptions
+class DDBCAverageDemandRateForecastElement(
+    GenDDBCAverageDemandRateForecastElement,
+    S2MessageComponent["DDBCAverageDemandRateForecastElement"],
+):
+    model_config = GenDDBCAverageDemandRateForecastElement.model_config
+    model_config["validate_assignment"] = True
+
+    duration: Duration = GenDDBCAverageDemandRateForecastElement.model_fields["duration"]  # type: ignore[assignment]
+    demand_rate_expected: float = GenDDBCAverageDemandRateForecastElement.model_fields[
+        "demand_rate_expected"
+    ]  # type: ignore[assignment]

src/s2python/ddbc/ddbc_instruction.py

@@ -0,0 +1,19 @@
+import uuid
+
+from s2python.generated.gen_s2 import DDBCInstruction as GenDDBCInstruction
+from s2python.validate_values_mixin import (
+    catch_and_convert_exceptions,
+    S2MessageComponent,
+)
+
+
+@catch_and_convert_exceptions
+class DDBCInstruction(GenDDBCInstruction, S2MessageComponent["DDBCInstruction"]):
+    model_config = GenDDBCInstruction.model_config
+    model_config["validate_assignment"] = True
+
+    message_id: uuid.UUID = GenDDBCInstruction.model_fields["message_id"]  # type: ignore[assignment]
+    actuator_id: uuid.UUID = GenDDBCInstruction.model_fields["actuator_id"]  # type: ignore[assignment]
+    operation_mode_id: uuid.UUID = GenDDBCInstruction.model_fields["operation_mode_id"]  # type: ignore[assignment]
+    operation_mode_factor: float = GenDDBCInstruction.model_fields["operation_mode_factor"]  # type: ignore[assignment]
+    abnormal_condition: bool = GenDDBCInstruction.model_fields["abnormal_condition"]  # type: ignore[assignment]

src/s2python/ddbc/ddbc_operation_mode.py

@@ -0,0 +1,26 @@
+from typing import List
+import uuid
+
+from s2python.generated.gen_s2 import DDBCOperationMode as GenDDBCOperationMode
+
+from s2python.common.power_range import PowerRange
+from s2python.common.number_range import NumberRange
+
+from s2python.validate_values_mixin import (
+    catch_and_convert_exceptions,
+    S2MessageComponent,
+)
+
+
+@catch_and_convert_exceptions
+class DDBCOperationMode(GenDDBCOperationMode, S2MessageComponent["DDBCOperationMode"]):
+    model_config = GenDDBCOperationMode.model_config
+    model_config["validate_assignment"] = True
+
+    # ? Id vs id
+    id: uuid.UUID = GenDDBCOperationMode.model_fields["Id"]  # type: ignore[assignment]
+    power_ranges: List[PowerRange] = GenDDBCOperationMode.model_fields["power_ranges"]  # type: ignore[assignment]
+    supply_ranges: List[NumberRange] = GenDDBCOperationMode.model_fields["supply_ranges"]  # type: ignore[assignment]
+    abnormal_condition_only: bool = GenDDBCOperationMode.model_fields[
+        "abnormal_condition_only"
+    ]  # type: ignore[assignment]

src/s2python/ddbc/ddbc_system_description.py

@@ -0,0 +1,29 @@
+from typing import List
+import uuid
+
+from s2python.generated.gen_s2 import (
+    DDBCSystemDescription as GenDDBCSystemDescription,
+)
+from s2python.common.number_range import NumberRange
+from s2python.ddbc.ddbc_actuator_description import DDBCActuatorDescription
+from s2python.validate_values_mixin import (
+    catch_and_convert_exceptions,
+    S2MessageComponent,
+)
+
+
+@catch_and_convert_exceptions
+class DDBCSystemDescription(GenDDBCSystemDescription, S2MessageComponent["DDBCSystemDescription"]):
+    model_config = GenDDBCSystemDescription.model_config
+    model_config["validate_assignment"] = True
+
+    message_id: uuid.UUID = GenDDBCSystemDescription.model_fields["message_id"]  # type: ignore[assignment]
+    actuators: List[DDBCActuatorDescription] = GenDDBCSystemDescription.model_fields[
+        "actuators"
+    ]  # type: ignore[assignment]
+    present_demand_rate: NumberRange = GenDDBCSystemDescription.model_fields[
+        "present_demand_rate"
+    ]  # type: ignore[assignment]
+    provides_average_demand_rate_forecast: bool = GenDDBCSystemDescription.model_fields[
+        "provides_average_demand_rate_forecast"
+    ]  # type: ignore[assignment]

src/s2python/ddbc/ddbc_timer_status.py

@@ -0,0 +1,18 @@
+import uuid
+
+from s2python.generated.gen_s2 import DDBCTimerStatus as GenDDBCTimerStatus
+
+from s2python.validate_values_mixin import (
+    catch_and_convert_exceptions,
+    S2MessageComponent,
+)
+
+
+@catch_and_convert_exceptions
+class DDBCTimerStatus(GenDDBCTimerStatus, S2MessageComponent["DDBCTimerStatus"]):
+    model_config = GenDDBCTimerStatus.model_config
+    model_config["validate_assignment"] = True
+
+    message_id: uuid.UUID = GenDDBCTimerStatus.model_fields["message_id"]  # type: ignore[assignment]
+    timer_id: uuid.UUID = GenDDBCTimerStatus.model_fields["timer_id"]  # type: ignore[assignment]
+    actuator_id: uuid.UUID = GenDDBCTimerStatus.model_fields["actuator_id"]  # type: ignore[assignment]

src/s2python/s2_control_type.py

@@ -66,6 +66,20 @@ def deactivate(self, conn: "S2Connection") -> None:
         """Overwrite with the actual deactivation logic of your Resource Manager for this particular control type."""
 
 
+class DDBControlType(S2ControlType):
+    def get_protocol_control_type(self) -> ProtocolControlType:
+        return ProtocolControlType.DEMAND_DRIVEN_BASED_CONTROL
+
+    def register_handlers(self, handlers: "MessageHandlers") -> None:
+        pass
+
+    @abc.abstractmethod
+    def activate(self, conn: "S2Connection") -> None: ...
+
+    @abc.abstractmethod
+    def deactivate(self, conn: "S2Connection") -> None: ...
+
+
 class NoControlControlType(S2ControlType):
     def get_protocol_control_type(self) -> ProtocolControlType:
         return ProtocolControlType.NOT_CONTROLABLE