Documentation for `PipelineElement`

PHOTONAI wrapper class for any transformer or estimator in the pipeline.

So called PHOTONAI PipelineElements can be added to the Hyperpipe, each of them being a data-processing method or a learning algorithm. By choosing, combining data-processing methods and algorithms, and arranging them with the PHOTONAI classes, simple and complex pipeline architectures can be designed rapidly.

The PHOTONAI PipelineElement implements several helpful features:

Saves the hyperparameters that should be tested and creates a grid of all hyperparameter configurations.
Enables fast and rapid instantiation of pipeline elements per string identifier, e.g 'svc' creates an sklearn.svm.SVC object.
Attaches a "disable" switch to every element in the pipeline in order to test a complete disable.

`iadd(self, pipe_element)` `special`

Add an element to the intern list of elements.

Parameters:

Name	Type	Description	Default
`pipe_element`	`PipelineElement`	The object to add, being either a transformer or an estimator.	required

Source code in photonai/base/photon_elements.py

def __iadd__(self, pipe_element):
    """
    Add an element to the intern list of elements.

    Parameters:
        pipe_element (PipelineElement):
            The object to add, being either a transformer or an estimator.

    """
    PipelineElement.sanity_check_element_type_for_building_photon_pipes(pipe_element, type(self))

    # check if that exact instance has been added before
    already_added_objects = len([i for i in self.elements if i is pipe_element])
    if already_added_objects > 0:
        error_msg = "Cannot add the same instance twice to " + self.name + " - " + str(type(self))
        logger.error(error_msg)
        raise ValueError(error_msg)

    # check for doubled names:
    already_existing_element_with_that_name = len([i for i in self.elements if i.name == pipe_element.name])

    if already_existing_element_with_that_name > 0:
        error_msg = "Already added a pipeline element with the name " + pipe_element.name + " to " + self.name
        logger.warning(error_msg)
        warnings.warn(error_msg)

        # check for other items that have been renamed
        nr_of_existing_elements_with_that_name = len([i for i in self.elements if i.name.startswith(pipe_element.name)])
        new_name = pipe_element.name + str(nr_of_existing_elements_with_that_name + 1)
        while len([i for i in self.elements if i.name == new_name]) > 0:
            nr_of_existing_elements_with_that_name += 1
            new_name = pipe_element.name + str(nr_of_existing_elements_with_that_name + 1)
        msg = "Renaming " + pipe_element.name + " in " + self.name + " to " + new_name + " in " + self.name
        logger.warning(msg)
        warnings.warn(msg)
        pipe_element.name = new_name

    self.elements.append(pipe_element)
    return self

`init(self, name, hyperparameters=None, test_disabled=False, disabled=False, base_element=None, batch_size=0, **kwargs)` `special`

Takes a string literal and transforms it into an object of the associated class (see PhotonCore.JSON).

Parameters:

Name	Type	Description	Default
`name`	`str`	A string literal encoding the class to be instantiated.	required
`hyperparameters`	`dict`	Which values/value range should be tested for the hyperparameter. In form of Dict: parameter_name -> HyperparameterElement.	`None`
`test_disabled`	`bool`	If the hyperparameter search should evaluate a complete disabling of the element.	`False`
`disabled`	`bool`	If true, the element is currently disabled and does nothing except return the data it received.	`False`
`base_element`	`BaseEstimator`	The underlying BaseEstimator. If not given the instantiation per string identifier takes place.	`None`
`batch_size`	`int`	Size of the division on which is calculated separately.	`0`
`**kwargs`		Any parameters that should be passed to the object to be instantiated, default parameters.	`{}`

Source code in photonai/base/photon_elements.py

