chore: init mapping

Author: jsolaas

src/libecalc/domain/process/entities/process_units/recirculation_loop.py

@@ -1,6 +1,7 @@
 from collections.abc import Sequence
 
 from libecalc.domain.component_validation_error import DomainValidationException
+from libecalc.domain.process.entities.process_units.compressor import Compressor
 from libecalc.domain.process.entities.process_units.mixer import Mixer
 from libecalc.domain.process.entities.process_units.splitter import Splitter
 from libecalc.domain.process.process_system.process_system import ProcessSystem, ProcessSystemId
@@ -57,7 +58,7 @@ class RecirculationLoop(ProcessSystem):
     def __init__(
         self,
         process_system_id: ProcessSystemId,
-        inner_process: ProcessSystem,
+        inner_process: ProcessSystem | ProcessUnit,
         fluid_service: FluidService,
         recirculation_rate: float = 0,
     ):
@@ -77,20 +78,36 @@ def __init__(
         )
 
     def _validate_inner_process(self):
-        assert isinstance(self._inner_process, ProcessSystem), "Recirculation loop should contain a ProcessSystem"
-        for process_unit in self._inner_process.get_process_units():
-            if isinstance(process_unit, Mixer | Splitter):
-                raise DomainValidationException("Recirculation loop cannot contain splitters or mixers")
+        if not isinstance(self._inner_process, ProcessSystem | ProcessUnit):
+            raise DomainValidationException(
+                "Recirculation loop should contain a ProcessSystem with a compressor or a single compressor"
+            )
+        if isinstance(self._inner_process, ProcessSystem):
+            for process_unit in self._inner_process.get_process_units():
+                if isinstance(process_unit, Splitter | Mixer):
+                    raise DomainValidationException("Recirculation loop cannot contain splitters or mixers")
+        if isinstance(self._inner_process, ProcessUnit):
+            if not isinstance(self._inner_process, Compressor):
+                raise DomainValidationException(
+                    "Recirculation loop should contain a ProcessSystem with a compressor or a single compressor"
+                )
 
     def get_id(self) -> ProcessSystemId:
         return self._id
 
     def get_process_units(self) -> Sequence[ProcessUnit | ProcessSystem]:
-        return [
-            self._mixer,
-            *self._inner_process.get_process_units(),
-            self._splitter,
-        ]
+        if isinstance(self._inner_process, ProcessSystem):
+            return [
+                self._mixer,
+                *self._inner_process.get_process_units(),
+                self._splitter,
+            ]
+        else:
+            return [
+                self._mixer,
+                self._inner_process,
+                self._splitter,
+            ]
 
     def set_recirculation_rate(self, rate: float):
         self._mixer.set_mix_rate(rate)

src/libecalc/domain/process/process_simulation.py

@@ -0,0 +1,51 @@
+from dataclasses import dataclass
+from typing import Literal
+
+from libecalc.domain.process.process_system.process_system import ProcessSystem, ProcessSystemId
+from libecalc.domain.process.stream_distribution.common_stream_distribution import Overflow
+from libecalc.domain.process.value_objects.fluid_stream.time_series_stream import TimeSeriesStream
+from libecalc.presentation.yaml.domain.time_series_expression import TimeSeriesExpression
+
+
+@dataclass
+class CommonStreamSettings:
+    rate_fractions: list[TimeSeriesExpression]
+    overflow: list[Overflow]
+
+
+@dataclass
+class CommonStreamDistributionConfig:
+    inlet_stream: TimeSeriesStream
+    settings: list[CommonStreamSettings]
+
+
+@dataclass
+class IndividualStreamDistributionConfig:
+    inlet_streams: list[TimeSeriesStream]
+
+
+@dataclass
+class Constraint:
+    process_system_id: ProcessSystemId
+    outlet_pressure: TimeSeriesExpression
+
+
+@dataclass
+class PressureControlConfig:
+    process_system_id: ProcessSystemId
+    type: Literal["UPSTREAM_CHOKE", "DOWNSTREAM_CHOKE", "COMMON_ASV", "INDIVIDUAL_ASV_RATE", "INDIVIDUAL_ASV_PRESSURE"]
+
+
+@dataclass
+class AntiSurgeConfig:
+    process_system_id: ProcessSystemId
+    type: Literal["INDIVIDUAL_ASV", "COMMON_ASV"]
+
+
+@dataclass
+class ProcessSimulation:
+    process_systems: list[ProcessSystem]
+    pressure_control_strategies: list[PressureControlConfig]
+    anti_surge_strategies: list[AntiSurgeConfig]
+    constraints: list[Constraint]
+    stream_distribution: IndividualStreamDistributionConfig | CommonStreamDistributionConfig

