refactor: add get_configuration to Solution

olelod · olelod · commit 1183c3cd4d62 · 2026-03-26T12:19:50.000+01:00
refactor: extract Configurations and SimulationUnitId from ProcessRunner to break import cycle

refactor: introduce OpertaingConfiguration as union of speed-, choke-, and recircconfigurations

i#refactor: use ConfigurationValue type alias broadly
diff --git a/src/libecalc/domain/process/process_solver/anti_surge/anti_surge_strategy.py b/src/libecalc/domain/process/process_solver/anti_surge/anti_surge_strategy.py
@@ -1,7 +1,7 @@
 from abc import ABC, abstractmethod
 from collections.abc import Sequence
 
-from libecalc.domain.process.process_solver.process_runner import Configuration
+from libecalc.domain.process.process_solver.configuration import Configuration
 from libecalc.domain.process.process_solver.solver import Solution
 from libecalc.domain.process.process_solver.solvers.recirculation_solver import RecirculationConfiguration
 from libecalc.domain.process.value_objects.fluid_stream import FluidStream
diff --git a/src/libecalc/domain/process/process_solver/anti_surge/common_asv.py b/src/libecalc/domain/process/process_solver/anti_surge/common_asv.py
@@ -2,7 +2,8 @@
 
 from libecalc.domain.process.entities.process_units.compressor import Compressor
 from libecalc.domain.process.process_solver.anti_surge.anti_surge_strategy import AntiSurgeStrategy
-from libecalc.domain.process.process_solver.process_runner import Configuration, ProcessRunner
+from libecalc.domain.process.process_solver.configuration import Configuration
+from libecalc.domain.process.process_solver.process_runner import ProcessRunner
 from libecalc.domain.process.process_solver.search_strategies import BinarySearchStrategy, RootFindingStrategy
 from libecalc.domain.process.process_solver.solver import Solution
 from libecalc.domain.process.process_solver.solvers.recirculation_solver import (
diff --git a/src/libecalc/domain/process/process_solver/anti_surge/individual_asv.py b/src/libecalc/domain/process/process_solver/anti_surge/individual_asv.py
@@ -3,7 +3,8 @@
 
 from libecalc.domain.process.entities.process_units.compressor import Compressor
 from libecalc.domain.process.process_solver.anti_surge.anti_surge_strategy import AntiSurgeStrategy
-from libecalc.domain.process.process_solver.process_runner import Configuration, ProcessRunner
+from libecalc.domain.process.process_solver.configuration import Configuration
+from libecalc.domain.process.process_solver.process_runner import ProcessRunner
 from libecalc.domain.process.process_solver.solver import Solution
 from libecalc.domain.process.process_solver.solvers.recirculation_solver import RecirculationConfiguration
 from libecalc.domain.process.process_system.process_error import RateTooHighError
diff --git a/src/libecalc/domain/process/process_solver/configuration.py b/src/libecalc/domain/process/process_solver/configuration.py
@@ -0,0 +1,49 @@
+from dataclasses import dataclass
+from functools import total_ordering
+from typing import Generic, TypeVar
+
+from libecalc.domain.process.entities.shaft.shaft import ShaftId
+from libecalc.domain.process.process_system.process_system import ProcessSystemId
+from libecalc.domain.process.process_system.process_unit import ProcessUnitId
+
+T_co = TypeVar("T_co", covariant=True)
+
+SimulationUnitId = ProcessUnitId | ProcessSystemId | ShaftId
+
+
+@dataclass(frozen=True)
+class Configuration(Generic[T_co]):
+    simulation_unit_id: SimulationUnitId
+    value: T_co
+
+
+@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
+
+
+@dataclass
+class ChokeConfiguration:
+    delta_pressure: float
+
+
+@dataclass
+class RecirculationConfiguration:
+    recirculation_rate: float
+
+    def __post_init__(self):
+        self.recirculation_rate = float(self.recirculation_rate)
+
+
+OperatingConfiguration = SpeedConfiguration | ChokeConfiguration | RecirculationConfiguration
diff --git a/src/libecalc/domain/process/process_solver/multi_pressure_solver.py b/src/libecalc/domain/process/process_solver/multi_pressure_solver.py
@@ -2,13 +2,13 @@
 from typing import Final
 
 from libecalc.domain.component_validation_error import DomainValidationException
+from libecalc.domain.process.process_solver.configuration import Configuration
 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
@@ -60,14 +60,6 @@ def _validate_pressure_control_placement(segments: list[OutletPressureSolver]) -
                     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],
