Skip to content

Trasformers

Transformer

The transformer is a high-level class that hold at least two transformation pipelines

  • One related to the transformation of the input of the model
  • The other related to the target of the model.

It allows accessing the information of the transformed data and is the object that uses the dataset iterators to transform the data before feeding it to the model.

Transformer

Transform each life

The transformer class is the highest level class of the transformer API. It contains Transformation Pipelines for the input data and the target, and provides mechanism to inspect the structure of the transformed data.

Parameters:

pipelineX: Pipeline that will be applied to the run-to-cycle data
pipelineY: Pipeline that will be applied to the target.
pipelineMetadata: Pipeline that will be used to extract additional
                    data from the lives information, by default None
Source code in ceruleo/transformation/functional/transformers.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
class Transformer:
    """Transform each life

    The transformer class is the highest level class of the transformer API.
    It contains Transformation Pipelines for the input data and the target,
    and provides mechanism to inspect the structure of the transformed data.

    Parameters:

        pipelineX: Pipeline that will be applied to the run-to-cycle data
        pipelineY: Pipeline that will be applied to the target.
        pipelineMetadata: Pipeline that will be used to extract additional
                            data from the lives information, by default None
    """

    def __init__(
        self,
        pipelineX: Union[Pipeline, TransformerStep],
        pipelineY: Optional[Union[Pipeline, TransformerStep]] = None,
        pipelineMetadata: Optional[Union[Pipeline, TransformerStep]] = None,
        cache_type: CacheStoreType = CacheStoreType.MEMORY,
    ):
        def ensure_pipeline(x, cache_type: CacheStoreType):
            if isinstance(x, Pipeline):
                return x
            return Pipeline(x, cache_type=cache_type)
        self.cache_type = cache_type
        self.pipelineX = ensure_pipeline(pipelineX, cache_type)
        if pipelineY is not None:
            self.pipelineY = ensure_pipeline(pipelineY, cache_type)
        else:
            self.pipelineY = None
        self.pipelineMetadata = (
            ensure_pipeline(pipelineMetadata, cache_type)
            if pipelineMetadata is not None
            else None
        )
        self.features = None
        self.fitted_ = False

    def _process_selected_features(self):
        if self.pipelineX["selector"] is not None:
            selected_columns = self.pipelineX["selector"].get_support(indices=True)
            self.features = [self.features[i] for i in selected_columns]

    def clone(self):
        return copy.deepcopy(self)

    def fit(self, dataset, show_progress: bool = False):
        """Fit the transformer with a given dataset.

        The transformer will fit the X transformer,
        the Y transformer and the metadata transformer

        Parameters:
            dataset:

        """
        logger.debug("Fitting Transformer")

        self.pipelineX.fit(dataset, show_progress=show_progress)
        if self.pipelineY is not None:
            self.pipelineY.fit(dataset, show_progress=show_progress)
        if self.pipelineMetadata is not None:
            self.pipelineMetadata.fit(dataset)

        if not isinstance(dataset, pd.DataFrame):
            self.minimal_df = dataset[0].head(n=20)
        else:
            self.minimal_df = dataset.head(n=20)
        X = self.pipelineX.transform(self.minimal_df)
        self.number_of_features_ = X.shape[1]
        self.fitted_ = True
        self.column_names = self._compute_column_names()
        return self

    def transform(self, life: pd.DataFrame):
        """Transform a life and obtain the input data, the target and the metadata

        Parameters:
            life: A life in a form of a DataFrame

        Returns:

            Tuple[np.array, np.array, np.array]
                * The first element consists of the input transformed
                * The second element consits of the target transformed
                * The third element consists of the metadata
        """
        check_is_fitted(self, "fitted_")
        return (
            self.transformX(life),
            self.transformY(life),
            self.transformMetadata(life),
        )

    def fit_map(self, dataset, show_progress: bool = False) -> "TransformedDataset":
        self.fit(dataset, show_progress=show_progress)
        return dataset.map(self)

    def transformMetadata(self, df: pd.DataFrame) -> Optional[any]:
        if self.pipelineMetadata is not None:
            return self.pipelineMetadata.transform(df)
        else:
            return None

    def transformY(self, life: pd.DataFrame) -> np.array:
        """Get the transformed target from a life

        Parameters

        life: A run-to-failrue cycle in a form of a DataFrame

        Returns
            t: Target obtained from the life
        """
        if self.pipelineY is not None:
            return self.pipelineY.transform(life)
        else:
            return None

    def transformX(self, life: pd.DataFrame) -> np.array:
        """Get the transformer input data

        Parameters

            life: A life i an form of a DataFrame

        Returns

            t: Input data transformed
        """
        return self.pipelineX.transform(life)

    def columns(self) -> List[str]:
        """Columns names after transformation

        Returns:

            c: columns
        """
        return self.column_names

    @property
    def n_features(self) -> int:
        """Number of features after transformation

        Returns:

            n: Number of features
        """
        return self.number_of_features_

    def _compute_column_names(self):
        return self.pipelineX.column_names

    def description(self):
        return {
            "features": self.features,
            "pipelineX": transformer_info(self.pipelineX),
            "pipelineY": transformer_info(self.pipelineY),
        }

    def __str__(self):
        return str(self.description())

    def get_params(self, deep: bool = False):
        params = {
            "pipelineX": self.pipelineX,
            "pipelineY": self.pipelineY,
            "pipelineMetadata": self.pipelineMetadata,
            "cache_type": self.cache_type,
        }
        if deep:
            paramsX = self.pipelineX.get_params(deep)
            paramsY = self.pipelineY.get_params(deep)
            for k in paramsX.keys():
                params[f"pipeline_X__{k}"] = paramsX[k]
            for k in paramsY.keys():
                params[f"pipeline_Y__{k}"] = paramsY[k]

        return params

    def set_params(self, **params):
        pipeline_X_params = {}
        pipeline_Y_params = {}
        for k in params.keys():
            if k.starts_with("pipeline_X__"):
                new_key = "__".join(k.split("__")[1:])
                pipeline_X_params[new_key] = params[k]
            elif k.starts_with("pipeline_Y__"):
                new_key = "__".join(k.split("__")[1:])
                pipeline_Y_params[new_key] = params[k]
        self.pipelineX = self.pipelineX.set_params(pipeline_X_params)
        self.pipelineY = self.pipelineY.set_params(pipeline_Y_params)
        return self

