feat(libecalc): add closed-form liquid pump process simulation

Introduce a liquid pump process graph under a dedicated PUMP_PROCESS_SIMULATIONS
YAML keyword, kept separate from the gas FluidStream pipeline.

- Extend the closed-form Pump result with efficiency, specific shaft work,
  shaft power and speed at the operating point (PumpEvaluationResult).
- Add PumpProcessSimulation: a fixed liquid graph of inlet, recirculation
  mixer, pump, recirculation splitter, choke and outlet with typed
  LiquidProcessUnit entities, connections and a recirculation loop. The domain
  is time-agnostic: evaluate() maps a sequence of PumpOperatingInput to a
  liquid stream on every connection plus the pump result, in order.
- Add the PUMP_PROCESS_SIMULATIONS "TYPE: PUMP" YAML root, its mapper and
  YamlModel.get_pump_process_simulations().

Refs: equinor/ecalc-internal#2059

Author: kjbrak

src/libecalc/presentation/yaml/mappers/pump_process_simulation_mapper.py

@@ -0,0 +1,164 @@
+from libecalc.common.errors.ecalc_validation_error import EcalcValidationException
+from libecalc.common.errors.exceptions import InvalidResourceException
+from libecalc.common.time_utils import Period
+from libecalc.common.variables import ExpressionEvaluator
+from libecalc.domain.process.value_objects.chart.chart import ChartData
+from libecalc.domain.regularity import Regularity
+from libecalc.domain.resource import Resources
+from libecalc.presentation.yaml.domain.expression_time_series_flow_rate import ExpressionTimeSeriesFlowRate
+from libecalc.presentation.yaml.domain.expression_time_series_fluid_density import ExpressionTimeSeriesFluidDensity
+from libecalc.presentation.yaml.domain.expression_time_series_pressure import ExpressionTimeSeriesPressure
+from libecalc.presentation.yaml.domain.reference_service import ReferenceService
+from libecalc.presentation.yaml.domain.time_series_expression import TimeSeriesExpression
+from libecalc.presentation.yaml.mappers.charts.user_defined_chart_data import UserDefinedChartData
+from libecalc.presentation.yaml.yaml_types.facility_model.yaml_facility_model import (
+    YamlPumpChartSingleSpeed,
+)
+from libecalc.presentation.yaml.yaml_types.process.yaml_process_simulation import YamlPumpProcessSimulation
+from libecalc.process.pump.liquid_stream import LiquidStream
+from libecalc.process.pump.pump import Pump
+from libecalc.process.pump.pump_process_simulation import (
+    PumpOperatingInput,
+    PumpProcessSimulation,
+)
+
+
+class PumpProcessSimulationMapper:
+    def __init__(
+        self,
+        expression_evaluator: ExpressionEvaluator,
+        reference_service: ReferenceService,
+        resources: Resources,
+        process_simulation_period: Period,
+    ):
+        self._expression_evaluator = expression_evaluator.get_subset_for_period(process_simulation_period)
+        self._reference_service = reference_service
+        self._resources = resources
+
+    def map(
+        self,
+        yaml_process_simulation: YamlPumpProcessSimulation,
+    ) -> tuple[PumpProcessSimulation, list[PumpOperatingInput], list[Period]]:
+        chart_data = self._get_chart_data(yaml_process_simulation.pump_model.chart)
+        pump = Pump(
+            pump_chart=chart_data,
+            minimum_flow_rate_m3_per_hour=yaml_process_simulation.pump_model.minimum_flow_rate,
+        )
+        simulation = PumpProcessSimulation(
+            pump=pump,
+            name=yaml_process_simulation.name,
+        )
+
+        regularity = Regularity(
+            expression_evaluator=self._expression_evaluator,
+            target_period=self._expression_evaluator.get_period(),
+        )
+        rate = ExpressionTimeSeriesFlowRate(
+            time_series_expression=TimeSeriesExpression(
+                expression_evaluator=self._expression_evaluator,
+                expression=yaml_process_simulation.inlet.rate,
+            ),
+            regularity=regularity,
+        )
+        rate_values = rate.get_stream_day_values()
+        suction_pressure = ExpressionTimeSeriesPressure(
+            time_series_expression=TimeSeriesExpression(
+                expression_evaluator=self._expression_evaluator,
+                expression=yaml_process_simulation.inlet.pressure,
+            ),
+        )
+        discharge_pressure = ExpressionTimeSeriesPressure(
+            time_series_expression=TimeSeriesExpression(
+                expression_evaluator=self._expression_evaluator,
+                expression=yaml_process_simulation.required_discharge_pressure,
+            ),
+        )
+        density = ExpressionTimeSeriesFluidDensity(
+            time_series_expression=TimeSeriesExpression(
+                expression_evaluator=self._expression_evaluator,
+                expression=yaml_process_simulation.inlet.density,
+            )
+        )
+
+        suction_values = suction_pressure.get_values()
+        discharge_values = discharge_pressure.get_values()
+        density_values = density.get_values()
+
+        # The domain is time-agnostic: build one physical input per period and keep the period
+        # vector here in the presentation layer, paired with the inputs (and the results) by index.
+        periods = list(rate.get_periods())
+        operating_inputs = [
+            self._to_operating_input(
+                rate_value=rate_value,
+                suction_pressure_bara=suction,
+                required_discharge_pressure_bara=discharge,
+                density_kg_per_m3=density_value,
+            )
+            for rate_value, suction, discharge, density_value in zip(
+                rate_values,
+                suction_values,
+                discharge_values,
+                density_values,
+                strict=True,
+            )
+        ]
+        if len(periods) != len(operating_inputs):
+            raise ValueError("Pump period vector and input vector length mismatch.")
+        return simulation, operating_inputs, periods
+
+    def _to_operating_input(
+        self,
+        rate_value: float,
+        suction_pressure_bara: float,
+        required_discharge_pressure_bara: float,
+        density_kg_per_m3: float,
+    ) -> PumpOperatingInput:
+        # Suction pressure and density describe the physical inlet fluid; they must be positive
+        # whether or not the pump runs.
+        self._require_positive(suction_pressure_bara, "suction pressure [bara]")
+        self._require_positive(density_kg_per_m3, "inlet density [kg/m3]")
+        inlet_stream = LiquidStream.from_volumetric_rate(
+            volumetric_rate_m3_per_day=rate_value,
+            pressure_bara=suction_pressure_bara,
+            density_kg_per_m3=density_kg_per_m3,
+        )
+
+        if rate_value > 0:
+            self._require_positive(required_discharge_pressure_bara, "required discharge pressure [bara]")
+            discharge = required_discharge_pressure_bara
+        else:
+            # Pump off (zero rate): the required discharge is a meaningless duty target. Keep the
+            # user's value when it is a valid absolute pressure, otherwise fall back to the inlet
+            # pressure (the pump delivers no head, so the outlet equals the inlet).
+            discharge = (
+                required_discharge_pressure_bara if required_discharge_pressure_bara > 0 else suction_pressure_bara
+            )
+
+        return PumpOperatingInput(
+            inlet_stream=inlet_stream,
+            required_discharge_pressure_bara=discharge,
+        )
+
+    @staticmethod
+    def _require_positive(value: float, subject: str) -> None:
+        if value <= 0:
+            raise EcalcValidationException(f"Pump {subject} must be greater than 0; got {value}.")
+
+    def _get_chart_data(self, reference: str) -> ChartData:
+        model = self._reference_service.get_pump_model(reference)
+        if model.head_margin != 0.0:
+            raise EcalcValidationException(
+                "HEAD_MARGIN is not supported by the new pump process domain "
+                "(points above the maximum head are flagged infeasible instead of snapped to it)."
+            )
+        resource = self._resources.get(model.file)
+        if resource is None:
+            raise EcalcValidationException(f"Pump chart resource '{model.file}' was not found.")
+        try:
+            return UserDefinedChartData.from_resource(
+                resource,
+                units=model.units,
+                is_single_speed=isinstance(model, YamlPumpChartSingleSpeed),
+            )
+        except InvalidResourceException as error:
+            raise EcalcValidationException(str(error)) from error

