def _initialize_task_env()

in evals/elsuite/hr_ml_agent_bench/environment.py [0:0]

• 39 lines of code
• 17 McCabe index (conditional complexity)

    def _initialize_task_env(self):
        work_dir = self.work_dir

        # remove the workspace folder if it exists
        if os.path.exists(work_dir):
            shutil.rmtree(work_dir)

        benchmark_dir = os.path.join(
            os.path.dirname(os.path.realpath(__file__)),
            "benchmarks",
            self.benchmark_folder_name,
        )

        # prepare if there is a prepare.py and it has not been prepared
        prepare_task(benchmark_dir, self.python_command)

        # copy the benchmarks folder to work_dir
        if os.path.exists(os.path.join(benchmark_dir, "env")):
            shutil.copytree(os.path.join(benchmark_dir, "env"), work_dir, symlinks=True)

        # find all read only files
        if os.path.exists(os.path.join(benchmark_dir, "scripts", "read_only_files.txt")):
            ignore_files = (
                open(os.path.join(benchmark_dir, "scripts", "read_only_files.txt"), "r")
                .read()
                .split("\n")
            )
            for path, subdirs, files in os.walk(os.path.join(work_dir)):
                relpath = os.path.relpath(path, work_dir)
                # filter out the files that are read only
                filenames = [os.path.join(relpath, filename) for filename in files]
                for ignore in ignore_files:
                    ignore_filenames = [n for n in filenames if fnmatch.fnmatch(n, ignore)]
                    self.read_only_files.extend(ignore_filenames)

        # init backup folder and remove all content if it exists
        if os.path.exists(os.path.join(work_dir, "backup")):
            shutil.rmtree(os.path.join(work_dir, "backup"))
        os.mkdir(os.path.join(work_dir, "backup"))

        if self.resume:
            shutil.rmtree(work_dir)
            resume_dir = os.path.join(
                self.resume,
                "env_log",
                "traces",
                f"step_{self.resume_step}_files",
            )
            logger.info(f"Restoring workspace ing from {resume_dir}")
            shutil.copytree(resume_dir, work_dir, symlinks=True)
            if not os.path.exists(os.path.join(work_dir, "backup")):
                os.mkdir(os.path.join(work_dir, "backup"))