feat: add reusable fluid definitions for process simulations

Author: frodehk

src/libecalc/ecalc_model/time_series_fluid_model.py

@@ -0,0 +1,54 @@
+from libecalc.common.ddd import value_object
+from libecalc.common.errors.exceptions import ProgrammingError
+from libecalc.common.time_utils import Period
+from libecalc.presentation.yaml.domain.time_series_expression import TimeSeriesExpression
+from libecalc.process.fluid_stream.fluid_model import EoSModel, FluidComposition, FluidModel
+
+
+@value_object
+class TimeSeriesFluidComposition:
+    water: TimeSeriesExpression
+    nitrogen: TimeSeriesExpression
+    CO2: TimeSeriesExpression
+    methane: TimeSeriesExpression
+    ethane: TimeSeriesExpression
+    propane: TimeSeriesExpression
+    i_butane: TimeSeriesExpression
+    n_butane: TimeSeriesExpression
+    i_pentane: TimeSeriesExpression
+    n_pentane: TimeSeriesExpression
+    n_hexane: TimeSeriesExpression
+
+    def get_value(self, period: Period) -> FluidComposition:
+        """Evaluate all component expressions and materialize the fluid composition for the given period."""
+        try:
+            index = self.methane.get_periods().index(period)
+        except ValueError as error:
+            raise ProgrammingError(f"Period {period} is not available for the fluid composition.") from error
+
+        return FluidComposition(
+            water=self.water.get_masked_values()[index],
+            nitrogen=self.nitrogen.get_masked_values()[index],
+            CO2=self.CO2.get_masked_values()[index],
+            methane=self.methane.get_masked_values()[index],
+            ethane=self.ethane.get_masked_values()[index],
+            propane=self.propane.get_masked_values()[index],
+            i_butane=self.i_butane.get_masked_values()[index],
+            n_butane=self.n_butane.get_masked_values()[index],
+            i_pentane=self.i_pentane.get_masked_values()[index],
+            n_pentane=self.n_pentane.get_masked_values()[index],
+            n_hexane=self.n_hexane.get_masked_values()[index],
+        )
+
+
+@value_object
+class TimeSeriesFluidModel:
+    eos_model: EoSModel
+    composition: TimeSeriesFluidComposition
+
+    def get_value(self, period: Period) -> FluidModel:
+        """Materialize a fluid model with the composition evaluated for the given period."""
+        return FluidModel(
+            eos_model=self.eos_model,
+            composition=self.composition.get_value(period),
+        )

src/libecalc/ecalc_model/time_series_stream.py

@@ -1,12 +1,12 @@
 from libecalc.common.ddd import value_object
+from libecalc.ecalc_model.time_series_fluid_model import TimeSeriesFluidModel
 from libecalc.presentation.yaml.domain.expression_time_series_flow_rate import ExpressionTimeSeriesFlowRate
 from libecalc.presentation.yaml.domain.time_series_expression import TimeSeriesExpression
-from libecalc.process.fluid_stream.fluid_model import FluidModel
 
 
 @value_object
 class TimeSeriesStream:
-    fluid_model: FluidModel
+    fluid_model: TimeSeriesFluidModel
     pressure_bara: TimeSeriesExpression
     temperature_kelvin: TimeSeriesExpression
     standard_rate_m3_per_day: ExpressionTimeSeriesFlowRate

src/libecalc/presentation/yaml/domain/reference_service.py

@@ -24,6 +24,7 @@
     YamlVariableSpeedCompressorTrain,
     YamlVariableSpeedCompressorTrainMultipleStreamsAndPressures,
 )
+from libecalc.presentation.yaml.yaml_types.process.yaml_fluid_definitions import YamlFluidDefinition
 from libecalc.presentation.yaml.yaml_types.process.yaml_process_pipeline import YamlProcessPipeline
 from libecalc.presentation.yaml.yaml_types.process.yaml_process_units import YamlProcessUnitDefinition
 from libecalc.presentation.yaml.yaml_types.streams.yaml_inlet_stream import YamlInletStream
@@ -55,6 +56,9 @@ def get_yaml_path(self, reference: str) -> YamlPath: ...
     @abc.abstractmethod
     def get_fluid(self, reference: str) -> YamlFluidModel: ...
 
