fix: allow negative rates or split fractions in splitter when within floating point precision

Author: olelod

src/libecalc/domain/process/compressor/core/train/stage.py

@@ -281,7 +281,7 @@ def split(
         ]
         split_streams = self.splitter.split_stream(
             stream=inlet_stream_stage,
-            split_fractions=all_rates_to_splitter,
+            rates=all_rates_to_splitter,
         )
         return split_streams[-1]  # The last stream goes to the compressor stage
 

src/libecalc/domain/process/compressor/core/train/utils/common.py

@@ -4,6 +4,7 @@
 from libecalc.domain.process.compressor.core.train.utils.numeric_methods import DampState, adaptive_pressure_update
 from libecalc.domain.process.value_objects.fluid_stream import FluidStream
 
+FLOATING_POINT_PRECISION = 1e-9
 EPSILON = 1e-5
 PRESSURE_CALCULATION_TOLERANCE = 1e-3
 POWER_CALCULATION_TOLERANCE = 1e-3

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

@@ -1,33 +1,74 @@
+from libecalc.common.errors.exceptions import IllegalStateException
+from libecalc.common.logger import logger
+from libecalc.domain.process.compressor.core.train.utils.common import FLOATING_POINT_PRECISION
 from libecalc.domain.process.value_objects.fluid_stream import FluidStream
 
 
 class Splitter:
     def __init__(self, number_of_outputs: int):
         self.number_of_outputs = number_of_outputs
 
-    def split_stream(self, stream: FluidStream, split_fractions: list[float]) -> list[FluidStream]:
+    def split_stream(
+        self, stream: FluidStream, split_fractions: list[float] | None = None, rates: list[float] | None = None
+    ) -> list[FluidStream]:
         """
         Splits the fluid stream into streams based on the split fractions.
 
         Args:
             stream (FluidStream): The fluid stream to be split.
-            split_fractions (Sequence[float]): The fractions of the stream to go to the different output streams (0.0 to 1.0).
+            split_fractions (list[float] | None): The fractions of the stream to go to the different output streams (0.0 to 1.0).
+                The split fractions should sum to 1.0.
+            rates (list[float] | None): The flow rates for each output stream.
 
         Returns:
             list[FluidStream]: A list containing the resulting FluidStreams.
         """
+        if (split_fractions is None) == (rates is None):
+            raise IllegalStateException("Either split_fractions or rates must be provided.")
+        if rates is not None:
+            split_fractions = self.convert_rates_to_split_fractions(rates)
+
         # make sure number of split fractions matches number of outputs
+        assert split_fractions is not None
         if len(split_fractions) != self.number_of_outputs:
             raise ValueError("Number of split fractions must match number of outputs.")
 
-        # normalize split fractions
-        total = sum(split_fractions)
-        normalized_fractions = [f / total for f in split_fractions]
+        # make sure split fractions sum to 1.0
+        total_fraction = sum(split_fractions)
+        if total_fraction != 1.0:
+            raise IllegalStateException("Split fractions must sum to 1.0.")
 
         return [
             FluidStream(
                 thermo_system=stream.thermo_system,
                 mass_rate_kg_per_h=stream.mass_rate_kg_per_h * split_fraction,
             )
-            for split_fraction in normalized_fractions
+            for split_fraction in split_fractions
         ]
+
+    @staticmethod
+    def convert_rates_to_split_fractions(rates: list[float]) -> list[float]:
+        """
+        Converts a list of flow rates to split fractions. Some of the flow rates may be very small negative
+        numbers due to floating point precision issues (conversion between standard rate and mass rate and back),
+        these are corrected to zero if they are less than .
+
+        Args:
+            rates (list[float]): The flow rates for each output stream.
+
+        Returns:
+            list[float]: The corresponding split fractions.
+        """
+        if any(rate < -FLOATING_POINT_PRECISION for rate in rates):
+            raise IllegalStateException(
+                f"Negative rate found when splitting stream in compressor stage splitter. "
+                f"Rates: {rates}. Correcting negative rate to zero."
+            )
+        elif any(-FLOATING_POINT_PRECISION < rate < 0 for rate in rates):
+            logger.warning(
+                f"Negative rate found when splitting stream in compressor stage splitter. "
+                f"Rates before correction: {rates}. Correcting negative rate to zero."
+            )
+            rates = [max(rate, 0.0) for rate in rates]
+
+        return [rate / sum(rates) for rate in rates]