n_features: int property

Number of features after transformation

Returns:

n: Number of features

columns()

Columns names after transformation

Returns:

c: columns
Source code in ceruleo/transformation/functional/transformers.py 
187
188
189
190
191
192
193
194
def columns(self) -> List[str]:
    """Columns names after transformation

    Returns:

        c: columns
    """
    return self.column_names

fit(dataset, show_progress=False)

Fit the transformer with a given dataset.

The transformer will fit the X transformer, the Y transformer and the metadata transformer

Parameters:

Name Type Description Default
dataset
required
Source code in ceruleo/transformation/functional/transformers.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def fit(self, dataset, show_progress: bool = False):
    """Fit the transformer with a given dataset.

    The transformer will fit the X transformer,
    the Y transformer and the metadata transformer

    Parameters:
        dataset:

    """
    logger.debug("Fitting Transformer")

    self.pipelineX.fit(dataset, show_progress=show_progress)
    if self.pipelineY is not None:
        self.pipelineY.fit(dataset, show_progress=show_progress)
    if self.pipelineMetadata is not None:
        self.pipelineMetadata.fit(dataset)

    if not isinstance(dataset, pd.DataFrame):
        self.minimal_df = dataset[0].head(n=20)
    else:
        self.minimal_df = dataset.head(n=20)
    X = self.pipelineX.transform(self.minimal_df)
    self.number_of_features_ = X.shape[1]
    self.fitted_ = True
    self.column_names = self._compute_column_names()
    return self

transform(life)

Transform a life and obtain the input data, the target and the metadata

Parameters:

Name Type Description Default
life DataFrame

A life in a form of a DataFrame

 required 
Tuple[np.array, np.array, np.array]
    * The first element consists of the input transformed
    * The second element consits of the target transformed
    * The third element consists of the metadata
Source code in ceruleo/transformation/functional/transformers.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def transform(self, life: pd.DataFrame):
    """Transform a life and obtain the input data, the target and the metadata

    Parameters:
        life: A life in a form of a DataFrame

    Returns:

        Tuple[np.array, np.array, np.array]
            * The first element consists of the input transformed
            * The second element consits of the target transformed
            * The third element consists of the metadata
    """
    check_is_fitted(self, "fitted_")
    return (
        self.transformX(life),
        self.transformY(life),
        self.transformMetadata(life),
    )

transformX(life)

Get the transformer input data

Parameters

life: A life i an form of a DataFrame

Returns

t: Input data transformed
Source code in ceruleo/transformation/functional/transformers.py 
174
175
176
177
178
179
180
181
182
183
184
185
def transformX(self, life: pd.DataFrame) -> np.array:
    """Get the transformer input data

    Parameters

        life: A life i an form of a DataFrame

    Returns

        t: Input data transformed
    """
    return self.pipelineX.transform(life)

transformY(life)

Get the transformed target from a life

Parameters

life: A run-to-failrue cycle in a form of a DataFrame

Returns t: Target obtained from the life

Source code in ceruleo/transformation/functional/transformers.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def transformY(self, life: pd.DataFrame) -> np.array:
    """Get the transformed target from a life

    Parameters

    life: A run-to-failrue cycle in a form of a DataFrame

    Returns
        t: Target obtained from the life
    """
    if self.pipelineY is not None:
        return self.pipelineY.transform(life)
    else:
        return None

TransformerIdentity(rul_column='RUL')

Return the Transformer

Parameters:

rul_column : Name of the RUL Column

Returns:

