Skip to content

eps_scheduler

Schedule perturbation radii and loss-mixture weights during certified training.

BaseScheduler

Store the perturbation and loss-mixture schedule shared by certified trainers.

The scheduler tracks raw-input epsilon together with kappa and beta. Epsilon can be returned in model coordinates using the supplied normalisation statistics. LinearScheduler and SmoothedScheduler implement the updates.

Source code in CTRAIN/train/certified/eps_scheduler.py 
 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
class BaseScheduler():

    """Store the perturbation and loss-mixture schedule shared by certified trainers.

    The scheduler tracks raw-input epsilon together with kappa and beta. Epsilon
    can be returned in model coordinates using the supplied normalisation
    statistics. LinearScheduler and SmoothedScheduler implement the updates.
    """
    def __init__(self, num_epochs, eps, mean, std, start_eps=0, start_kappa=1, end_kappa=0, start_beta=0, end_beta=0, eps_schedule_unit='batch', eps_schedule=(0, 20), batches_per_epoch=None, start_epoch=-1):
        """Configure perturbation and mixture schedules with input-normalisation metadata.

        Args:
            num_epochs (int): The number of epochs for training.
            eps (float): The epsilon value for the scheduler.
            mean (float): The mean value for normalisation.
            std (float): The standard deviation value for normalisation.
            start_eps (float, optional): The starting epsilon value.
            start_kappa (float, optional): The starting kappa value.
            end_kappa (float, optional): The ending kappa value.
            start_beta (float, optional): The starting beta value.
            end_beta (float, optional): The ending beta value.
            eps_schedule_unit (str, optional): The unit for epsilon scheduling ('batch' or
                'epoch').
            eps_schedule (tuple, optional): Warm-up, ramp-up and remaining training durations in
                the selected scheduling unit.
            batches_per_epoch (int, optional): Positive number of training batches per epoch,
                required for both scheduling units.
            start_epoch (int, optional): Starting epoch used to initialise the batch counter; -1
                denotes a new schedule.

        Raises:
            ValueError: Invalid schedule durations, epoch budget, batch count, normalisation statistics or mixture endpoints.
        """

        if eps_schedule_unit not in ('epoch', 'batch'):
            raise ValueError("eps_schedule_unit must be 'epoch' or 'batch'")
        if len(eps_schedule) not in (2, 3) or any(type(v) is not int or v < 0 for v in eps_schedule):
            raise ValueError('eps_schedule must contain two or three nonnegative integer durations')
        if num_epochs is None and eps_schedule_unit == 'epoch':
            num_epochs = sum(eps_schedule)
        if type(num_epochs) is not int or num_epochs <= 0:
            raise ValueError('num_epochs must be a positive integer')
        if type(batches_per_epoch) is not int or batches_per_epoch <= 0:
            raise ValueError('batches_per_epoch must be a positive integer')
        if eps_schedule_unit == 'epoch' and (sum(eps_schedule) > num_epochs or
                (len(eps_schedule) == 3 and sum(eps_schedule) != num_epochs)):
            raise ValueError('eps_schedule is incompatible with num_epochs')
        if not (0 <= start_eps <= eps < float('inf')):
            raise ValueError('epsilon values must be finite and satisfy 0 <= start_eps <= eps')
        if any(not 0 <= v <= 1 for v in (start_kappa, end_kappa, start_beta, end_beta)):
            raise ValueError('kappa and beta values must be finite and in [0, 1]')
        if not torch.isfinite(torch.as_tensor(std)).all() or (torch.as_tensor(std) <= 0).any():
            raise ValueError('std must be finite and positive')

        self.num_epochs = num_epochs
        if len(eps_schedule) == 2:
            self.warm_up, self.ramp_up = eps_schedule
        elif len(eps_schedule) == 3:
            self.warm_up, self.ramp_up, _ = eps_schedule

        self.cur_eps = start_eps
        self.cur_kappa = self.start_kappa = start_kappa
        self.end_kappa = end_kappa
        self.eps = eps
        self.start_eps = start_eps
        self.batches_per_epoch = batches_per_epoch
        self.start_beta = self.cur_beta = start_beta
        self.end_beta = end_beta
        self.mean = mean
        self.std = std

        if eps_schedule_unit == 'epoch':
            self.warm_up *= batches_per_epoch
            self.ramp_up *= batches_per_epoch

        self.training_steps = num_epochs * batches_per_epoch

        self.no_batches = 0

        if start_epoch > 0:
            self.no_batches = self.batches_per_epoch * start_epoch - 1            

    def get_cur_eps(self, normalise=True):
        """Return the current perturbation radius, optionally divided by the input standard deviation."""
        max_eps = self.get_max_eps(normalise=False)
        cur_eps = torch.as_tensor(self.cur_eps)
        # Check needed to mitigate numerical instabilities
        if (torch.as_tensor(max_eps) - cur_eps < 1e-7).all():
            self.cur_eps = max_eps
            cur_eps = torch.as_tensor(self.cur_eps)
        return cur_eps / torch.as_tensor(self.std) if normalise else self.cur_eps

    def get_cur_kappa(self):
        """Return the current natural-loss mixture weight."""
        return self.cur_kappa

    def get_cur_beta(self):
        """Return the current CROWN-IBP bound-mixture weight."""
        return self.cur_beta

    def get_max_eps(self, normalise=True):
        """Return the target perturbation radius, optionally divided by the input standard deviation."""
        return torch.as_tensor(self.eps) / torch.as_tensor(self.std) if normalise else self.eps

    def batch_step(self, metric=None):
        """Define the batch-step interface; derived schedulers implement the numerical update."""
        raise NotImplementedError

