Skip to content

gradient_aware_harmonisation.convergence#

Implements the decay/convergence methods

Classes:

Name Description
CosineDecaySplineHelper

Spline that supports being used as a cosine-decay between splines
CosineDecaySplineHelperDerivative

Derivative of CosineDecaySplineHelper

Functions:

Name Description
get_cosine_decay_harmonised_spline

Generate the harmonised spline based on a cosine-decay

CosineDecaySplineHelper #

Spline that supports being used as a cosine-decay between splines

Between initial_time and final_time, we return values based on a cosine-decay between 1 and 0 if self.apply_to_convergence is False, otherwise we return values based on a cosine-increase between 0 and 1.

Methods:

Name Description
__call__

Evaluate the spline at a given x-value
antiderivative

Calculate the anti-derivative/integral of self
derivative

Calculate the derivative of self

Attributes:

Name Type Description
apply_to_convergence bool

Is this helper being applied to the convergence spline?
final_time Union[float, int]

At and after this time, we return values from final
initial_time Union[float, int]

At and before this time, we return values from initial
Source code in src/gradient_aware_harmonisation/convergence.py 
 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
 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
@define
class CosineDecaySplineHelper:
    """
    Spline that supports being used as a cosine-decay between splines

    Between `initial_time` and `final_time`,
    we return values based on a cosine-decay between 1 and 0
    if `self.apply_to_convergence` is `False`,
    otherwise we return values based on a cosine-increase between 0 and 1.
    """

    initial_time: Union[float, int]
    """
    At and before this time, we return values from `initial`
    """

    final_time: Union[float, int]
    """
    At and after this time, we return values from `final`
    """

    apply_to_convergence: bool = False
    """
    Is this helper being applied to the convergence spline?

    If `True`, we return 1 - the weights, rather than the weights.
    """

    # domain: ClassVar[list[float, float]] = [
    #     np.finfo(np.float64).tiny,
    #     np.finfo(np.float64).max,
    # ]
    # """Domain of spline"""

    @overload
    def __call__(self, x: int | float) -> int | float: ...

    @overload
    def __call__(self, x: NP_FLOAT_OR_INT) -> NP_FLOAT_OR_INT: ...

    @overload
    def __call__(self, x: NP_ARRAY_OF_FLOAT_OR_INT) -> NP_ARRAY_OF_FLOAT_OR_INT: ...

    def __call__(
        self, x: int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT
    ) -> int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT:
        """
        Evaluate the spline at a given x-value

        Parameters
        ----------
        x
            x-value

        Returns
        -------
        :
            Value of the spline at `x`
        """

        def calc_gamma(
            x: int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT,
        ) -> int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT:
            """Get cosine-decay derivative"""
            # compute weight (here: gamma) according to a cosine-decay
            angle = (
                np.pi * (x - self.initial_time) / (self.final_time - self.initial_time)
            )

            gamma_decaying = 0.5 * (1 + np.cos(angle))

            return gamma_decaying

        if not isinstance(x, np.ndarray):
            if x <= self.initial_time:
                gamma: float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT = 1.0
            elif x >= self.final_time:
                gamma = 0.0
            else:
                gamma = calc_gamma(x)

            # The weighted sum for computing the harmonised AND converged
            # function has the form: "gamma * harmonised + (1-gamma) * convergence".
            # Depending on which product we want to compute (LHS or RHS of sum),
            # we need gamma or 1-gamma, therefore we include the following condition
            # in all our return statements.
            if self.apply_to_convergence:
                return 1.0 - gamma

            return gamma

        # apply decay function only to values that lie between harmonisation
        # time and convergence-time
        x_gte_final_time = np.where(x >= self.final_time)
        x_decay = np.logical_and(x >= self.initial_time, x < self.final_time)
        gamma = np.ones_like(x, dtype=np.floating)
        gamma[x_gte_final_time] = 0.0
        gamma[x_decay] = calc_gamma(x[x_decay])

        if self.apply_to_convergence:
            return 1.0 - gamma

        return gamma

    def derivative(self) -> CosineDecaySplineHelperDerivative:
        """
        Calculate the derivative of self

        Returns
        -------
        :
            Derivative of self

        """
        return CosineDecaySplineHelperDerivative(
            initial_time=self.initial_time,
            final_time=self.final_time,
            apply_to_convergence=self.apply_to_convergence,
        )

    def antiderivative(self) -> CosineDecaySplineHelperDerivative:
        """
        Calculate the anti-derivative/integral of self

        Returns
        -------
        :
            Anti-derivative of self
        """
        raise NotImplementedError

apply_to_convergence class-attribute instance-attribute #

apply_to_convergence: bool = False