TransformerIdentity: An identity f(x)=x transformer
Source code in ceruleo/transformation/functional/transformers.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def TransformerIdentity(rul_column: str = "RUL") -> Transformer:
    """Return the Transformer

    Parameters:

        rul_column : Name of the RUL Column

    Returns:

        TransformerIdentity: An identity f(x)=x transformer
    """
    from ceruleo.transformation.features.selection import ByNameFeatureSelector
    from ceruleo.transformation.utils import IdentityTransformerStep

    return Transformer(
        IdentityTransformerStep(), ByNameFeatureSelector(features=[rul_column])
    )

transformer_info(transformer)

Obtains the transformer information in a serializable format

Parameters:

transformer: The transformer step, or pipeline to obtain their underlying information

Returns:

Type Description

dict

 
ValueError
    If the transformer passed as an argument doesn't have
    the get_params method.
Source code in ceruleo/transformation/functional/transformers.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def transformer_info(transformer: Optional[Pipeline]):
    """Obtains the transformer information in a serializable format

    Parameters:

        transformer: The transformer step, or pipeline to obtain their underlying information

    Returns:
        dict

    Raises:

        ValueError
            If the transformer passed as an argument doesn't have
            the get_params method.
    """
    if transformer is None:
        return "Missing"

    data = []
    Q = topological_sort_iterator(transformer)
    for q in Q:
        data.append(q.description())
    return data

Transformer Step

Transformer step is the base class of all transformers

The pipeline will use the steps to fit and transform the run-to-failure cycles

TransformerStep

Bases: TransformerStepMixin, TransformerMixin

Base class of all transformation step

Source code in ceruleo/transformation/functional/transformerstep.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class TransformerStep(TransformerStepMixin, TransformerMixin):
    """Base class of all transformation step

    """
    def partial_fit(self, X:pd.DataFrame, y=None) -> "TransformerStep":
        """Fit a single run-to-failure cycle

        Parameters:

            X: Features of the run-to-failure cycle


        Returns:
            TransformerStep: The same step
        """
        return self

    def fit(self, X, y=None)  -> "TransformerStep":
        """Fit the complete set of run-to-failure cycles

        Parameters:

            X: Features of the all the run-to-failure cycles


        Returns:
            TransformerStep: The same step
        """
        return self

    def find_feature(self, X: pd.DataFrame, name: str) -> Optional[str]:
        """Find the feature that best maches the columns in X

        Parameters:
            X: A run-to-failure cycle
            name: The name of the feature to find

        Returns:
            The name of the columns if it was found, else None

        """
        matches = [c for c in X.columns if name in c]
        if len(matches) > 0:
            return matches[0]
        else:
            return None

    def description(self):
        return f"{self.name}"




    def __add__(self, other):
        from ceruleo.transformation.features.operations import Sum
        from ceruleo.transformation.utils import ensure_step
        return Sum()([self, ensure_step(other)])

    def __truediv__(self, other):
        from ceruleo.transformation.features.operations import Divide
        from ceruleo.transformation.utils import ensure_step
        return Divide()([self, ensure_step(other)])

find_feature(X, name)

Find the feature that best maches the columns in X

Parameters:

Name Type Description Default
X DataFrame

A run-to-failure cycle

 required
name str

The name of the feature to find

 required

Returns:

Type Description
Optional[str]

The name of the columns if it was found, else None
Source code in ceruleo/transformation/functional/transformerstep.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def find_feature(self, X: pd.DataFrame, name: str) -> Optional[str]:
    """Find the feature that best maches the columns in X

    Parameters:
        X: A run-to-failure cycle
        name: The name of the feature to find

    Returns:
        The name of the columns if it was found, else None

    """
    matches = [c for c in X.columns if name in c]
    if len(matches) > 0:
        return matches[0]
    else:
        return None

fit(X, y=None)

Fit the complete set of run-to-failure cycles

Parameters:

X: Features of the all the run-to-failure cycles

Returns:

Name Type Description
TransformerStep TransformerStep

The same step
Source code in ceruleo/transformation/functional/transformerstep.py 
31
32
33
34
35
36
37
38
39
40
41
42
def fit(self, X, y=None)  -> "TransformerStep":
    """Fit the complete set of run-to-failure cycles

    Parameters:

        X: Features of the all the run-to-failure cycles


    Returns:
        TransformerStep: The same step
    """
    return self

partial_fit(X, y=None)

Fit a single run-to-failure cycle

Parameters:

X: Features of the run-to-failure cycle

Returns:

Name Type Description
TransformerStep TransformerStep

The same step
Source code in ceruleo/transformation/functional/transformerstep.py 
18
19
20
21
22
23
24
25
26
27
28
29
def partial_fit(self, X:pd.DataFrame, y=None) -> "TransformerStep":
    """Fit a single run-to-failure cycle

    Parameters:

        X: Features of the run-to-failure cycle


    Returns:
        TransformerStep: The same step
    """
    return self