+    @abc.abstractmethod
+    def get_fluid_definition(self, reference: str) -> YamlFluidDefinition: ...
+
     @abc.abstractmethod
     def get_turbine(self, reference: str) -> YamlTurbine: ...
 

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

@@ -1,6 +1,22 @@
+from typing import assert_never
+
 from libecalc.common.errors.ecalc_validation_error import EcalcValidationException
+from libecalc.common.variables import ExpressionEvaluator
+from libecalc.ecalc_model.time_series_fluid_model import TimeSeriesFluidComposition, TimeSeriesFluidModel
+from libecalc.expression.expression import ExpressionType
+from libecalc.presentation.yaml.domain.time_series_expression import TimeSeriesExpression
 from libecalc.presentation.yaml.yaml_keywords import EcalcYamlKeywords
-from libecalc.presentation.yaml.yaml_types.models.yaml_fluid import YamlCompositionFluidModel, YamlPredefinedFluidModel
+from libecalc.presentation.yaml.yaml_types.models.yaml_fluid import (
+    YamlCompositionFluidModel,
+    YamlEosModel,
+    YamlPredefinedFluidModel,
+)
+from libecalc.presentation.yaml.yaml_types.process.yaml_fluid_definitions import (
+    YamlCompositionFluidDefinition,
+    YamlFluidComposition,
+    YamlFluidDefinition,
+    YamlPredefinedFluidDefinition,
+)
 from libecalc.process.fluid_stream.fluid_model import EoSModel, FluidComposition, FluidModel
 
 """
@@ -121,3 +137,69 @@ def composition_fluid_model_mapper(
             n_hexane=user_defined_composition.n_hexane,
         ),
     )
+
+
+def fluid_definition_mapper(
+    definition: YamlFluidDefinition,
+    expression_evaluator: ExpressionEvaluator,
+) -> TimeSeriesFluidModel:
+    match definition:
+        case YamlPredefinedFluidDefinition():
+            return predefined_fluid_definition_mapper(definition, expression_evaluator)
+        case YamlCompositionFluidDefinition():
+            return composition_fluid_definition_mapper(definition, expression_evaluator)
+        case _:
+            assert_never(definition)
+
+
+def predefined_fluid_definition_mapper(
+    model_config: YamlPredefinedFluidDefinition,
+    expression_evaluator: ExpressionEvaluator,
+) -> TimeSeriesFluidModel:
+    predefined_composition = _predefined_fluid_composition_mapper[model_config.gas_type]
+
+    return _create_time_series_fluid_model(
+        eos_model_type=model_config.eos_model,
+        composition=predefined_composition,
+        expression_evaluator=expression_evaluator,
+    )
+
+
+def composition_fluid_definition_mapper(
+    model_config: YamlCompositionFluidDefinition,
+    expression_evaluator: ExpressionEvaluator,
+) -> TimeSeriesFluidModel:
+    return _create_time_series_fluid_model(
+        eos_model_type=model_config.eos_model,
+        composition=model_config.composition,
+        expression_evaluator=expression_evaluator,
+    )
+
+
+def _create_time_series_fluid_model(
+    eos_model_type: YamlEosModel,
+    composition: FluidComposition | YamlFluidComposition,
+    expression_evaluator: ExpressionEvaluator,
+) -> TimeSeriesFluidModel:
+    def time_series_expression(expression: ExpressionType) -> TimeSeriesExpression:
+        return TimeSeriesExpression(
+            expression=expression,
+            expression_evaluator=expression_evaluator,
+        )
+
+    return TimeSeriesFluidModel(
+        eos_model=_eos_model_mapper[eos_model_type],
+        composition=TimeSeriesFluidComposition(
+            water=time_series_expression(composition.water),
+            nitrogen=time_series_expression(composition.nitrogen),
+            CO2=time_series_expression(composition.CO2),
+            methane=time_series_expression(composition.methane),
+            ethane=time_series_expression(composition.ethane),
+            propane=time_series_expression(composition.propane),
+            i_butane=time_series_expression(composition.i_butane),
+            n_butane=time_series_expression(composition.n_butane),
+            i_pentane=time_series_expression(composition.i_pentane),
+            n_pentane=time_series_expression(composition.n_pentane),
+            n_hexane=time_series_expression(composition.n_hexane),
+        ),
+    )

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

@@ -26,25 +26,23 @@
     TimeSeriesSplitterConfiguration,
     TimeSeriesTemperatureSetterConfiguration,
 )
+from libecalc.ecalc_model.time_series_fluid_model import TimeSeriesFluidModel
 from libecalc.ecalc_model.time_series_stream import TimeSeriesStream
 from libecalc.expression.expression import ExpressionType
 from libecalc.presentation.yaml.domain.expression_time_series_flow_rate import ExpressionTimeSeriesFlowRate
 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.mappers.consumer_function_mapper import handle_condition_list
-from libecalc.presentation.yaml.mappers.fluid_mapper import (
-    composition_fluid_model_mapper,
-    predefined_fluid_model_mapper,
-)
+from libecalc.presentation.yaml.mappers.fluid_mapper import fluid_definition_mapper
 from libecalc.presentation.yaml.mappers.model import InvalidChartResourceException
 from libecalc.presentation.yaml.mappers.process.build_sections import ProcessSectionBuilder
 from libecalc.presentation.yaml.yaml_types.components.yaml_expression_type import YamlExpressionType
-from libecalc.presentation.yaml.yaml_types.models import YamlFluidModel
 from libecalc.presentation.yaml.yaml_types.models.yaml_compressor_stages import YamlControlMarginUnits
-from libecalc.presentation.yaml.yaml_types.models.yaml_fluid import YamlCompositionFluidModel, YamlPredefinedFluidModel
+from libecalc.presentation.yaml.yaml_types.process.yaml_fluid_definitions import YamlFluidDefinition
 from libecalc.presentation.yaml.yaml_types.process.yaml_process_pipeline import YamlProcessPipeline
 from libecalc.presentation.yaml.yaml_types.process.yaml_process_references import (
+    DefinitionReference,
     InstanceReference,
 )
 from libecalc.presentation.yaml.yaml_types.process.yaml_process_simulation import (
@@ -62,7 +60,6 @@
 )
 from libecalc.presentation.yaml.yaml_types.streams.yaml_inlet_stream import YamlInletStream, YamlInletStreamRate
 from libecalc.presentation.yaml.yaml_types.yaml_data_or_file import YamlFile
-from libecalc.process.fluid_stream.fluid_model import FluidModel
 from libecalc.process.fluid_stream.fluid_service import FluidService
 from libecalc.process.fluid_stream.fluid_stream import FluidStream
 from libecalc.process.process_pipeline.process_pipeline import (
@@ -186,11 +183,13 @@ def _resolve_stream_reference(self, ref: str | YamlInletStream) -> YamlInletStre
         else:
             return ref
 
-    def _resolve_fluid_model_reference(self, ref: str | YamlFluidModel) -> YamlFluidModel:
-        if isinstance(ref, str):
-            return self._reference_service.get_fluid(reference=ref)
-        else:
-            return ref
+    def _resolve_fluid_definition(
+        self,
+        reference: DefinitionReference | YamlFluidDefinition,
+    ) -> YamlFluidDefinition:
+        if isinstance(reference, str):
+            return self._reference_service.get_fluid_definition(reference)
+        return reference
 
     def _map_conditions(
         self, conditions: YamlExpressionType | list[YamlExpressionType] | None
@@ -235,14 +234,14 @@ def _map_temperature(self, temperature: ExpressionType) -> TimeSeriesExpression:
             expression=temperature,
         )
 
-    def _map_fluid_model(self, yaml_fluid_model: YamlFluidModel) -> FluidModel:
-        match yaml_fluid_model:
-            case YamlPredefinedFluidModel():
-                return predefined_fluid_model_mapper(yaml_fluid_model)
-            case YamlCompositionFluidModel():
-                return composition_fluid_model_mapper(yaml_fluid_model)
-            case _:
-                assert_never(yaml_fluid_model)
+    def _map_fluid_definition(
+        self,
+        fluid_definition: YamlFluidDefinition,
+    ) -> TimeSeriesFluidModel:
+        return fluid_definition_mapper(
+            fluid_definition,
+            expression_evaluator=self._expression_evaluator,
+        )
 
     def _validate_and_map_pipeline_events(
         self,
@@ -347,14 +346,14 @@ def map_process_simulation(
 
                     case YamlMixerDefinition():
                         yaml_stream = self._resolve_stream_reference(yaml_process_unit.sidestream)
-                        yaml_fluid_model = self._resolve_fluid_model_reference(yaml_stream.fluid_model)
+                        yaml_fluid_definition = self._resolve_fluid_definition(yaml_stream.fluid)
                         unit = Mixer(fluid_service=self._fluid_service)
                         problem_time_series_configurations[unit.get_id()] = TimeSeriesMixerConfiguration(
                             sidestream=TimeSeriesStream(
                                 pressure_bara=self._map_pressure(yaml_stream.pressure),
                                 standard_rate_m3_per_day=self._map_rate(yaml_stream.rate),
                                 temperature_kelvin=self._map_temperature(yaml_stream.temperature),
-                                fluid_model=self._map_fluid_model(yaml_fluid_model),
+                                fluid_model=self._map_fluid_definition(yaml_fluid_definition),
                             )
                         )
 
@@ -539,12 +538,12 @@ def map_process_simulation(
                     )
 
                 yaml_stream = self._resolve_stream_reference(yaml_stream_distribution.inlet_stream)
-                yaml_fluid_model = self._resolve_fluid_model_reference(yaml_stream.fluid_model)
+                yaml_fluid_definition = self._resolve_fluid_definition(yaml_stream.fluid)
                 inlet_stream = TimeSeriesStream(
                     pressure_bara=self._map_pressure(yaml_stream.pressure),
                     standard_rate_m3_per_day=self._map_rate(yaml_stream.rate),
                     temperature_kelvin=self._map_temperature(yaml_stream.temperature),
-                    fluid_model=self._map_fluid_model(yaml_fluid_model),
+                    fluid_model=self._map_fluid_definition(yaml_fluid_definition),
                 )
                 stream_distribution = CommonStreamDistributionConfig(
                     inlet_stream=inlet_stream,
@@ -554,13 +553,13 @@ def map_process_simulation(
                 inlet_streams = []
                 for inlet_stream in yaml_stream_distribution.inlet_streams:
                     yaml_stream = self._resolve_stream_reference(inlet_stream)
-                    yaml_fluid_model = self._resolve_fluid_model_reference(yaml_stream.fluid_model)
+                    yaml_fluid_definition = self._resolve_fluid_definition(yaml_stream.fluid)
                     inlet_streams.append(
                         TimeSeriesStream(
                             pressure_bara=self._map_pressure(yaml_stream.pressure),
                             standard_rate_m3_per_day=self._map_rate(yaml_stream.rate),
                             temperature_kelvin=self._map_temperature(yaml_stream.temperature),
-                            fluid_model=self._map_fluid_model(yaml_fluid_model),
+                            fluid_model=self._map_fluid_definition(yaml_fluid_definition),
                         )
                     )
                 stream_distribution = IndividualStreamDistributionConfig(

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

@@ -31,6 +31,7 @@
 from libecalc.presentation.yaml.yaml_types.facility_model.yaml_facility_model import YamlFacilityModel
 from libecalc.presentation.yaml.yaml_types.fuel_type.yaml_fuel_type import YamlFuelType
 from libecalc.presentation.yaml.yaml_types.models import YamlConsumerModel, YamlFluidModel
+from libecalc.presentation.yaml.yaml_types.process.yaml_fluid_definitions import YamlFluidDefinition
 from libecalc.presentation.yaml.yaml_types.process.yaml_process_pipeline import YamlProcessPipeline
 from libecalc.presentation.yaml.yaml_types.process.yaml_process_simulation import (
     YamlEcalcEvent,
@@ -51,6 +52,7 @@
 _PROCESS_PIPELINES_KEY = "PROCESS_PIPELINES"
 _INLET_STREAMS_KEY = "INLET_STREAMS"
 _FLUID_MODELS_KEY = "FLUID_MODELS"
+_FLUIDS_KEY = "FLUIDS"
 _PROCESS_SIMULATIONS_KEY = "PROCESS_SIMULATIONS"
 _ECALC_EVENTS_KEY = "ECALC_EVENTS"
 _PROCESS_EVENTS_KEY = "PROCESS_EVENTS"
@@ -456,17 +458,25 @@ def fluid_models(self) -> dict[str, YamlFluidModel]:
     @property
     def definitions(self) -> YamlDefinitions:
         process_units: dict[str, YamlProcessUnitDefinition] = {}
+        fluids: dict[str, YamlFluidDefinition] = {}
         definitions = (
             self._internal_datamodel.get(_DEFINITIONS_KEY, {}) if isinstance(self._internal_datamodel, dict) else {}
         )
-        raw = definitions.get(_PROCESS_UNITS_KEY, {}) if isinstance(definitions, dict) else {}
+        raw_process_units = definitions.get(_PROCESS_UNITS_KEY, {}) if isinstance(definitions, dict) else {}
+        raw_fluids = definitions.get(_FLUIDS_KEY, {}) if isinstance(definitions, dict) else {}
 
-        for name, unit_data in raw.items():
+        for name, unit_data in raw_process_units.items():
             try:
                 process_units[name] = TypeAdapter(YamlProcessUnitDefinition).validate_python(unit_data)
             except PydanticValidationError:
                 pass
-        return YamlDefinitions(process_units=process_units)
+        for name, fluid_data in raw_fluids.items():
+            try:
+                fluids[name] = TypeAdapter(YamlFluidDefinition).validate_python(fluid_data)
+            except PydanticValidationError:
+                pass
+
+        return YamlDefinitions(process_units=process_units, fluids=fluids)
 
     @property
     def process_pipelines(self) -> dict[str, YamlProcessPipeline]:

src/libecalc/presentation/yaml/yaml_reference_service.py

@@ -25,6 +25,7 @@
     YamlTurbine,
 )
 from libecalc.presentation.yaml.yaml_types.models.yaml_enums import YamlModelType
+from libecalc.presentation.yaml.yaml_types.process.yaml_fluid_definitions import YamlFluidDefinition
 from libecalc.presentation.yaml.yaml_types.process.yaml_process_pipeline import YamlProcessPipeline
 from libecalc.presentation.yaml.yaml_types.process.yaml_process_simulation import (
     YamlProcessSimulation,
@@ -45,6 +46,7 @@
     | YamlProcessSimulation
     | YamlPumpProcessSimulation
     | YamlProcessUnitDefinition
+    | YamlFluidDefinition
 )
 
 # Some models are referenced by other models, for example a compressor model will reference compressor chart models
@@ -117,6 +119,12 @@ def __init__(
             references[process_unit_key] = process_unit
             reference_yaml_context[process_unit_key] = process_unit_path
 
+        fluids_path = YamlPath(keys=("DEFINITIONS", "FLUIDS"))
+        for fluid_key, fluid in configuration.definitions.fluids.items():
+            fluid_path = fluids_path.append(fluid_key)
+            references[fluid_key] = fluid
+            reference_yaml_context[fluid_key] = fluid_path
+
         process_pipelines_path = YamlPath(keys=("PROCESS_PIPELINES",))
         for process_pipeline_key, process_pipeline in configuration.process_pipelines.items():
             process_pipeline_path = process_pipelines_path.append(process_pipeline_key)
@@ -154,6 +162,12 @@ def get_fluid(self, reference: str) -> YamlFluidModel:
             raise InvalidReferenceException("fluid model", reference)
         return model
 
+    def get_fluid_definition(self, reference: str) -> YamlFluidDefinition:
+        fluid = self._resolve_yaml_reference(reference, "fluid definition")
+        if not isinstance(fluid, get_args(get_args(YamlFluidDefinition)[0])):
+            raise InvalidReferenceException("fluid definition", reference)
+        return fluid
+
     def get_turbine(self, reference: str) -> YamlTurbine:
         model = self._resolve_yaml_reference(reference, "turbine model")
         if not isinstance(model, YamlTurbine):