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 |
|
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