chore: introduce evaluation input for compressor sampled

frodehk · frodehk · commit b37b6817e6e4 · 2025-11-19T14:39:54.000+01:00
diff --git a/src/libecalc/domain/process/evaluation_input.py b/src/libecalc/domain/process/evaluation_input.py
@@ -29,10 +29,10 @@ class CompressorEvaluationInput:
     def __init__(
         self,
         rate_expression: TimeSeriesFlowRate | list[TimeSeriesFlowRate],
-        fluid_factory: FluidFactoryInterface | list[FluidFactoryInterface] | None,
+        fluid_factory: FluidFactoryInterface | list[FluidFactoryInterface],
+        suction_pressure_expression: TimeSeriesPressure,
+        discharge_pressure_expression: TimeSeriesPressure,
         power_loss_factor: TimeSeriesPowerLossFactor | None = None,
-        suction_pressure_expression: TimeSeriesPressure | None = None,
-        discharge_pressure_expression: TimeSeriesPressure | None = None,
         intermediate_pressure_expression: TimeSeriesPressure | None = None,
     ):
         self._rate_expression = rate_expression
@@ -43,15 +43,15 @@ def __init__(
         self._intermediate_pressure_expression = intermediate_pressure_expression
 
     @property
-    def suction_pressure_expression(self) -> TimeSeriesPressure | None:
+    def suction_pressure_expression(self) -> TimeSeriesPressure:
         return self._suction_pressure_expression
 
     @property
-    def discharge_pressure_expression(self) -> TimeSeriesPressure | None:
+    def discharge_pressure_expression(self) -> TimeSeriesPressure:
         return self._discharge_pressure_expression
 
     @property
-    def power_loss_factor(self) -> TimeSeriesPowerLossFactor:
+    def power_loss_factor(self) -> TimeSeriesPowerLossFactor | None:
         return self._power_loss_factor
 
     @property
@@ -74,6 +74,68 @@ def apply_to_model(self, compressor_model: CompressorModel):
             if self._intermediate_pressure_expression is not None
             else None
         )
