Skip to content

Documentation for Switch

This class encapsulates several PipelineElements that belong at the same step of the pipeline, competing for being the best choice.

If for example you want to find out if Preprocessing A or Preprocessing B is better at this position in the pipe. Or you want to test if a random forest outperforms the good old SVM.

ATTENTION: This class is a construct that may be convenient but is not suitable for any complex optimizations. Currently optimization works for grid_search, random search and smac and the specializes switch optimizer.

Examples:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
from photonai.base import PipelineElement, Switch
from photonai.optimization import IntegerRange
# Estimator Switch
svm = PipelineElement('SVC',
                      hyperparameters={'kernel': ['rbf', 'linear']})

tree = PipelineElement('DecisionTreeClassifier',
                       hyperparameters={'min_samples_split': IntegerRange(2, 5),
                                        'min_samples_leaf': IntegerRange(1, 5),
                                        'criterion': ['gini', 'entropy']})

my_pipe += Switch('EstimatorSwitch', [svm, tree])
Source code in photonai/base/photon_elements.py
class Switch(PipelineElement):
    """
    This class encapsulates several PipelineElements that
    belong at the same step of the pipeline, competing for
    being the best choice.

    If for example you want to find out if Preprocessing A
    or Preprocessing B is better at this position in the pipe.
    Or you want to test if a random forest outperforms the good old SVM.

    ATTENTION: This class is a construct that may be convenient
    but is not suitable for any complex optimizations.
    Currently optimization works for grid_search, random search and
    smac and the specializes switch optimizer.

    Example:
        ``` python
        from photonai.base import PipelineElement, Switch
        from photonai.optimization import IntegerRange
        # Estimator Switch
        svm = PipelineElement('SVC',
                              hyperparameters={'kernel': ['rbf', 'linear']})

        tree = PipelineElement('DecisionTreeClassifier',
                               hyperparameters={'min_samples_split': IntegerRange(2, 5),
                                                'min_samples_leaf': IntegerRange(1, 5),
                                                'criterion': ['gini', 'entropy']})

        my_pipe += Switch('EstimatorSwitch', [svm, tree])
        ```

    """

    def __init__(self, name: str, elements: List[PipelineElement] = None, estimator_name: str = ''):
        """
        Creates a new Switch object and generated the hyperparameter combination grid.

        Parameters:
            name:
                How the element is called in the pipeline.

            elements:
                The competing pipeline elements.

            estimator_name:
                -

        """
        self._name = name
        self.initial_name = self._name
        self.sklearn_name = self.name + "__current_element"
        self._hyperparameters = {}
        self._current_element = (1, 1)
        self.pipeline_element_configurations = []
        self.base_element = None
        self.disabled = False
        self.test_disabled = False
        self.batch_size = 0
        self.estimator_name = estimator_name

        self.needs_y = True
        self.needs_covariates = True
        # we assume we test models against each other, but only guessing
        self.is_estimator = True
        self.is_transformer = True
        self.identifier = "SWITCH:"
        self._random_state = False

        self.elements_dict = {}

        if elements:
            self.elements = elements
            self.generate_private_config_grid()
            for p_element in elements:
                self.elements_dict[p_element.name] = p_element
        else:
            self.elements = []

    def __iadd__(self, pipeline_element: PipelineElement):
        """
        Add a new estimator or transformer object to the switch container.
        All items change their positions during testing.

        Parameters:
            pipeline_element:
                Item that should be tested against other
                competing elements at that position in the pipeline.

        """
        super(Switch, self).__iadd__(pipeline_element)
        self.elements_dict[pipeline_element.name] = pipeline_element
        self.generate_private_config_grid()
        return self

    def add(self, pipeline_element: PipelineElement):
        """
        Add a new estimator or transformer object to the switch container.
        All items change their positions during testing.

        Parameters:
            pipeline_element:
                Item that should be tested against other
                competing elements at that position in the pipeline.

        """
        self.__iadd__(pipeline_element)

    @property
    def hyperparameters(self):
        # Todo: return actual hyperparameters of all pipeline elements??
        return self._hyperparameters

    @hyperparameters.setter
    def hyperparameters(self, value):
        pass

    def generate_private_config_grid(self):
        # reset
        self.pipeline_element_configurations = []

        # calculate anew
        hyperparameters = []
        # generate possible combinations for each item respectively - do not mix hyperparameters across items
        for i, pipe_element in enumerate(self.elements):
            # distinct_values_config = create_global_config([pipe_element])
            # add pipeline switch name in the config so that the hyperparameters can be set from other classes
            # pipeline switch will give the hyperparameters to the respective child
            # distinct_values_config_copy = {}
            # for config_key, config_value in distinct_values_config.items():
            #     distinct_values_config_copy[self.name + "__" + config_key] = config_value

            if hasattr(pipe_element, 'generate_config_grid'):
                element_configurations = pipe_element.generate_config_grid()
                final_configuration_list = []
                if len(element_configurations) == 0:
                    final_configuration_list.append({})
                # else:
                for dict_item in element_configurations:
                    # copy_of_dict_item = {}
                    # for key, value in dict_item.items():
                    #     copy_of_dict_item[self.name + '__' + key] = value
                    final_configuration_list.append(dict(dict_item))

            self.pipeline_element_configurations.append(final_configuration_list)
            hyperparameters += [(i, nr) for nr in range(len(final_configuration_list))]

        self._hyperparameters = {self.sklearn_name: hyperparameters}

    @property
    def current_element(self):
        return self._current_element

    @current_element.setter
    def current_element(self, value):
        self._current_element = value
        self.base_element = self.elements[self.current_element[0]]

    def get_params(self, deep: bool=True):
        if self.base_element:
            return self.base_element.get_params(deep)
        else:
            return {}

    def set_params(self, **kwargs):
        """
        The optimization process sees the amount of possible
        combinations and chooses one of them. Then this class activates
        the belonging element and prepared the element with the
        particular chosen configuration.

        """
        config_nr = None
        config = None
        self.estimator_name = ''
        # copy dict for adaptations
        params = dict(kwargs)

        # in case we are operating with grid search
        if self.sklearn_name in params:
            config_nr = params[self.sklearn_name]
        elif 'current_element' in params:
            config_nr = params['current_element']

        if "estimator_name" in kwargs:
            self.estimator_name = params["estimator_name"]
            del params["estimator_name"]
            self.base_element = self.elements_dict[self.estimator_name]

        if params is not None:
            config = params

        # todo: raise Warning that Switch could not identify which estimator to set when estimator
        #  has no params to optimize

        # in case we are operating with grid search or any derivates
        if config_nr is not None:

            if not isinstance(config_nr, (tuple, list)):
                logger.error('ValueError: current_element must be of type Tuple')
                raise ValueError('current_element must be of type Tuple')

            # grid search hack
            self.current_element = config_nr
            config = self.pipeline_element_configurations[config_nr[0]][config_nr[1]]
        # if we don't use the specialized switch optimizer
        # we need to identify the element to activate by checking for which element the optimizer gave params
        elif not self.estimator_name:
            # ugly hack because subscription is somehow not possible, we use the for loop but break
            for kwargs_key, kwargs_value in params.items():
                first_element_name = kwargs_key.split("__")[0]
                self.base_element = self.elements_dict[first_element_name]
                break

        # so now the element to be activated is found and taken care of,
        # let's move on to give the base element the config to set
        if config:
            # remove name
            unnamed_config = {}
            for config_key, config_value in config.items():
                key_split = config_key.split('__')
                unnamed_config['__'.join(key_split[1::])] = config_value
            self.base_element.set_params(**unnamed_config)
        return self

    def copy_me(self):

        ps = Switch(self.name)
        ps._random_state = self._random_state
        for element in self.elements:
            new_element = element.copy_me()
            ps += new_element
        ps._current_element = self._current_element
        return ps

    def prettify_config_output(self, config_name, config_value, return_dict=False) -> str:
        """
        Makes the sklearn configuration dictionary human readable.

        Returns:
            Configuration as prettified string or configuration as
            dict with prettified keys.

        """
        if isinstance(config_value, tuple):
            output = self.pipeline_element_configurations[config_value[0]][config_value[1]]
            if not output:
                if return_dict:
                    return {self.elements[config_value[0]].name: None}
                else:
                    return self.elements[config_value[0]].name
            else:
                if return_dict:
                    return output
                return str(output)
        else:
            return super(Switch, self).prettify_config_output(config_name, config_value)

    def predict_proba(self, X: np.ndarray, **kwargs) -> Union[np.ndarray, None]:
        """
        Predict probabilities. Base element needs predict_proba()
        function, otherwise return None.

        Parameters:
            X:
                The array-like data with shape=[N, D], where N is the number
                of samples and D is the number of features.

            **kwargs:
                Keyword arguments, not in use yet.

        Returns:
            Probabilities.

        """
        if not self.disabled:
            if hasattr(self.base_element.base_element, 'predict_proba'):
                return self.base_element.predict_proba(X)
            else:
                return None
        return X

    def _check_hyper(self,BaseEstimator):
        pass

    def inverse_transform(self, X: np.ndarray, y: np.ndarray = None, **kwargs):
        """
        Calls inverse_transform on the base element.

        For a dimension preserving transformer without inverse,
        the value is returned untreated.

        Parameters:
            X:
                The array-like data with shape=[N, D], where N
                is the number of samples and D is the number of features.

            y:
                The truth array-like values with shape=[N], where N is
                the number of samples.

            **kwargs:
                Keyword arguments, passed to base_element.transform.

        Returns:
            (X, y, kwargs) in back-transformed version if possible.

            """
        if hasattr(self.base_element, 'inverse_transform'):
            # todo: check this
            X, y, kwargs = self.adjusted_delegate_call(self.base_element.inverse_transform, X, y, **kwargs)
        return X, y, kwargs

    @property
    def _estimator_type(self):
        estimator_types = list()
        for element in self.elements:
            estimator_types.append(getattr(element, '_estimator_type'))

        unique_types = set(estimator_types)
        if len(unique_types) > 1:
            raise NotImplementedError("Switch should only contain elements of a single type (transformer, classifier, "
                                      "regressor). Found multiple types: {}".format(unique_types))
        elif len(unique_types) == 1:
            return list(unique_types)[0]
        else:
            return

    @property
    def feature_importances_(self):
        if hasattr(self.base_element, 'feature_importances_'):
            return getattr(self.base_element, 'feature_importances_')