__init__(num_epochs, eps, mean, std, start_eps=0, start_kappa=1, end_kappa=0, start_beta=0, end_beta=0, eps_schedule_unit='batch', eps_schedule=(0, 20), batches_per_epoch=None, start_epoch=-1)

Configure perturbation and mixture schedules with input-normalisation metadata.

Parameters:

Name Type Description Default
num_epochs int

The number of epochs for training.

 required
eps float

The epsilon value for the scheduler.

 required
mean float

The mean value for normalisation.

 required
std float

The standard deviation value for normalisation.

 required
start_eps float

The starting epsilon value.

 0
start_kappa float

The starting kappa value.

 1
end_kappa float

The ending kappa value.

 0
start_beta float

The starting beta value.

 0
end_beta float

The ending beta value.

 0
eps_schedule_unit str

The unit for epsilon scheduling ('batch' or 'epoch').

 'batch'
eps_schedule tuple

Warm-up, ramp-up and remaining training durations in the selected scheduling unit.

 (0, 20)
batches_per_epoch int

Positive number of training batches per epoch, required for both scheduling units.

 None
start_epoch int

Starting epoch used to initialise the batch counter; -1 denotes a new schedule.

 -1

Raises:

Type Description
ValueError

Invalid schedule durations, epoch budget, batch count, normalisation statistics or mixture endpoints.
Source code in CTRAIN/train/certified/eps_scheduler.py 
 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
