integrated pathsim-chem

example_graphs/festim_two_walls.json

@@ -18,6 +18,8 @@
         "n_vertices": "10",
         "surface_area": "",
         "temperature": "300",
+        "T": "0.1",
+        "tau": "0",
         "thickness": "2",
         "nodeColor": "#DDE6ED"
       },
@@ -63,6 +65,8 @@
         "surface_area": "",
         "temperature": "300",
         "thickness": "2",
+        "T": "0.1",
+        "tau": "0",
         "nodeColor": "#DDE6ED"
       },
       "measured": {

pyproject.toml

@@ -25,6 +25,7 @@ classifiers = [
 requires-python = ">=3.8"
 dependencies = [
     "pathsim>=0.8.2",
+    "pathsim-chem",
     "matplotlib>=3.7.0",
     "numpy>=1.24.0",
     "plotly>=6.0",

src/python/custom_pathsim_blocks.py

@@ -1,7 +1,7 @@
-from pathsim.blocks import Block, ODE, Wrapper
+from pathsim.blocks import ODE, Wrapper
+from pathsim_chem import Splitter
 import pathsim.blocks
 import pathsim.events
-from pathsim import Subsystem, Interface, Connection
 import numpy as np
 
 
@@ -41,33 +41,14 @@ def update(self, t):
         self.outputs.update_from_array(outputs)
 
 
-class Splitter(Block):
-    """Splitter block that splits the input signal into multiple outputs based on specified fractions."""
-
-    def __init__(self, n=2, fractions=None):
-        super().__init__()
-        self.n = n  # number of splits
-        self.fractions = np.ones(n) / n if fractions is None else np.array(fractions)
-        assert len(self.fractions) == n, "Fractions must match number of outputs"
-        assert np.isclose(np.sum(self.fractions), 1), (
-            f"Fractions must sum to 1, not {np.sum(self.fractions)}"
-        )
-
-    def update(self, t):
-        # get the input from port '0'
-        u = self.inputs[0]
-        # mult by fractions and update outputs
-        self.outputs.update_from_array(self.fractions * u)
-
-
 class Splitter2(Splitter):
     _port_map_out = {"source1": 0, "source2": 1}
 
     def __init__(self, f1=0.5, f2=0.5):
         """
         Splitter with two outputs, fractions are f1 and f2.
         """
-        super().__init__(n=2, fractions=[f1, f2])
+        super().__init__(fractions=[f1, f2])
 
 
 class Splitter3(Splitter):
@@ -77,7 +58,7 @@ def __init__(self, f1=1 / 3, f2=1 / 3, f3=1 / 3):
         """
         Splitter with three outputs, fractions are f1, f2 and f3.
         """
-        super().__init__(n=3, fractions=[f1, f2, f3])
+        super().__init__(fractions=[f1, f2, f3])
 
 
 class Integrator(pathsim.blocks.Integrator):
@@ -154,147 +135,6 @@ def func_act(_):
         return [event]
 
 
-# BUBBLER SYSTEM
-
-
-class Bubbler(Subsystem):
-    """
-    Subsystem representing a tritium bubbling system with 4 vials.
-
-    Args:
-        conversion_efficiency: Conversion efficiency from insoluble to soluble (between 0 and 1).
-        vial_efficiency: collection efficiency of each vial (between 0 and 1).
-        replacement_times: List of times at which each vial is replaced. If None, no replacement
-            events are created. If a single value is provided, it is used for all vials.
-            If a single list of floats is provided, it will be used for all vials.
-            If a list of lists is provided, each sublist corresponds to the replacement times for each vial.
-    """
-
-    vial_efficiency: float
-    conversion_efficiency: float
-    n_soluble_vials: float
-    n_insoluble_vials: float
-
-    _port_map_out = {
-        "vial1": 0,
-        "vial2": 1,
-        "vial3": 2,
-        "vial4": 3,
-        "sample_out": 4,
-    }
-    _port_map_in = {
-        "sample_in_soluble": 0,
-        "sample_in_insoluble": 1,
-    }
-
-    def __init__(
-        self,
-        conversion_efficiency=0.9,
-        vial_efficiency=0.9,
-        replacement_times=None,
-    ):
-        self.reset_times = replacement_times
-        self.n_soluble_vials = 2
-        self.n_insoluble_vials = 2
-        self.vial_efficiency = vial_efficiency
-        col_eff1 = Splitter(n=2, fractions=[vial_efficiency, 1 - vial_efficiency])
-        vial_1 = pathsim.blocks.Integrator()
-        col_eff2 = Splitter(n=2, fractions=[vial_efficiency, 1 - vial_efficiency])
-        vial_2 = pathsim.blocks.Integrator()
-
-        conversion_eff = Splitter(
-            n=2, fractions=[conversion_efficiency, 1 - conversion_efficiency]
-        )
-
-        col_eff3 = Splitter(n=2, fractions=[vial_efficiency, 1 - vial_efficiency])
-        vial_3 = pathsim.blocks.Integrator()
-        col_eff4 = Splitter(n=2, fractions=[vial_efficiency, 1 - vial_efficiency])
-        vial_4 = pathsim.blocks.Integrator()
-
-        add1 = pathsim.blocks.Adder()
-        add2 = pathsim.blocks.Adder()
-        interface = Interface()
-
-        self.vials = [vial_1, vial_2, vial_3, vial_4]
-
-        blocks = [
-            vial_1,
-            col_eff1,
-            vial_2,
-            col_eff2,
-            conversion_eff,
-            vial_3,
-            col_eff3,
-            vial_4,
-            col_eff4,
-            add1,
-            add2,
-            interface,
-        ]
-        connections = [
-            Connection(interface[0], col_eff1),
-            Connection(col_eff1[0], vial_1),
-            Connection(col_eff1[1], col_eff2),
-            Connection(col_eff2[0], vial_2),
-            Connection(col_eff2[1], conversion_eff),
-            Connection(conversion_eff[0], add1[0]),
-            Connection(conversion_eff[1], add2[0]),
-            Connection(interface[1], add1[1]),
-            Connection(add1, col_eff3),
-            Connection(col_eff3[0], vial_3),
-            Connection(col_eff3[1], col_eff4),
-            Connection(col_eff4[0], vial_4),
-            Connection(col_eff4[1], add2[1]),
-            Connection(vial_1, interface[0]),
-            Connection(vial_2, interface[1]),
-            Connection(vial_3, interface[2]),
-            Connection(vial_4, interface[3]),
-            Connection(add2, interface[4]),
-        ]
-        super().__init__(blocks, connections)
-
-    def _create_reset_events_one_vial(
-        self, block, reset_times
-    ) -> pathsim.events.ScheduleList:
-        def reset_itg(_):
-            block.reset()
-
-        event = pathsim.events.ScheduleList(times_evt=reset_times, func_act=reset_itg)
-        return event
-
-    def create_reset_events(self) -> list[pathsim.events.ScheduleList]:
-        """Create reset events for all vials based on the replacement times.
-
-        Raises:
-            ValueError: If reset_times is not valid.
-
-        Returns:
-            list of reset events.
-        """
-        reset_times = self.reset_times
-        events = []
-        # if reset_times is a single list use it for all vials
-        if reset_times is None:
-            return events
-        if isinstance(reset_times, (int, float)):
-            reset_times = [reset_times]
-        # if it's a flat list use it for all vials
-        elif isinstance(reset_times, list) and all(
-            isinstance(t, (int, float)) for t in reset_times
-        ):
-            reset_times = [reset_times] * len(self.vials)
-        elif isinstance(reset_times, np.ndarray) and reset_times.ndim == 1:
-            reset_times = [reset_times.tolist()] * len(self.vials)
-        elif isinstance(reset_times, list) and len(reset_times) != len(self.vials):
-            raise ValueError(
-                "reset_times must be a single value or a list with the same length as the number of vials"
-            )
-        for i, vial in enumerate(self.vials):
-            events.append(self._create_reset_events_one_vial(vial, reset_times[i]))
-
-        return events
-
-
 # FESTIM wall
 from pathsim.utils.register import Register
 

src/python/pathsim_utils.py

@@ -47,13 +47,12 @@
 from pathsim.blocks.noise import WhiteNoise, PinkNoise
 from .custom_pathsim_blocks import (
     Process,
-    Splitter,
     Splitter2,
     Splitter3,
-    Bubbler,
     FestimWall,
     Integrator,
 )
+from pathsim_chem import Bubbler4, Splitter
 import inspect
 
 NAME_TO_SOLVER = {
@@ -119,7 +118,7 @@
     "function2to2": Function,
     "delay": Delay,
     "ode": ODE,
-    "bubbler": Bubbler,
+    "bubbler": Bubbler4,
     "wall": FestimWall,
     "white_noise": WhiteNoise,
     "pink_noise": PinkNoise,

src/python/templates/template_with_macros.py

@@ -3,6 +3,7 @@
 import numpy as np
 import matplotlib.pyplot as plt
 import pathview
+import pathsim_chem
 {# Import macros #}
 {% from 'block_macros.py' import create_block, create_integrator_block, create_bubbler_block, create_connections, create_event -%}
 

test/test_backend.py

@@ -3,9 +3,9 @@
     Process,
     Splitter2,
     Splitter3,
-    Bubbler,
     Integrator,
 )
+from pathsim_chem import Bubbler4
 
 import pathsim.blocks
 
@@ -137,7 +137,7 @@ def _create_node(block_type: str, id: str = "1", data_overrides: dict = None):
         ("splitter3", Splitter3),
         ("white_noise", pathsim.blocks.noise.WhiteNoise),
         ("pink_noise", pathsim.blocks.noise.PinkNoise),
-        ("bubbler", Bubbler),
+        ("bubbler", Bubbler4),
         ("integrator", Integrator),
         ("scope", pathsim.blocks.Scope),
     ],
@@ -163,7 +163,7 @@ def test_auto_block_construction(node_factory, block_type, expected_class):
         ("process", Process),
         ("white_noise", pathsim.blocks.noise.WhiteNoise),
         ("pink_noise", pathsim.blocks.noise.PinkNoise),
-        ("bubbler", Bubbler),
+        ("bubbler", Bubbler4),
         ("integrator", Integrator),
     ],
 )

test/test_custom_blocks.py

@@ -1,11 +1,10 @@
 import pathsim.blocks
 from pathsim import Simulation, Connection
-from pathview.custom_pathsim_blocks import Bubbler
+from pathsim_chem import Bubbler4
 
 
 def test_bubbler():
-    my_bubbler = Bubbler()
-
+    my_bubbler = Bubbler4()
     source_soluble = pathsim.blocks.Constant(1)
     source_insoluble = pathsim.blocks.Constant(0.5)
     environment = pathsim.blocks.Integrator()