+        suction_pressure = np.asarray(self._suction_pressure_expression.get_values(), dtype=np.float64)
+        discharge_pressure = np.asarray(self._discharge_pressure_expression.get_values(), dtype=np.float64)
+
+        compressor_model.set_evaluation_input(
+            rate=stream_day_rate,
+            fluid_factory=self._fluid_factory,
+            suction_pressure=suction_pressure,
+            discharge_pressure=discharge_pressure,
+            intermediate_pressure=intermediate_pressure,
+        )
+
+
+class CompressorSampledEvaluationInput:
+    """
+    Encapsulates all input data required to configure and apply evaluation parameters to a sampled compressor model.
+
+    Attributes:
+        rate_expression (TimeSeriesFlowRate | list[TimeSeriesFlowRate]): the flow rate expression(s).
+        power_loss_factor (TimeSeriesPowerLossFactor | None): expression for power loss factor.
+        suction_pressure_expression (TimeSeriesPressure | None): expression for the inlet (suction) pressure.
+        discharge_pressure_expression (TimeSeriesPressure | None): expression for the outlet (discharge) pressure.
+    """
+
+    def __init__(
+        self,
+        rate_expression: TimeSeriesFlowRate | list[TimeSeriesFlowRate],
+        power_loss_factor: TimeSeriesPowerLossFactor | None = None,
+        suction_pressure_expression: TimeSeriesPressure | None = None,
+        discharge_pressure_expression: TimeSeriesPressure | None = None,
+    ):
+        self._rate_expression = rate_expression
+        self._power_loss_factor = power_loss_factor
+        self._suction_pressure_expression = suction_pressure_expression
+        self._discharge_pressure_expression = discharge_pressure_expression
+
+    @property
+    def suction_pressure_expression(self) -> TimeSeriesPressure | None:
+        return self._suction_pressure_expression
+
+    @property
+    def discharge_pressure_expression(self) -> TimeSeriesPressure | None:
+        return self._discharge_pressure_expression
+
+    @property
+    def power_loss_factor(self) -> TimeSeriesPowerLossFactor | None:
+        return self._power_loss_factor
+
+    @property
+    def periods(self) -> Periods:
+        if isinstance(self._rate_expression, list):
+            return self._rate_expression[0].get_periods()
+        return self._rate_expression.get_periods()
+
+    def apply_to_model(self, compressor_model: CompressorModel):
+        rate_expr = self._rate_expression if isinstance(self._rate_expression, list) else [self._rate_expression]
+
+        if not isinstance(compressor_model, CompressorTrainCommonShaftMultipleStreamsAndPressures):
+            assert len(rate_expr) == 1
+            stream_day_rate = np.asarray(rate_expr[0].get_stream_day_values(), dtype=np.float64)
+        else:
+            stream_day_rate = np.array([rate.get_stream_day_values() for rate in rate_expr], dtype=np.float64)
+
         suction_pressure = (
             np.asarray(self._suction_pressure_expression.get_values(), dtype=np.float64)
             if self._suction_pressure_expression is not None
@@ -87,10 +149,10 @@ def apply_to_model(self, compressor_model: CompressorModel):
 
         compressor_model.set_evaluation_input(
             rate=stream_day_rate,
-            fluid_factory=self._fluid_factory,
+            fluid_factory=None,
             suction_pressure=suction_pressure,
             discharge_pressure=discharge_pressure,
-            intermediate_pressure=intermediate_pressure,
+            intermediate_pressure=None,
         )
 
 
@@ -110,7 +172,7 @@ def __init__(
         self._power_loss_factor = power_loss_factor
 
     @property
-    def power_loss_factor(self) -> TimeSeriesPowerLossFactor:
+    def power_loss_factor(self) -> TimeSeriesPowerLossFactor | None:
         return self._power_loss_factor
 
     @property
diff --git a/src/libecalc/domain/process/process_service.py b/src/libecalc/domain/process/process_service.py
@@ -5,7 +5,11 @@
 from libecalc.domain.ecalc_component import EcalcComponent
 from libecalc.domain.process.compressor.core.base import CompressorModel, CompressorWithTurbineModel
 from libecalc.domain.process.compressor.core.sampled import CompressorModelSampled
-from libecalc.domain.process.evaluation_input import CompressorEvaluationInput, PumpEvaluationInput
+from libecalc.domain.process.evaluation_input import (
+    CompressorEvaluationInput,
+    CompressorSampledEvaluationInput,
+    PumpEvaluationInput,
+)
 from libecalc.domain.process.pump.pump import PumpModel
 from libecalc.presentation.yaml.domain.ecalc_components import ProcessSystemComponent, SimplifiedProcessUnitComponent
 
@@ -24,7 +28,7 @@ def register_simplified_process_unit(
         self,
         ecalc_component: EcalcComponent,
         simplified_process_unit: CompressorModelSampled | PumpModel | CompressorWithTurbineModel,
-        evaluation_input: CompressorEvaluationInput | PumpEvaluationInput,
+        evaluation_input: CompressorSampledEvaluationInput | PumpEvaluationInput,
     ): ...
 
     @abstractmethod
@@ -36,4 +40,6 @@ def map_model_to_consumer(
     ): ...
 
     @abstractmethod
-    def get_evaluation_input(self, process_system_id: UUID) -> CompressorEvaluationInput | PumpEvaluationInput: ...
+    def get_evaluation_input(
+        self, process_system_id: UUID
+    ) -> CompressorEvaluationInput | CompressorSampledEvaluationInput | PumpEvaluationInput: ...
diff --git a/src/libecalc/presentation/yaml/domain/default_process_service.py b/src/libecalc/presentation/yaml/domain/default_process_service.py
@@ -5,7 +5,11 @@
 from libecalc.domain.process.compressor.core.base import CompressorModel, CompressorWithTurbineModel
 from libecalc.domain.process.compressor.core.sampled import CompressorModelSampled
 from libecalc.domain.process.compressor.core.train.base import CompressorTrainModel
-from libecalc.domain.process.evaluation_input import CompressorEvaluationInput, PumpEvaluationInput
+from libecalc.domain.process.evaluation_input import (
+    CompressorEvaluationInput,
+    CompressorSampledEvaluationInput,
+    PumpEvaluationInput,
+)
 from libecalc.domain.process.process_service import ProcessService
 from libecalc.domain.process.pump.pump import PumpModel
 from libecalc.presentation.yaml.domain.ecalc_components import ProcessSystemComponent, SimplifiedProcessUnitComponent
@@ -15,12 +19,16 @@ class DefaultProcessService(ProcessService):
     def __init__(self):
         self._process_systems: dict[UUID, CompressorTrainModel] = {}
         self._simplified_process_units: dict[UUID, CompressorModelSampled | PumpModel | CompressorWithTurbineModel] = {}
-        self._evaluation_inputs: dict[UUID, CompressorEvaluationInput | PumpEvaluationInput] = {}
+        self._evaluation_inputs: dict[
+            UUID, CompressorEvaluationInput | CompressorSampledEvaluationInput | PumpEvaluationInput
+        ] = {}
         self._consumer_to_model_map: dict[tuple[UUID, Period], UUID] = {}
         self._ecalc_components: dict[UUID, EcalcComponent] = {}
 
     @property
-    def evaluation_inputs(self) -> dict[UUID, CompressorEvaluationInput | PumpEvaluationInput]:
+    def evaluation_inputs(
+        self,
+    ) -> dict[UUID, CompressorEvaluationInput | CompressorSampledEvaluationInput | PumpEvaluationInput]:
         return self._evaluation_inputs
 
     @property
@@ -53,7 +61,7 @@ def register_simplified_process_unit(
         self,
         ecalc_component: EcalcComponent,
         simplified_process_unit: CompressorModelSampled | PumpModel | CompressorWithTurbineModel,
-        evaluation_input: CompressorEvaluationInput | PumpEvaluationInput,
+        evaluation_input: CompressorEvaluationInput | CompressorSampledEvaluationInput | PumpEvaluationInput,
     ):
         self._ecalc_components[ecalc_component.id] = ecalc_component
         self.simplified_process_units[ecalc_component.id] = simplified_process_unit
@@ -67,5 +75,7 @@ def map_model_to_consumer(
     ):
         self.consumer_to_model_map[(consumer_id, period)] = ecalc_component.id
 
-    def get_evaluation_input(self, model_id: UUID) -> CompressorEvaluationInput | PumpEvaluationInput:
+    def get_evaluation_input(
+        self, model_id: UUID
+    ) -> CompressorEvaluationInput | CompressorSampledEvaluationInput | PumpEvaluationInput:
         return self.evaluation_inputs.get(model_id)
diff --git a/src/libecalc/presentation/yaml/mappers/consumer_function_mapper.py b/src/libecalc/presentation/yaml/mappers/consumer_function_mapper.py
@@ -48,7 +48,11 @@
 from libecalc.domain.process.compressor.core.train.types import FluidStreamObjectForMultipleStreams
 from libecalc.domain.process.compressor.dto import InterstagePressureControl
 from libecalc.domain.process.entities.shaft import SingleSpeedShaft, VariableSpeedShaft
-from libecalc.domain.process.evaluation_input import CompressorEvaluationInput, PumpEvaluationInput
+from libecalc.domain.process.evaluation_input import (
+    CompressorEvaluationInput,
+    CompressorSampledEvaluationInput,
+    PumpEvaluationInput,
+)
 from libecalc.domain.process.pump.pump import PumpModel
 from libecalc.domain.process.value_objects.chart.chart import ChartData
 from libecalc.domain.process.value_objects.chart.compressor import CompressorChart
@@ -182,6 +186,14 @@ def _all_equal(items: set) -> bool:
     return len(items) <= 1
 
 
+def _is_sampled_compressor(model: CompressorModel) -> bool:
+    if isinstance(model, CompressorModelSampled):
+        return True
+    if isinstance(model, CompressorWithTurbineModel) and isinstance(model.compressor_model, CompressorModelSampled):
+        return True
+    return False
+
+
 @overload
 def _create_fluid_factory(fluid_model: None) -> None: ...
 
@@ -1188,6 +1200,7 @@ def _map_multiple_streams_compressor(
             intermediate_pressure=interstage_control_pressure,
         )
 
+        assert fluid_factories is not None
         evaluation_input = CompressorEvaluationInput(
             rate_expression=rates_per_stream,
             fluid_factory=fluid_factories,
@@ -1279,40 +1292,34 @@ def _map_compressor(
                 discharge_pressure=discharge_pressure,
             )
 
-        evaluation_input = CompressorEvaluationInput(
-            rate_expression=stream_day_rate,
-            fluid_factory=fluid_factory,
-            power_loss_factor=power_loss_factor,
-            suction_pressure_expression=suction_pressure,
-            discharge_pressure_expression=discharge_pressure,
-            intermediate_pressure_expression=None,
-        )
-
         model_id = uuid4()
         # Register the compressor model and its evaluation input in the process service
         # - If it is a sampled model, or a turbine model wrapping a sampled model, treat as non-process.
         # - Otherwise, treat as a process model.
-        if isinstance(compressor_model, CompressorModelSampled):
+        if _is_sampled_compressor(compressor_model):
+            evaluation_input = CompressorSampledEvaluationInput(
+                rate_expression=stream_day_rate,
+                power_loss_factor=power_loss_factor,
+                suction_pressure_expression=suction_pressure,
+                discharge_pressure_expression=discharge_pressure,
+            )
             component = SimplifiedProcessUnitComponent(id=model_id, name=model.energy_function, type=model.type)
+            assert isinstance(compressor_model, CompressorModelSampled | CompressorWithTurbineModel)
             self._process_service.register_simplified_process_unit(
                 ecalc_component=component, simplified_process_unit=compressor_model, evaluation_input=evaluation_input
             )
-        elif isinstance(compressor_model, CompressorWithTurbineModel):
-            if isinstance(compressor_model.compressor_model, CompressorModelSampled):
-                component = SimplifiedProcessUnitComponent(id=model_id, name=model.energy_function, type=model.type)
-                self._process_service.register_simplified_process_unit(
-                    ecalc_component=component,
-                    simplified_process_unit=compressor_model,
-                    evaluation_input=evaluation_input,
-                )
-            else:
-                component = ProcessSystemComponent(id=model_id, name=model.energy_function, type=model.type)
-                self._process_service.register_process_system(
-                    ecalc_component=component, process_system=compressor_model, evaluation_input=evaluation_input
-                )
-
         else:
+            assert suction_pressure is not None and discharge_pressure is not None and fluid_factory is not None
+            evaluation_input = CompressorEvaluationInput(
+                rate_expression=stream_day_rate,
+                fluid_factory=fluid_factory,
+                power_loss_factor=power_loss_factor,
+                suction_pressure_expression=suction_pressure,
+                discharge_pressure_expression=discharge_pressure,
+                intermediate_pressure_expression=None,
+            )
             component = ProcessSystemComponent(id=model_id, name=model.energy_function, type=model.type)
+            assert isinstance(evaluation_input, CompressorEvaluationInput)
             self._process_service.register_process_system(
                 ecalc_component=component, process_system=compressor_model, evaluation_input=evaluation_input
             )
diff --git a/src/libecalc/presentation/yaml/model.py b/src/libecalc/presentation/yaml/model.py
@@ -23,7 +23,11 @@
 from libecalc.domain.process.compressor.core.base import CompressorWithTurbineModel
 from libecalc.domain.process.compressor.core.sampled import CompressorModelSampled
 from libecalc.domain.process.core.results import CompressorTrainResult, PumpModelResult
-from libecalc.domain.process.evaluation_input import CompressorEvaluationInput, PumpEvaluationInput
+from libecalc.domain.process.evaluation_input import (
+    CompressorEvaluationInput,
+    CompressorSampledEvaluationInput,
+    PumpEvaluationInput,
+)
 from libecalc.domain.process.pump.pump import PumpModel
 from libecalc.dto import ResultOptions
 from libecalc.dto.component_graph import ComponentGraph
@@ -390,7 +394,9 @@ def get_process_service(self) -> DefaultProcessService:
     @staticmethod
     def evaluate_domain_models(
         models: dict[UUID, Any],
-        evaluation_inputs: dict[UUID, CompressorEvaluationInput | PumpEvaluationInput],
+        evaluation_inputs: dict[
+            UUID, CompressorEvaluationInput | CompressorSampledEvaluationInput | PumpEvaluationInput
+        ],
     ) -> dict[UUID, CompressorTrainResult | PumpModelResult]:
         """
         Evaluates domain models and returns a mapping: model_id -> evaluated_result.
@@ -401,7 +407,11 @@ def evaluate_domain_models(
         assert all(
             isinstance(model, allowed_model_types) for model in models.values()
         ), "All models must be of allowed types"
-        allowed_evaluation_input_types = (CompressorEvaluationInput, PumpEvaluationInput)
+        allowed_evaluation_input_types = (
+            CompressorEvaluationInput,
+            CompressorSampledEvaluationInput,
+            PumpEvaluationInput,
+        )
         assert all(
             isinstance(evaluation_input, allowed_evaluation_input_types)
             for evaluation_input in evaluation_inputs.values()