src/libecalc/presentation/yaml/model.py

@@ -45,6 +45,7 @@
 from libecalc.presentation.yaml.domain.time_series_resource import TimeSeriesResource
 from libecalc.presentation.yaml.mappers.component_mapper import EcalcModelMapper
 from libecalc.presentation.yaml.mappers.process_simulation_mapper import ProcessSimulationMapper
+from libecalc.presentation.yaml.mappers.pump_process_simulation_mapper import PumpProcessSimulationMapper
 from libecalc.presentation.yaml.mappers.variables_mapper import map_yaml_to_variables
 from libecalc.presentation.yaml.mappers.variables_mapper.get_global_time_vector import (
     InvalidEndDate,
@@ -68,6 +69,10 @@
     YamlModelValidationContextNames,
 )
 from libecalc.process.process_pipeline.process_pipeline import ProcessPipeline
+from libecalc.process.pump.pump_process_simulation import (
+    PumpOperatingInput,
+    PumpProcessSimulation,
+)
 
 DEFAULT_START_TIME = datetime(1900, 1, 1)
 
@@ -165,6 +170,22 @@ def get_process_simulations(self) -> tuple[list[ProcessPipeline], list[ProcessSi
 
         return process_pipelines, process_simulations
 
+    def get_pump_process_simulations(
+        self,
+    ) -> list[tuple[PumpProcessSimulation, list[PumpOperatingInput], list[Period]]]:
+        self.validate_for_run()
+        facility_resources, _ = self._resource_service.get_facility_resources()
+        mapper = PumpProcessSimulationMapper(
+            expression_evaluator=self.get_expression_evaluator(),
+            process_simulation_period=self.period,
+            resources=facility_resources,
+            reference_service=self._get_reference_service(),
+        )
+        return [
+            mapper.map(yaml_pump_process_simulation)
+            for yaml_pump_process_simulation in self._configuration.pump_process_simulations
+        ]
+
     def get_periods(self) -> list[Period]:
         """
         Get the global timevector for this model

src/libecalc/presentation/yaml/yaml_models/pyyaml_yaml_model.py

@@ -36,6 +36,7 @@
     YamlEcalcEvent,
     YamlProcessEvent,
     YamlProcessSimulation,
+    YamlPumpProcessSimulation,
 )
 from libecalc.presentation.yaml.yaml_types.process.yaml_process_units import YamlProcessUnit
 from libecalc.presentation.yaml.yaml_types.streams.yaml_inlet_stream import YamlInletStream
@@ -52,6 +53,7 @@
 _PROCESS_SIMULATIONS_KEY = "PROCESS_SIMULATIONS"
 _ECALC_EVENTS_KEY = "ECALC_EVENTS"
 _PROCESS_EVENTS_KEY = "PROCESS_EVENTS"
+_PUMP_PROCESS_SIMULATIONS_KEY = "PUMP_PROCESS_SIMULATIONS"
 _NEW_SECTIONS_WITH_FILE_REFS: tuple[str, ...] = (
     "PROCESS_UNITS",
     "PROCESS_PIPELINES",
@@ -506,6 +508,17 @@ def process_events(self) -> list[YamlProcessEvent]:
                 pass
         return process_events
 
+    @property
+    def pump_process_simulations(self) -> list[YamlPumpProcessSimulation]:
+        pump_process_simulations: list[YamlPumpProcessSimulation] = []
+        adapter = TypeAdapter(YamlPumpProcessSimulation)
+        for pump_process_simulation in self._get_yaml_list_or_empty(_PUMP_PROCESS_SIMULATIONS_KEY):
+            try:
+                pump_process_simulations.append(adapter.validate_python(pump_process_simulation))
+            except PydanticValidationError:
+                pass
+        return pump_process_simulations
+
     @property
     def start(self) -> datetime.datetime | None:
         start_value = self._internal_datamodel.get(EcalcYamlKeywords.start)

src/libecalc/presentation/yaml/yaml_models/yaml_model.py

@@ -21,6 +21,7 @@
     YamlEcalcEvent,
     YamlProcessEvent,
     YamlProcessSimulation,
+    YamlPumpProcessSimulation,
 )
 from libecalc.presentation.yaml.yaml_types.process.yaml_process_units import YamlProcessUnit
 from libecalc.presentation.yaml.yaml_types.streams.yaml_inlet_stream import YamlInletStream
@@ -112,6 +113,11 @@ def ecalc_events(self) -> list[YamlEcalcEvent]:
     def process_events(self) -> list[YamlProcessEvent]:
         pass
 
+    @property
+    @abc.abstractmethod
+    def pump_process_simulations(self) -> Iterable[YamlPumpProcessSimulation]:
+        pass
+
     @property
     @abc.abstractmethod
     def installations(self) -> Iterable[YamlInstallation]:

src/libecalc/presentation/yaml/yaml_reference_service.py

@@ -26,7 +26,10 @@
 )
 from libecalc.presentation.yaml.yaml_types.models.yaml_enums import YamlModelType
 from libecalc.presentation.yaml.yaml_types.process.yaml_process_pipeline import YamlProcessPipeline
-from libecalc.presentation.yaml.yaml_types.process.yaml_process_simulation import YamlProcessSimulation
+from libecalc.presentation.yaml.yaml_types.process.yaml_process_simulation import (
+    YamlProcessSimulation,
+    YamlPumpProcessSimulation,
+)
 from libecalc.presentation.yaml.yaml_types.process.yaml_process_units import YamlProcessUnit
 from libecalc.presentation.yaml.yaml_types.streams.yaml_inlet_stream import YamlInletStream
 
@@ -35,7 +38,13 @@
 YamlModel = YamlConsumerModel | YamlFacilityModel
 
 ReferenceType = (
-    YamlModel | YamlFuelType | YamlInletStream | YamlProcessPipeline | YamlProcessSimulation | YamlProcessUnit
+    YamlModel
+    | YamlFuelType
+    | YamlInletStream
+    | YamlProcessPipeline
+    | YamlProcessSimulation
+    | YamlPumpProcessSimulation
+    | YamlProcessUnit
 )
 
 # Some models are referenced by other models, for example a compressor model will reference compressor chart models
@@ -122,6 +131,12 @@ def __init__(
             references[process_simulation.name] = process_simulation
             reference_yaml_context[process_simulation.name] = process_simulation_path
 
+        pump_process_simulations_path = YamlPath(keys=("PUMP_PROCESS_SIMULATIONS",))
+        for pump_process_simulation_index, pump_process_simulation in enumerate(configuration.pump_process_simulations):
+            pump_process_simulation_path = pump_process_simulations_path.append(pump_process_simulation_index)
+            references[pump_process_simulation.name] = pump_process_simulation
+            reference_yaml_context[pump_process_simulation.name] = pump_process_simulation_path
+
         fluid_models_path = YamlPath(keys=("FLUID_MODELS",))
 
         for fluid_model_key, fluid_model in configuration.fluid_models.items():

src/libecalc/presentation/yaml/yaml_types/components/yaml_asset.py

@@ -12,6 +12,7 @@
     YamlEcalcEvent,
     YamlProcessEvent,
     YamlProcessSimulation,
+    YamlPumpProcessSimulation,
 )
 from libecalc.presentation.yaml.yaml_types.process.yaml_process_units import YamlProcessUnit
 from libecalc.presentation.yaml.yaml_types.streams.yaml_inlet_stream import YamlInletStream
@@ -93,6 +94,11 @@ class YamlAsset(YamlBase):
         title="PROCESS_EVENTS",
         description="Defines process-specific events that reference global eCalc events.",
     )
+    pump_process_simulations: list[YamlPumpProcessSimulation] = Field(
+        default_factory=list,
+        title="PUMP_PROCESS_SIMULATIONS",
+        description="Defines one or more liquid pump process simulations to be run.",
+    )
     installations: list[YamlInstallation] = Field(
         ...,
         title="INSTALLATIONS",
@@ -189,6 +195,10 @@ def validate_unique_references(self):
             for process_simulation in self.process_simulations:
                 references.append(process_simulation.name)
 
+        if self.pump_process_simulations is not None:
+            for pump_process_simulation in self.pump_process_simulations:
+                references.append(pump_process_simulation.name)
+
         # TODO: Add ecalc events references?
 
         if self.fluid_models is not None:

src/libecalc/presentation/yaml/yaml_types/process/yaml_process_references.py

@@ -7,3 +7,4 @@
 type ProcessUnitReference = str
 type EcalcEventReference = str
 type ProcessEventReference = str
+type PumpChartReference = str