(
        self, requested_target_capacity: float, force: bool = False, no_scale_down: bool = False,
    )
return constrained_target_capacity
Signals can return arbitrary values, so make sure we don't add or remove too much capacity
Signals can return arbitrary values, so make sure we don't add or remove too much capacity
Signals can return arbitrary values so make sure we don t add or remove too much capacity
def _constrain_target_capacity(
        self, requested_target_capacity: float, force: bool = False, no_scale_down: bool = False,
    ) -> float:
        """ Signals can return arbitrary values, so make sure we don't add or remove too much capacity """

        requested_delta = requested_target_capacity - self.target_capacity

        # first, determine whether or not the delta is actually positive or negative.
        # for example, if the current target capacity is above the maximum, the resulting delta
        # will be negative, even if the requested delta is positive because we take the min
        # of the two. This is good because using this delta means moving towards the
        # limit, in the case of the example, towards the maximum, since the target capacity
        # is currently above the maximum.
        if requested_delta > 0:
            delta = min(self.max_capacity - self.target_capacity, requested_delta)
        elif requested_delta < 0:
            delta = max(self.min_capacity - self.target_capacity, requested_delta)
        else:
            delta = 0

        # second, constrain the delta by the max weight to change, depending on if it
        # it is positive or negative.
        if delta > 0:
            delta = min(self.max_weight_to_add, delta)
        elif delta < 0:
            delta = max(-self.max_weight_to_remove, delta)

        # third, do not allow delta to be negative if scaling down is currently prohibited
        if no_scale_down:
            delta = max(delta, 0)

        constrained_target_capacity = self.target_capacity + delta
        if requested_delta != delta:
            if force:
                forced_target_capacity = self.target_capacity + requested_delta
                logger.warning(
                    f"Forcing target capacity to {forced_target_capacity} even though "
                    f"scaling limits would restrict to {constrained_target_capacity}."
                )
                return forced_target_capacity
            else:
                logger.warning(
                    f"Requested target capacity {requested_target_capacity}; "
                    f"restricting to {constrained_target_capacity} due to scaling limits."
                )
        return constrained_target_capacity
def _constrain_target_capacity ( self , requested_target_capacity : float , force : bool = False , no_scale_down : bool = False , ) -> float : requested_delta = requested_target_capacity - self . target_capacity # first, determine whether or not the delta is actually positive or negative. # for example, if the current target capacity is above the maximum, the resulting delta # will be negative, even if the requested delta is positive because we take the min # of the two. This is good because using this delta means moving towards the # limit, in the case of the example, towards the maximum, since the target capacity # is currently above the maximum. if requested_delta > 0 : delta = min ( self . max_capacity - self . target_capacity , requested_delta ) elif requested_delta < 0 : delta = max ( self . min_capacity - self . target_capacity , requested_delta ) else : delta = 0 # second, constrain the delta by the max weight to change, depending on if it # it is positive or negative. if delta > 0 : delta = min ( self . max_weight_to_add , delta ) elif delta < 0 : delta = max ( - self . max_weight_to_remove , delta ) # third, do not allow delta to be negative if scaling down is currently prohibited if no_scale_down : delta = max ( delta , 0 ) constrained_target_capacity = self . target_capacity + delta if requested_delta != delta : if force : forced_target_capacity = self . target_capacity + requested_delta logger . warning ( f"Forcing target capacity to {forced_target_capacity} even though " f"scaling limits would restrict to {constrained_target_capacity}." ) return forced_target_capacity else : logger . warning ( f"Requested target capacity {requested_target_capacity}; " f"restricting to {constrained_target_capacity} due to scaling limits." ) return constrained_target_capacity