Source code for openpnm.algorithms._algorithm

import logging
import numpy as np
from openpnm.core import ParserMixin, LabelMixin, Base2
from openpnm.utils import Docorator


__all__ = ['Algorithm']


logger = logging.getLogger(__name__)
docstr = Docorator()


@docstr.get_sections(base='AlgorithmSettings', sections=docstr.all_sections)
@docstr.dedent
class AlgorithmSettings:
    r"""

    Parameters
    ----------
    %(BaseSettings.parameters)s

    """



[docs]
@docstr.get_sections(base='Algorithm', sections=['Parameters'])
@docstr.dedent
class Algorithm(ParserMixin, LabelMixin, Base2):
    r"""
    Generic class to define the foundation of Algorithms

    Parameters
    ----------
    %(Base.parameters)s

    """

    def __init__(self, network, name='alg_?', **kwargs):
        super().__init__(network=network, name=name, **kwargs)
        self.settings._update(AlgorithmSettings())
        self['pore.all'] = np.ones([network.Np, ], dtype=bool)
        self['throat.all'] = np.ones([network.Nt, ], dtype=bool)

    # @functools.cached_property
    @property
    def iterative_props(self):
        r"""
        Finds and returns properties that need to be iterated while
        running the algorithm.
        """
        import networkx as nx
        phase = self.project[self.settings.phase]
        # Generate global dependency graph
        dg = nx.compose_all([x.models.dependency_graph(deep=True)
                             for x in [phase]])
        variable_props = self.settings["variable_props"].copy()
        variable_props.add(self.settings["quantity"])
        base = list(variable_props)
        # Find all props downstream that depend on base props
        dg = nx.DiGraph(nx.edge_dfs(dg, source=base))
        if len(dg.nodes) == 0:
            return []
        iterative_props = list(nx.dag.lexicographical_topological_sort(dg))
        # "variable_props" should be in the returned list but not "quantity"
        if self.settings.quantity in iterative_props:
            iterative_props.remove(self.settings["quantity"])
        return iterative_props

    def _update_iterative_props(self, iterative_props=None):
        """
        Regenerates phase, geometries, and physics objects using the
        current value of ``quantity``.

        Notes
        -----
        The algorithm directly writes the value of 'quantity' into the
        phase, which is against one of the OpenPNM rules of objects not
        being able to write into each other.

        """
        if iterative_props is None:
            iterative_props = self.iterative_props
        if not iterative_props:
            return
        # Fetch objects associated with the algorithm
        phase = self.project[self.settings.phase]
        # Update 'quantity' on phase with the most recent value
        quantity = self.settings['quantity']
        phase[quantity] = self.x
        # Regenerate all associated objects
        phase.regenerate_models(propnames=iterative_props)


[docs]
    def clear_BCs(self, bctype=[]):
        r"""
        Clear all BCs of the given type(s)

        Parameters
        ----------
        bctype : str or list(str)
            The specific type of boundary condition to clear. If not provided
            all types will be cleared.
        """
        self.set_BC(pores=None, bctype=bctype, mode='remove')