src/libecalc/domain/process/process_solver/anti_surge/individual_asv.py

@@ -29,8 +29,8 @@ class IndividualASVAntiSurgeStrategy(AntiSurgeStrategy):
 
     def __init__(
         self,
-        recirculation_loop_ids: list[ProcessSystemId],
-        compressors: list[Compressor],
+        recirculation_loop_ids: Sequence[ProcessSystemId],
+        compressors: Sequence[Compressor],
         simulator: ProcessRunner,
     ):
         assert len(recirculation_loop_ids) == len(compressors)

src/libecalc/domain/process/process_solver/pressure_control/individual_asv.py

@@ -33,8 +33,8 @@ class IndividualASVPressureControlStrategy(PressureControlStrategy):
     def __init__(
         self,
         simulator: ProcessRunner,
-        recirculation_loop_ids: list[ProcessSystemId],
-        compressors: list[Compressor],
+        recirculation_loop_ids: Sequence[ProcessSystemId],
+        compressors: Sequence[Compressor],
         root_finding_strategy: RootFindingStrategy,
     ):
         self._simulator = simulator
@@ -131,8 +131,8 @@ class IndividualASVRateControlStrategy(PressureControlStrategy):
     def __init__(
         self,
         simulator: ProcessRunner,
-        recirculation_loop_ids: list[ProcessSystemId],
-        compressors: list[Compressor],
+        recirculation_loop_ids: Sequence[ProcessSystemId],
+        compressors: Sequence[Compressor],
     ):
         self._simulator = simulator
         self._recirculation_loop_ids = recirculation_loop_ids
@@ -222,8 +222,8 @@ def get_outlet_stream(rate_fraction: float) -> FluidStream:
 
 def _minimum_achievable_pressure(
     simulator: ProcessRunner,
-    recirculation_loop_ids: list[ProcessSystemId],
-    compressors: list[Compressor],
+    recirculation_loop_ids: Sequence[ProcessSystemId],
+    compressors: Sequence[Compressor],
     inlet_stream: FluidStream,
 ) -> Sequence[Configuration[RecirculationConfiguration | ChokeConfiguration]]:
     """Propagate with maximum recirculation on every stage to find the lowest achievable pressure."""

src/libecalc/domain/process/process_solver/process_system_runner.py

@@ -1,4 +1,4 @@
-from collections.abc import Iterable
+from collections.abc import Iterable, Sequence
 
 from libecalc.domain.process.entities.process_units.choke import Choke
 from libecalc.domain.process.entities.process_units.recirculation_loop import RecirculationLoop
@@ -15,7 +15,7 @@
 
 
 class ProcessSystemRunner(ProcessRunner):
-    def __init__(self, shaft: Shaft, units: list[ProcessUnit | ProcessSystem]):
+    def __init__(self, shaft: Shaft, units: Sequence[ProcessUnit | ProcessSystem]):
         self._shaft = shaft
         self._units = {unit.get_id(): unit for unit in units}
         self._configurations: dict[ProcessUnitId | ProcessSystemId | ShaftId, Configuration] = {}

src/libecalc/domain/process/stream_distribution/common_stream_distribution.py

@@ -1,12 +1,14 @@
 import abc
 from collections import defaultdict
 from collections.abc import Hashable, Iterable
+from collections.abc import Mapping
 from dataclasses import dataclass
 from typing import Generic, TypeVar
 
 import networkx as nx
-from libecalc.domain.process.stream_distribution.stream_distribution import StreamDistribution
+
 from libecalc.domain.component_validation_error import DomainValidationException
+from libecalc.domain.process.stream_distribution.stream_distribution import StreamDistribution
 from libecalc.domain.process.value_objects.fluid_stream import FluidService, FluidStream
 
 
