chore: introduce StreamPropagator

Common StreamPropagator interface to better capture propagate_stream behavior.

Also make ProcessSystem an interface, since we want different kinds of ProcessSystems, recirculation loop and compressor train.

Author: jsolaas

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

@@ -1,17 +1,16 @@
+from libecalc.domain.component_validation_error import DomainValidationException
 from libecalc.domain.process.entities.process_units.mixer.mixer import Mixer
 from libecalc.domain.process.entities.process_units.splitter.splitter import Splitter
 from libecalc.domain.process.process_system.process_system import ProcessSystem
-from libecalc.domain.process.process_system.process_unit import ProcessUnit, ProcessUnitId
+from libecalc.domain.process.process_system.process_unit import ProcessUnitId
 from libecalc.domain.process.value_objects.fluid_stream import FluidService, FluidStream
 
-InnerProcess = ProcessSystem | ProcessUnit
 
-
-class RecirculationLoop(ProcessUnit):
+class RecirculationLoop(ProcessSystem):
     def __init__(
         self,
         process_unit_id: ProcessUnitId,
-        inner_process: InnerProcess,
+        inner_process: ProcessSystem,
         fluid_service: FluidService,
         recirculation_rate: float = 0,
     ):
@@ -22,18 +21,16 @@ def __init__(
         self._validate_inner_process()
 
     def _validate_inner_process(self):
-        if isinstance(self._inner_process, Splitter):
-            raise ValueError("Inner process cannot be a splitter")
-        if isinstance(self._inner_process, Mixer):
-            raise ValueError("Inner process cannot be a mixer")
-        if isinstance(self._inner_process, ProcessSystem) and self._inner_process.has_multiple_streams:
-            raise ValueError("Inner process cannot have multiple streams")
+        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, Splitter | Mixer):
+                raise DomainValidationException("Recirculation loop cannot contain splitters or mixers")
 
     def get_id(self) -> ProcessUnitId:
         return self._id
 
-    def get_inner_process(self) -> InnerProcess:
-        return self._inner_process
+    def get_process_units(self):
+        return self._inner_process.get_process_units()
 
     def set_recirculation_rate(self, rate: float):
         self._recirculation_rate = rate

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

@@ -26,8 +26,8 @@
 from libecalc.domain.process.process_solver.solvers.speed_solver import SpeedConfiguration, SpeedSolver
 from libecalc.domain.process.process_system.compressor_stage_process_unit import CompressorStageProcessUnit
 from libecalc.domain.process.process_system.process_error import RateTooLowError
-from libecalc.domain.process.process_system.process_system import ProcessSystem
 from libecalc.domain.process.process_system.process_unit import create_process_unit_id
+from libecalc.domain.process.process_system.serial_process_system import SerialProcessSystem
 from libecalc.domain.process.value_objects.fluid_stream import FluidService, FluidStream
 
 
