refactor(process): split SpeedSolver EOS recovery into helpers; add BinarySearchResult

Author: olelod

src/libecalc/process/process_solver/search_strategies.py

@@ -1,5 +1,6 @@
 import abc
 from collections.abc import Callable
+from typing import NamedTuple
 
 from scipy.optimize import root_scalar
 
@@ -9,6 +10,17 @@
 CONVERGENCE_TOLERANCE = 1e-5
 
 
+class BinarySearchResult(NamedTuple):
+    higher: bool
+    accepted: bool
+
+
+ACCEPT_AND_GO_HIGHER = BinarySearchResult(higher=True, accepted=True)
+REJECT_AND_GO_LOWER = BinarySearchResult(higher=False, accepted=False)
+ACCEPT_AND_GO_LOWER = BinarySearchResult(higher=False, accepted=True)
+REJECT_AND_GO_HIGHER = BinarySearchResult(higher=True, accepted=False)
+
+
 class DidNotConvergeError(EcalcError):
     def __init__(
         self,

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

@@ -9,7 +9,15 @@
 )
 from libecalc.process.process_solver.boundary import Boundary
 from libecalc.process.process_solver.configuration import SpeedConfiguration
-from libecalc.process.process_solver.search_strategies import RootFindingStrategy, SearchStrategy
+from libecalc.process.process_solver.search_strategies import (
+    ACCEPT_AND_GO_HIGHER,
+    ACCEPT_AND_GO_LOWER,
+    REJECT_AND_GO_HIGHER,
+    REJECT_AND_GO_LOWER,
+    BinarySearchResult,
+    RootFindingStrategy,
+    SearchStrategy,
+)
 from libecalc.process.process_solver.solver import (
     OutletFluidNotAchievableEvent,
     OutsideCapacityEvent,
@@ -39,20 +47,6 @@ def solve(self, func: Callable[[SpeedConfiguration], FluidStream]) -> Solution[S
         def get_outlet_stream(speed: float) -> FluidStream:
             return func(SpeedConfiguration(speed=speed))
 
-        def _fluid_not_achievable_solution(
-            e: OutletFluidNotAchievableError, configuration: SpeedConfiguration
-        ) -> Solution[SpeedConfiguration]:
-            logger.debug(f"Outlet fluid not achievable at speed {configuration.speed}", exc_info=e)
-            return Solution(
-                success=False,
-                configuration=configuration,
-                failure_event=OutletFluidNotAchievableEvent(
-                    status=SolverFailureStatus.OUTLET_FLUID_NOT_ACHIEVABLE,
-                    source_id=e.process_unit_id,
-                    unachievable_operating_point=e.unachievable_operating_point,
-                ),
-            )
-
         max_speed_configuration = SpeedConfiguration(speed=self._boundary.max)
         try:
             maximum_speed_outlet_stream = func(max_speed_configuration)
@@ -81,33 +75,11 @@ def _fluid_not_achievable_solution(
                 ),
             )
         except OutletFluidNotAchievableError:
-            # EOS fails at max speed — find the highest speed where a valid outlet can still be produced.
-            logger.debug(
-                "Outlet fluid not achievable at maximum speed %.1f — searching for highest feasible speed.",
-                self._boundary.max,
-            )
-
-            # Pre-check: if boundary.min also fails, no speed in the range produces a valid outlet.
-            try:
-                get_outlet_stream(speed=self._boundary.min)
-            except OutletFluidNotAchievableError as e:
-                return _fluid_not_achievable_solution(e, SpeedConfiguration(speed=self._boundary.min))
-
-            def max_speed_fluid_func(x: float) -> tuple[bool, bool]:
-                try:
-                    get_outlet_stream(speed=x)
-                    return True, True  # EOS succeeded — can try higher
-                except OutletFluidNotAchievableError:
-                    return False, False  # EOS failed — must go lower
-                except (RateTooHighError, RateTooLowError):
-                    return False, False  # rate error at this speed — also unusable
-
-            max_fluid_achievable_speed = self._search_strategy.search(
-                boundary=self._boundary,
-                func=max_speed_fluid_func,
-            )
-            max_speed_configuration = SpeedConfiguration(speed=max_fluid_achievable_speed)
-            maximum_speed_outlet_stream = get_outlet_stream(speed=max_fluid_achievable_speed)
+            result = self._search_highest_eos_feasible_speed(get_outlet_stream)
+            if isinstance(result, Solution):
+                return result
+            max_speed_configuration = result
+            maximum_speed_outlet_stream = get_outlet_stream(speed=result.speed)
 
         if maximum_speed_outlet_stream.pressure_bara < self._target_pressure:
             return Solution(
@@ -143,33 +115,11 @@ def bool_speed_func(x: float):
             minimum_speed_configuration = SpeedConfiguration(speed=minimum_speed_within_capacity)
             minimum_speed_outlet_stream = func(minimum_speed_configuration)
         except OutletFluidNotAchievableError:
-            # EOS fails at min speed — find the lowest speed where a valid outlet can be produced.
-            logger.debug(
-                "Outlet fluid not achievable at minimum speed %.1f — searching for lowest feasible speed.",
-                self._boundary.min,
-            )
-
-            # Pre-check: if boundary.max also fails, no speed in the range produces a valid outlet.
-            try:
-                get_outlet_stream(speed=self._boundary.max)
-            except OutletFluidNotAchievableError as e:
-                return _fluid_not_achievable_solution(e, SpeedConfiguration(speed=self._boundary.max))
-
-            def min_speed_fluid_func(x: float) -> tuple[bool, bool]:
-                try:
-                    get_outlet_stream(speed=x)
-                    return False, True  # EOS succeeded — can try lower
-                except OutletFluidNotAchievableError:
-                    return True, False  # EOS failed — must go higher
-                except (RateTooHighError, RateTooLowError):
-                    return True, False  # rate error at this speed — also unusable
-
-            min_fluid_achievable_speed = self._search_strategy.search(
-                boundary=self._boundary,
-                func=min_speed_fluid_func,
-            )
-            minimum_speed_configuration = SpeedConfiguration(speed=min_fluid_achievable_speed)
-            minimum_speed_outlet_stream = get_outlet_stream(speed=min_fluid_achievable_speed)
+            result = self._search_lowest_eos_feasible_speed(get_outlet_stream)
+            if isinstance(result, Solution):
+                return result
+            minimum_speed_configuration = result
+            minimum_speed_outlet_stream = get_outlet_stream(speed=result.speed)
 
         if minimum_speed_outlet_stream.pressure_bara > self._target_pressure:
             # Solution 2, target pressure is too low
@@ -202,6 +152,79 @@ def root_speed_func(x: float) -> float:
                 func=root_speed_func,
             )
         except OutletFluidNotAchievableError as e:
-            return _fluid_not_achievable_solution(e, SpeedConfiguration(speed=e.unachievable_operating_point.speed))
+            return self._fluid_not_achievable_solution(
+                e, SpeedConfiguration(speed=e.unachievable_operating_point.speed)
+            )
 
         return Solution(success=True, configuration=SpeedConfiguration(speed=speed))
+
+    def _fluid_not_achievable_solution(
+        self,
+        e: OutletFluidNotAchievableError,
+        configuration: SpeedConfiguration,
+    ) -> Solution[SpeedConfiguration]:
+        logger.warning(f"Outlet fluid not achievable at speed {configuration.speed}", exc_info=e)
+        return Solution(
+            success=False,
+            configuration=configuration,
+            failure_event=OutletFluidNotAchievableEvent(
+                status=SolverFailureStatus.OUTLET_FLUID_NOT_ACHIEVABLE,
+                source_id=e.process_unit_id,
+                unachievable_operating_point=e.unachievable_operating_point,
+            ),
+        )
+
+    def _search_highest_eos_feasible_speed(
+        self,
+        get_outlet_stream: Callable[[float], FluidStream],
+    ) -> SpeedConfiguration | Solution[SpeedConfiguration]:
+        """Binary-search downward for the highest speed where the EOS produces a valid outlet.
+
+        Pre-checks that boundary.min is feasible; if not, returns a failed Solution
+        (the search would otherwise exhaust iterations without converging).
+        """
+        logger.debug(
+            "Outlet fluid not achievable at maximum speed %.1f — searching for highest feasible speed.",
+            self._boundary.max,
+        )
+        try:
+            get_outlet_stream(self._boundary.min)
+        except OutletFluidNotAchievableError as e:
+            return self._fluid_not_achievable_solution(e, SpeedConfiguration(speed=self._boundary.min))
+
+        def feasible(x: float) -> BinarySearchResult:
+            try:
+                get_outlet_stream(x)
+                return ACCEPT_AND_GO_HIGHER
+            except (OutletFluidNotAchievableError, RateTooHighError, RateTooLowError):
+                return REJECT_AND_GO_LOWER
+
+        speed = self._search_strategy.search(boundary=self._boundary, func=feasible)
+        return SpeedConfiguration(speed=speed)
+
+    def _search_lowest_eos_feasible_speed(
+        self,
+        get_outlet_stream: Callable[[float], FluidStream],
+    ) -> SpeedConfiguration | Solution[SpeedConfiguration]:
+        """Binary-search upward for the lowest speed where the EOS produces a valid outlet.
+
+        Pre-checks that boundary.max is feasible; if not, returns a failed Solution.
+        """
+        logger.debug(
+            "Outlet fluid not achievable at minimum speed %.1f — searching for lowest feasible speed.",
+            self._boundary.min,
+        )
+        try:
+            get_outlet_stream(self._boundary.max)
+        except OutletFluidNotAchievableError as e:
+            return self._fluid_not_achievable_solution(e, SpeedConfiguration(speed=self._boundary.max))
+
+        def feasible(x: float) -> BinarySearchResult:
+            try:
+                get_outlet_stream(x)
+                return ACCEPT_AND_GO_LOWER
+            except (OutletFluidNotAchievableError, RateTooHighError, RateTooLowError):
+                return REJECT_AND_GO_HIGHER
+
+        speed = self._search_strategy.search(boundary=self._boundary, func=feasible)
+        return SpeedConfiguration(speed=speed)