feat(process_solver): MultiPressureSolver for N ordered pressure targets on a shared shaft (#1452)

Implements MultiPressureSolver for compressor trains where N segments
share a shaft and must each meet an independent outlet pressure target
(e.g. interstage pressure control).

Author: olelod

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

@@ -0,0 +1,129 @@
+from collections.abc import Sequence
+from typing import Final
+
+from libecalc.domain.component_validation_error import DomainValidationException
+from libecalc.domain.process.process_solver.float_constraint import FloatConstraint
+from libecalc.domain.process.process_solver.outlet_pressure_solver import OutletPressureSolver
+from libecalc.domain.process.process_solver.pressure_control.downstream_choke import (
+    DownstreamChokePressureControlStrategy,
+)
+from libecalc.domain.process.process_solver.pressure_control.upstream_choke import UpstreamChokePressureControlStrategy
+from libecalc.domain.process.process_solver.process_runner import Configuration
+from libecalc.domain.process.process_solver.solver import Solution
+from libecalc.domain.process.process_solver.solvers.speed_solver import SpeedConfiguration
+from libecalc.domain.process.value_objects.fluid_stream import FluidStream
+
+
+class MultiPressureSolver:
+    """Tries to find the shaft speed satisfying N ordered pressure targets, one per segment. Segments share a
+    single physical shaft. Each segment's runner covers a disjoint sub-sequence of the stream propagation chain.
+
+    Independent OutletPressureSolver per segment (sequential) against its own pressure target. This gives the
+    speed each segment would require if unconstrained. The segment requiring the highest speed is the binding
+    constraint. At binding speed, non-binding segments produce more pressure than needed and must have it reduced
+    by their pressure-control strategy.
+
+    All segments are re-run in sequence at binding_speed. Anti-surge is applied first. If a segment's outlet
+    still exceeds its target, the pressure-control strategy reduces it. The outlet of each segment feeds the inlet
+    of the next. For all but one segment, speed changed from the individual evaluations. Therefore, recirculation
+    is re-evaluated from scratch.
+    """
+
+    def __init__(
+        self,
+        segments: list[OutletPressureSolver],
+    ) -> None:
+        if len(segments) < 2:
+            raise DomainValidationException("MultiPressureSolver requires at least 2 segments.")
+        shaft_ids = {segment.shaft_id for segment in segments}
+        if len(shaft_ids) != 1:
+            raise DomainValidationException("All segments must share the same shaft_id.")
+        self._shaft_id: Final = next(iter(shaft_ids))
+        self._validate_pressure_control_placement(segments)
+        self._segments: Final = segments
+
+    @staticmethod
+    def _validate_pressure_control_placement(segments: list[OutletPressureSolver]) -> None:
+        """Upstream choke is only valid on the first segment; downstream choke only on the last."""
+        for i, segment in enumerate(segments):
+            strategy = segment.pressure_control_strategy
+            is_first = i == 0
+            is_last = i == len(segments) - 1
+            if isinstance(strategy, UpstreamChokePressureControlStrategy) and not is_first:
+                raise DomainValidationException(
+                    f"UpstreamChokePressureControlStrategy is only valid for the first segment "
+                    f"(segment {i} of {len(segments)})."
+                )
+            if isinstance(strategy, DownstreamChokePressureControlStrategy) and not is_last:
+                raise DomainValidationException(
+                    f"DownstreamChokePressureControlStrategy is only valid for the last segment "
+                    f"(segment {i} of {len(segments)})."
+                )
+
+    @staticmethod
+    def _extract_speed_configuration(solution: Solution[Sequence[Configuration]]) -> SpeedConfiguration:
+        """Find speed configuration in a sequence of configurations."""
+        for config in solution.configuration:
+            if isinstance(config.value, SpeedConfiguration):
+                return config.value
+        raise DomainValidationException("No SpeedConfiguration found in solution.")
+
+    def find_solution(
+        self,
+        pressure_targets: list[FloatConstraint],
+        inlet_stream: FluidStream,
+    ) -> Solution[Sequence[Configuration]]:
+        if len(pressure_targets) != len(self._segments):
+            raise DomainValidationException(
+                f"Number of pressure targets ({len(pressure_targets)}) must match "
+                f"number of segments ({len(self._segments)})."
+            )
+
+        speed_configurations: list[SpeedConfiguration] = []
+        current_inlet = inlet_stream
+        for segment, target in zip(self._segments, pressure_targets):
+            solution_for_segment = segment.find_solution(pressure_constraint=target, inlet_stream=current_inlet)
+            speed_configurations.append(self._extract_speed_configuration(solution_for_segment))
+            segment.runner.apply_configurations(solution_for_segment.configuration)
+            current_inlet = segment.runner.run(inlet_stream=current_inlet)
+
+        shaft_config = Configuration(
+            simulation_unit_id=self._shaft_id,
+            value=max(speed_configurations),
+        )
+        all_configurations: dict = {self._shaft_id: shaft_config}
+
+        current_inlet = inlet_stream
+        overall_success = True
+
+        for segment, target in zip(self._segments, pressure_targets):
+            segment.runner.apply_configuration(shaft_config)
+
+            segment.anti_surge_strategy.reset()
+            anti_surge_solution = segment.anti_surge_strategy.apply(inlet_stream=current_inlet)
+            segment.runner.apply_configurations(anti_surge_solution.configuration)
+            for config in anti_surge_solution.configuration:
+                all_configurations[config.simulation_unit_id] = config
+
+            outlet = segment.runner.run(inlet_stream=current_inlet)
+
+            if outlet.pressure_bara > target:
+                pressure_control_solution = segment.pressure_control_strategy.apply(
+                    target_pressure=target,
+                    inlet_stream=current_inlet,
+                )
+                for config in pressure_control_solution.configuration:
+                    all_configurations[config.simulation_unit_id] = config
+                segment.runner.apply_configurations(pressure_control_solution.configuration)
+                outlet = segment.runner.run(inlet_stream=current_inlet)
+                if not pressure_control_solution.success:
+                    overall_success = False
+            elif outlet.pressure_bara < target:
+                overall_success = False
+
+            current_inlet = outlet
+
+        return Solution(
+            success=overall_success,
+            configuration=list(all_configurations.values()),
+        )

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

@@ -1,7 +1,6 @@
 from collections.abc import Sequence
 from typing import Final
 
-from libecalc.domain.process.entities.shaft import Shaft
 from libecalc.domain.process.entities.shaft.shaft import ShaftId
 from libecalc.domain.process.process_solver.anti_surge.anti_surge_strategy import AntiSurgeStrategy
 from libecalc.domain.process.process_solver.boundary import Boundary
@@ -37,14 +36,14 @@ class OutletPressureSolver:
 
     def __init__(
         self,
-        shaft: Shaft,
+        shaft_id: ShaftId,
         runner: ProcessRunner,
         anti_surge_strategy: AntiSurgeStrategy,
         pressure_control_strategy: PressureControlStrategy,
         root_finding_strategy: RootFindingStrategy,
         speed_boundary: Boundary,
     ) -> None:
-        self._shaft: Final = shaft
+        self._shaft_id: Final = shaft_id
         self._root_finding_strategy: Final = root_finding_strategy
         self._anti_surge_strategy: Final = anti_surge_strategy
         self._simulator: Final = runner
@@ -53,6 +52,22 @@ def __init__(
 
         self._anti_surge_solution: Solution[Sequence[Configuration[RecirculationConfiguration]]] | None = None
 
+    @property
+    def runner(self) -> ProcessRunner:
+        return self._simulator
+
+    @property
+    def anti_surge_strategy(self) -> AntiSurgeStrategy:
+        return self._anti_surge_strategy
+
+    @property
+    def pressure_control_strategy(self) -> PressureControlStrategy:
+        return self._pressure_control_strategy
+
+    @property
+    def shaft_id(self) -> ShaftId:
+        return self._shaft_id
+
     def _get_initial_speed_boundary(self) -> Boundary:
         return self._speed_boundary
 
@@ -69,9 +84,7 @@ def _find_speed_solution(
         )
 
         def speed_func(configuration: SpeedConfiguration) -> FluidStream:
-            self._simulator.apply_configuration(
-                Configuration(simulation_unit_id=self._shaft.get_id(), value=configuration)
-            )
+            self._simulator.apply_configuration(Configuration(simulation_unit_id=self._shaft_id, value=configuration))
             self._anti_surge_strategy.reset()
             try:
                 return self._simulator.run(inlet_stream=inlet_stream)
@@ -103,8 +116,8 @@ def find_solution(
         """
         configurations: dict[ShaftId | ProcessUnitId | ProcessSystemId, Configuration] = {}
         speed_solution = self._find_speed_solution(pressure_constraint=pressure_constraint, inlet_stream=inlet_stream)
-        configurations[self._shaft.get_id()] = Configuration(
-            simulation_unit_id=self._shaft.get_id(),
+        configurations[self._shaft_id] = Configuration(
+            simulation_unit_id=self._shaft_id,
             value=speed_solution.configuration,
         )
 

src/libecalc/domain/process/process_solver/solvers/speed_solver.py

@@ -1,6 +1,7 @@
 import logging
 from collections.abc import Callable
 from dataclasses import dataclass
+from functools import total_ordering
 
 from libecalc.domain.process.process_solver.boundary import Boundary
 from libecalc.domain.process.process_solver.search_strategies import RootFindingStrategy, SearchStrategy
@@ -11,10 +12,21 @@
 logger = logging.getLogger(__name__)
 
 
+@total_ordering
 @dataclass
 class SpeedConfiguration:
     speed: float
 
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, SpeedConfiguration):
+            return NotImplemented
+        return self.speed == other.speed
+
+    def __lt__(self, other: object) -> bool:
+        if not isinstance(other, SpeedConfiguration):
+            return NotImplemented
+        return self.speed < other.speed
+
 
 class SpeedSolver(Solver[SpeedConfiguration]):
     def __init__(

tests/libecalc/domain/process/process_solver/conftest.py

@@ -6,6 +6,7 @@
 from libecalc.domain.process.process_solver.anti_surge.common_asv import CommonASVAntiSurgeStrategy
 from libecalc.domain.process.process_solver.anti_surge.individual_asv import IndividualASVAntiSurgeStrategy
 from libecalc.domain.process.process_solver.boundary import Boundary
+from libecalc.domain.process.process_solver.multi_pressure_solver import MultiPressureSolver
 from libecalc.domain.process.process_solver.outlet_pressure_solver import OutletPressureSolver
 from libecalc.domain.process.process_solver.pressure_control.common_asv import CommonASVPressureControlStrategy
 from libecalc.domain.process.process_solver.pressure_control.downstream_choke import (
@@ -20,6 +21,7 @@
 from libecalc.domain.process.process_solver.process_runner import ProcessRunner
 from libecalc.domain.process.process_system.process_system import ProcessSystemId
 from libecalc.domain.process.process_system.process_unit import ProcessUnit, ProcessUnitId
+from libecalc.domain.process.value_objects.chart import ChartCurve
 
 
 @pytest.fixture
@@ -70,7 +72,7 @@ def create_outlet_pressure_solver(
         speed_boundary: Boundary,
     ):
         return OutletPressureSolver(
-            shaft=shaft,
+            shaft_id=shaft.get_id(),
             runner=runner,
             anti_surge_strategy=anti_surge_strategy,
             pressure_control_strategy=pressure_control_strategy,
@@ -191,3 +193,33 @@ def create(
         )
 
     return create
+
+
+@pytest.fixture
+def multi_pressure_solver_factory():
+    def create_multi_pressure_solver(
+        segments: list[OutletPressureSolver],
+    ) -> MultiPressureSolver:
+        return MultiPressureSolver(segments=segments)
+
+    return create_multi_pressure_solver
+
+
+def make_variable_speed_chart_data(chart_data_factory, *, min_rate, max_rate, head_hi, head_lo, eff):
+    return chart_data_factory.from_curves(
+        curves=[
+            ChartCurve(
+                speed_rpm=75.0,
+                rate_actual_m3_hour=[min_rate, max_rate],
+                polytropic_head_joule_per_kg=[head_hi, head_lo],
+                efficiency_fraction=[eff, eff],
+            ),
+            ChartCurve(
+                speed_rpm=105.0,
+                rate_actual_m3_hour=[min_rate, max_rate],
+                polytropic_head_joule_per_kg=[head_hi * 1.05, head_lo * 1.05],
+                efficiency_fraction=[eff, eff],
+            ),
+        ],
+        control_margin=0.0,
+    )