Class BasePreprocessing

datasets/preprocessing/base_preprocessing.py:17–184

class BasePreprocessing:
    def __init__(
        self,
        data_dir: str = "./data/raw/",
        save_dir: str = "./data/processed/",
        modes: tuple = ("train", "validation", "test"),
        n_jobs: int = -1,
    ):
        self.data_dir = Path(data_dir)
        self.save_dir = Path(save_dir)
        self.n_jobs = n_jobs
        self.modes = modes

        if not self.data_dir.exists():
            logger.error("data folder doesn't exist")
            raise FileNotFoundError
        if self.save_dir.exists() is False:
            self.save_dir.mkdir(parents=True, exist_ok=True)

        self.files = {}
        for data_type in self.modes:
            self.files.update({data_type: []})

    @logger.catch
    def preprocess(self):
        self.n_jobs = multiprocessing.cpu_count() if self.n_jobs == -1 else self.n_jobs
        for mode in self.modes:
            database = []
            logger.info(f"Tasks for {mode}: {len(self.files[mode])}")
            parallel_results = Parallel(n_jobs=self.n_jobs, verbose=10)(
                delayed(self.process_file)(file, mode) for file in self.files[mode]
            )
            for filebase in parallel_results:
                if filebase is not None:
                    database.append(filebase)
            self.save_database(database, mode)
        self.fix_bugs_in_labels()
        # self.joint_database()
        self.compute_color_mean_std(
            train_database_path=(self.save_dir / "validation_database.yaml")
        )

    def preprocess_sequential(self):
        for mode in self.modes:
            database = []
            for filepath in tqdm(self.files[mode], unit="file"):
                filebase = self.process_file(filepath, mode)
                if filebase is not None:
                    database.append(filebase)
            self.save_database(database, mode)
        self.fix_bugs_in_labels()
        # self.joint_database()
        self.compute_color_mean_std(
            train_database_path=(self.save_dir / "validation_database.yaml")
        )

    def process_file(self, filepath, mode):
        """process_file.