@@ -63,8 +63,8 @@ def __init__(
             [
                 RecirculationLoop(
                     process_unit_id=create_process_unit_id(),
-                    inner_process=ProcessSystem(
-                        process_units=[compressor],
+                    inner_process=SerialProcessSystem(
+                        propagators=[compressor],
                     ),
                     fluid_service=self._fluid_service,
                 )
@@ -74,8 +74,8 @@ def __init__(
             else [
                 RecirculationLoop(
                     process_unit_id=create_process_unit_id(),
-                    inner_process=ProcessSystem(
-                        process_units=self._compressors,
+                    inner_process=SerialProcessSystem(
+                        propagators=self._compressors,
                     ),
                     fluid_service=self._fluid_service,
                 )

src/libecalc/domain/process/process_system/process_system.py

@@ -1,37 +1,10 @@
-from collections.abc import Sequence
+import abc
 
-from libecalc.domain.process.entities.process_units.mixer.mixer import Mixer
-from libecalc.domain.process.entities.process_units.splitter.splitter import Splitter
-from libecalc.domain.process.process_system.process_unit import ProcessUnit
-from libecalc.domain.process.value_objects.fluid_stream import FluidStream
+from libecalc.domain.process.process_system.stream_propagator import StreamPropagator
 
 PORT_ID = str
 
 
-class ProcessSystem:
-    def __init__(
-        self,
-        process_units: Sequence[ProcessUnit],
-    ):
-        self._process_units = process_units
-
-    def get_process_units(self):
-        return self._process_units
-
-    def propagate_stream(self, inlet_stream: FluidStream) -> FluidStream:
-        current_inlet = inlet_stream
-        for process_unit in self._process_units:
-            current_inlet = process_unit.propagate_stream(inlet_stream=current_inlet)
-        return current_inlet
-
-    @property
-    def contains_splitter(self):
-        return any(isinstance(process_unit, Splitter) for process_unit in self._process_units)
-
-    @property
-    def contains_mixer(self):
-        return any(isinstance(process_unit, Mixer) for process_unit in self._process_units)
-
-    @property
-    def has_multiple_streams(self):
-        return self.contains_splitter or self.contains_mixer
+class ProcessSystem(StreamPropagator, abc.ABC):
+    @abc.abstractmethod
+    def get_process_units(self): ...

src/libecalc/domain/process/process_system/process_unit.py

@@ -3,7 +3,7 @@
 from typing import NewType
 from uuid import UUID
 
-from libecalc.domain.process.value_objects.fluid_stream import FluidStream
+from libecalc.domain.process.process_system.stream_propagator import StreamPropagator
 
 ProcessUnitId = NewType("ProcessUnitId", UUID)
 
@@ -12,9 +12,6 @@ def create_process_unit_id() -> ProcessUnitId:
     return ProcessUnitId(uuid.uuid4())
 
 
-class ProcessUnit(abc.ABC):
+class ProcessUnit(StreamPropagator, abc.ABC):
     @abc.abstractmethod
     def get_id(self) -> ProcessUnitId: ...
-
-    @abc.abstractmethod
-    def propagate_stream(self, inlet_stream: FluidStream) -> FluidStream: ...

src/libecalc/domain/process/process_system/serial_process_system.py

@@ -0,0 +1,28 @@
+from collections.abc import Sequence
+
+from libecalc.domain.process.process_system.process_system import ProcessSystem
+from libecalc.domain.process.process_system.process_unit import ProcessUnit
+from libecalc.domain.process.value_objects.fluid_stream import FluidStream
+
+
+class SerialProcessSystem(ProcessSystem):
+    def __init__(
+        self,
+        propagators: Sequence[ProcessUnit | ProcessSystem],
+    ):
+        self._propagators = propagators
+
+    def get_process_units(self):
+        process_units = []
+        for propagator in self._propagators:
+            if isinstance(propagator, ProcessSystem):
+                process_units.extend(propagator.get_process_units())
+            else:
+                process_units.append(propagator)
+        return process_units
+
+    def propagate_stream(self, inlet_stream: FluidStream) -> FluidStream:
+        current_inlet = inlet_stream
+        for process_unit in self._propagators:
+            current_inlet = process_unit.propagate_stream(inlet_stream=current_inlet)
+        return current_inlet

src/libecalc/domain/process/process_system/stream_propagator.py

@@ -0,0 +1,8 @@
+import abc
+
+from libecalc.domain.process.value_objects.fluid_stream import FluidStream
+
+
+class StreamPropagator(abc.ABC):
+    @abc.abstractmethod
+    def propagate_stream(self, inlet_stream: FluidStream) -> FluidStream: ...

tests/libecalc/conftest.py

@@ -22,7 +22,8 @@
 from libecalc.domain.process.entities.process_units.temperature_setter import TemperatureSetter
 from libecalc.domain.process.entities.shaft import Shaft, SingleSpeedShaft, VariableSpeedShaft
 from libecalc.domain.process.process_system.process_system import ProcessSystem
-from libecalc.domain.process.process_system.process_unit import ProcessUnitId
+from libecalc.domain.process.process_system.process_unit import ProcessUnit, ProcessUnitId
+from libecalc.domain.process.process_system.serial_process_system import SerialProcessSystem
 from libecalc.domain.process.value_objects.chart import ChartCurve
 from libecalc.domain.process.value_objects.chart.chart import Chart, ChartData
 from libecalc.domain.process.value_objects.chart.compressor import CompressorChart
@@ -202,12 +203,26 @@ def create_liquid_remover(process_unit_id: ProcessUnitId = None):
 
 
 @pytest.fixture
-def recirculation_loop_factory(fluid_service):
+def process_system_factory(compressor_stage_factory, fluid_service):
+    def create_process_system(
+        process_units: list[ProcessUnit],
+    ):
+        return SerialProcessSystem(
+            propagators=process_units,
+        )
+
+    return create_process_system
+
+
+@pytest.fixture
+def recirculation_loop_factory(fluid_service, process_system_factory):
     def create_recirculation_loop(
-        inner_process: ProcessSystem,
+        inner_process: ProcessSystem | ProcessUnit,
         process_unit_id: ProcessUnitId = None,
         recirculation_rate: float = 0,
     ):
+        if isinstance(inner_process, ProcessUnit):
+            inner_process = process_system_factory([inner_process])
         return RecirculationLoop(
             inner_process=inner_process,
             process_unit_id=process_unit_id or uuid.uuid4(),

tests/libecalc/domain/process/conftest.py

@@ -13,7 +13,6 @@
 from libecalc.domain.process.process_solver.stream_constraint import PressureStreamConstraint
 from libecalc.domain.process.process_system.compressor_stage_process_unit import CompressorStageProcessUnit
 from libecalc.domain.process.process_system.process_error import OutsideCapacityError, RateTooHighError, RateTooLowError
-from libecalc.domain.process.process_system.process_system import ProcessSystem
 from libecalc.domain.process.process_system.process_unit import ProcessUnit, ProcessUnitId, create_process_unit_id
 from libecalc.domain.process.value_objects.chart.chart import ChartData
 from libecalc.domain.process.value_objects.chart.chart_area_flag import ChartAreaFlag
@@ -235,20 +234,6 @@ def propagate_stream(self, inlet_stream: FluidStream) -> FluidStream:
         return result.outlet_stream
 
 
-@pytest.fixture
-def process_system_factory(compressor_stage_factory, fluid_service):
-    def create_process_system(
-        process_units: list[ProcessUnit] | None = None,
-    ):
-        if process_units is None:
-            process_units = [simple_process_unit_factory(fluid_service)]
-        return ProcessSystem(
-            process_units=process_units,
-        )
-
-    return create_process_system
-
-
 @pytest.fixture
 def stream_constraint_factory():
     def create_stream_constraint(pressure: float):

tests/libecalc/domain/process/entities/process_units/test_recirculation_loop.py

@@ -1,8 +1,9 @@
 import pytest
+from inline_snapshot import snapshot
 
+from libecalc.domain.component_validation_error import DomainValidationException
 from libecalc.domain.process.entities.process_units.mixer.mixer import Mixer
 from libecalc.domain.process.entities.process_units.splitter.splitter import Splitter
-from libecalc.domain.process.process_system.process_system import ProcessSystem
 from libecalc.domain.process.value_objects.fluid_stream import (
     Fluid,
     FluidModel,
@@ -47,30 +48,43 @@ def fluid_service() -> FluidService:
     return DummyFluidService()
 
 
-def test_recirculation_loop_around_splitter_raises_valueerror(fluid_service, recirculation_loop_factory):
+@pytest.mark.inlinesnapshot
+@pytest.mark.snapshot
+def test_recirculation_loop_around_splitter_raises_exception(fluid_service, recirculation_loop_factory):
     process_unit = Splitter(number_of_outputs=2)
-    with pytest.raises(ValueError, match="Inner process cannot be a splitter"):
+    with pytest.raises(Exception) as exc_info:
         recirculation_loop_factory(inner_process=process_unit)
 
+    assert str(exc_info.value) == snapshot("Recirculation loop should contain a ProcessSystem")
 
-def test_recirculation_loop_around_mixer_raises_valueerror(
+
+@pytest.mark.inlinesnapshot
+@pytest.mark.snapshot
+def test_recirculation_loop_around_mixer_raises_exception(
     fluid_service,
     recirculation_loop_factory,
 ):
     process_unit = Mixer(number_of_inputs=2, fluid_service=fluid_service)
-    with pytest.raises(ValueError, match="Inner process cannot be a mixer"):
+    with pytest.raises(Exception) as exc_info:
         recirculation_loop_factory(inner_process=process_unit)
 
+    assert str(exc_info.value) == snapshot("Recirculation loop should contain a ProcessSystem")
+
 
-def test_recirculation_loop_around_process_system_with_multiple_streams_raises_valueerror(
+@pytest.mark.inlinesnapshot
+@pytest.mark.snapshot
+def test_recirculation_loop_around_process_system_with_multiple_streams_raises_exception(
     choke_factory,
     liquid_remover_factory,
     recirculation_loop_factory,
     fluid_service,
+    process_system_factory,
 ):
     liquid_remover = liquid_remover_factory()
     choke = choke_factory(pressure_change=2)
     mixer = Mixer(number_of_inputs=2, fluid_service=fluid_service)
-    process_system = ProcessSystem(process_units=[liquid_remover, choke, mixer])
-    with pytest.raises(ValueError, match="Inner process cannot have multiple streams"):
+    process_system = process_system_factory(process_units=[liquid_remover, choke, mixer])
+    with pytest.raises(DomainValidationException) as exc_info:
         recirculation_loop_factory(inner_process=process_system)
+
+    assert str(exc_info.value) == snapshot("Recirculation loop cannot contain splitters or mixers")