def __init__(self, num_epochs, eps, mean, std, start_eps=0, start_kappa=1, end_kappa=0, start_beta=0, end_beta=0, eps_schedule_unit='batch', eps_schedule=(0, 20), batches_per_epoch=None, start_epoch=-1):
    """Configure perturbation and mixture schedules with input-normalisation metadata.

    Args:
        num_epochs (int): The number of epochs for training.
        eps (float): The epsilon value for the scheduler.
        mean (float): The mean value for normalisation.
        std (float): The standard deviation value for normalisation.
        start_eps (float, optional): The starting epsilon value.
        start_kappa (float, optional): The starting kappa value.
        end_kappa (float, optional): The ending kappa value.
        start_beta (float, optional): The starting beta value.
        end_beta (float, optional): The ending beta value.
        eps_schedule_unit (str, optional): The unit for epsilon scheduling ('batch' or
            'epoch').
        eps_schedule (tuple, optional): Warm-up, ramp-up and remaining training durations in
            the selected scheduling unit.
        batches_per_epoch (int, optional): Positive number of training batches per epoch,
            required for both scheduling units.
        start_epoch (int, optional): Starting epoch used to initialise the batch counter; -1
            denotes a new schedule.

    Raises:
        ValueError: Invalid schedule durations, epoch budget, batch count, normalisation statistics or mixture endpoints.
    """

    if eps_schedule_unit not in ('epoch', 'batch'):
        raise ValueError("eps_schedule_unit must be 'epoch' or 'batch'")
    if len(eps_schedule) not in (2, 3) or any(type(v) is not int or v < 0 for v in eps_schedule):
        raise ValueError('eps_schedule must contain two or three nonnegative integer durations')
    if num_epochs is None and eps_schedule_unit == 'epoch':
        num_epochs = sum(eps_schedule)
    if type(num_epochs) is not int or num_epochs <= 0:
        raise ValueError('num_epochs must be a positive integer')
    if type(batches_per_epoch) is not int or batches_per_epoch <= 0:
        raise ValueError('batches_per_epoch must be a positive integer')
    if eps_schedule_unit == 'epoch' and (sum(eps_schedule) > num_epochs or
            (len(eps_schedule) == 3 and sum(eps_schedule) != num_epochs)):
        raise ValueError('eps_schedule is incompatible with num_epochs')
    if not (0 <= start_eps <= eps < float('inf')):
        raise ValueError('epsilon values must be finite and satisfy 0 <= start_eps <= eps')
    if any(not 0 <= v <= 1 for v in (start_kappa, end_kappa, start_beta, end_beta)):
        raise ValueError('kappa and beta values must be finite and in [0, 1]')
    if not torch.isfinite(torch.as_tensor(std)).all() or (torch.as_tensor(std) <= 0).any():
        raise ValueError('std must be finite and positive')

    self.num_epochs = num_epochs
    if len(eps_schedule) == 2:
        self.warm_up, self.ramp_up = eps_schedule
    elif len(eps_schedule) == 3:
        self.warm_up, self.ramp_up, _ = eps_schedule

    self.cur_eps = start_eps
    self.cur_kappa = self.start_kappa = start_kappa
    self.end_kappa = end_kappa
    self.eps = eps
    self.start_eps = start_eps
    self.batches_per_epoch = batches_per_epoch
    self.start_beta = self.cur_beta = start_beta
    self.end_beta = end_beta
    self.mean = mean
    self.std = std

    if eps_schedule_unit == 'epoch':
        self.warm_up *= batches_per_epoch
        self.ramp_up *= batches_per_epoch

    self.training_steps = num_epochs * batches_per_epoch

    self.no_batches = 0

    if start_epoch > 0:
        self.no_batches = self.batches_per_epoch * start_epoch - 1

batch_step(metric=None)

Define the batch-step interface; derived schedulers implement the numerical update.

Source code in CTRAIN/train/certified/eps_scheduler.py 
126
127
128
def batch_step(self, metric=None):
    """Define the batch-step interface; derived schedulers implement the numerical update."""
    raise NotImplementedError

get_cur_beta()

Return the current CROWN-IBP bound-mixture weight.

Source code in CTRAIN/train/certified/eps_scheduler.py 
118
119
120
def get_cur_beta(self):
    """Return the current CROWN-IBP bound-mixture weight."""
    return self.cur_beta

get_cur_eps(normalise=True)

Return the current perturbation radius, optionally divided by the input standard deviation.

Source code in CTRAIN/train/certified/eps_scheduler.py 
104
105
106
107
108
109
110
111
112
def get_cur_eps(self, normalise=True):
    """Return the current perturbation radius, optionally divided by the input standard deviation."""
    max_eps = self.get_max_eps(normalise=False)
    cur_eps = torch.as_tensor(self.cur_eps)
    # Check needed to mitigate numerical instabilities
    if (torch.as_tensor(max_eps) - cur_eps < 1e-7).all():
        self.cur_eps = max_eps
        cur_eps = torch.as_tensor(self.cur_eps)
    return cur_eps / torch.as_tensor(self.std) if normalise else self.cur_eps

get_cur_kappa()

Return the current natural-loss mixture weight.

Source code in CTRAIN/train/certified/eps_scheduler.py 
114
115
116
def get_cur_kappa(self):
    """Return the current natural-loss mixture weight."""
    return self.cur_kappa

get_max_eps(normalise=True)

Return the target perturbation radius, optionally divided by the input standard deviation.

