Add StreamSplitter block for ratio-based stream splitting

Author: milanofthe

src/pathsim_chem/process/stream_splitter.py

@@ -0,0 +1,56 @@
+#########################################################################################
+##
+##                              Stream Splitter Block
+##
+#########################################################################################
+
+# IMPORTS ===============================================================================
+
+from pathsim.blocks.function import Function
+
+# BLOCKS ================================================================================
+
+class StreamSplitter(Function):
+    """Algebraic stream splitter that divides one stream into two by a fixed ratio.
+
+    Named ``StreamSplitter`` to avoid collision with ``tritium.Splitter``.
+
+    Mathematical Formulation
+    -------------------------
+    .. math::
+
+        F_1 = \\mathrm{split} \\cdot F_{in}, \\quad F_2 = (1 - \\mathrm{split}) \\cdot F_{in}
+
+    .. math::
+
+        T_1 = T_2 = T_{in}
+
+    Parameters
+    ----------
+    split : float
+        Fraction of feed sent to first outlet (0 to 1). Default 0.5.
+    """
+
+    input_port_labels = {
+        "F_in": 0,
+        "T_in": 1,
+    }
+
+    output_port_labels = {
+        "F_1": 0,
+        "T_1": 1,
+        "F_2": 2,
+        "T_2": 3,
+    }
+
+    def __init__(self, split=0.5):
+        if not 0.0 <= split <= 1.0:
+            raise ValueError(f"'split' must be in [0, 1] but is {split}")
+
+        self.split = split
+        super().__init__(func=self._eval)
+
+    def _eval(self, F_in, T_in):
+        F_1 = self.split * F_in
+        F_2 = (1.0 - self.split) * F_in
+        return (F_1, T_in, F_2, T_in)

tests/process/test_stream_splitter.py

@@ -0,0 +1,106 @@
+########################################################################################
+##
+##                                  TESTS FOR
+##                        'process.stream_splitter.py'
+##
+########################################################################################
+
+# IMPORTS ==============================================================================
+
+import unittest
+
+from pathsim_chem.process import StreamSplitter
+
+
+# TESTS ================================================================================
+
+class TestStreamSplitter(unittest.TestCase):
+    """Test the stream splitter block."""
+
+    def test_init_default(self):
+        """Test default split ratio."""
+        S = StreamSplitter()
+        self.assertEqual(S.split, 0.5)
+
+    def test_init_custom(self):
+        """Test custom split ratio."""
+        S = StreamSplitter(split=0.3)
+        self.assertEqual(S.split, 0.3)
+
+    def test_init_validation(self):
+        """Test input validation."""
+        with self.assertRaises(ValueError):
+            StreamSplitter(split=-0.1)
+        with self.assertRaises(ValueError):
+            StreamSplitter(split=1.5)
+
+    def test_port_labels(self):
+        """Test port label definitions."""
+        self.assertEqual(StreamSplitter.input_port_labels["F_in"], 0)
+        self.assertEqual(StreamSplitter.input_port_labels["T_in"], 1)
+        self.assertEqual(StreamSplitter.output_port_labels["F_1"], 0)
+        self.assertEqual(StreamSplitter.output_port_labels["T_1"], 1)
+        self.assertEqual(StreamSplitter.output_port_labels["F_2"], 2)
+        self.assertEqual(StreamSplitter.output_port_labels["T_2"], 3)
+
+    def test_mass_balance(self):
+        """F_1 + F_2 should equal F_in."""
+        S = StreamSplitter(split=0.7)
+        S.inputs[0] = 10.0   # F_in
+        S.inputs[1] = 350.0  # T_in
+
+        S.update(None)
+
+        F_1 = S.outputs[0]
+        F_2 = S.outputs[2]
+        self.assertAlmostEqual(F_1 + F_2, 10.0, places=8)
+
+    def test_split_ratio(self):
+        """Check split produces correct flow fractions."""
+        S = StreamSplitter(split=0.3)
+        S.inputs[0] = 10.0
+        S.inputs[1] = 350.0
+
+        S.update(None)
+
+        self.assertAlmostEqual(S.outputs[0], 3.0)   # F_1 = 0.3 * 10
+        self.assertAlmostEqual(S.outputs[2], 7.0)   # F_2 = 0.7 * 10
+
+    def test_temperature_unchanged(self):
+        """Both outlet temperatures should equal inlet."""
+        S = StreamSplitter(split=0.4)
+        S.inputs[0] = 10.0
+        S.inputs[1] = 375.0
+
+        S.update(None)
+
+        self.assertAlmostEqual(S.outputs[1], 375.0)  # T_1
+        self.assertAlmostEqual(S.outputs[3], 375.0)  # T_2
+
+    def test_split_zero(self):
+        """With split=0, all flow goes to second outlet."""
+        S = StreamSplitter(split=0.0)
+        S.inputs[0] = 10.0
+        S.inputs[1] = 350.0
+
+        S.update(None)
+
+        self.assertAlmostEqual(S.outputs[0], 0.0)   # F_1
+        self.assertAlmostEqual(S.outputs[2], 10.0)  # F_2
+
+    def test_split_one(self):
+        """With split=1, all flow goes to first outlet."""
+        S = StreamSplitter(split=1.0)
+        S.inputs[0] = 10.0
+        S.inputs[1] = 350.0
+
+        S.update(None)
+
+        self.assertAlmostEqual(S.outputs[0], 10.0)  # F_1
+        self.assertAlmostEqual(S.outputs[2], 0.0)   # F_2
+
+
+# RUN TESTS LOCALLY ====================================================================
+
+if __name__ == '__main__':
+    unittest.main(verbosity=2)