Is this helper being applied to the convergence spline?

If True, we return 1 - the weights, rather than the weights.

final_time instance-attribute #

final_time: Union[float, int]

At and after this time, we return values from final

initial_time instance-attribute #

initial_time: Union[float, int]

At and before this time, we return values from initial

__call__ #

__call__(x: int | float) -> int | float

__call__(x: NP_FLOAT_OR_INT) -> NP_FLOAT_OR_INT

__call__(
    x: NP_ARRAY_OF_FLOAT_OR_INT,
) -> NP_ARRAY_OF_FLOAT_OR_INT

__call__(
    x: int
    | float
    | NP_FLOAT_OR_INT
    | NP_ARRAY_OF_FLOAT_OR_INT,
) -> (
    int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT
)

Evaluate the spline at a given x-value

Parameters:

Name Type Description Default
x int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT

x-value

 required

Returns:

Type Description
int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT

Value of the spline at x
Source code in src/gradient_aware_harmonisation/convergence.py 
 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
def __call__(
    self, x: int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT
) -> int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT:
    """
    Evaluate the spline at a given x-value

    Parameters
    ----------
    x
        x-value

    Returns
    -------
    :
        Value of the spline at `x`
    """

    def calc_gamma(
        x: int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT,
    ) -> int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT:
        """Get cosine-decay derivative"""
        # compute weight (here: gamma) according to a cosine-decay
        angle = (
            np.pi * (x - self.initial_time) / (self.final_time - self.initial_time)
        )

        gamma_decaying = 0.5 * (1 + np.cos(angle))

        return gamma_decaying

    if not isinstance(x, np.ndarray):
        if x <= self.initial_time:
            gamma: float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT = 1.0
        elif x >= self.final_time:
            gamma = 0.0
        else:
            gamma = calc_gamma(x)

        # The weighted sum for computing the harmonised AND converged
        # function has the form: "gamma * harmonised + (1-gamma) * convergence".
        # Depending on which product we want to compute (LHS or RHS of sum),
        # we need gamma or 1-gamma, therefore we include the following condition
        # in all our return statements.
        if self.apply_to_convergence:
            return 1.0 - gamma

        return gamma

    # apply decay function only to values that lie between harmonisation
    # time and convergence-time
    x_gte_final_time = np.where(x >= self.final_time)
    x_decay = np.logical_and(x >= self.initial_time, x < self.final_time)
    gamma = np.ones_like(x, dtype=np.floating)
    gamma[x_gte_final_time] = 0.0
    gamma[x_decay] = calc_gamma(x[x_decay])

    if self.apply_to_convergence:
        return 1.0 - gamma

    return gamma

antiderivative #

antiderivative() -> CosineDecaySplineHelperDerivative

Calculate the anti-derivative/integral of self

Returns:

Type Description
CosineDecaySplineHelperDerivative

Anti-derivative of self
Source code in src/gradient_aware_harmonisation/convergence.py 
141
142
143
144
145
146
147
148
149
150
def antiderivative(self) -> CosineDecaySplineHelperDerivative:
    """
    Calculate the anti-derivative/integral of self

    Returns
    -------
    :
        Anti-derivative of self
    """
    raise NotImplementedError

derivative #

derivative() -> CosineDecaySplineHelperDerivative

Calculate the derivative of self

Returns:

Type Description
CosineDecaySplineHelperDerivative

Derivative of self
Source code in src/gradient_aware_harmonisation/convergence.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def derivative(self) -> CosineDecaySplineHelperDerivative:
    """
    Calculate the derivative of self

    Returns
    -------
    :
        Derivative of self

    """
    return CosineDecaySplineHelperDerivative(
        initial_time=self.initial_time,
        final_time=self.final_time,
        apply_to_convergence=self.apply_to_convergence,
    )

CosineDecaySplineHelperDerivative #

Derivative of CosineDecaySplineHelper

Methods:

Name Description
__call__

Evaluate the spline at a given x-value
antiderivative

Calculate the anti-derivative/integral of self
derivative

Calculate the derivative of self

Attributes:

Name Type Description
apply_to_convergence bool

Is this helper being applied to the convergence spline?
final_time Union[float, int]

Final time of the cosine-decay
initial_time Union[float, int]

