Skip to content

Resamplers

Resamplers

IndexMeanResampler

Bases: TransformerStep

Resample When the index of the run-to-failure cycle is a time feature

Parameters:

Name Type Description Default
rule str

Time frequency or rule according to which the data should be resampled

 required
Source code in ceruleo/transformation/features/resamplers.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class IndexMeanResampler(TransformerStep):
    """Resample 
    When the index of the run-to-failure cycle is a time feature

    Parameters:
        rule: Time frequency or rule according to which the data should be resampled
    """
    def __init__(self, *, rule: str,  **kwargs):
        super().__init__(**kwargs)
        self.rule = rule

    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
        """ 
        Apply the transformation to the input life

        Parameters:
            X: The input life

        Returns: 
            The resampled DataFrame
        """
        return X.resample(self.rule).mean().dropna()

transform(X)

Apply the transformation to the input life

Parameters:

Name Type Description Default
X DataFrame

The input life

 required

Returns:

Type Description
DataFrame

The resampled DataFrame
Source code in ceruleo/transformation/features/resamplers.py 
27
28
29
30
31
32
33
34
35
36
37
def transform(self, X: pd.DataFrame) -> pd.DataFrame:
    """ 
    Apply the transformation to the input life

    Parameters:
        X: The input life

    Returns: 
        The resampled DataFrame
    """
    return X.resample(self.rule).mean().dropna()

IntegerIndexResamplerTransformer

Bases: TransformerStep

Resample the time series with an integer index and interpolate linearly the values

Parameters:

Name Type Description Default
time_feature str

Time feature

 required
steps int

Number of steps

 required
drop_time_feature bool

Drop the time feature

 required
Source code in ceruleo/transformation/features/resamplers.py 
 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
class IntegerIndexResamplerTransformer(TransformerStep):
    """
    Resample the time series with an integer index and interpolate linearly the values

    Parameters:
        time_feature: Time feature
        steps: Number of steps
        drop_time_feature: Drop the time feature
    """

    def __init__(self, *args, time_feature: str, steps: int, drop_time_feature: bool):
        super().__init__(*args)
        self._time_feature_name = time_feature
        self._time_feature = None
        self.steps = steps
        self.drop_time_feature = drop_time_feature

    def partial_fit(self, X: pd.DataFrame):
        """
        Obtain the name of the feature used as time

        Parameters:
            X: The current time-series to be fitted

        Returns:
            Instance of class IntegerIndexResamplerTransformer
        """
        if self._time_feature is None:
            self._time_feature = self.find_feature(X, self._time_feature_name)
        return self

    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
        """ 
        Apply the transformation to the input life

        Parameters:
            X: The input life

        Returns: 
            The resampled DataFrame
        """
        Told = X[self._time_feature].values
        Tnew = np.arange(Told.min(), Told.max(), self.steps)
        new_columns = X.columns.values
        if self.drop_time_feature:
            new_columns = np.delete(
                new_columns, np.where(new_columns == self._time_feature)
            )

        new_X = pd.DataFrame(index=Tnew, columns=new_columns)
        for c in new_columns:
            try:
                F = interp1d(Told, X[c])
                new_X[c] = F(Tnew)
            except ValueError:
                # https://stackoverflow.com/questions/62015823/interpolating-categorical-data-in-python-nearest-previous-value/62016097#62016097
                y = X[c].values
                f = interp1d(
                    Told,
                    range(len(y)),
                    kind="nearest",
                    fill_value=(0, len(y) - 1),
                    bounds_error=False,
                )
                y_idx = f(Tnew)
                new_X[c] = [y[int(i)] for i in y_idx]
        return new_X

partial_fit(X)

Obtain the name of the feature used as time

Parameters:

Name Type Description Default
X DataFrame

The current time-series to be fitted

 required

Returns:

Type Description

Instance of class IntegerIndexResamplerTransformer
Source code in ceruleo/transformation/features/resamplers.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
def partial_fit(self, X: pd.DataFrame):
    """
    Obtain the name of the feature used as time

    Parameters:
        X: The current time-series to be fitted

    Returns:
        Instance of class IntegerIndexResamplerTransformer
    """
    if self._time_feature is None:
        self._time_feature = self.find_feature(X, self._time_feature_name)
    return self

transform(X)

