preprocess.py Source File

preprocess.py Source File#

Mobilint SDK qb Compiler: preprocess.py Source File
Mobilint SDK qb Compiler v0.12.0.0
MCS002-KR
 
 
import copy
import gc
import numpy as np
import cv2
import os
import inspect
from typing import Tuple, Union, List, Optional, Dict
import torch
import torchvision.transforms.functional as F
from functools import partial
from pathlib import Path
 
from .utils_preprocess import *
 
 
 
DTYPE_GETDATA_MAP = {"Image": "GetImage", "Npy": "GetNpy", "LLM": "GetText"}
__preprocs__ = {"Image": {}, "Npy": {}, "LLM": {}}
 
 
def preprocess_fnc(fnc, Datatype="Image"):
    if not inspect.isclass(fnc) and not inspect.isfunction(fnc):
        raise TypeError(f"module must be a class or a function, but got {type(fnc)}.")
    __preprocs__[Datatype][fnc.__name__.lower()] = fnc
    return fnc
 
 
@partial(preprocess_fnc, Datatype="Npy")
class GetNpy:
    """
    @brief Load NumPy data from an in-memory array or a .npy file.
    """
 
    def __init__(self):
        super().__init__()
 
    def __call__(self, npy_input: Union[np.ndarray, str]) -> np.ndarray:
        """
        Args:
            npy_input (Union[np.ndarray, str]): Input .npy path or numpy data.
        """
        if isinstance(npy_input, str):
            assert os.path.isfile(npy_input), f"Can not find npy file from {npy_input}."
            npy = np.load(npy_input)
        elif isinstance(npy_input, np.ndarray):
            npy = npy_input
        else:
            raise TypeError(
                f"Expected str or np.ndarray npy_input, but got {type(npy_input)}."
            )
        return npy
 
 
@partial(preprocess_fnc, Datatype="Npy")
class NpyNormalize:
    def __init__(
        self,
        mean: Union[List[float], np.ndarray],
        std: Union[List[float], np.ndarray],
        to_float_div255: bool = False,
    ):
        super().__init__()
        assert all(
            [isinstance(element, (np.ndarray, list, tuple)) for element in [mean, std]]
        ), f"We support List and np.ndarry data type for mean and std values. But type mean: {type(mean)}, type std: {type(std)}"
        self.mean = np.array(mean) if isinstance(mean, (list, tuple)) else mean
        self.std = np.array(std) if isinstance(std, (list, tuple)) else std
        self.to_float_div255 = to_float_div255
 
    def __call__(self, img: np.ndarray):
        img = img.copy().astype(np.float32)
        if self.to_float_div255:
            img /= 255.0
        try:
            normalize_(img, self.mean, self.std)
        except:
            raise Exception(
                f"Calling the '{self.__class__.__name__}', normalization process faild. check the img shape:{img.shape}, mean:{self.mean}, and std:{self.std}"
            )
 
 