[docs]
    def set_BC(self, pores=None, bctype=[], bcvalues=[], mode='add'):
        r"""
        The main method for setting and adjusting boundary conditions.

        This method is called by other more convenient wrapper functions like
        ``set_value_BC``.

        Parameters
        ----------
        pores : array_like
            The pores where the boundary conditions should be applied. If
            ``None`` is given then *all* pores are assumed.  This is useful
            when ``mode='remove'``.
        bctype : str
            Specifies the type or the name of boundary condition to apply. This
            can be anything, but normal options are 'rate' and 'value'. If
            an empty list is provided, then all bc types will be assumed. This
            is useful for clearing all bcs if ``mode='remove'`` and ``pores=
            None``.
        bcvalues : int or array_like
            The boundary value to apply, such as concentration or rate.
            If a single value is given, it's assumed to apply to all
            locations. Different values can be applied to all pores in the
            form of an array of the same length as ``pores``. Note that using
            ``mode='add'`` and ``values=np.nan`` is equivalent to removing
            bcs from the given ``pores``.
        mode : str or list of str, optional
            Controls how the boundary conditions are applied. Options are:

            ============ =====================================================
            mode         meaning
            ============ =====================================================
            'add'        (default) Adds the supplied boundary conditions to
                         the given locations. Raises an exception if values
                         of any type already exist in the given locations.
            'overwrite'  Adds supplied boundary conditions to the given
                         locations, including overwriting conditions of the
                         given type or any other type that may be present in
                         the given locations.
            'remove'     Removes boundary conditions of the specified type
                         from the specified locations. If ``bctype`` is not
                         specified then *all* types are removed. If no
                         locations are given then values are remvoed from
                         *all* locations.
            ============ =====================================================

            If a list of strings is provided, then each mode in the list is
            handled in order, so that ``['remove', 'add']`` will give the same
            results add ``'overwrite'``.

        Notes
        -----
        It is not possible to have multiple boundary conditions for a
        specified location in one algorithm.

        """
        # If a list of modes was given, handle them each in order
        if not isinstance(mode, str):
            for item in mode:
                self.set_BC(pores=pores, bctype=bctype,
                            bcvalues=bcvalues, mode=item)
            return
        # If a list of bctypes was given, handle them each in order
        if len(bctype) == 0:
            bctype = self['pore.bc'].keys()
        if not isinstance(bctype, str):
            for item in bctype:
                self.set_BC(pores=pores, bctype=item,
                            bcvalues=bcvalues, mode=mode)
            return

        # Begin method
        bc_types = list(self['pore.bc'].keys())
        other_types = np.setdiff1d(bc_types, bctype).tolist()

        mode = self._parse_mode(
            mode,
            allowed=['overwrite', 'add', 'remove'],
            single=True)

        # Infer the value that indicates "no bc" based on array dtype
        no_bc = get_no_bc(self[f'pore.bc.{bctype}'])

        if pores is None:
            pores = self.Ps
        pores = self._parse_indices(pores)

        # Deal with size of the given bcvalues
        values = np.array(bcvalues)
        if values.size == 1:
            values = np.ones_like(pores, dtype=values.dtype)*values
        # Ensure values and pores are the same size
        if values.size > 1 and values.size != pores.size:
            raise Exception('The number of values must match the number of locations')

        # Finally adjust the BCs according to mode
        if mode == 'add':
            mask = np.ones_like(pores, dtype=bool)  # Indices of pores to keep
            for item in self['pore.bc'].keys():  # Remove pores that are taken
                mask[isfinite(self[f'pore.bc.{item}'][pores])] = False
            if not np.all(mask):  # Raise exception if some conflicts found
                msg = "Some of the given locations already have BCs, " \
                    + "either use mode='remove' first or " \
                    + "use mode='overwrite' instead"
                raise Exception(msg)
            self[f"pore.bc.{bctype}"][pores[mask]] = values[mask]
        elif mode == 'overwrite':
            # Put given values in specified BC, sort out conflicts below
            self[f"pore.bc.{bctype}"][pores] = values
            # Collect indices that are present for other BCs for removal
            mask = np.ones_like(pores, dtype=bool)
            for item in other_types:
                self[f"pore.bc.{item}"][pores] = get_no_bc(self[f"pore.bc.{item}"])
                # Make a note of any BCs values of other types
                mask[isfinite(self[f'pore.bc.{item}'][pores])] = False
            if not np.all(mask):  # Warn that other values were overwritten
                msg = 'Some of the given locations already have BCs of ' \
                    + 'another type, these will be overwritten'
                logger.warning(msg)
        elif mode == 'remove':
            self[f"pore.bc.{bctype}"][pores] = no_bc




def get_no_bc(arr):
    no_bc = np.nan if arr.dtype in (float, int) else False
    return no_bc


def isfinite(arr, inf=None):
    if arr.dtype in (bool, ):
        results = arr == True
    elif arr.dtype in (float, ):
        results = ~np.isnan(arr)
    else:
        results = arr != inf
    return results