@@ -28,7 +30,7 @@ class CommonStreamDistribution(StreamDistribution, Generic[T]):
     def __init__(
         self,
         inlet_stream: FluidStream,
-        items: dict[T, HasExcessRate],
+        items: Mapping[T, HasExcessRate],
         rate_fractions: list[float],
         overflows: list[Overflow[T]],
         fluid_service: FluidService,

src/libecalc/domain/process/stream_distribution/individual_stream_distribution.py

@@ -0,0 +1,15 @@
+from collections.abc import Sequence
+
+from libecalc.domain.process.stream_distribution.stream_distribution import StreamDistribution
+from libecalc.domain.process.value_objects.fluid_stream import FluidStream
+
+
+class IndividualStreamDistribution(StreamDistribution):
+    def __init__(self, streams: Sequence[FluidStream]):
+        self._streams = streams
+
+    def get_number_of_streams(self) -> int:
+        return len(self._streams)
+
+    def get_streams(self) -> list[FluidStream]:
+        return list(self._streams)

src/libecalc/domain/process/stream_distribution/priorities_stream_distribution.py

@@ -1,4 +1,5 @@
 import abc
+from collections.abc import Sequence
 
 from libecalc.domain.process.stream_distribution.stream_distribution import StreamDistribution
 from libecalc.domain.process.value_objects.fluid_stream import FluidStream
@@ -9,7 +10,7 @@ class HasValidity(abc.ABC):
     def is_valid(self, inlet_stream: FluidStream) -> bool: ...
 
 
-def find_first_valid(stream_distributions: list[StreamDistribution], items: list[HasValidity]) -> list[FluidStream]:
+def find_first_valid(stream_distributions: list[StreamDistribution], items: Sequence[HasValidity]) -> list[FluidStream]:
     assert len(stream_distributions) > 0
     for stream_distribution in stream_distributions:
         streams = stream_distribution.get_streams()
@@ -20,7 +21,7 @@ def find_first_valid(stream_distributions: list[StreamDistribution], items: list
 
 
 class PrioritiesStreamDistribution(StreamDistribution):
-    def __init__(self, stream_distributions: list[StreamDistribution], items: list[HasValidity]):
+    def __init__(self, stream_distributions: list[StreamDistribution], items: Sequence[HasValidity]):
         self._stream_distributions = stream_distributions
         self._items = items
 

src/libecalc/domain/process/value_objects/fluid_stream/time_series_stream.py

@@ -0,0 +1,13 @@
+from dataclasses import dataclass
+
+from libecalc.domain.process.value_objects.fluid_stream import FluidModel
+from libecalc.domain.time_series_flow_rate import TimeSeriesFlowRate
+from libecalc.presentation.yaml.domain.time_series_expression import TimeSeriesExpression
+
+
+@dataclass(frozen=True)
+class TimeSeriesStream:
+    fluid_model: FluidModel
+    pressure_bara: TimeSeriesExpression
+    temperature_kelvin: TimeSeriesExpression
+    standard_rate_m3_per_day: TimeSeriesFlowRate

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

@@ -4,6 +4,11 @@
 
 from libecalc.domain.component_validation_error import DomainValidationException
 from libecalc.presentation.yaml.mappers.yaml_path import YamlPath
+from libecalc.presentation.yaml.yaml_types.components.yaml_process_system import (
+    YamlCompressor,
+    YamlCompressorStageProcessSystem,
+    YamlSerialProcessSystem,
+)
 from libecalc.presentation.yaml.yaml_types.facility_model.yaml_facility_model import (
     YamlCompressorTabularModel,
     YamlGeneratorSetModel,
@@ -24,6 +29,7 @@
     YamlVariableSpeedCompressorTrain,
     YamlVariableSpeedCompressorTrainMultipleStreamsAndPressures,
 )
+from libecalc.presentation.yaml.yaml_types.streams.yaml_inlet_stream import YamlInletStream
 
 
 class InvalidReferenceException(DomainValidationException):
@@ -72,3 +78,15 @@ def get_pump_model(self, reference: str) -> YamlPumpChartSingleSpeed | YamlPumpC
 
     @abc.abstractmethod
     def get_tabulated_model(self, reference: str) -> YamlTabularModel: ...
+
+    @abc.abstractmethod
+    def get_process_system(self, reference: str) -> YamlSerialProcessSystem: ...
+
+    @abc.abstractmethod
+    def get_compressor_stage(self, reference: str) -> YamlCompressorStageProcessSystem: ...
+
+    @abc.abstractmethod
+    def get_compressor(self, reference: str) -> YamlCompressor: ...
+
+    @abc.abstractmethod
+    def get_stream(self, reference: str) -> YamlInletStream: ...