Initial time of the cosine-decay
Source code in src/gradient_aware_harmonisation/convergence.py 
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
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
@define
class CosineDecaySplineHelperDerivative:
    """
    Derivative of [CosineDecaySplineHelper][(m).CosineDecaySplineHelper]
    """

    initial_time: Union[float, int]
    """
    Initial time of the cosine-decay
    """

    final_time: Union[float, int]
    """
    Final time of the cosine-decay
    """

    apply_to_convergence: bool = False
    """
    Is this helper being applied to the convergence spline?
    """

    # domain: ClassVar[list[float, float]] = [
    #     np.finfo(np.float64).tiny,
    #     np.finfo(np.float64).max,
    # ]
    # """Domain of spline"""

    @overload
    def __call__(self, x: int | float) -> int | float: ...

    @overload
    def __call__(self, x: NP_FLOAT_OR_INT) -> NP_FLOAT_OR_INT: ...

    @overload
    def __call__(self, x: NP_ARRAY_OF_FLOAT_OR_INT) -> NP_ARRAY_OF_FLOAT_OR_INT: ...

    def __call__(
        self, x: int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT
    ) -> int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT:
        """
        Evaluate the spline at a given x-value

        Parameters
        ----------
        x
            x-value

        Returns
        -------
        :
            Value of the spline at `x`
        """

        # compute weight (here: gamma) according to a cosine-decay
        def calc_gamma_rising_derivative(
            x: int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT,
        ) -> int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT:
            """Get cosine-decay derivative"""
            # compute derivative of gamma according to a cosine-decay
            angle = (
                np.pi * (x - self.initial_time) / (self.final_time - self.initial_time)
            )
            gamma_decaying_derivative = -0.5 * np.sin(angle)

            return gamma_decaying_derivative

        if not isinstance(x, np.ndarray):
            if x <= self.initial_time or x >= self.final_time:
                return 0.0

            gamma_rising_derivative = calc_gamma_rising_derivative(x)

            # The weighted sum for computing the harmonised AND converged
            # function has the form: "gamma * harmonised + (1-gamma) * convergence".
            # Depending on which product we want to compute (LHS or RHS of sum),
            # we need gamma or 1-gamma, therefore we include the following condition
            # in all our return statements.
            if self.apply_to_convergence:
                return -gamma_rising_derivative

            return gamma_rising_derivative

        # apply decay function only to values that lie between harmonisation
        # time and convergence-time
        x_decay = np.where(np.logical_and(x > self.initial_time, x < self.final_time))
        gamma_rising_derivative = np.zeros_like(x, dtype=np.floating)
        gamma_rising_derivative[x_decay] = calc_gamma_rising_derivative(x[x_decay])

        if self.apply_to_convergence:
            return -gamma_rising_derivative
        return gamma_rising_derivative

    def derivative(self) -> CosineDecaySplineHelperDerivative:
        """
        Calculate the derivative of self

        Returns
        -------
        :
            Derivative of self
        """
        raise NotImplementedError

    def antiderivative(self) -> CosineDecaySplineHelperDerivative:
        """
        Calculate the anti-derivative/integral of self

        Returns
        -------
        :
            Anti-derivative of self

        Raises
        ------
        NotImplementedError
        """
        raise NotImplementedError

apply_to_convergence class-attribute instance-attribute #

apply_to_convergence: bool = False

Is this helper being applied to the convergence spline?

final_time instance-attribute #

final_time: Union[float, int]

Final time of the cosine-decay

initial_time instance-attribute #

initial_time: Union[float, int]

Initial time of the cosine-decay

__call__ #

__call__(x: int | float) -> int | float

__call__(x: NP_FLOAT_OR_INT) -> NP_FLOAT_OR_INT

__call__(
    x: NP_ARRAY_OF_FLOAT_OR_INT,
) -> NP_ARRAY_OF_FLOAT_OR_INT

__call__(
    x: int
    | float
    | NP_FLOAT_OR_INT
    | NP_ARRAY_OF_FLOAT_OR_INT,
) -> (
    int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT
)

Evaluate the spline at a given x-value

Parameters:

Name Type Description Default
x int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT

x-value

 required

Returns:

Type Description
int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT

