Source code for CADETProcess.stationarity

"""
======================================================
Cyclic Stationarity (:mod:`CADETProcess.stationarity`)
======================================================

.. currentmodule:: CADETProcess.stationarity

Module to evaluate cyclic stationarity of succeeding cycles.

.. autosummary::
    :toctree: generated/

    RelativeArea
    NRMSE
    StationarityEvaluator

"""

from addict import Dict
import numpy as np

from CADETProcess import log
from CADETProcess.dataStructure import Structure, UnsignedFloat
from CADETProcess import SimulationResults
from CADETProcess.comparison import Comparator
from CADETProcess.processModel import Inlet


__all__ = ['RelativeArea', 'NRMSE', 'StationarityEvaluator']


class CriterionBase(Structure):
    threshold = UnsignedFloat(default=1e-3)

    def __str__(self):
        return self.__class__.__name__




class RelativeArea(CriterionBase):
    """Class to evaluate difference in relative area as stationarity critereon."""

    pass





class NRMSE(CriterionBase):
    """Class to evaluate NRMSE as stationarity critereon."""

    pass





class StationarityEvaluator(Comparator):
    """Class for checking two succeding chromatograms for stationarity."""

    valid_criteria = ['RelativeArea', 'NRMSE']

    def __init__(
            self,
            criteria=None,
            log_level='WARNING',
            *args, **kwargs):
        """Initialize the stationarity evaluator.

        Parameters
        ----------
        criteria : List[CriterionBase], optional
            List of criteria for stationarity evaluation, by default None
        log_level : str, optional
            The logging level, by default 'WARNING'
        args : list
            Additional arguments.
        kwargs : dict
            Additional keyword arguments.
        """
        super().__init__(*args, **kwargs)

        self.logger = log.get_logger('StationarityEvaluator', level=log_level)

        self._criteria = []

    @property
    def criteria(self):
        """list: List of criteria."""
        return self._criteria



    def add_criterion(self, criterion):
        """Add a criterion to the list of criteria.

        Parameters
        ----------
        criterion : CriterionBase
            Criterion to add to the list of criteria.

        """
        if not isinstance(criterion, CriterionBase):
            raise TypeError("Expected CriterionBase.")

        self._criteria.append(criterion)




    def assert_stationarity(self, simulation_results):
        """Check stationarity of two succeeding cycles.

        Parameters
        ----------
        simulation_results : SimulationResults
            Results of current cycle.

        Returns
        -------
        bool
            True if stationarity is reached. False otherwise.

        Raises
        ------
        TypeError
            If simulation_results is not a SimulationResults object.

        """
        self._metrics = []
        criteria = Dict()
        if not isinstance(simulation_results, SimulationResults):
            raise TypeError('Expcected SimulationResults')

        stationarity = True
        for unit, solution in simulation_results.solution_cycles.items():
            if isinstance(simulation_results.process.flow_sheet[unit], Inlet):
                continue
            solution_previous = solution.outlet[-2]
            solution_this = solution.outlet[-1]
            self.add_reference(solution_previous, update=True, smooth=False)

            for c in self.criteria:
                metric = self.add_difference_metric(
                    str(c), unit, f'{unit}.outlet', smooth=False
                )
                criteria[unit][str(c)]['threshold'] = c.threshold
                diff = metric.evaluate(solution_this)
                criteria[unit][str(c)]['metric'] = diff
                if not np.all(diff <= c.threshold):
                    s = False
                    stationarity = s
                else:
                    s = True
                criteria[unit][str(c)]['stationarity'] = s

        self.logger.debug(f'Stationrity criteria: {criteria}')

        return stationarity