Method reproduce

dvc/repo/experiments/executor/base.py:449–539

Run dvc repro and return the result. Returns tuple of (exp_hash, exp_ref, force) where exp_hash is the experiment hash (or None on error), exp_ref is the experiment ref, and force is a bool specifying whether or not this experiment should force overwrite

(
        cls,
        info: "ExecutorInfo",
        rev: str,
        queue: Optional["Queue"] = None,
        infofile: Optional[str] = None,
        log_errors: bool = True,
        log_level: Optional[int] = None,
        copy_paths: Optional[list[str]] = None,
        message: Optional[str] = None,
        **kwargs,
    )

    @classmethod
    def reproduce(
        cls,
        info: "ExecutorInfo",
        rev: str,
        queue: Optional["Queue"] = None,
        infofile: Optional[str] = None,
        log_errors: bool = True,
        log_level: Optional[int] = None,
        copy_paths: Optional[list[str]] = None,
        message: Optional[str] = None,
        **kwargs,
    ) -> "ExecutorResult":
        """Run dvc repro and return the result.

        Returns tuple of (exp_hash, exp_ref, force) where exp_hash is the
            experiment hash (or None on error), exp_ref is the experiment ref,
            and force is a bool specifying whether or not this experiment
            should force overwrite any existing duplicates.
        """
        from dvc.repo.checkout import checkout as dvc_checkout
        from dvc.ui import ui

        if queue is not None:
            queue.put((rev, os.getpid()))

        log_ctx = cls._set_log_level(log_level) if log_errors else nullcontext()

        exp_hash: Optional[str] = None
        exp_ref: Optional[ExpRefInfo] = None
        repro_force: bool = False

        if info.name:
            ui.write(f"Reproducing experiment '{info.name}'")

        with (
            log_ctx,
            cls._repro_dvc(
                info,
                infofile,
                log_errors=log_errors,
                copy_paths=copy_paths,
                message=message,
                **kwargs,
            ) as dvc,
        ):
            args, kwargs = cls._repro_args(dvc)
            if args:
                targets: Optional[Union[list, str]] = args[0]
            else:
                targets = kwargs.get("targets")

            repro_force = kwargs.get("force", False)
            logger.trace("Executor repro with force = '%s'", str(repro_force))

            repro_dry = kwargs.get("dry")

            if not repro_dry:
                dvc_checkout(