@partial(preprocess_fnc, Datatype="Npy")
class NpyPad:
    """
    @brief Pad NumPy arrays to a desired shape.
 
    @param shape list[int]. Expected output shape after padding.
    @param padding_type string. Padding mode, either "constant" or "reflect".
    @param pad_val float. Fill value when padding_mode is 'constant'. Defaults to 0.
    """
 
    def __init__(
        self, shape: List[int], padding_type: str, pad_val: Optional[float] = 0
    ):
        assert padding_type in [
            "constant",
            "reflect",
        ], f"padding_type is not correct, padding_type must be 'constant', 'reflect', but input padding type is {padding_type}"
        self.shape = np.array(shape)
        self.padding_type = padding_type
        self.pad_val = pad_val
 
    def __call__(self, data: np.ndarray):
        assert (
            len(self.shape) == data.ndim
        ), f"Dimension of padded data shape is not equal to the npy data. dimension of padded shape: {self.shape}, data dimension: {data.ndim}"
        residual_pad = self.shape - np.array(data.shape)
 
        assert (
            residual_pad
        ).min() >= 0, "Padded data shape must be larger than original shape"
        if self.padding_type == "constant":
            data = np.pad(
                data,
                residual_pad.max() // 2 + 1,
                "constant",
                constant_values=self.pad_val,
            )
        elif self.padding_type == "reflect":
            data = np.pad(data, residual_pad.max() // 2 + 1, "reflect")
        else:
            raise NotImplementedError(
                f"padding_type: {self.padding_type}, padding_type must be 'constant' or 'reflect'"
            )
 
        return self.slice_padded_data(data)
 
    def slice_padded_data(self, data):
        residual_slice = np.array(data.shape) - self.shape
        slice_list = [
            slice(residual_slice[i] // 2, residual_slice[i] // 2 + self.shape[i], None)
            for i in range(data.ndim)
        ]
 
        return data[tuple(slice_list)]
 
 
@preprocess_fnc
class GetImage:
    """
    @brief Retrieve an image tensor from a file path or NumPy array using OpenCV.
 
    @param to_float32 bool. Cast the output image to float32 when True. Defaults to False.
    @param channel_order string. Channel order to enforce ("bgr", "rgb", "bw", "rgba", or "brga"). Defaults to "bgr".
    """
 
    def __init__(self, to_float32: bool = False, channel_order: str = "bgr"):
        super().__init__()
        self.to_float32 = to_float32
        assert channel_order.lower() in ["bgr", "rgb", "bw", "rgba", "brga"]
        self.channel_order = channel_order.lower()
 
    def __call__(self, img_input: Union[np.ndarray, str]) -> np.ndarray:
        """
        Args:
            img_input (Union[np.ndarray, str]): Input image path or numpy tensor.
 
        Returns:
             keepdims: Contains image numpy tensor and image meta such as "ori_shape".
        """
        if isinstance(img_input, str):
            img = self.get_image_data(img_input)
            if self.channel_order == "bw":
                if self.num_channel == 1:
                    pass
                elif self.num_channel == 3:
                    img = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
                else:
                    img = cv2.cvtColor(img, cv2.COLOR_BGRA2GRAY)
                # (H, w) -> (H, W, 1)
                img = np.expand_dims(img, axis=-1)
            elif self.channel_order == "bgr":
                if self.num_channel == 1:
                    raise ValueError(
                        f"The input image is grayscale with single channel. Therefore, data cannot be loaded in 'bgr' format."
                    )
                elif self.num_channel == 3:
                    pass
                else:
                    img = cv2.cvtColor(img, cv2.COLOR_BGRA2BGR)
            elif self.channel_order == "rgb":
                if self.num_channel == 1:
                    raise ValueError(
                        f"The input image is grayscale with single channel. Therefore, data cannot be loaded in 'rgb' format."
                    )
                elif self.num_channel == 3:
                    img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
                else:
                    img = cv2.cvtColor(img, cv2.COLOR_BGRA2RGB)
            elif self.channel_order == "rgba":
                if self.num_channel == 1:
                    raise ValueError(
                        f"The input image is grayscale with single channel. Therefore, data cannot be loaded in 'rgba' format."
                    )
                elif self.num_channel == 3:
                    img = cv2.cvtColor(img, cv2.COLOR_BGR2RGBA)
                    print(
                        f"The input image has three channels. As the alpha channel is added to the image, the values of the last transparency channel are filled with 255."
                    )
                else:
                    img = cv2.cvtColor(img, cv2.COLOR_BGRA2RGBA)
            elif self.channel_order == "brga":
                if self.num_channel == 1:
                    raise ValueError(
                        f"The input image is grayscale with single channel. Therefore, data cannot be loaded in 'bgra' format."
                    )
                elif self.num_channel == 3:
                    img = cv2.cvtColor(img, cv2.COLOR_BGR2BGRA)
                    print(
                        f"The input image has three channels. As the alpha channel is added to the image, the values of the last transparency channel are filled with 255."
                    )
                else:
                    pass
            else:
                raise NotImplementedError(
                    f"Got unsupported channel_order: {self.channel_order}."
                )
        elif isinstance(img_input, np.ndarray):
            img = img_input
        else:
            raise TypeError(
                f"Expected str or np.ndarray img_input, but got {type(img_input)}."
            )
 
        if self.to_float32:
            img = img.astype(np.float32)
 
        return img
 
    def get_image_data(self, img_input: str):
        assert os.path.isfile(img_input), f"Can not find img file fro {img_input}"
        img = cv2.imread(img_input, cv2.IMREAD_COLOR)
        if len(img.shape) == 2:
            self.num_channel = 1
        elif len(img.shape) == 3:
            if img.shape[-1] == 3:
                self.num_channel = 3
            elif img.shape[-1] == 4:
                self.num_channel = 4
        else:
            raise ValueError(f"the number of image channel must be 1 or 3 or 4.")
 
        return img
 
 
@partial(preprocess_fnc, Datatype="LLM")
class GetText:
    """@class GetText
    @brief Extract text strings from raw inputs.
    @param text_key Optional key to pull from dictionary inputs.
    @param strip Remove leading and trailing whitespace when true.
    """
 
    def __init__(self, text_key: Optional[str] = None, strip: bool = True):
        super().__init__()
        self.text_key = text_key
        self.strip = strip
 
    def __call__(self, text_input: Union[str, Dict[str, str]]):
        if isinstance(text_input, dict):
            if self.text_key:
                if self.text_key not in text_input:
                    raise KeyError(
                        f"Key '{self.text_key}' not found in text input dictionary."
                    )
                text = text_input[self.text_key]
            else:
                candidates = [v for v in text_input.values() if isinstance(v, str)]
                if not candidates:
                    raise ValueError(
                        "No string value found in input dictionary for text extraction."
                    )
                text = candidates[0]
        elif isinstance(text_input, str):
            if os.path.isfile(text_input):
                with open(text_input, "r", encoding="utf-8") as f:
                    text = f.read()
            else:
                text = text_input
        else:
            raise TypeError(
                f"Expected str or dict[str, str] text_input, but got {type(text_input)}."
            )
 
        if self.strip:
            text = text.strip()
        return text
 
 
@partial(preprocess_fnc, Datatype="LLM")
class TokenizeToEmbedding:
    """@class TokenizeToEmbedding
    @brief Convert tokenized text into embeddings using a Hugging Face model.
    @param tokenizer_dir Directory containing tokenizer resources.
    @param model_dir Optional directory for the model weights; defaults to tokenizer_dir.
    @param device Device string determining where inference runs.
    @param trust_remote_code Whether to load custom code from the model repo.
    @param min_seqlen Minimum token length required for a sample.
    @param max_seqlen Maximum token length; longer sequences are truncated.
    @param pad_to_max When true, pad sequences up to max_seqlen with zeros.
    @param dtype NumPy dtype used for exported embeddings.
    """
 
    def __init__(
        self,
        tokenizer_dir: str,
        model_dir: Optional[str] = None,
        device: Optional[str] = None,
        trust_remote_code: bool = False,
        min_seqlen: int = 0,
        max_seqlen: Optional[int] = None,
        pad_to_max: bool = False,
        dtype: str = "float32",
    ):
        super().__init__()
        self.tokenizer_dir = tokenizer_dir
        try:
            from transformers import AutoTokenizer, AutoModelForCausalLM
        except:
            raise ImportError(
                "Failed to import transformers. Please refer to the Mobilint Docker environment and install an appropriate version of transformers."
            )
        self.tokenizer = AutoTokenizer.from_pretrained(
            tokenizer_dir, trust_remote_code=trust_remote_code
        )
        model_path = model_dir or tokenizer_dir
        llm_model = AutoModelForCausalLM.from_pretrained(
            model_path, trust_remote_code=trust_remote_code
        )
        if device is None:
            device = "cuda" if torch.cuda.is_available() else "cpu"
        self.device = torch.device(device)
 
        embedding_module = llm_model.get_input_embeddings()
        if embedding_module is None:
            raise ValueError("Loaded model does not expose an input embedding layer.")
 
        embedding_copy = copy.deepcopy(embedding_module)
        del embedding_module
        embedding_copy.to(self.device)
        embedding_copy.weight.requires_grad_(False)
        self.embedding_layer = embedding_copy.eval()
 
        del llm_model
        gc.collect()
        if self.device.type == "cuda":
            torch.cuda.empty_cache()
 
        self.min_seqlen = max(0, int(min_seqlen))
        self.max_seqlen = int(max_seqlen) if max_seqlen is not None else None
        self.pad_to_max = pad_to_max
        if self.pad_to_max and self.max_seqlen is None:
            raise ValueError("pad_to_max=True requires max_seqlen to be specified.")
 
        supported_dtypes = {
            "float32": np.float32,
            "float16": np.float16,
        }
        if dtype not in supported_dtypes:
            raise ValueError(
                f"Unsupported dtype '{dtype}'. Supported values: {sorted(supported_dtypes)}"
            )
        self.np_dtype = supported_dtypes[dtype]
        self.model_name = Path(model_path).name
 
    def __call__(self, text: str):
        if not isinstance(text, str):
            raise TypeError(f"Expected text input as str, but got {type(text)}.")
 
        token_ids = self.tokenizer(text, return_tensors="pt")["input_ids"]
        if token_ids.numel() == 0:
            return None
 
        token_ids = token_ids.squeeze(0).to(self.device)
        embedded_text = self.embedding_layer(token_ids)
 
        if embedded_text.ndim == 1:
            embedded_text = embedded_text.unsqueeze(0).unsqueeze(0)
        elif embedded_text.ndim == 2:
            embedded_text = embedded_text.unsqueeze(0)
 
        seq_len = embedded_text.shape[1]
        if seq_len < self.min_seqlen:
            return None
 
        if self.max_seqlen is not None and seq_len > self.max_seqlen:
            embedded_text = embedded_text[:, : self.max_seqlen, :]
            seq_len = embedded_text.shape[1]
 
        if self.pad_to_max and seq_len < self.max_seqlen:
            pad_len = self.max_seqlen - seq_len
            pad_tensor = torch.zeros(
                embedded_text.shape[0],
                pad_len,
                embedded_text.shape[2],
                device=embedded_text.device,
                dtype=embedded_text.dtype,
            )
            embedded_text = torch.cat([embedded_text, pad_tensor], dim=1)
 
        result = (
            embedded_text.detach().to("cpu").numpy().astype(self.np_dtype, copy=False)
        )
        return result
 
 
@preprocess_fnc
class Pad:
    """
    @brief Pad images to a target shape or to meet a size divisor.
 
    @param shape tuple[int]. Expected output shape (h, w). Defaults to None.
    @param size_divisor int. Ensures width and height are divisible by this value. Defaults to None.
    @param pad_val float. Fill value used during padding. Defaults to 0.0.
    @param right_bottom bool. When True, pad on the right and bottom only. Defaults to False.
    """
 
    def __init__(
        self,
        shape=None,
        size_divisor: int = None,
        pad_val: float = 0.0,
        right_bottom: bool = False,
    ):
        super().__init__()
        self.pad_val = pad_val
        self.size_divisor = size_divisor
        self.shape = shape
        self.right_bottom = right_bottom
 
    def __call__(self, img: np.ndarray):
        if self.shape is not None:
            padded_img = impad(
                img,
                shape=self.shape,
                pad_val=self.pad_val,
                right_bottom=self.right_bottom,
            )
        elif self.size_divisor is not None:
            padded_img = imgpad_to_multiple(
                img,
                self.size_divisor,
                pad_val=self.pad_val,
                right_bottom=self.right_bottom,
            )
        else:
            raise NotImplementedError(f"Unsupported pad operation.")
 
        return padded_img
 
 
@preprocess_fnc
class Normalize:
    """
    @brief Apply mean/std normalization to images.
 
    @param mean list[float] or np.ndarray. Normalization mean.
    @param std list[float] or np.ndarray. Normalization standard deviation.
    @param to_float_div255 bool. Divide by 255 before normalization when True. Defaults to False.
    """
 
    def __init__(
        self,
        mean: Union[List[float], np.ndarray],
        std: Union[List[float], np.ndarray],
        to_float_div255: bool = False,
    ):
        super().__init__()
        if not isinstance(mean, (np.ndarray, list, tuple)):
            raise TypeError(
                f"Expected mean to be np.ndarray or List[float], but got {type(mean)}."
            )
 
        if not isinstance(std, (np.ndarray, list, tuple)):
            raise TypeError(
                f"Expected std to be np.ndarray or List[float], but got {type(std)}."
            )
 
        mean = np.array(mean, dtype=np.float32) if isinstance(mean, List) else mean
        self.mean = np.float32(mean.reshape(1, -1))
        std = np.array(std, dtype=np.float32) if isinstance(std, List) else std
        self.std = np.float32(std.reshape(1, -1))
        self.to_float_div255 = to_float_div255
 
    def __call__(self, img: np.ndarray):
        img = img.copy().astype(np.float32)
        if self.to_float_div255:
            img /= 255.0
 
        normalize_(img, self.mean, self.std)
        return img
 
 
@preprocess_fnc
class ResizeTorch:
    def __init__(self, size: List[int], interpolation: str):
        super().__init__()
        self.size = size  # h, w
        self.interpolation = interpolation
 
    def __call__(self, img: np.ndarray):
        img_tensor = img_npy_to_torch_tensor(img)
        resized_img_tensor = F.resize(
            img_tensor,
            size=self.size,
            interpolation=TORCH_INTERP_CODES[self.interpolation],
        )
        img = torch_tensor_to_img_npy(resized_img_tensor)
        return img
 
 
@preprocess_fnc
class Resize:
    """
    @brief Resize images with optional aspect-ratio preservation.
 
    @param img_scale float or tuple[int, int]. Scaling factor or target size (h, w).
    @param keep_ratio bool. Preserve aspect ratio when True. Defaults to False.
    @param interpolation string. Interpolation mode ("nearest", "bilinear", "bicubic", "area", or "lanczos").
    """
 
    def __init__(
        self,
        img_scale: Union[float, Tuple[int, int]],
        keep_ratio: bool,
        interpolation: str,
    ):
        super().__init__()
        self.img_scale = img_scale
        self.keep_ratio = keep_ratio
        self.interpolation = interpolation
 
    def __call__(self, img: np.ndarray):
        if self.keep_ratio:
            h, w = img.shape[:2]
            img = imrescale(
                img,
                self.img_scale,
                return_scale=False,
                interpolation=self.interpolation,
            )
            new_h, new_w = img.shape[:2]
            w_scale = new_w / w
            h_scale = new_h / h
        else:
            img, w_scale, h_scale = imresize(
                img, self.img_scale, return_scale=True, interpolation=self.interpolation
            )
 
        return img
 
 
@preprocess_fnc
class CenterCrop:
    """
    @brief Perform a centered crop on the image.
 
    @param size list[int]. Target height and width.
    """
 
    def __init__(self, size: List[int]):
        super().__init__()
        assert has_same_type(size), "size must consist of same type elements."
        type_check(size[0], (int))
        if len(size) == 1:
            self.size = size * 2
        elif len(size) == 2:
            self.size = size
        else:
            raise AssertionError(f"Invalid shape of size; {size}")
 
    def __call__(self, img: np.ndarray):
        # img_tensor = torch.from_numpy(img).permute(2, 0, 1).unsqueeze(0)
        # img_tensor = F.center_crop(img_tensor, self.size)
        # result.img = img_tensor.squeeze(0).permute(1, 2, 0).numpy().astype(np.uint8)
 
        # h, w = img.shape[:2] # h, w, c
        # h_target = min(self.size[0], h)
        # w_target = min(self.size[1], w)
 
        # if h_target == h and w_target == w:
        #     return result
 
        # x_begin = round(w - w_target/2)
        # x_end = x_begin + w_target
        # y_begin = round(h - h_target/2)
        # y_end = y_begin + h_target
 
        # result.img = img[y_begin:y_end, x_begin:x_end, :]
        h, w = img.shape[:2]
        if (self.size[1] > w) or (self.size[0] > h):
            padding = [
                (self.size[1] - w) // 2 if self.size[1] > w else 0,
                (self.size[0] - h) // 2 if self.size[0] > h else 0,
                (self.size[1] - w + 1) // 2 if self.size[1] > w else 0,
                (self.size[0] - h + 1) // 2 if self.size[0] > h else 0,
            ]
            img = impad(img, padding=tuple(padding))
            h, w = img.shape[:2]
            if (self.size[1] == w) and (self.size[0] == h):
                return img
        crop_top = round((h - self.size[0]) / 2.0)
        crop_left = round((w - self.size[1]) / 2.0)
        img = img[
            crop_top : crop_top + self.size[0], crop_left : crop_left + self.size[1], :
        ]
        return img
 
 
@preprocess_fnc
class YoloPre:
    """@class YoloPre
    @brief Apply YOLO-style resize, padding, and normalization to images.
    @param size Target image size provided as [height, width].
    """
 
    def __init__(self, size: List[int]):
        super().__init__()
        assert has_same_type(size), "size must consist of same type elements."
        type_check(size[0], (int))
        if len(size) == 1:
            self.size = size * 2
        elif len(size) == 2:
            self.size = size
        else:
            raise AssertionError(f"Invalid shape of size; {size}")
 
    def __call__(self, img: np.ndarray):
        if not isinstance(img, np.ndarray):
            raise TypeError(f"Expected numpy.ndarray as input, but got {type(img)}.")
 
        h0, w0 = img.shape[:2]
        r = min(self.size[0] / h0, self.size[1] / w0)
        new_unpad = (int(round(w0 * r)), int(round(h0 * r)))
        dh = self.size[0] - new_unpad[1]
        dw = self.size[1] - new_unpad[0]
 
        dw /= 2
        dh /= 2
 
        if (img.shape[1], img.shape[0]) != new_unpad:
            img = cv2.resize(img, new_unpad, interpolation=cv2.INTER_LINEAR)
 
        top = int(round(dh - 0.1))
        bottom = int(round(dh + 0.1))
        left = int(round(dw - 0.1))
        right = int(round(dw + 0.1))
 
        img = cv2.copyMakeBorder(
            img,
            top,
            bottom,
            left,
            right,
            cv2.BORDER_CONSTANT,
            value=(114, 114, 114),
        )
 
        img = (img / 255.0).astype(np.float32)
        return img
 
 
@preprocess_fnc
class SetOrder:
    def __init__(self, shape: str = "HWC"):
        super().__init__()
        assert shape.lower() in ["hwc", "chw", "bhwc", "bchw"]
        self.shape = shape
 
    def __call__(self, img: np.ndarray):
        if "hwc" in self.shape.lower():
            pass
        elif "chw" in self.shape.lower():
            img = img.transpose(2, 0, 1)
        if "b" in self.shape.lower():
            img = np.expand_dims(img, axis=0)
        return img
 
 
class preproc_builder:
    """
    @brief Build a calibration preprocessing pipeline from a YAML or dict configuration.
 
    @details Supported preprocessing types:
    - `Datatype`: Selects how input data is loaded. Supported value: `Image`.
    - `GetImage`: Loads an image tensor via OpenCV or PIL. Must appear before other operators.
    - `Pre-Order`: Explicit list of preprocessing stages to execute sequentially; entries must match the
      preprocessing definitions.
    - `Pad`: Adds padding to the tensor.
    - `Normalize`: Applies mean/std normalization.
    - `ResizeTorch`: Resizes images using `torchvision.transforms.functional.resize`.
    - `Resize`: Resizes images using `cv2.resize`.
    - `CenterCrop`: Performs a center crop.
    - `SetOrder`: Reorders axes; must be the final operator.
 
    @par YAML structure
    @code{.yaml}
    [Pre-processing Type]
        [Parameter]: [Argument]
        ...
 
    # Example
    Datatype: Image
    GetImage:
      to_float32: false
      channel_order: RGB
    Pre-Order: [ResizeTorch, CenterCrop, Normalize, SetOrder]
    Pre-processing:
        ResizeTorch:
            size: [256, 256]
            interpolation: blinear
        CenterCrop:
            size: [224, 224]
        Normalize:
            mean: [0.485, 0.456, 0.406]
            std: [0.229, 0.224, 0.225]
            to_float_div255: true
        SetOrder:
            shape: HWC
    @endcode
 
    @par Operator parameters
    - `GetImage`: `to_float32` (bool, default False); `channel_order` (string, default "bgr").
    - `Pad`: `shape` (tuple[int]); `size_divisor` (int); `pad_val` (float, default 0); `right_bottom` (bool, default False).
    - `Normalize`: `mean` (list/ndarray); `std` (list/ndarray); `to_float_div255` (bool, default False).
    - `ResizeTorch`: `size` (list[int]); `interpolation` (nearest|bilinear|bicubic|box|hamming|lanczos).
    - `Resize`: `img_scale` (float or tuple[int, int]); `keep_ratio` (bool, default False); `interpolation`
      (nearest|bilinear|bicubic|area|lanczos).
    - `CenterCrop`: `size` (list[int]).
    - `SetOrder`: `shape` (HWC|CHW|BHWC|BCHW, default HWC).
 
    @param args dict. Parsed preprocessing configuration.
    @param to_tensor bool. When True, converts the resulting NumPy array to a Torch tensor.
    """
 
    def __init__(self, args, to_tensor=False):
        self.preprocess = []
        self.to_tensor = to_tensor
        self.args = args
        self.get_pre_list(args)
 
    def __call__(self, img):
        for fnc in self.preprocess:
            img = fnc(img)
        if self.to_tensor:
            img = img_npy_to_torch_tensor(img)
        return img
 
    def __repre__(self):
        msg = "Preprocess Function\n"
        for fnc_type, fnc_args in self.args.items():
            msg += f"- {fnc_type}\n"
            for k, v in fnc_args.items():
                msg += f"\t{k}: {v}\n"
        msg += f"- Return as Tensor: {self.to_tensor}"
        return msg
 
    def __str__(self):
        return self.__repre__()
 
    def get_pre_list(self, args):
        order = self.check_yaml_pre(args)
        self.preprocess.append(
            __preprocs__[self.datatype][self.loadtype.lower()](**args[self.loadtype])
        )
        for fnc_type in order:
            fnc_args = args["Pre-processing"][fnc_type]
            fnc_type = fnc_type.lower()
            if fnc_type not in __preprocs__[self.datatype].keys():
                raise KeyError(
                    f"{fnc_type} is already registered in {__preprocs__.keys()}."
                )
            fnc = (
                __preprocs__[self.datatype][fnc_type](**fnc_args)
                if inspect.isclass(__preprocs__[self.datatype][fnc_type])
                else partial(__preprocs__[self.datatype][fnc_type], **fnc_args)
            )
            self.preprocess.append(fnc)
 
    def check_yaml_pre(self, args):
        assert (
            "Pre-processing" in args
        ), "Write the Pre-processing block in the yaml file."
 
        preproc_cfg = args["Pre-processing"]
        pre_order = args.get("Pre-Order")
 
        if pre_order is None:
            pre_order = list(preproc_cfg.keys())
        else:
            assert not set(pre_order) - set(
                preproc_cfg.keys()
            ), "All elements of Pre-Order must be included in pre-processing. check the yaml file."
        assert args.get(
            "Datatype", 0
        ), "Write the Datatype in the yaml file, Datatype is not writen in the yaml file and we support 'Image' and 'Npy'."
 
        self.datatype = args["Datatype"]
        self.loadtype = DTYPE_GETDATA_MAP[self.datatype]
 
        pre_set = set([name.lower() for name in pre_order])
        pre_not_include = pre_set - set(__preprocs__[self.datatype].keys())
 
        if self.loadtype not in args.keys():
            raise Exception(
                f"Datatype is {self.datatype}, but getting data method is not {self.loadtype}"
            )
        elif pre_not_include:
            raise Exception(
                f"{pre_not_include} is not supported if data type is {self.datatype}"
            )
        else:
            return pre_order
 
 