Skip to content

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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
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