Source code in CTRAIN/train/certified/eps_scheduler.py 
122
123
124
def get_max_eps(self, normalise=True):
    """Return the target perturbation radius, optionally divided by the input standard deviation."""
    return torch.as_tensor(self.eps) / torch.as_tensor(self.std) if normalise else self.eps

LinearScheduler

Bases: BaseScheduler

Increase epsilon linearly while interpolating the loss-mixture weights.

Parameters:

Name Type Description Default
num_epochs int

Total number of epochs for training.

 required
eps float

The target epsilon value.

 required
mean float

The mean value for normalisation.

 required
std float

The standard deviation value for normalisation.

 required
start_eps float

The starting epsilon value.

 0
start_kappa float

The starting kappa value.

 1
end_kappa float

The ending kappa value.

 0
start_beta float

The starting beta value.

 1
end_beta float

The ending beta value.

 0
eps_schedule_unit str

The unit for epsilon scheduling ('batch' or 'epoch').

 'batch'
eps_schedule tuple

Warm-up, ramp-up and remaining training durations in the selected scheduling unit.

 (0, 20)
batches_per_epoch int

Positive number of training batches per epoch, required for both scheduling units.

 None
start_epoch int

Starting epoch used to initialise the batch counter; -1 denotes a new schedule.

 -1

Methods:

Name Description
batch_step

Adjusts the current epsilon, kappa, and beta values based on the current batch number.
Source code in CTRAIN/train/certified/eps_scheduler.py 
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
class LinearScheduler(BaseScheduler):
    """Increase epsilon linearly while interpolating the loss-mixture weights.

    Args:
        num_epochs (int): Total number of epochs for training.
        eps (float): The target epsilon value.
        mean (float): The mean value for normalisation.
        std (float): The standard deviation value for normalisation.
        start_eps (float, optional): The starting epsilon value.
        start_kappa (float, optional): The starting kappa value.
        end_kappa (float, optional): The ending kappa value.
        start_beta (float, optional): The starting beta value.
        end_beta (float, optional): The ending beta value.
        eps_schedule_unit (str, optional): The unit for epsilon scheduling ('batch' or
            'epoch').
        eps_schedule (tuple, optional): Warm-up, ramp-up and remaining training durations in
            the selected scheduling unit.
        batches_per_epoch (int, optional): Positive number of training batches per epoch,
            required for both scheduling units.
        start_epoch (int, optional): Starting epoch used to initialise the batch counter; -1
            denotes a new schedule.

    Methods:
        batch_step():
            Adjusts the current epsilon, kappa, and beta values based on the current batch number.
    """
    def __init__(self, num_epochs, eps, mean, std,start_eps=0, start_kappa=1, end_kappa=0, start_beta=1, end_beta=0, eps_schedule_unit='batch', eps_schedule=(0, 20), batches_per_epoch=None, start_epoch=-1):
        super().__init__(
            num_epochs=num_epochs, 
            eps=eps, 
            mean=mean, 
            std=std,
            start_eps=start_eps, 
            start_kappa=start_kappa, 
            end_kappa=end_kappa, 
            eps_schedule_unit=eps_schedule_unit, 
            eps_schedule=eps_schedule, 
            batches_per_epoch=batches_per_epoch, 
            start_beta=start_beta, 
            end_beta=end_beta,
            start_epoch=start_epoch,
        )

        if start_epoch > 0 or (self.warm_up == 0 and self.ramp_up == 0):
            self.batch_step()

    def batch_step(self, metric=None):
        """Advance epsilon and mixture weights by one training batch."""
        progress = (float(self.no_batches >= self.warm_up) if self.ramp_up == 0 else
                    min(1., max(0., (self.no_batches - self.warm_up) / self.ramp_up)))
        self.cur_eps = self.start_eps + (self.eps - self.start_eps) * progress
        self.cur_kappa = self.start_kappa + (self.end_kappa - self.start_kappa) * progress
        self.cur_beta = self.start_beta + (self.end_beta - self.start_beta) * progress
        self.no_batches += 1

batch_step(metric=None)

Advance epsilon and mixture weights by one training batch.

