def __call__()

in syne_tune/stopping_criterion.py [0:0]

• 31 lines of code
• 21 McCabe index (conditional complexity)

    def __call__(self, status: TuningStatus) -> bool:
        if self.max_wallclock_time is not None and status.wallclock_time > self.max_wallclock_time:
            logger.info(f"reaching max wallclock time ({self.max_wallclock_time}), stopping there.")
            return True
        if self.max_num_trials_started is not None and status.num_trials_started > self.max_num_trials_started:
            logger.info(f"reaching max number of trials started ({self.max_num_trials_started}), stopping there.")
            return True
        if self.max_num_trials_completed is not None and status.num_trials_completed > self.max_num_trials_completed:
            logger.info(f"reaching max number of trials completed ({self.max_num_trials_completed}), stopping there.")
            return True
        if self.max_num_trials_finished is not None and status.num_trials_finished > self.max_num_trials_finished:
            logger.info(f"reaching max number of trials finished ({self.max_num_trials_finished}), stopping there.")
            return True
        if self.max_cost is not None and status.cost > self.max_cost:
            logger.info(f"reaching max cost ({self.max_cost}), stopping there.")
            return True

        if self.max_metric_value is not None and status.overall_metric_statistics.count > 0:
            max_metrics_observed = status.overall_metric_statistics.max_metrics
            for metric, max_metric_accepted in self.max_metric_value.items():
                if metric in max_metrics_observed and max_metrics_observed[metric] > max_metric_accepted:
                    logger.info(f"found {metric} with value ({max_metrics_observed[metric]}), "
                                f"above the provided threshold {max_metric_accepted} stopping there.")
                    return True

        if self.min_metric_value is not None and status.overall_metric_statistics.count > 0:
            min_metrics_observed = status.overall_metric_statistics.min_metrics
            for metric, min_metric_accepted in self.min_metric_value.items():
                if metric in min_metrics_observed and min_metrics_observed[metric] < min_metric_accepted:
                    logger.info(f"found {metric} with value ({min_metrics_observed[metric]}), "
                                f"bellow the provided threshold {min_metric_accepted} stopping there.")

                    return True
        return False