[WIP] Overhaul Fractionator

Author: schmoelder

CADETProcess/fractionation/fractionationOptimizer.py

@@ -125,7 +125,7 @@ def _setup_fractionator(
         Fractionator
             The Fractionator object that has been set up using the provided arguments.
         """
-        frac = Fractionator(
+        frac = Fractionator.from_simulation_results(
             simulation_results,
             components=components,
             use_total_concentration_components=use_total_concentration_components,

CADETProcess/fractionation/fractionator.py

@@ -1,4 +1,3 @@
-import os
 from collections import defaultdict
 from functools import wraps
 from typing import Any, Callable, Optional
@@ -7,21 +6,20 @@
 from addict import Dict
 from matplotlib.axes import Axes
 
-from CADETProcess import CADETProcessError, plotting, settings
+from CADETProcess import CADETProcessError, plotting
 from CADETProcess.dataStructure import String
 from CADETProcess.dynamicEvents import Event, EventHandler
 from CADETProcess.fractionation.fractions import Fraction, FractionPool
 from CADETProcess.performance import Performance
-from CADETProcess.processModel import ComponentSystem, Process
-from CADETProcess.simulationResults import SimulationResults
+from CADETProcess.processModel import ComponentSystem, ProcessMetaInformation
 from CADETProcess.solution import SolutionIO, slice_solution
 
 __all__ = ["Fractionator"]
 
 
 class Fractionator(EventHandler):
     """
-    Class for Chromatogram Fractionation.
+    Class for fractionation of Chromatograms.
 
     This class is responsible for setting events for starting and ending fractionation,
     handling multiple chromatograms, and calculating various performance metrics.
@@ -33,6 +31,7 @@ class Fractionator(EventHandler):
     performance_keys : list
         Keys for performance metrics including mass, concentration, purity, recovery,
         productivity, and eluent consumption.
+
     """
 
     name = String(default="Fractionator")
@@ -47,9 +46,10 @@ class Fractionator(EventHandler):
 
     def __init__(
         self,
-        simulation_results: SimulationResults,
+        chromatograms: SolutionIO | list[SolutionIO],
+        process_meta_information: ProcessMetaInformation,
         components: Optional[list[str]] = None,
-        use_total_concentration_components: bool = True,
+        use_total_concentration_components: Optional[bool] = True,
         *args: Any,
         **kwargs: Any,
     ) -> None:
@@ -58,83 +58,75 @@ def __init__(
 
         Parameters
         ----------
-        simulation_results : SimulationResults
-            Simulation results containing chromatograms.
-        components : list, optional
-            List of components to be fractionated. Default is None.
-        use_total_concentration_components : bool, optional
-            Use total concentration components. Default is True.
+        chromatograms : SolutionIO | list[SolutionIO]
+            Chromatograms to be fractionated.
+        process_meta_information: ProcessMetaInformation
+            Process meta information.
+        components : Optional[list[str]]
+            List of components to be fractionated. If None, all components are
+            considered. The default is None.
+        use_total_concentration_components : Optional[bool]
+            If True, use the total concentration of components, i.e. the sum of all
+            subspecies of each component. The default is True.
         *args
             Variable length argument list.
         **kwargs
             Arbitrary keyword arguments.
-        """
-        self.components: Optional[list[str]] = components
-        self.use_total_concentration_components: bool = use_total_concentration_components
-        self.simulation_results = simulation_results
-
-        super().__init__(*args, **kwargs)
 
-    @property
-    def simulation_results(self) -> SimulationResults:
-        """SimulationResults: The simulation results containing the chromatograms."""
-        return self._simulation_results
-
-    @simulation_results.setter
-    def simulation_results(self, simulation_results: SimulationResults) -> None:
         """
-        Set the simulation results.
+        if not isinstance(chromatograms, list):
+            chromatograms = [chromatograms]
 
-        Parameters
-        ----------
-        simulation_results : SimulationResults
-            Simulation results containing chromatograms.
+        self.process_meta_information = process_meta_information
 
-        Raises
-        ------
-        TypeError
-            If simulation_results is not of type SimulationResults.
-        CADETProcessError
-            If the simulation results do not contain any chromatograms.
-        """
-        if not isinstance(simulation_results, SimulationResults):
-            raise TypeError("Expected SimulationResults")
+        for chrom in chromatograms:
+            if chrom.component_system is not self.component_system:
+                raise CADETProcessError("Component systems do not match.")
 
-        if len(simulation_results.chromatograms) == 0:
-            raise CADETProcessError("Simulation results do not contain chromatogram")
+        component_system = chromatograms[0].component_system
 
-        self._simulation_results = simulation_results
+        if components is not None:
+            for comp in components:
+                if comp not in component_system:
+                    raise CADETProcessError(
+                        f"Could not find component {comp} in component system."
+                    )
+
+        self.components = components
+        self.use_total_concentration_components = use_total_concentration_components
 
         self._chromatograms = [
             slice_solution(
                 chrom,
                 components=self.components,
                 use_total_concentration_components=self.use_total_concentration_components,
             )
-            for chrom in simulation_results.chromatograms
+            for chrom in chromatograms
         ]
 
         m_feed = np.zeros((self.component_system.n_comp,))
         counter = 0
-        for comp, indices in simulation_results.component_system.indices.items():
+        for comp, indices in self.component_system.indices.items():
             if comp in self.component_system.names:
-                m_feed_comp = simulation_results.process.m_feed[indices]
+                m_feed_comp = process_meta_information.m_feed[indices]
                 if self.use_total_concentration_components:
                     m_feed[counter] = np.sum(m_feed_comp)
                     counter += 1
                 else:
                     n_species = len(indices)
                     m_feed[counter : counter + n_species] = m_feed_comp
                     counter += n_species
-        self.m_feed: np.ndarray = m_feed
+        self.m_feed = m_feed
 
         self._fractionation_states = Dict({chrom: [] for chrom in self.chromatograms})
         self._chromatogram_events = Dict({chrom: [] for chrom in self.chromatograms})
 
-        self._cycle_time = self.process.cycle_time
+        self._cycle_time = self.process_meta_information.cycle_time
 
         self.reset()
 
+        super().__init__(*args, **kwargs)
+
     @property
     def component_system(self) -> ComponentSystem:
         """ComponentSystem: The component system of the chromatograms."""
@@ -196,13 +188,23 @@ def chromatogram_events(self) -> dict[SolutionIO, list[Event]]:
         return chrom_events
 
     @property
-    def process(self) -> Process:
-        """Process: The process from the simulation results."""
-        return self.simulation_results.process
+    def process_meta_information(self) -> ProcessMetaInformation:
+        """Process:  from the simulation results."""
+        return self._process_meta_information
+
+    @process_meta_information.setter
+    def process_meta_information(
+        self,
+        process_meta_information: ProcessMetaInformation
+    ) -> None:
+        if not isinstance(process_meta_information, ProcessMetaInformation):
+            raise TypeError("Expected ProcessMetaInformation.")
+
+        self._process_meta_information = process_meta_information
 
     @property
     def n_comp(self) -> int:
-        """int: Number of components to be fractionized."""
+        """int: Number of components to be fractionated."""
         return self.chromatograms[0].n_comp
 
     @property
@@ -745,41 +747,6 @@ def section_dependent_parameters(self) -> Dict:
         """dict: Section dependent parameters of the fractionator."""
         return self.parameters
 
-    def save(
-        self,
-        case_dir: str,
-        start: float = 0,
-        end: Optional[float] = None,
-    ) -> None:
-        """
-        Save chromatogram and purity plots to a specified directory.
-
-        Parameters
-        ----------
-        case_dir : str
-            Directory name within the working directory to save plots.
-        start : float, optional
-            Start time for plotting purity, default is 0.
-        end : Optional[float]
-            End time for plotting purity. If None, includes all data.
-        """
-        path = os.path.join(settings.working_directory, case_dir)
-
-        for index, chrom in enumerate(self.chromatograms):
-            chrom.plot(save_path=path + f"/chrom_{index}.png")
-            chrom.plot_purity(
-                start=start, end=end, save_path=path + "/chrom_purity.png"
-            )
-
-        for chrom in enumerate(self.chromatograms):
-            self.plot_fraction_signal(
-                chromatogram=chrom,
-                start=start,
-                end=end,
-                save_path=path + f"/fractionation_signal_{index}.png",
-                index=index,
-            )
-
     def __str__(self) -> str:
         """str: String representation of the fractionator."""
         return self.__class__.__name__

tests/fractionation/test_fractionator.py

@@ -0,0 +1,23 @@
+import unittest
+
+from CADETProcess.fractionation import Fractionator
+from CADETProcess.processModel import ComponentSystem
+from solution_fixtures import TestSolutionIOConstant, TestSolutionIOGaussian
+
+component_system = ComponentSystem(['A', 'B'])
+
+test_solution_const = TestSolutionIOConstant(component_system)
+test_solution_gauss = TestSolutionIOGaussian(component_system)
+
+
+frac = Fractionator()
+
+
+class Test_Fractionator(unittest.TestCase):
+
+    def __init__(self, methodName='runTest'):
+        super().__init__(methodName)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_fractions.py renamed to tests/fractionation/test_fractions.py

@@ -1,36 +1,37 @@
 import unittest
 
-import CADETProcess
 import numpy as np
+from CADETProcess import CADETProcessError
+from CADETProcess.fractionation import Fraction, FractionPool
 
 
 class Test_Fractions(unittest.TestCase):
     def create_fractions(self):
         m_0 = np.array([0, 0])
-        frac0 = CADETProcess.fractionation.Fraction(m_0, 1)
+        frac0 = Fraction(m_0, 1)
         m_1 = np.array([3, 0])
-        frac1 = CADETProcess.fractionation.Fraction(m_1, 2)
+        frac1 = Fraction(m_1, 2)
         m_2 = np.array([1, 2])
-        frac2 = CADETProcess.fractionation.Fraction(m_2, 3)
+        frac2 = Fraction(m_2, 3)
         m_3 = np.array([0, 2])
-        frac3 = CADETProcess.fractionation.Fraction(m_3, 3)
+        frac3 = Fraction(m_3, 3)
         m_4 = np.array([0, 0])
-        frac4 = CADETProcess.fractionation.Fraction(m_4, 0)
+        frac4 = Fraction(m_4, 0)
 
         return frac0, frac1, frac2, frac3, frac4
 
     def create_pools(self):
         fractions = self.create_fractions()
 
-        pool_waste = CADETProcess.fractionation.FractionPool(n_comp=2)
+        pool_waste = FractionPool(n_comp=2)
         pool_waste.add_fraction(fractions[0])
         pool_waste.add_fraction(fractions[1])
         pool_waste.add_fraction(fractions[2])
 
-        pool_1 = CADETProcess.fractionation.FractionPool(n_comp=2)
+        pool_1 = FractionPool(n_comp=2)
         pool_1.add_fraction(fractions[3])
 
-        pool_2 = CADETProcess.fractionation.FractionPool(n_comp=2)
+        pool_2 = FractionPool(n_comp=2)
         pool_2.add_fraction(fractions[4])
 
         return pool_waste, pool_1, pool_2
@@ -69,9 +70,9 @@ def test_n_comp(self):
         self.assertEqual(pools[0].fractions[0].n_comp, 2)
 
         m_wrong_n_comp = np.array([0, 0, 0])
-        frac_wrong_n_comp = CADETProcess.fractionation.Fraction(m_wrong_n_comp, 1)
+        frac_wrong_n_comp = Fraction(m_wrong_n_comp, 1)
 
-        with self.assertRaises(CADETProcess.CADETProcessError):
+        with self.assertRaises(CADETProcessError):
             pools[0].add_fraction(frac_wrong_n_comp)
 
     def test_pool_mass(self):

tests/test_solution.py

@@ -67,7 +67,7 @@ def __init__(self, *args, breaks=None, values=None, **kwargs):
 
         if breaks is None:
             mid = self.cycle_time / 2
-            breaks = [(mid - mid / (i+2), mid + mid / (i+2)) for i in range(self.n_comp)]
+            breaks = [(mid - mid / (i + 2), mid + mid / (i + 2)) for i in range(self.n_comp)]
 
         if values is None:
             values = np.arange(self.n_comp) + 1
@@ -89,13 +89,14 @@ def __init__(self, *args, mu=None, sigma=None, **kwargs):
         super().__init__(*args, **kwargs)
 
         if mu is None:
-            mu = [(i+1) * self.cycle_time / (self.n_comp + 1) for i in range(self.n_comp)]
+            mu = [(i + 1) * self.cycle_time / (self.n_comp + 1) for i in range(self.n_comp)]
         if sigma is None:
             sigma = np.arange(self.n_comp) + 1
 
         for comp_i, (mu_i, sigma_i) in enumerate(zip(mu, sigma)):
             self.solution[:, comp_i] = stats.norm.pdf(self.time, mu_i, sigma_i)
 
+
 comp_2 = ComponentSystem(2)
 
 comp_2_3_species = ComponentSystem()