Source code in CTRAIN/train/certified/eps_scheduler.py 
178
179
180
181
182
183
184
185
def batch_step(self, metric=None):
    """Advance epsilon and mixture weights by one training batch."""
    progress = (float(self.no_batches >= self.warm_up) if self.ramp_up == 0 else
                min(1., max(0., (self.no_batches - self.warm_up) / self.ramp_up)))
    self.cur_eps = self.start_eps + (self.eps - self.start_eps) * progress
    self.cur_kappa = self.start_kappa + (self.end_kappa - self.start_kappa) * progress
    self.cur_beta = self.start_beta + (self.end_beta - self.start_beta) * progress
    self.no_batches += 1

SmoothedScheduler

Bases: BaseScheduler

Increase epsilon through a polynomial segment followed by a matched linear ramp.

Parameters:

Name Type Description Default
num_epochs int

Number of epochs for training.

 required
eps float

Final epsilon value.

 required
mean float

Mean value for normalisation.

 required
std float

Standard deviation value for normalisation.

 required
start_eps float

Initial epsilon value.

 0
start_kappa float

Initial kappa value.

 1
end_kappa float

Final kappa value.

 0
start_beta float

Initial beta value.

 1
end_beta float

Final beta value.

 0
eps_schedule_unit str

Unit for epsilon scheduling ('batch' or 'epoch').

 'batch'
batches_per_epoch int

Positive number of training batches per epoch, required for both scheduling units.

 None
start_epoch int

Starting epoch used to initialise the batch counter; -1 denotes a new schedule.

 -1
eps_schedule tuple

Warm-up, ramp-up and remaining training durations in the selected scheduling unit.

 (0, 20)
midpoint float

Ramp fraction where epsilon changes from polynomial growth to a matched linear segment.

 0.25
exponent float

Power of the initial polynomial epsilon-growth segment.

 4.0

Methods:

Name Description
batch_step

Updates the current epsilon, kappa, and beta values based on the current batch number.
Source code in CTRAIN/train/certified/eps_scheduler.py 
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
class SmoothedScheduler(BaseScheduler):
    """Increase epsilon through a polynomial segment followed by a matched linear ramp.

    Args:
        num_epochs (int): Number of epochs for training.
        eps (float): Final epsilon value.
        mean (float): Mean value for normalisation.
        std (float): Standard deviation value for normalisation.
        start_eps (float, optional): Initial epsilon value.
        start_kappa (float, optional): Initial kappa value.
        end_kappa (float, optional): Final kappa value.
        start_beta (float, optional): Initial beta value.
        end_beta (float, optional): Final beta value.
        eps_schedule_unit (str, optional): Unit for epsilon scheduling ('batch' or 'epoch').
        batches_per_epoch (int, optional): Positive number of training batches per epoch,
            required for both scheduling units.
        start_epoch (int, optional): Starting epoch used to initialise the batch counter; -1
            denotes a new schedule.
        eps_schedule (tuple, optional): Warm-up, ramp-up and remaining training durations in
            the selected scheduling unit.
        midpoint (float, optional): Ramp fraction where epsilon changes from polynomial
            growth to a matched linear segment.
        exponent (float, optional): Power of the initial polynomial epsilon-growth segment.

    Methods:
        batch_step():
            Updates the current epsilon, kappa, and beta values based on the current batch number.
    """
    def __init__(self, num_epochs, eps, mean, std, start_eps=0, start_kappa=1, end_kappa=0, start_beta=1, end_beta=0, eps_schedule_unit='batch', batches_per_epoch=None, start_epoch=-1, eps_schedule=(0, 20), midpoint=.25, exponent=4.0):
        super().__init__(
            num_epochs=num_epochs, 
            eps=eps, 
            mean=mean,
            std=std,
            start_eps=start_eps, 
            start_kappa=start_kappa, 
            end_kappa=end_kappa, 
            eps_schedule_unit=eps_schedule_unit, 
            eps_schedule=eps_schedule, 
            batches_per_epoch=batches_per_epoch, 
            start_epoch=start_epoch,
            start_beta=start_beta, 
            end_beta=end_beta)
        if not 0 < midpoint < 1 or not 1 < exponent < float('inf'):
            raise ValueError('midpoint must be in (0, 1) and exponent must be finite and greater than 1')
        self.midpoint = midpoint
        self.exponent = exponent
        if start_epoch > 0 or (self.warm_up == 0 and self.ramp_up == 0):
            self.batch_step()


    def batch_step(self, metric=None):
        """Advance epsilon and mixture weights by one training batch."""
        step = self.no_batches
        init_step = self.warm_up + 1
        final_step = self.warm_up + self.ramp_up
        mid_step = int((final_step - init_step) * self.midpoint) + init_step

        if self.ramp_up == 0:
            progress = float(step >= self.warm_up)
        elif step <= self.warm_up:
            progress = 0.
        elif step >= final_step:
            progress = 1.
        elif mid_step <= init_step:
            # Short ramps have no exponential segment; use a finite linear ramp.
            progress = (step - self.warm_up) / self.ramp_up
        elif step <= init_step:
            progress = 0.
        else:
            power = (mid_step - init_step) ** (self.exponent - 1)
            alpha = 1 / ((final_step - mid_step) * self.exponent * power +
                         (mid_step - init_step) * power)
            mid_value = alpha * (mid_step - init_step) ** self.exponent
            progress = (alpha * (step - init_step) ** self.exponent if step < mid_step else
                        mid_value + (1 - mid_value) * (step - mid_step) / (final_step - mid_step))

        self.cur_eps = self.start_eps + (self.eps - self.start_eps) * progress
        self.cur_kappa = self.start_kappa + (self.end_kappa - self.start_kappa) * progress
        self.cur_beta = self.start_beta + (self.end_beta - self.start_beta) * progress
        self.no_batches += 1