__iadd__(self, pipeline_element) special

Add a new estimator or transformer object to the switch container. All items change their positions during testing.

Parameters:

Name Type Description Default
pipeline_element PipelineElement

Item that should be tested against other competing elements at that position in the pipeline.

required
Source code in photonai/base/photon_elements.py
def __iadd__(self, pipeline_element: PipelineElement):
    """
    Add a new estimator or transformer object to the switch container.
    All items change their positions during testing.

    Parameters:
        pipeline_element:
            Item that should be tested against other
            competing elements at that position in the pipeline.

    """
    super(Switch, self).__iadd__(pipeline_element)
    self.elements_dict[pipeline_element.name] = pipeline_element
    self.generate_private_config_grid()
    return self

__init__(self, name, elements=None, estimator_name='') special

Creates a new Switch object and generated the hyperparameter combination grid.

Parameters:

Name Type Description Default
name str

How the element is called in the pipeline.

required
elements List[photonai.base.photon_elements.PipelineElement]

The competing pipeline elements.

None
estimator_name str

-

''
Source code in photonai/base/photon_elements.py
def __init__(self, name: str, elements: List[PipelineElement] = None, estimator_name: str = ''):
    """
    Creates a new Switch object and generated the hyperparameter combination grid.

    Parameters:
        name:
            How the element is called in the pipeline.

        elements:
            The competing pipeline elements.

        estimator_name:
            -

    """
    self._name = name
    self.initial_name = self._name
    self.sklearn_name = self.name + "__current_element"
    self._hyperparameters = {}
    self._current_element = (1, 1)
    self.pipeline_element_configurations = []
    self.base_element = None
    self.disabled = False
    self.test_disabled = False
    self.batch_size = 0
    self.estimator_name = estimator_name

    self.needs_y = True
    self.needs_covariates = True
    # we assume we test models against each other, but only guessing
    self.is_estimator = True
    self.is_transformer = True
    self.identifier = "SWITCH:"
    self._random_state = False

    self.elements_dict = {}

    if elements:
        self.elements = elements
        self.generate_private_config_grid()
        for p_element in elements:
            self.elements_dict[p_element.name] = p_element
    else:
        self.elements = []