Value of the spline at x
Source code in src/gradient_aware_harmonisation/convergence.py 
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
def __call__(
    self, x: int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT
) -> int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT:
    """
    Evaluate the spline at a given x-value

    Parameters
    ----------
    x
        x-value

    Returns
    -------
    :
        Value of the spline at `x`
    """

    # compute weight (here: gamma) according to a cosine-decay
    def calc_gamma_rising_derivative(
        x: int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT,
    ) -> int | float | NP_FLOAT_OR_INT | NP_ARRAY_OF_FLOAT_OR_INT:
        """Get cosine-decay derivative"""
        # compute derivative of gamma according to a cosine-decay
        angle = (
            np.pi * (x - self.initial_time) / (self.final_time - self.initial_time)
        )
        gamma_decaying_derivative = -0.5 * np.sin(angle)

        return gamma_decaying_derivative

    if not isinstance(x, np.ndarray):
        if x <= self.initial_time or x >= self.final_time:
            return 0.0

        gamma_rising_derivative = calc_gamma_rising_derivative(x)

        # The weighted sum for computing the harmonised AND converged
        # function has the form: "gamma * harmonised + (1-gamma) * convergence".
        # Depending on which product we want to compute (LHS or RHS of sum),
        # we need gamma or 1-gamma, therefore we include the following condition
        # in all our return statements.
        if self.apply_to_convergence:
            return -gamma_rising_derivative

        return gamma_rising_derivative

    # apply decay function only to values that lie between harmonisation
    # time and convergence-time
    x_decay = np.where(np.logical_and(x > self.initial_time, x < self.final_time))
    gamma_rising_derivative = np.zeros_like(x, dtype=np.floating)
    gamma_rising_derivative[x_decay] = calc_gamma_rising_derivative(x[x_decay])

    if self.apply_to_convergence:
        return -gamma_rising_derivative
    return gamma_rising_derivative

antiderivative #

antiderivative() -> CosineDecaySplineHelperDerivative

Calculate the anti-derivative/integral of self

Returns:

Type Description
CosineDecaySplineHelperDerivative

Anti-derivative of self

Raises:

Type Description
NotImplementedError
Source code in src/gradient_aware_harmonisation/convergence.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def antiderivative(self) -> CosineDecaySplineHelperDerivative:
    """
    Calculate the anti-derivative/integral of self

    Returns
    -------
    :
        Anti-derivative of self

    Raises
    ------
    NotImplementedError
    """
    raise NotImplementedError

derivative #

derivative() -> CosineDecaySplineHelperDerivative

Calculate the derivative of self

Returns:

Type Description
CosineDecaySplineHelperDerivative

Derivative of self
Source code in src/gradient_aware_harmonisation/convergence.py 
245
246
247
248
249
250
251
252
253
254
def derivative(self) -> CosineDecaySplineHelperDerivative:
    """
    Calculate the derivative of self

    Returns
    -------
    :
        Derivative of self
    """
    raise NotImplementedError

get_cosine_decay_harmonised_spline #

get_cosine_decay_harmonised_spline(
    harmonisation_time: Union[int, float],
    convergence_time: Union[int, float],
    harmonised_spline_no_convergence: Spline,
    convergence_spline: Spline,
) -> SumOfSplines

Generate the harmonised spline based on a cosine-decay

Parameters:

Name Type Description Default
harmonisation_time Union[int, float]

Harmonisation time

This is the time at and before which the solution should be equal to harmonised_spline_no_convergence.

 required
convergence_time Union[int, float]

Convergence time

This is the time at and after which the solution should be equal to convergence_spline.

 required
harmonised_spline_no_convergence Spline

Harmonised spline that does not consider convergence

 required
convergence_spline Spline

The spline to which the result should converge

 required

Returns:

Type Description
SumOfSplines

Harmonised spline
Source code in src/gradient_aware_harmonisation/convergence.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
def get_cosine_decay_harmonised_spline(
    harmonisation_time: Union[int, float],
    convergence_time: Union[int, float],
    harmonised_spline_no_convergence: Spline,
    convergence_spline: Spline,
) -> SumOfSplines:
    """
    Generate the harmonised spline based on a cosine-decay

    Parameters
    ----------
    harmonisation_time
        Harmonisation time

        This is the time at and before which
        the solution should be equal to `harmonised_spline_no_convergence`.

    convergence_time
        Convergence time

        This is the time at and after which
        the solution should be equal to `convergence_spline`.

    harmonised_spline_no_convergence
        Harmonised spline that does not consider convergence

    convergence_spline
        The spline to which the result should converge

    Returns
    -------
    :
        Harmonised spline
    """
    # The harmonised spline is considered as the spline that match
    # the target-spline at the harmonisation time (wrt to zero-and
    # first order derivative). Then we use a decay function to let
    # the harmonised spline converge to the convergence-spline.
    # This decay function has the form of a weighted sum:
    # weight * harmonised_spline + (1-weight) * convergence_spline
    # With weights decaying from 1 to 0 whereby the decay trajectory
    # is determined by the cosine decay.
    return SumOfSplines(
        ProductOfSplines(
            CosineDecaySplineHelper(
                initial_time=harmonisation_time,
                final_time=convergence_time,
                apply_to_convergence=False,
            ),
            harmonised_spline_no_convergence,
        ),
        ProductOfSplines(
            CosineDecaySplineHelper(
                initial_time=harmonisation_time,
                final_time=convergence_time,
                apply_to_convergence=True,
            ),
            convergence_spline,
        ),
    )