def __init__(self, name: str, hyperparameters: dict = None, test_disabled: bool = False,
             disabled: bool = False, base_element: BaseEstimator = None, batch_size: int = 0, **kwargs) -> None:
    """
    Takes a string literal and transforms it into an object
    of the associated class (see PhotonCore.JSON).

    Parameters:
        name:
            A string literal encoding the class to be instantiated.

        hyperparameters:
            Which values/value range should be tested for the
            hyperparameter.
            In form of Dict: parameter_name -> HyperparameterElement.

        test_disabled:
            If the hyperparameter search should evaluate a
            complete disabling of the element.

        disabled:
            If true, the element is currently disabled and
            does nothing except return the data it received.

        base_element:
            The underlying BaseEstimator. If not given the
            instantiation per string identifier takes place.

        batch_size:
            Size of the division on which is calculated separately.

        **kwargs:
            Any parameters that should be passed to the object
            to be instantiated, default parameters.

    """
    if hyperparameters is None:
        hyperparameters = {}

    if base_element is None:

        # Registering Pipeline Elements
        if len(PhotonRegistry.ELEMENT_DICTIONARY) == 0:
            registry = PhotonRegistry

        if name not in PhotonRegistry.ELEMENT_DICTIONARY:
            # try to reload
            PhotonRegistry.ELEMENT_DICTIONARY = PhotonRegistry().get_package_info()

        if name in PhotonRegistry.ELEMENT_DICTIONARY:
            try:
                desired_class_info = PhotonRegistry.ELEMENT_DICTIONARY[name]
                desired_class_home = desired_class_info[0]
                desired_class_name = desired_class_info[1]
                imported_module = importlib.import_module(desired_class_home)
                desired_class = getattr(imported_module, desired_class_name)
                self.base_element = desired_class(**kwargs)
            except AttributeError as ae:
                logger.error('ValueError: Could not find according class:'
                             + str(PhotonRegistry.ELEMENT_DICTIONARY[name]))
                raise ValueError('Could not find according class:', PhotonRegistry.ELEMENT_DICTIONARY[name])
        else:
            # if even after reload the element does not appear, it is not supported
            logger.error('Element not supported right now:' + name)
            raise NameError('Element not supported right now:', name)
    else:
        self.base_element = base_element

    self.is_transformer = hasattr(self.base_element, "transform")
    self.reduce_dimension = False  # boolean - set on transform method
    self.is_estimator = hasattr(self.base_element, "predict")
    self._name = name
    self.initial_name = str(name)
    self.kwargs = kwargs
    self.current_config = None
    self.batch_size = batch_size
    self.test_disabled = test_disabled
    self.initial_hyperparameters = dict(hyperparameters)

    self._sklearn_disabled = self.name + '__disabled'
    self._hyperparameters = hyperparameters
    if len(hyperparameters) > 0:
        key_0 = next(iter(hyperparameters))
        if self.name not in key_0:
            self.hyperparameters = hyperparameters
    else:
        self.hyperparameters = hyperparameters
    # self.initalize_hyperparameters = hyperparameters
    # check if hyperparameters are already in sklearn style

    # check if hyperparameters are members of the class
    if self.is_transformer or self.is_estimator:
        self._check_hyperparameters(BaseEstimator)

    self.disabled = disabled

    # check if self.base element needs y for fitting and transforming
    if hasattr(self.base_element, 'needs_y'):
        self.needs_y = self.base_element.needs_y
    else:
        self.needs_y = False
    # or if it maybe needs covariates for fitting and transforming
    if hasattr(self.base_element, 'needs_covariates'):
        self.needs_covariates = self.base_element.needs_covariates
    else:
        self.needs_covariates = False

    self._random_state = False

`create(name, base_element, hyperparameters, test_disabled=False, disabled=False, **kwargs)` `classmethod`

Takes an instantiated object and encapsulates it into the PHOTONAI structure. Add the disabled function and attaches information about the hyperparameters that should be tested.

Parameters:

Name	Type	Description	Default
`name`	`str`	A string literal encoding the class to be instantiated.	required
`base_element`	`BaseEstimator`	The underlying transformer or estimator class.	required
`hyperparameters`	`dict`	Which values/value range should be tested for the hyperparameter. In form of Dict: parameter_name -> HyperparameterElement.	required
`test_disabled`	`bool`	If the hyperparameter search should evaluate a complete disabling of the element.	`False`
`disabled`	`bool`	If true, the element is currently disabled and does nothing except return the data it received.	`False`
`**kwargs`		Any parameters that should be passed to the object to be instantiated, default parameters.	`{}`