add(self, pipeline_element)

Add a new estimator or transformer object to the switch container. All items change their positions during testing.

Parameters:

Name Type Description Default
pipeline_element PipelineElement

Item that should be tested against other competing elements at that position in the pipeline.

required
Source code in photonai/base/photon_elements.py
def add(self, pipeline_element: PipelineElement):
    """
    Add a new estimator or transformer object to the switch container.
    All items change their positions during testing.

    Parameters:
        pipeline_element:
            Item that should be tested against other
            competing elements at that position in the pipeline.

    """
    self.__iadd__(pipeline_element)

predict_proba(self, X, **kwargs)

Predict probabilities. Base element needs predict_proba() function, otherwise return None.

Parameters:

Name Type Description Default
X ndarray

The array-like data with shape=[N, D], where N is the number of samples and D is the number of features.

required
**kwargs

Keyword arguments, not in use yet.

{}

Returns:

Type Description
Optional[numpy.ndarray]

Probabilities.

Source code in photonai/base/photon_elements.py
def predict_proba(self, X: np.ndarray, **kwargs) -> Union[np.ndarray, None]:
    """
    Predict probabilities. Base element needs predict_proba()
    function, otherwise return None.

    Parameters:
        X:
            The array-like data with shape=[N, D], where N is the number
            of samples and D is the number of features.

        **kwargs:
            Keyword arguments, not in use yet.

    Returns:
        Probabilities.

    """
    if not self.disabled:
        if hasattr(self.base_element.base_element, 'predict_proba'):
            return self.base_element.predict_proba(X)
        else:
            return None
    return X