def init_git_repo()

in optimum/graphcore/trainer.py [0:0]

• 35 lines of code
• 10 McCabe index (conditional complexity)

    def init_git_repo(self, at_init: bool = False):
        """
        Initializes a Git repo in `self.args.hub_model_id`.

        Args:
            at_init (`bool`, *optional*, defaults to `False`):
                If `True`, this function is called before any training. If
                `self.args.overwrite_output_dir` is `True` and `at_init` is
                `True`, the path to the repo (which is `self.args.output_dir`)
                might be wiped out.
        """
        if not self.is_world_process_zero():
            return
        use_auth_token = True if self.args.hub_token is None else self.args.hub_token
        if self.args.hub_model_id is None:
            repo_name = Path(self.args.output_dir).absolute().name
        else:
            repo_name = self.args.hub_model_id
        if "/" not in repo_name:
            repo_name = get_full_repo_name(repo_name, token=self.args.hub_token)

        try:
            self.repo = Repository(
                self.args.output_dir,
                clone_from=repo_name,
                use_auth_token=use_auth_token,
                private=self.args.hub_private_repo,
            )
        except EnvironmentError:
            if self.args.overwrite_output_dir and at_init:
                # Try again after wiping output_dir
                shutil.rmtree(self.args.output_dir)
                self.repo = Repository(
                    self.args.output_dir,
                    clone_from=repo_name,
                    use_auth_token=use_auth_token,
                )
            else:
                raise

        self.repo.git_pull()

        # By default, ignore the checkpoint folders
        if (
            not os.path.exists(os.path.join(self.args.output_dir, ".gitignore"))
            and self.args.hub_strategy != HubStrategy.ALL_CHECKPOINTS
        ):
            with open(os.path.join(self.args.output_dir, ".gitignore"), "w", encoding="utf-8") as writer:
                writer.writelines(["checkpoint-*/"])

        self.push_in_progress = None