Examples:

class RD(BaseEstimator, TransformerMixin):

    def fit(self, X, y, **kwargs):
        pass

    def fit_transform(self, X, y=None, **fit_params):
        return self.transform(X)

    def transform(self, X):
        return X[:, :3]

trans = PipelineElement.create('MyTransformer', base_element=RD(), hyperparameters={})

Source code in photonai/base/photon_elements.py

@classmethod
def create(cls, name: str, base_element: BaseEstimator, hyperparameters: dict, test_disabled: bool = False,
           disabled: bool = False, **kwargs):
    """
    Takes an instantiated object and encapsulates it
    into the PHOTONAI structure.
    Add the disabled function and attaches
    information about the hyperparameters that should be tested.

    Parameters:
        name:
            A string literal encoding the class to be instantiated.

        base_element:
            The underlying transformer or estimator class.

        hyperparameters:
            Which values/value range should be tested for the
            hyperparameter.
            In form of Dict: parameter_name -> HyperparameterElement.

        test_disabled:
            If the hyperparameter search should evaluate a
            complete disabling of the element.

        disabled:
            If true, the element is currently disabled and
            does nothing except return the data it received.

        **kwargs:
            Any parameters that should be passed to the object
            to be instantiated, default parameters.

    Example:
        ``` python
        class RD(BaseEstimator, TransformerMixin):

            def fit(self, X, y, **kwargs):
                pass

            def fit_transform(self, X, y=None, **fit_params):
                return self.transform(X)

            def transform(self, X):
                return X[:, :3]

        trans = PipelineElement.create('MyTransformer', base_element=RD(), hyperparameters={})
        ```

    """
    if isinstance(base_element, type):
        raise ValueError("Base element should be an instance but is a class.")
    return PipelineElement(name, hyperparameters, test_disabled, disabled, base_element=base_element, **kwargs)

`fit(self, X, y=None, **kwargs)`

Calls the fit function of the base element.

Parameters:

Name	Type	Description	Default
`X`	`ndarray`	The array-like training and test data with shape=[N, D], where N is the number of samples and D is the number of features.	required
`y`	`ndarray`	The truth array-like values with shape=[N], where N is the number of samples.	`None`
`**kwargs`		Keyword arguments, passed to base_element.predict.	`{}`

Returns:

Type	Description
	Fitted self.

Source code in photonai/base/photon_elements.py

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

    Parameters:
        X:
            The array-like training and test 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.predict.

    Returns:
        Fitted self.

    """
    if not self.disabled:
        obj = self.base_element
        arg_list = inspect.signature(obj.fit)
        if len(arg_list.parameters) > 2:
            vals = arg_list.parameters.values()
            kwargs_param = list(vals)[-1]
            if kwargs_param.kind == kwargs_param.VAR_KEYWORD:
                obj.fit(X, y, **kwargs)
                return self
        obj.fit(X, y)
    return self

`inverse_transform(self, X, y=None, **kwargs)`

Calls inverse_transform on the base element.

When the dimension is preserved: transformers without inverse returns original input.

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
`y`	`ndarray`	The truth array-like values with shape=[N], where N is the number of samples.	`None`
`**kwargs`		Keyword arguments, passed to base_element.transform.	`{}`

Returns:

Type	Description
`(<class 'numpy.ndarray'>, <class 'numpy.ndarray'>, <class 'dict'>)`	(X, y, kwargs) in back-transformed version.

Source code in photonai/base/photon_elements.py

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

    When the dimension is preserved: transformers
    without inverse returns original input.

    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.

    Raises:
        NotImplementedError:
            Thrown when there is a dimensional reduction but no inverse is defined.

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

    """
    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)
    elif self.is_transformer and self.reduce_dimension:
        msg = "{} has no inverse_transform, but element reduce dimesions.".format(self.name)
        logger.error(msg)
        raise NotImplementedError(msg)
    return X, y, kwargs