batch_step(metric=None)

Advance epsilon and mixture weights by one training batch.

Source code in CTRAIN/train/certified/eps_scheduler.py 
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
def batch_step(self, metric=None):
    """Advance epsilon and mixture weights by one training batch."""
    step = self.no_batches
    init_step = self.warm_up + 1
    final_step = self.warm_up + self.ramp_up
    mid_step = int((final_step - init_step) * self.midpoint) + init_step

    if self.ramp_up == 0:
        progress = float(step >= self.warm_up)
    elif step <= self.warm_up:
        progress = 0.
    elif step >= final_step:
        progress = 1.
    elif mid_step <= init_step:
        # Short ramps have no exponential segment; use a finite linear ramp.
        progress = (step - self.warm_up) / self.ramp_up
    elif step <= init_step:
        progress = 0.
    else:
        power = (mid_step - init_step) ** (self.exponent - 1)
        alpha = 1 / ((final_step - mid_step) * self.exponent * power +
                     (mid_step - init_step) * power)
        mid_value = alpha * (mid_step - init_step) ** self.exponent
        progress = (alpha * (step - init_step) ** self.exponent if step < mid_step else
                    mid_value + (1 - mid_value) * (step - mid_step) / (final_step - mid_step))

    self.cur_eps = self.start_eps + (self.eps - self.start_eps) * progress
    self.cur_kappa = self.start_kappa + (self.end_kappa - self.start_kappa) * progress
    self.cur_beta = self.start_beta + (self.end_beta - self.start_beta) * progress
    self.no_batches += 1

linear_schedule_fraction(epoch, batch, batches, warm_up_epochs, ramp_up_epochs, *, at_batch_end=False)

Return progress through a linear ramp at the requested batch boundary.

Warm-up returns zero and completed ramp-up returns one. at_batch_end selects whether the current batch contributes to the ramp fraction.

Source code in CTRAIN/train/certified/eps_scheduler.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
def linear_schedule_fraction(epoch, batch, batches, warm_up_epochs, ramp_up_epochs,
                             *, at_batch_end=False):
    """Return progress through a linear ramp at the requested batch boundary.

    Warm-up returns zero and completed ramp-up returns one. at_batch_end selects
    whether the current batch contributes to the ramp fraction.
    """
    if epoch < warm_up_epochs:
        return 0.
    elapsed = (epoch - warm_up_epochs + (batch + 1) / batches if at_batch_end
               else epoch + batch / batches - warm_up_epochs)
    return min(1., max(0., elapsed / ramp_up_epochs)) if ramp_up_epochs else 1.