def sweep()

in plugins/hydra_nevergrad_sweeper/hydra_plugins/hydra_nevergrad_sweeper/_impl.py [0:0]

• 73 lines of code
• 19 McCabe index (conditional complexity)

    def sweep(self, arguments: List[str]) -> None:

        assert self.config is not None
        assert self.launcher is not None
        assert self.job_idx is not None
        direction = -1 if self.opt_config.maximize else 1
        name = "maximization" if self.opt_config.maximize else "minimization"
        # Override the parametrization from commandline
        params = dict(self.parametrization)

        parser = OverridesParser.create()
        parsed = parser.parse_overrides(arguments)

        for override in parsed:
            params[
                override.get_key_element()
            ] = create_nevergrad_parameter_from_override(override)

        parametrization = ng.p.Dict(**params)
        parametrization.function.deterministic = not self.opt_config.noisy
        parametrization.random_state.seed(self.opt_config.seed)
        # log and build the optimizer
        opt = self.opt_config.optimizer
        remaining_budget = self.opt_config.budget
        nw = self.opt_config.num_workers
        log.info(
            f"NevergradSweeper(optimizer={opt}, budget={remaining_budget}, "
            f"num_workers={nw}) {name}"
        )
        log.info(f"with parametrization {parametrization}")
        log.info(f"Sweep output dir: {self.config.hydra.sweep.dir}")
        optimizer = ng.optimizers.registry[opt](parametrization, remaining_budget, nw)
        # loop!
        all_returns: List[Any] = []
        best: Tuple[float, ng.p.Parameter] = (float("inf"), parametrization)
        while remaining_budget > 0:
            batch = min(nw, remaining_budget)
            remaining_budget -= batch
            candidates = [optimizer.ask() for _ in range(batch)]
            overrides = list(
                tuple(f"{x}={y}" for x, y in c.value.items()) for c in candidates
            )
            self.validate_batch_is_legal(overrides)
            returns = self.launcher.launch(overrides, initial_job_idx=self.job_idx)
            # would have been nice to avoid waiting for all jobs to finish
            # aka batch size Vs steady state (launching a new job whenever one is done)
            self.job_idx += len(returns)
            # check job status and prepare losses
            failures = 0
            for cand, ret in zip(candidates, returns):
                if ret.status == utils.JobStatus.COMPLETED:
                    rectified_loss = direction * ret.return_value
                else:
                    rectified_loss = math.inf
                    failures += 1
                    try:
                        ret.return_value
                    except Exception as e:
                        log.warning(f"Returning infinity for failed experiment: {e}")
                optimizer.tell(cand, rectified_loss)
                if rectified_loss < best[0]:
                    best = (rectified_loss, cand)
            # raise if too many failures
            if failures / len(returns) > self.opt_config.max_failure_rate:
                log.error(
                    f"Failed {failures} times out of {len(returns)} "
                    f"with max_failure_rate={self.opt_config.max_failure_rate}"
                )
                for ret in returns:
                    ret.return_value  # delegate raising to JobReturn, with actual traceback
            all_returns.extend(returns)
        recom = optimizer.provide_recommendation()
        results_to_serialize = {
            "name": "nevergrad",
            "best_evaluated_params": best[1].value,
            "best_evaluated_result": direction * best[0],
        }
        OmegaConf.save(
            OmegaConf.create(results_to_serialize),
            f"{self.config.hydra.sweep.dir}/optimization_results.yaml",
        )
        log.info(
            "Best parameters: %s", " ".join(f"{x}={y}" for x, y in recom.value.items())
        )