`predict(self, X, **kwargs)`

Calls the predict function of the underlying base_element.

Parameters:

Name	Type	Description	Default
`X`	`ndarray`	The array-like training and test data with shape=[N, D], where N is the number of samples and D is the number of features.	required
`**kwargs`		Keyword arguments, passed to base_element.predict.	`{}`

Returns:

Type	Description
`ndarray`	Predictions values.

Source code in photonai/base/photon_elements.py

def predict(self, X: np.ndarray, **kwargs) -> np.ndarray:
    """
    Calls the predict function of the underlying base_element.

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

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

    Returns:
        Predictions values.

    """
    if self.batch_size == 0:
        return self.__predict(X, **kwargs)
    else:
        return self.__batch_predict(self.__predict, X, **kwargs)

`score(self, X_test, y_test)`

Calls the score function on the base element.

Parameters:

Name	Type	Description	Default
`X_test`	`ndarray`	Input test data to score on.	required
`y_test`	`ndarray`	Input true targets to score on.	required

Returns:

Type	Description
`float`	A goodness of fit measure or a likelihood of unseen data.

Source code in photonai/base/photon_elements.py

def score(self, X_test: np.ndarray, y_test: np.ndarray) -> float:
    """
    Calls the score function on the base element.

    Parameters:
        X_test:
            Input test data to score on.

        y_test:
            Input true targets to score on.

    Returns:
        A goodness of fit measure or a likelihood of unseen data.

    """
    return self.base_element.score(X_test, y_test)

`transform(self, X, y=None, **kwargs)`

Calls transform on the base element.

In case there is no transform method, calls predict. This is used if we are using an estimator as a preprocessing step.

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
`y`	`ndarray`	The truth array-like values with shape=[N], where N is the number of samples.	`None`
`**kwargs`		Keyword arguments, passed to base_element.transform.	`{}`

Returns:

Type	Description
`(<class 'numpy.ndarray'>, <class 'numpy.ndarray'>, <class 'dict'>)`	(X, y) in transformed version and original kwargs.

Source code in photonai/base/photon_elements.py

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

    In case there is no transform method, calls predict.
    This is used if we are using an estimator as a preprocessing step.

    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) in transformed version and original kwargs.

    """
    if self.batch_size == 0:
        Xt, yt, kwargs = self.__transform(X, y, **kwargs)
    else:
        Xt, yt, kwargs = self.__batch_transform(X, y, **kwargs)
    if all(hasattr(data, "shape") for data in [X, Xt]) and all(len(data.shape) > 1 for data in [X, Xt]):
        self.reduce_dimension = (Xt.shape[1] < X.shape[1])
    return Xt, yt, kwargs

Documentation for PipelineElement

__iadd__(self, pipe_element) special

__init__(self, name, hyperparameters=None, test_disabled=False, disabled=False, base_element=None, batch_size=0, **kwargs) special

create(name, base_element, hyperparameters, test_disabled=False, disabled=False, **kwargs) classmethod

fit(self, X, y=None, **kwargs)

inverse_transform(self, X, y=None, **kwargs)

predict(self, X, **kwargs)

score(self, X_test, y_test)

transform(self, X, y=None, **kwargs)

Documentation for `PipelineElement`

`iadd(self, pipe_element)` `special`

`init(self, name, hyperparameters=None, test_disabled=False, disabled=False, base_element=None, batch_size=0, **kwargs)` `special`

`create(name, base_element, hyperparameters, test_disabled=False, disabled=False, **kwargs)` `classmethod`

`fit(self, X, y=None, **kwargs)`

`inverse_transform(self, X, y=None, **kwargs)`

`predict(self, X, **kwargs)`

`score(self, X_test, y_test)`

`transform(self, X, y=None, **kwargs)`