Apply the transformation to the input life

Parameters:

Name Type Description Default
X DataFrame

The input life

 required

Returns:

Type Description
DataFrame

The resampled DataFrame
Source code in ceruleo/transformation/features/resamplers.py 
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
def transform(self, X: pd.DataFrame) -> pd.DataFrame:
    """ 
    Apply the transformation to the input life

    Parameters:
        X: The input life

    Returns: 
        The resampled DataFrame
    """
    Told = X[self._time_feature].values
    Tnew = np.arange(Told.min(), Told.max(), self.steps)
    new_columns = X.columns.values
    if self.drop_time_feature:
        new_columns = np.delete(
            new_columns, np.where(new_columns == self._time_feature)
        )

    new_X = pd.DataFrame(index=Tnew, columns=new_columns)
    for c in new_columns:
        try:
            F = interp1d(Told, X[c])
            new_X[c] = F(Tnew)
        except ValueError:
            # https://stackoverflow.com/questions/62015823/interpolating-categorical-data-in-python-nearest-previous-value/62016097#62016097
            y = X[c].values
            f = interp1d(
                Told,
                range(len(y)),
                kind="nearest",
                fill_value=(0, len(y) - 1),
                bounds_error=False,
            )
            y_idx = f(Tnew)
            new_X[c] = [y[int(i)] for i in y_idx]
    return new_X

SubsamplerTransformer

Bases: TransformerStep

IntegerIndexResamplerTransformer

Resample the time series with an integer index and interpolate linearly the values

Parameters:

Name Type Description Default
time_feature str

Time feature

 required
steps int

Number of steps

 required
drop_time_feature bool

Drop the time feature

 required
Source code in ceruleo/transformation/features/resamplers.py 
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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
class SubsamplerTransformer(TransformerStep):
    """IntegerIndexResamplerTransformer

    Resample the time series with an integer index and interpolate linearly the values

    Parameters:
        time_feature: Time feature
        steps:  Number of steps
        drop_time_feature: Drop the time feature
    """

    def __init__(self, *args, time_feature: str, steps: int, drop_time_feature: bool):
        super().__init__(*args)
        self._time_feature_name = time_feature
        self._time_feature = None
        self.steps = steps
        self.drop_time_feature = drop_time_feature

    def partial_fit(self, X: pd.DataFrame):
        """Obtain the name of the feature used as time

        Parameters:
            X: The current time-series to be fitted

        Returns:
            Instance of class IntegerIndexResamplerTransformer

        """
        if self._time_feature is None:
            self._time_feature = self.find_feature(X, self._time_feature_name)
            if self._time_feature is None:
                raise ValueError("Time feature not found")

        return self

    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
        """ 
        Apply the transformation to the input life

        Parameters:
            X: The input life

        Returns: 
            The resampled DataFrame
        """
        X = X.groupby(X[self._time_feature] // self.steps, sort=False).mean()
        if self.drop_time_feature:
            X = X.drop(columns=[self._time_feature])

        return X

partial_fit(X)

Obtain the name of the feature used as time

Parameters:

Name Type Description Default
X DataFrame

The current time-series to be fitted

 required

Returns:

Type Description

Instance of class IntegerIndexResamplerTransformer
Source code in ceruleo/transformation/features/resamplers.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def partial_fit(self, X: pd.DataFrame):
    """Obtain the name of the feature used as time

    Parameters:
        X: The current time-series to be fitted

    Returns:
        Instance of class IntegerIndexResamplerTransformer

    """
    if self._time_feature is None:
        self._time_feature = self.find_feature(X, self._time_feature_name)
        if self._time_feature is None:
            raise ValueError("Time feature not found")

    return self

transform(X)

Apply the transformation to the input life

Parameters:

Name Type Description Default
X DataFrame

The input life

 required

Returns:

Type Description
DataFrame

The resampled DataFrame
Source code in ceruleo/transformation/features/resamplers.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def transform(self, X: pd.DataFrame) -> pd.DataFrame:
    """ 
    Apply the transformation to the input life

    Parameters:
        X: The input life

    Returns: 
        The resampled DataFrame
    """
    X = X.groupby(X[self._time_feature] // self.steps, sort=False).mean()
    if self.drop_time_feature:
        X = X.drop(columns=[self._time_feature])

    return X