@@ -83,7 +75,9 @@ def find_solution(
         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))
+            speed_configuration = solution_for_segment.get_configuration(self._shaft_id)
+            assert isinstance(speed_configuration, SpeedConfiguration)
+            speed_configurations.append(speed_configuration)
             segment.runner.apply_configurations(solution_for_segment.configuration)
             current_inlet = segment.runner.run(inlet_stream=current_inlet)
 
diff --git a/src/libecalc/domain/process/process_solver/outlet_pressure_solver.py b/src/libecalc/domain/process/process_solver/outlet_pressure_solver.py
@@ -4,9 +4,10 @@
 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
+from libecalc.domain.process.process_solver.configuration import Configuration
 from libecalc.domain.process.process_solver.float_constraint import FloatConstraint
 from libecalc.domain.process.process_solver.pressure_control.pressure_control_strategy import PressureControlStrategy
-from libecalc.domain.process.process_solver.process_runner import Configuration, ProcessRunner
+from libecalc.domain.process.process_solver.process_runner import ProcessRunner
 from libecalc.domain.process.process_solver.search_strategies import BinarySearchStrategy, RootFindingStrategy
 from libecalc.domain.process.process_solver.solver import Solution
 from libecalc.domain.process.process_solver.solvers.recirculation_solver import (
diff --git a/src/libecalc/domain/process/process_solver/pressure_control/common_asv.py b/src/libecalc/domain/process/process_solver/pressure_control/common_asv.py
@@ -1,9 +1,10 @@
 from collections.abc import Sequence
 
 from libecalc.domain.process.entities.process_units.compressor import Compressor
+from libecalc.domain.process.process_solver.configuration import Configuration
 from libecalc.domain.process.process_solver.float_constraint import FloatConstraint
 from libecalc.domain.process.process_solver.pressure_control.pressure_control_strategy import PressureControlStrategy
-from libecalc.domain.process.process_solver.process_runner import Configuration, ProcessRunner
+from libecalc.domain.process.process_solver.process_runner import ProcessRunner
 from libecalc.domain.process.process_solver.search_strategies import BinarySearchStrategy, RootFindingStrategy
 from libecalc.domain.process.process_solver.solver import Solution
 from libecalc.domain.process.process_solver.solvers.downstream_choke_solver import ChokeConfiguration
diff --git a/src/libecalc/domain/process/process_solver/pressure_control/downstream_choke.py b/src/libecalc/domain/process/process_solver/pressure_control/downstream_choke.py
@@ -1,8 +1,9 @@
 from collections.abc import Sequence
 
+from libecalc.domain.process.process_solver.configuration import Configuration
 from libecalc.domain.process.process_solver.float_constraint import FloatConstraint
 from libecalc.domain.process.process_solver.pressure_control.pressure_control_strategy import PressureControlStrategy
-from libecalc.domain.process.process_solver.process_runner import Configuration, ProcessRunner
+from libecalc.domain.process.process_solver.process_runner import ProcessRunner
 from libecalc.domain.process.process_solver.solver import Solution
 from libecalc.domain.process.process_solver.solvers.downstream_choke_solver import (
     ChokeConfiguration,
diff --git a/src/libecalc/domain/process/process_solver/pressure_control/individual_asv.py b/src/libecalc/domain/process/process_solver/pressure_control/individual_asv.py
@@ -2,9 +2,10 @@
 
 from libecalc.domain.process.compressor.core.train.utils.numeric_methods import find_root
 from libecalc.domain.process.entities.process_units.compressor import Compressor
+from libecalc.domain.process.process_solver.configuration import Configuration
 from libecalc.domain.process.process_solver.float_constraint import FloatConstraint
 from libecalc.domain.process.process_solver.pressure_control.pressure_control_strategy import PressureControlStrategy
-from libecalc.domain.process.process_solver.process_runner import Configuration, ProcessRunner
+from libecalc.domain.process.process_solver.process_runner import ProcessRunner
 from libecalc.domain.process.process_solver.search_strategies import BinarySearchStrategy, RootFindingStrategy
 from libecalc.domain.process.process_solver.solver import Solution
 from libecalc.domain.process.process_solver.solvers.downstream_choke_solver import ChokeConfiguration
diff --git a/src/libecalc/domain/process/process_solver/pressure_control/pressure_control_strategy.py b/src/libecalc/domain/process/process_solver/pressure_control/pressure_control_strategy.py
@@ -1,8 +1,8 @@
 from abc import ABC, abstractmethod
 from collections.abc import Sequence
 
+from libecalc.domain.process.process_solver.configuration import Configuration
 from libecalc.domain.process.process_solver.float_constraint import FloatConstraint
-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.downstream_choke_solver import ChokeConfiguration
 from libecalc.domain.process.process_solver.solvers.recirculation_solver import RecirculationConfiguration
diff --git a/src/libecalc/domain/process/process_solver/pressure_control/upstream_choke.py b/src/libecalc/domain/process/process_solver/pressure_control/upstream_choke.py
@@ -2,9 +2,10 @@
 
 from libecalc.domain.process.compressor.core.train.utils.common import PRESSURE_CALCULATION_TOLERANCE
 from libecalc.domain.process.process_solver.boundary import Boundary
+from libecalc.domain.process.process_solver.configuration import Configuration
 from libecalc.domain.process.process_solver.float_constraint import FloatConstraint
 from libecalc.domain.process.process_solver.pressure_control.pressure_control_strategy import PressureControlStrategy
-from libecalc.domain.process.process_solver.process_runner import Configuration, ProcessRunner
+from libecalc.domain.process.process_solver.process_runner import ProcessRunner
 from libecalc.domain.process.process_solver.search_strategies import RootFindingStrategy
 from libecalc.domain.process.process_solver.solver import Solution
 from libecalc.domain.process.process_solver.solvers.downstream_choke_solver import ChokeConfiguration
diff --git a/src/libecalc/domain/process/process_solver/process_runner.py b/src/libecalc/domain/process/process_solver/process_runner.py
@@ -1,39 +1,17 @@
 import abc
 from collections.abc import Sequence
-from dataclasses import dataclass
-from typing import Generic, TypeVar
 
-from libecalc.domain.process.entities.shaft.shaft import ShaftId
-from libecalc.domain.process.process_solver.solvers.downstream_choke_solver import ChokeConfiguration
-from libecalc.domain.process.process_solver.solvers.recirculation_solver import RecirculationConfiguration
-from libecalc.domain.process.process_solver.solvers.speed_solver import SpeedConfiguration
-from libecalc.domain.process.process_system.process_system import ProcessSystemId
-from libecalc.domain.process.process_system.process_unit import ProcessUnitId
+from libecalc.domain.process.process_solver.configuration import Configuration, OperatingConfiguration, SimulationUnitId
 from libecalc.domain.process.value_objects.fluid_stream import FluidStream
 
-T_co = TypeVar("T_co", covariant=True)
-
-SimulationUnitId = ProcessUnitId | ProcessSystemId | ShaftId
-
-
-@dataclass(frozen=True)
-class Configuration(Generic[T_co]):
-    simulation_unit_id: SimulationUnitId
-    value: T_co
-
 
 class ProcessRunner(abc.ABC):
     @abc.abstractmethod
-    def apply_configuration(
-        self, configuration: Configuration[ChokeConfiguration | RecirculationConfiguration | SpeedConfiguration]
-    ):
+    def apply_configuration(self, configuration: Configuration[OperatingConfiguration]):
         """Apply the given configuration to the process system."""
         ...
 
-    def apply_configurations(
-        self,
-        configurations: Sequence[Configuration[ChokeConfiguration | RecirculationConfiguration | SpeedConfiguration]],
-    ):
+    def apply_configurations(self, configurations: Sequence[Configuration[OperatingConfiguration]]):
         """Apply the given configurations to the process system."""
         for configuration in configurations:
             self.apply_configuration(configuration)
diff --git a/src/libecalc/domain/process/process_solver/process_system_runner.py b/src/libecalc/domain/process/process_solver/process_system_runner.py
@@ -4,7 +4,8 @@
 from libecalc.domain.process.entities.process_units.recirculation_loop import RecirculationLoop
 from libecalc.domain.process.entities.shaft import Shaft
 from libecalc.domain.process.entities.shaft.shaft import ShaftId
-from libecalc.domain.process.process_solver.process_runner import Configuration, ProcessRunner, SimulationUnitId
+from libecalc.domain.process.process_solver.configuration import Configuration, SimulationUnitId
+from libecalc.domain.process.process_solver.process_runner import ProcessRunner
 from libecalc.domain.process.process_solver.solvers.downstream_choke_solver import ChokeConfiguration
 from libecalc.domain.process.process_solver.solvers.recirculation_solver import RecirculationConfiguration
 from libecalc.domain.process.process_solver.solvers.speed_solver import SpeedConfiguration
diff --git a/src/libecalc/domain/process/process_solver/solver.py b/src/libecalc/domain/process/process_solver/solver.py
@@ -1,8 +1,9 @@
 import abc
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
 from dataclasses import dataclass
-from typing import Generic, TypeVar
+from typing import Any, Generic, TypeVar
 
+from libecalc.domain.process.process_solver.configuration import Configuration, OperatingConfiguration, SimulationUnitId
 from libecalc.domain.process.value_objects.fluid_stream import FluidStream
 
 TConfiguration = TypeVar("TConfiguration")
@@ -13,6 +14,16 @@ class Solution(Generic[TConfiguration]):
     success: bool
     configuration: TConfiguration
 
+    def get_configuration(
+        self: "Solution[Sequence[Configuration[OperatingConfiguration]]]",
+        unit_id: SimulationUnitId,
+    ) -> OperatingConfiguration:
+        """Find a configuration value by unit ID."""
+        for config in self.configuration:
+            if config.simulation_unit_id == unit_id:
+                return config.value  # type: ignore[return-value]
+        raise ValueError(f"No configuration found for unit {unit_id}.")
+
 
 class Solver(abc.ABC, Generic[TConfiguration]):
     @abc.abstractmethod
diff --git a/src/libecalc/domain/process/process_solver/solvers/downstream_choke_solver.py b/src/libecalc/domain/process/process_solver/solvers/downstream_choke_solver.py
@@ -1,15 +1,10 @@
 from collections.abc import Callable
-from dataclasses import dataclass
 
+from libecalc.domain.process.process_solver.configuration import ChokeConfiguration
 from libecalc.domain.process.process_solver.solver import Solution, Solver
 from libecalc.domain.process.value_objects.fluid_stream import FluidStream
 
 
-@dataclass
-class ChokeConfiguration:
-    delta_pressure: float
-
-
 class DownstreamChokeSolver(Solver[ChokeConfiguration]):
     def __init__(self, target_pressure: float):
         self._target_pressure = target_pressure
diff --git a/src/libecalc/domain/process/process_solver/solvers/recirculation_solver.py b/src/libecalc/domain/process/process_solver/solvers/recirculation_solver.py
@@ -1,23 +1,15 @@
 from collections.abc import Callable
-from dataclasses import dataclass
 from typing import Literal
 
 from libecalc.domain.process.process_solver.boundary import Boundary
+from libecalc.domain.process.process_solver.configuration import RecirculationConfiguration
 from libecalc.domain.process.process_solver.float_constraint import FloatConstraint
 from libecalc.domain.process.process_solver.search_strategies import RootFindingStrategy, SearchStrategy
 from libecalc.domain.process.process_solver.solver import Solution, Solver
 from libecalc.domain.process.process_system.process_error import RateTooHighError, RateTooLowError
 from libecalc.domain.process.value_objects.fluid_stream import FluidStream
 
 
-@dataclass
-class RecirculationConfiguration:
-    recirculation_rate: float
-
-    def __post_init__(self):
-        self.recirculation_rate = float(self.recirculation_rate)
-
-
 class RecirculationSolver(Solver):
     def __init__(
         self,
diff --git a/src/libecalc/domain/process/process_solver/solvers/speed_solver.py b/src/libecalc/domain/process/process_solver/solvers/speed_solver.py
@@ -1,9 +1,8 @@
 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.configuration import SpeedConfiguration
 from libecalc.domain.process.process_solver.search_strategies import RootFindingStrategy, SearchStrategy
 from libecalc.domain.process.process_solver.solver import Solution, Solver
 from libecalc.domain.process.process_system.process_error import ProcessError, RateTooHighError, RateTooLowError
@@ -12,22 +11,6 @@
 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__(
         self,
diff --git a/tests/libecalc/domain/process/process_solver/test_individual_asv_solver_vs_legacy_train.py b/tests/libecalc/domain/process/process_solver/test_individual_asv_solver_vs_legacy_train.py
@@ -4,8 +4,8 @@
 from libecalc.domain.process.compressor.core.train.train_evaluation_input import CompressorTrainEvaluationInput
 from libecalc.domain.process.entities.shaft import VariableSpeedShaft
 from libecalc.domain.process.process_solver.boundary import Boundary
+from libecalc.domain.process.process_solver.configuration import Configuration
 from libecalc.domain.process.process_solver.float_constraint import FloatConstraint
-from libecalc.domain.process.process_solver.process_runner import Configuration
 from libecalc.domain.process.process_solver.solvers.recirculation_solver import RecirculationConfiguration
 from libecalc.domain.process.process_solver.solvers.speed_solver import SpeedConfiguration
 from libecalc.domain.process.value_objects.chart import ChartCurve
