dataset

Module diagram

classDiagram
  class dataset {
  }
  class base_dataset {
  }
  class builder {
  }
  class platform_reader {
  }
  class saige_vision_reader {
  }
  class srproj_dataset {
  }
  class srproj_reader {
  }
  dataset --> builder
  builder --> base_dataset
  builder --> srproj_dataset
  srproj_dataset --> base_dataset
  srproj_dataset --> srproj_reader

data.dataset

커스텀 데이터셋 구현과 각종 데이터 로드 메서드를 제공합니다.

모든 데이터셋 구현은 SaigeDataset 클래스를 상속받아야 합니다.

SaigeDataset

SaigeDataset(crop: Optional[dict] = None, transform: Optional[dict] = None, collate: Optional[Callable[[List[dict]], dict]] = None, device: device = torch.device('cpu'), **neglect: dict)

Bases: Dataset

Saige's basic dataset class.

Attributes:

• crop_fn (Optional[Callable[[dict], List[dict]]) –

  function cropping image data into image patch data.

• transform (Type[Transform]) –

  transform class for transforming to a single image data.

• to_tensor (Callable[[Union[Image.Image, np.ndarray], torch.Tensor]) –

  data returning as Tensor type

• n_patch (int) –

  number of patches when crop_fn is used.

• collate_function (Callable[[List[dict]], dict]) –

  function list of data collating into one batch data

initializing Saige base dataset.

Parameters:

• crop (Optional[dict], default: None ) –

  config for crop function. Defaults to None.

• transform (Optional[dict], default: None ) –

  config for transforming to a single raw data

• collate (Optional[Callable[[List[dict]], dict]], default: None ) –

  list of data collating into one batch data. if None, torch.utils basic collate function would be used. Defaults to None.

• device (device, default: device('cpu') ) –

  gpu device

Source code in SaigeToolkit/data/dataset/base_dataset.py

def __init__(
    self,
    crop: Optional[dict] = None,
    transform: Optional[dict] = None,
    collate: Optional[Callable[[List[dict]], dict]] = None,
    device: torch.device = torch.device("cpu"),
    **neglect: dict,
) -> None:
    """initializing Saige base dataset.

    Args:
        crop (Optional[dict], optional):
            config for crop function. Defaults to None.
        transform (Optional[dict], optional):
            config for transforming to a single raw data
        collate (Optional[Callable[[List[dict]], dict]], optional):
            list of data collating into one batch data.
            if None, torch.utils basic collate function would be used. Defaults to None.
        device (torch.device, optional):
            gpu device
    """
    if neglect:
        logger.warning(f"Neglected arguments: {neglect}")

    # Config options
    self.crop_fn = get_crop_fn(**(crop or {}))
    self.has_crop_fn = crop is not None

    self.transform = Transform(**(transform or {}))

    self.device = device

    if collate is not None:
        assert collate in ["resize", "padding"]
        self.collate_function = {
            "resize": self.resize_collate,
            "padding": self.padding_collate,
        }[collate]
        logger.info(f'[{"DATA".center(9)}] [Base dataset collate] {collate}')

crop_fn instance-attribute

crop_fn = get_crop_fn(**crop or {})

has_crop_fn instance-attribute

has_crop_fn = crop is not None

transform instance-attribute

transform = Transform(**transform or {})

device instance-attribute

device = device

collate_function instance-attribute

collate_function = {'resize': resize_collate, 'padding': padding_collate}[collate]

files property writable

files

__len__

__len__()

Source code in SaigeToolkit/data/dataset/base_dataset.py

def __len__(self):
    return len(self.files)

data_on_memory

data_on_memory()

To store data in RAM memory (faster data loading), put all loaded data ( use self.load_raw_data() ) in list self.data_on_memory. Or define self.data_on_memory as empty list to load data from filesystem everytime.

Source code in SaigeToolkit/data/dataset/base_dataset.py

@abstract_attribute
def data_on_memory(self):
    """
    To store data in RAM memory (faster data loading),
    put all loaded data ( use self.load_raw_data() ) in list `self.data_on_memory`.
    Or define self.data_on_memory as empty list to load data from filesystem everytime.
    """
    pass

stack_data_on_memory

stack_data_on_memory() -> List[dict]

Stack all data on RAM memory for quick-loading purpose. output would be saved on class attribute.

Returns:

• List[dict] –

  List[dict]: Entire data list-dict.

Source code in SaigeToolkit/data/dataset/base_dataset.py

def stack_data_on_memory(self) -> List[dict]:
    """Stack all data on RAM memory for quick-loading purpose.
    output would be saved on class attribute.

    Returns:
        List[dict]: Entire data list-dict.
    """
    data_on_memory = []
    for idx in range(len(self)):
        data_on_memory.append(self.load_raw_data(idx))

    logger.info(f'[{"DATA".center(9)}] [DATA_ON_MEMORY] activated')

    return data_on_memory

load_raw_data

load_raw_data(index)

Source code in SaigeToolkit/data/dataset/base_dataset.py

def load_raw_data(self, index):
    # load image
    file = self.files[index]
    data = self.load_image(file)
    data.update(self.load_label(file))

    return data

load_image

load_image(file)

Source code in SaigeToolkit/data/dataset/base_dataset.py

def load_image(self, file):
    # open image
    if "image_path" in file:
        image = file["image_path"]
    elif "path" in file:
        image = file["path"]
    elif "Path" in file:
        image = file["Path"]
    else:
        raise NotImplementedError(f"no known image path item in {file.keys()}")

    data = self.transform.image_loader(image)

    return data

load_label abstractmethod

load_label(file)

Source code in SaigeToolkit/data/dataset/base_dataset.py

@abstractmethod
def load_label(self, file):
    pass

__getitem__

__getitem__(index: int) -> Union[dict, List[dict]]

dataset default getitem function. data is first loaded, then processed as following order: resize, augmentation, transform to torch.Tensor. data could be loaded from file_system (load_raw_data) or RAM memory (data_on_memory).

Parameters:

• index (int) –

  data item index.

Returns:

• Union[dict, List[dict]] –

  Union[dict, List[dict]]: single data, or cropped data list when crop_fn exists.

Source code in SaigeToolkit/data/dataset/base_dataset.py

def __getitem__(self, index: int) -> Union[dict, List[dict]]:
    """dataset default __getitem__ function.
    data is first loaded, then processed as following order:
        resize, augmentation, transform to torch.Tensor.
    data could be loaded from file_system (load_raw_data) or RAM memory (data_on_memory).

    Args:
        index (int): data item index.

    Returns:
        Union[dict, List[dict]]:
            single data, or cropped data list when crop_fn exists.
    """

    # Load Images and Text or Image Labels
    if self.data_on_memory:
        data = self.data_on_memory[index]
    else:
        data = self.load_raw_data(index)

    # ROI
    if self.transform.roi_handler is not None:
        if self.has_crop_fn:
            data = self.transform.roi_handler(**data)
        else:
            data = self.transform.roi_handler.apply_crop(**data)

    assert "image" in data, "SaigeDataset currently only supports image dataset"

    if not self.has_crop_fn:
        data = self.process_data(data)

        return data

    else:
        l_data_raw = self.crop_fn(**data)

        batch = []
        for data_patch in l_data_raw:
            batch.append(self.process_data(data_patch))

        return batch

process_data

process_data(data: dict) -> dict

data processing after raw loading.

- resize: Resize Image and Label
- btw_resize_aug: Dummy function for user customization
- augmentation: Process augmentation defined in util/augmentation
- btw_aug_transform: Dummy function for user customization
- to_tensor: Transform data to torch tensor

Parameters:

• data (dict) –

  raw data dict

Returns:

• dict ( dict ) –

  processed data dict

Source code in SaigeToolkit/data/dataset/base_dataset.py

def process_data(self, data: dict) -> dict:
    """data processing after raw loading.

        - resize: Resize Image and Label
        - btw_resize_aug: Dummy function for user customization
        - augmentation: Process augmentation defined in util/augmentation
        - btw_aug_transform: Dummy function for user customization
        - to_tensor: Transform data to torch tensor

    Args:
        data (dict): raw data dict

    Returns:
        dict: processed data dict
    """
    if self.transform.base_resizer is not None:
        data = self.transform.base_resizer(**data)

    if self.transform.roi_handler is not None and not self.has_crop_fn:
        data = self.transform.roi_handler.apply_mask(**data)

    if self.transform.augmentation is not None:
        data = self.transform.augmentation(**data)

    data = self.to_tensor(**data)

    return data

to_tensor

to_tensor(**data: dict) -> dict

to_tensor items in data into torch.Tensor (if convertible)

Parameters:

• data (dict, default: {} ) –

  raw data dict

Returns:

• dict ( dict ) –

  tensorized data dict

Source code in SaigeToolkit/data/dataset/base_dataset.py

def to_tensor(self, **data: dict) -> dict:
    """to_tensor items in data into torch.Tensor (if convertible)

    Args:
        data (dict): raw data dict

    Returns:
        dict: tensorized data dict
    """

    def _to_tensor_from_numpy(image: np.ndarray):
        return torch.from_numpy(image).to(self.device).contiguous()

    def _to_tensor_from_pil_image(image: Image.Image):
        return _to_tensor_from_numpy(np.array(image))

    for k, v in data.items():
        if k == "org_image":
            continue

        if isinstance(v, Image.Image):
            data[k] = _to_tensor_from_pil_image(v)

        elif isinstance(v, np.ndarray):
            data[k] = _to_tensor_from_numpy(v)

        elif isinstance(v, int):
            data[k] = torch.tensor(v).to(self.device).contiguous()

        else:
            continue

        if k == "image":
            data[k] = data[k].permute((2, 0, 1)).div(255)
        else:
            data[k] = data[k].long()

    return data

resize_collate

resize_collate(batch: List[dict]) -> dict

Example collate function with resizing items.

Parameters:

• batch (List[dict]) –

  list of data dicts

Returns:

• dict ( dict ) –

  resized-and-collated data dict

Source code in SaigeToolkit/data/dataset/base_dataset.py

def resize_collate(self, batch: List[dict]) -> dict:
    """Example collate function with resizing items.

    Args:
        batch (List[dict]): list of data dicts

    Returns:
        dict: resized-and-collated data dict
    """

    img_size = (
        int(np.mean([data["image"].shape[-2] for data in batch])),
        int(np.mean([data["image"].shape[-1] for data in batch])),
    )

    # 이미 tensor 형태로 들어온 상태기 때문에 F.interpolate 함수 사용해보자
    data_out = {key: [] for key in batch[0].keys()}
    for data in batch:
        data_out["image"].append(F.interpolate(data["image"][None], (img_size[0], img_size[1])))

        for k, v in data.items():
            if k == "class":
                data_out[k].append(v.unsqueeze(0))
            elif k == "boxes":
                data_out[k].append(v)
            elif k == "mask":
                data_out[k].append(
                    F.interpolate(v[None, None].float(), (img_size[0], img_size[1]))
                    .long()
                    .squeeze(1)
                )

    for key, value in data_out.items():
        data_out[key] = torch.cat(value)

    return data_out

padding_collate

padding_collate(batch: List[dict]) -> dict

Example collate function with padding items.

Parameters:

• batch (List[dict]) –

  list of data dicts

Returns:

• dict ( dict ) –

  padded-and-collated data dict

Source code in SaigeToolkit/data/dataset/base_dataset.py

def padding_collate(self, batch: List[dict]) -> dict:
    """Example collate function with padding items.

    Args:
        batch (List[dict]): list of data dicts

    Returns:
        dict: padded-and-collated data dict
    """

    max_h = np.max([d["image"].shape[-2] for d in batch])
    max_w = np.max([d["image"].shape[-1] for d in batch])

    data_out = {key: [] for key in batch[0].keys()}
    for data in batch:
        data_out["image"].append(
            F.pad(
                data["image"][None],
                (0, max_w - data["image"].shape[-1], 0, max_h - data["image"].shape[-2]),
            )
        )

        for k, v in data.items():
            if k == "class":
                data_out[k].append(v.unsqueeze(0))
            elif k == "boxes":
                data_out[k].append(v)
            elif k == "mask":
                data_out[k].append(F.pad(v[None], (0, max_w - v.shape[-1], 0, max_h - v.shape[-2])))

    for key, value in data_out.items():
        data_out[key] = torch.cat(value)

    return data_out

calculate_patch_number

calculate_patch_number(files: Optional[List[dict]] = None) -> int

When crop_fn is activated, calculate number of text patches.

Parameters:

• files (Optional[List[dict]], default: None ) –

  list of meta data dict. if None, calculate patch_number from self.files attribute. Defaults to None.

Returns:

• int ( int ) –

  number of patches in meta data list files

Source code in SaigeToolkit/data/dataset/base_dataset.py

def calculate_patch_number(self, files: Optional[List[dict]] = None) -> int:
    """When crop_fn is activated, calculate number of text patches.

    Args:
        files (Optional[List[dict]], optional): list of meta data dict.
            if None, calculate patch_number from self.files attribute. Defaults to None.

    Returns:
        int: number of patches in meta data list `files`
    """
    if not self.has_crop_fn:
        return len(self)

    if files is None:
        files = self.files

    n_patch = 0
    for file in files:
        labels = self.load_label(file)
        if "polygons" in labels:
            n_patch += self.crop_fn.get_n_patch(labels["polygons"])
        elif "labels" in labels:
            n_patch += len(labels["labels"])
        elif "labels_count" in labels:
            n_patch += labels["labels_count"]
        else:
            n_patch += 1

    logger.info(f"total patch number in files: {n_patch}")

    return n_patch

build_dataset

build_dataset(_target_, **cfg_dataset: dict) -> SaigeDataset

build dataset from config["dataset"]

Parameters:

• cfg_dataset (dict, default: {} ) –

  config dict for building dataset

Returns:

• SaigeDataset ( SaigeDataset ) –

  dataset object

Source code in SaigeToolkit/data/dataset/builder.py

def build_dataset(_target_, **cfg_dataset: dict) -> SaigeDataset:
    """build dataset from config["dataset"]

    Args:
        cfg_dataset (dict): config dict for building dataset

    Returns:
        SaigeDataset: dataset object
    """

    # Setup Dataset
    dataset_class = DATASET_IMPLEMENTED[_target_]
    dataset: SaigeDataset = dataset_class(**cfg_dataset)

    return dataset

base_dataset

모든 데이터셋 구현에 사용되는 기본 클래스를 정의합니다.

logger module-attribute

logger = getLogger('SaigeResearch')

DummyAttribute

ABCMeta

Bases: ABCMeta

"abstract_attribute" from:\ https://stackoverflow.com/questions/23831510/abstract-attribute-not-property/50381071#50381071

__call__

__call__(*args, **kwargs)

Source code in SaigeToolkit/data/dataset/base_dataset.py

def __call__(cls, *args, **kwargs):
    instance = NativeABCMeta.__call__(cls, *args, **kwargs)
    abstract_attributes = {
        name
        for name in dir(instance)
        if getattr(getattr(instance, name), "__is_abstract_attribute__", False)
    }
    if abstract_attributes:
        raise NotImplementedError(
            f"Can't instantiate abstract class {cls.__name__} with abstract attributes: {', '.join(abstract_attributes)}"
        )
    return instance

SaigeDataset

SaigeDataset(crop: Optional[dict] = None, transform: Optional[dict] = None, collate: Optional[Callable[[List[dict]], dict]] = None, device: device = torch.device('cpu'), **neglect: dict)

Bases: Dataset

Saige's basic dataset class.

Attributes:

• crop_fn (Optional[Callable[[dict], List[dict]]) –

  function cropping image data into image patch data.

• transform (Type[Transform]) –

  transform class for transforming to a single image data.

• to_tensor (Callable[[Union[Image.Image, np.ndarray], torch.Tensor]) –

  data returning as Tensor type

• n_patch (int) –

  number of patches when crop_fn is used.

• collate_function (Callable[[List[dict]], dict]) –

  function list of data collating into one batch data

initializing Saige base dataset.

Parameters:

• crop (Optional[dict], default: None ) –

  config for crop function. Defaults to None.

• transform (Optional[dict], default: None ) –

  config for transforming to a single raw data

• collate (Optional[Callable[[List[dict]], dict]], default: None ) –

  list of data collating into one batch data. if None, torch.utils basic collate function would be used. Defaults to None.

• device (device, default: device('cpu') ) –

  gpu device

Source code in SaigeToolkit/data/dataset/base_dataset.py

def __init__(
    self,
    crop: Optional[dict] = None,
    transform: Optional[dict] = None,
    collate: Optional[Callable[[List[dict]], dict]] = None,
    device: torch.device = torch.device("cpu"),
    **neglect: dict,
) -> None:
    """initializing Saige base dataset.

    Args:
        crop (Optional[dict], optional):
            config for crop function. Defaults to None.
        transform (Optional[dict], optional):
            config for transforming to a single raw data
        collate (Optional[Callable[[List[dict]], dict]], optional):
            list of data collating into one batch data.
            if None, torch.utils basic collate function would be used. Defaults to None.
        device (torch.device, optional):
            gpu device
    """
    if neglect:
        logger.warning(f"Neglected arguments: {neglect}")

    # Config options
    self.crop_fn = get_crop_fn(**(crop or {}))
    self.has_crop_fn = crop is not None

    self.transform = Transform(**(transform or {}))

    self.device = device

    if collate is not None:
        assert collate in ["resize", "padding"]
        self.collate_function = {
            "resize": self.resize_collate,
            "padding": self.padding_collate,
        }[collate]
        logger.info(f'[{"DATA".center(9)}] [Base dataset collate] {collate}')

crop_fn instance-attribute

crop_fn = get_crop_fn(**crop or {})

has_crop_fn instance-attribute

has_crop_fn = crop is not None

transform instance-attribute

transform = Transform(**transform or {})

device instance-attribute

device = device

collate_function instance-attribute

collate_function = {'resize': resize_collate, 'padding': padding_collate}[collate]

files property writable

files

__len__

__len__()

Source code in SaigeToolkit/data/dataset/base_dataset.py

def __len__(self):
    return len(self.files)

data_on_memory

data_on_memory()

To store data in RAM memory (faster data loading), put all loaded data ( use self.load_raw_data() ) in list self.data_on_memory. Or define self.data_on_memory as empty list to load data from filesystem everytime.

Source code in SaigeToolkit/data/dataset/base_dataset.py

@abstract_attribute
def data_on_memory(self):
    """
    To store data in RAM memory (faster data loading),
    put all loaded data ( use self.load_raw_data() ) in list `self.data_on_memory`.
    Or define self.data_on_memory as empty list to load data from filesystem everytime.
    """
    pass

stack_data_on_memory

stack_data_on_memory() -> List[dict]

Stack all data on RAM memory for quick-loading purpose. output would be saved on class attribute.

Returns:

• List[dict] –

  List[dict]: Entire data list-dict.

Source code in SaigeToolkit/data/dataset/base_dataset.py

def stack_data_on_memory(self) -> List[dict]:
    """Stack all data on RAM memory for quick-loading purpose.
    output would be saved on class attribute.

    Returns:
        List[dict]: Entire data list-dict.
    """
    data_on_memory = []
    for idx in range(len(self)):
        data_on_memory.append(self.load_raw_data(idx))

    logger.info(f'[{"DATA".center(9)}] [DATA_ON_MEMORY] activated')

    return data_on_memory

load_raw_data

load_raw_data(index)

Source code in SaigeToolkit/data/dataset/base_dataset.py

def load_raw_data(self, index):
    # load image
    file = self.files[index]
    data = self.load_image(file)
    data.update(self.load_label(file))

    return data

load_image

load_image(file)

Source code in SaigeToolkit/data/dataset/base_dataset.py

def load_image(self, file):
    # open image
    if "image_path" in file:
        image = file["image_path"]
    elif "path" in file:
        image = file["path"]
    elif "Path" in file:
        image = file["Path"]
    else:
        raise NotImplementedError(f"no known image path item in {file.keys()}")

    data = self.transform.image_loader(image)

    return data

load_label abstractmethod

load_label(file)

Source code in SaigeToolkit/data/dataset/base_dataset.py

@abstractmethod
def load_label(self, file):
    pass

__getitem__

__getitem__(index: int) -> Union[dict, List[dict]]

dataset default getitem function. data is first loaded, then processed as following order: resize, augmentation, transform to torch.Tensor. data could be loaded from file_system (load_raw_data) or RAM memory (data_on_memory).

Parameters:

• index (int) –

  data item index.

Returns:

• Union[dict, List[dict]] –

  Union[dict, List[dict]]: single data, or cropped data list when crop_fn exists.

Source code in SaigeToolkit/data/dataset/base_dataset.py

def __getitem__(self, index: int) -> Union[dict, List[dict]]:
    """dataset default __getitem__ function.
    data is first loaded, then processed as following order:
        resize, augmentation, transform to torch.Tensor.
    data could be loaded from file_system (load_raw_data) or RAM memory (data_on_memory).

    Args:
        index (int): data item index.

    Returns:
        Union[dict, List[dict]]:
            single data, or cropped data list when crop_fn exists.
    """

    # Load Images and Text or Image Labels
    if self.data_on_memory:
        data = self.data_on_memory[index]
    else:
        data = self.load_raw_data(index)

    # ROI
    if self.transform.roi_handler is not None:
        if self.has_crop_fn:
            data = self.transform.roi_handler(**data)
        else:
            data = self.transform.roi_handler.apply_crop(**data)

    assert "image" in data, "SaigeDataset currently only supports image dataset"

    if not self.has_crop_fn:
        data = self.process_data(data)

        return data

    else:
        l_data_raw = self.crop_fn(**data)

        batch = []
        for data_patch in l_data_raw:
            batch.append(self.process_data(data_patch))

        return batch

process_data

process_data(data: dict) -> dict

data processing after raw loading.

- resize: Resize Image and Label
- btw_resize_aug: Dummy function for user customization
- augmentation: Process augmentation defined in util/augmentation
- btw_aug_transform: Dummy function for user customization
- to_tensor: Transform data to torch tensor

Parameters:

• data (dict) –

  raw data dict

Returns:

• dict ( dict ) –

  processed data dict

Source code in SaigeToolkit/data/dataset/base_dataset.py

def process_data(self, data: dict) -> dict:
    """data processing after raw loading.

        - resize: Resize Image and Label
        - btw_resize_aug: Dummy function for user customization
        - augmentation: Process augmentation defined in util/augmentation
        - btw_aug_transform: Dummy function for user customization
        - to_tensor: Transform data to torch tensor

    Args:
        data (dict): raw data dict

    Returns:
        dict: processed data dict
    """
    if self.transform.base_resizer is not None:
        data = self.transform.base_resizer(**data)

    if self.transform.roi_handler is not None and not self.has_crop_fn:
        data = self.transform.roi_handler.apply_mask(**data)

    if self.transform.augmentation is not None:
        data = self.transform.augmentation(**data)

    data = self.to_tensor(**data)

    return data

to_tensor

to_tensor(**data: dict) -> dict

to_tensor items in data into torch.Tensor (if convertible)

Parameters:

• data (dict, default: {} ) –

  raw data dict

Returns:

• dict ( dict ) –

  tensorized data dict

Source code in SaigeToolkit/data/dataset/base_dataset.py

def to_tensor(self, **data: dict) -> dict:
    """to_tensor items in data into torch.Tensor (if convertible)

    Args:
        data (dict): raw data dict

    Returns:
        dict: tensorized data dict
    """

    def _to_tensor_from_numpy(image: np.ndarray):
        return torch.from_numpy(image).to(self.device).contiguous()

    def _to_tensor_from_pil_image(image: Image.Image):
        return _to_tensor_from_numpy(np.array(image))

    for k, v in data.items():
        if k == "org_image":
            continue

        if isinstance(v, Image.Image):
            data[k] = _to_tensor_from_pil_image(v)

        elif isinstance(v, np.ndarray):
            data[k] = _to_tensor_from_numpy(v)

        elif isinstance(v, int):
            data[k] = torch.tensor(v).to(self.device).contiguous()

        else:
            continue

        if k == "image":
            data[k] = data[k].permute((2, 0, 1)).div(255)
        else:
            data[k] = data[k].long()

    return data

resize_collate

resize_collate(batch: List[dict]) -> dict

Example collate function with resizing items.

Parameters:

• batch (List[dict]) –

  list of data dicts

Returns:

• dict ( dict ) –

  resized-and-collated data dict

Source code in SaigeToolkit/data/dataset/base_dataset.py

def resize_collate(self, batch: List[dict]) -> dict:
    """Example collate function with resizing items.

    Args:
        batch (List[dict]): list of data dicts

    Returns:
        dict: resized-and-collated data dict
    """

    img_size = (
        int(np.mean([data["image"].shape[-2] for data in batch])),
        int(np.mean([data["image"].shape[-1] for data in batch])),
    )

    # 이미 tensor 형태로 들어온 상태기 때문에 F.interpolate 함수 사용해보자
    data_out = {key: [] for key in batch[0].keys()}
    for data in batch:
        data_out["image"].append(F.interpolate(data["image"][None], (img_size[0], img_size[1])))

        for k, v in data.items():
            if k == "class":
                data_out[k].append(v.unsqueeze(0))
            elif k == "boxes":
                data_out[k].append(v)
            elif k == "mask":
                data_out[k].append(
                    F.interpolate(v[None, None].float(), (img_size[0], img_size[1]))
                    .long()
                    .squeeze(1)
                )

    for key, value in data_out.items():
        data_out[key] = torch.cat(value)

    return data_out

padding_collate

padding_collate(batch: List[dict]) -> dict

Example collate function with padding items.

Parameters:

• batch (List[dict]) –

  list of data dicts

Returns:

• dict ( dict ) –

  padded-and-collated data dict

Source code in SaigeToolkit/data/dataset/base_dataset.py

def padding_collate(self, batch: List[dict]) -> dict:
    """Example collate function with padding items.

    Args:
        batch (List[dict]): list of data dicts

    Returns:
        dict: padded-and-collated data dict
    """

    max_h = np.max([d["image"].shape[-2] for d in batch])
    max_w = np.max([d["image"].shape[-1] for d in batch])

    data_out = {key: [] for key in batch[0].keys()}
    for data in batch:
        data_out["image"].append(
            F.pad(
                data["image"][None],
                (0, max_w - data["image"].shape[-1], 0, max_h - data["image"].shape[-2]),
            )
        )

        for k, v in data.items():
            if k == "class":
                data_out[k].append(v.unsqueeze(0))
            elif k == "boxes":
                data_out[k].append(v)
            elif k == "mask":
                data_out[k].append(F.pad(v[None], (0, max_w - v.shape[-1], 0, max_h - v.shape[-2])))

    for key, value in data_out.items():
        data_out[key] = torch.cat(value)

    return data_out

calculate_patch_number

calculate_patch_number(files: Optional[List[dict]] = None) -> int

When crop_fn is activated, calculate number of text patches.

Parameters:

• files (Optional[List[dict]], default: None ) –

  list of meta data dict. if None, calculate patch_number from self.files attribute. Defaults to None.

Returns:

• int ( int ) –

  number of patches in meta data list files

Source code in SaigeToolkit/data/dataset/base_dataset.py

def calculate_patch_number(self, files: Optional[List[dict]] = None) -> int:
    """When crop_fn is activated, calculate number of text patches.

    Args:
        files (Optional[List[dict]], optional): list of meta data dict.
            if None, calculate patch_number from self.files attribute. Defaults to None.

    Returns:
        int: number of patches in meta data list `files`
    """
    if not self.has_crop_fn:
        return len(self)

    if files is None:
        files = self.files

    n_patch = 0
    for file in files:
        labels = self.load_label(file)
        if "polygons" in labels:
            n_patch += self.crop_fn.get_n_patch(labels["polygons"])
        elif "labels" in labels:
            n_patch += len(labels["labels"])
        elif "labels_count" in labels:
            n_patch += labels["labels_count"]
        else:
            n_patch += 1

    logger.info(f"total patch number in files: {n_patch}")

    return n_patch

abstract_attribute

abstract_attribute(obj=None)

Source code in SaigeToolkit/data/dataset/base_dataset.py

def abstract_attribute(obj=None):
    if obj is None:
        obj = DummyAttribute()
    obj.__is_abstract_attribute__ = True
    return obj

builder

DATASET_IMPLEMENTED module-attribute

DATASET_IMPLEMENTED = {'srproj': SrprojDataset}

build_dataset

build_dataset(_target_, **cfg_dataset: dict) -> SaigeDataset

build dataset from config["dataset"]

Parameters:

• cfg_dataset (dict, default: {} ) –

  config dict for building dataset

Returns:

• SaigeDataset ( SaigeDataset ) –

  dataset object

Source code in SaigeToolkit/data/dataset/builder.py

def build_dataset(_target_, **cfg_dataset: dict) -> SaigeDataset:
    """build dataset from config["dataset"]

    Args:
        cfg_dataset (dict): config dict for building dataset

    Returns:
        SaigeDataset: dataset object
    """

    # Setup Dataset
    dataset_class = DATASET_IMPLEMENTED[_target_]
    dataset: SaigeDataset = dataset_class(**cfg_dataset)

    return dataset

platform_reader

request_vision_projects

request_vision_projects(ip: str, port: int, project_ids: List[int]) -> List[dict]

Data platform으로부터 project 정보들을 가져옵니다.

Parameters:

• ip (str) –

  API서버에 접근하기 위한 IP 주소입니다.

• port (int) –

  API서버에 접근하기 위한 PORT 번호입니다.

• project_ids (list) –

  project 정보를 가져오기 위한 project_id의 list입니다.

Returns:

• List[dict] –

  List[dict]: project 정보를 dict형태로 저장한 list를 반환합니다.

  [
      {
          "project_id": int,
          "class_info": List[str],
          "dataset": {
              "train_images": {
                  "{image_id}": {
                      "path": str,
                      "width": int,
                      "height": int,
                      "client": str,    # Dataset metadata
                      "end_user": str,  # Dataset metadata
                      "domain": str,    # Dataset metadata
                      "class_index": int,  # CLS only have this key
                      "labels" : [         # DET, SEG only have this key
                          {
                              "class_index": int,
                              "bounding_box": List[int],   # DET only have this key
                              "contours": List[List[int]]  # SEG only have this key
                          },
                          ..., # times number of labels
                      ]
                  },
                  ..., # times number of images
              },
              "validation_images": {...},
              "Not split: {...}
          }
      },
      ...,    # times number of projects
  ]
  

Usage

projects = request_vision_projects("platform.saige.in", 8502, project_ids)

Source code in SaigeToolkit/data/dataset/platform_reader.py

def request_vision_projects(ip: str, port: int, project_ids: List[int]) -> List[dict]:
    """Data platform으로부터 project 정보들을 가져옵니다.

    Args:
        ip (str): API서버에 접근하기 위한 IP 주소입니다.
        port (int): API서버에 접근하기 위한 PORT 번호입니다.
        project_ids (list): project 정보를 가져오기 위한 project_id의 list입니다.

    Returns:
        List[dict]: project 정보를 dict형태로 저장한 list를 반환합니다.
            ```python
            [
                {
                    "project_id": int,
                    "class_info": List[str],
                    "dataset": {
                        "train_images": {
                            "{image_id}": {
                                "path": str,
                                "width": int,
                                "height": int,
                                "client": str,    # Dataset metadata
                                "end_user": str,  # Dataset metadata
                                "domain": str,    # Dataset metadata
                                "class_index": int,  # CLS only have this key
                                "labels" : [         # DET, SEG only have this key
                                    {
                                        "class_index": int,
                                        "bounding_box": List[int],   # DET only have this key
                                        "contours": List[List[int]]  # SEG only have this key
                                    },
                                    ..., # times number of labels
                                ]
                            },
                            ..., # times number of images
                        },
                        "validation_images": {...},
                        "Not split: {...}
                    }
                },
                ...,    # times number of projects
            ]
            ```

    Usage:
        ``` python
        projects = request_vision_projects("platform.saige.in", 8502, project_ids)
        ```
    """
    projects = []
    for project_id in project_ids:
        response = requests.get(f"http://{ip}:{port}/get_project/{project_id}")
        if response.status_code == 200:
            exported_project = response.json()
            projects.append(exported_project)
        else:
            print(response.text)

    return projects

saige_vision_reader

load_labels

load_labels(file: Dict) -> List[Dict]

Source code in SaigeToolkit/data/dataset/saige_vision_reader.py

def load_labels(file: Dict) -> List[Dict]:
    if "labels" in file:
        labels = file["labels"]
    elif "labels_path" in file:
        with open(file["labels_path"], "rb") as f:
            labels = pickle.load(f)
    else:
        labels = []

    return labels

load_label_cls

load_label_cls(file: Dict) -> Dict

Source code in SaigeToolkit/data/dataset/saige_vision_reader.py

def load_label_cls(file: Dict) -> Dict:
    return {"class": file["class_index"]}

load_label_iad

load_label_iad(file: Dict) -> Dict

Source code in SaigeToolkit/data/dataset/saige_vision_reader.py

def load_label_iad(file: Dict) -> Dict:
    return {"is_ng": file["labels"]["is_ng"]}

load_label_det

load_label_det(file: Dict) -> Dict

박스 별 dict를 concat 되어있는 bboxes, labels, scores로 변환합니다.

Source code in SaigeToolkit/data/dataset/saige_vision_reader.py

def load_label_det(file: Dict) -> Dict:
    """박스 별 dict를 concat 되어있는 bboxes, labels, scores로 변환합니다."""
    bboxes = file["labels"]

    label = {
        "bboxes": [bbox["bounding_box"] for bbox in bboxes],
        "labels": [bbox["class_index"] for bbox in bboxes],
    }
    if bboxes and "score" in bboxes[0]:
        label["scores"] = [bbox["score"] for bbox in bboxes]

    return label

load_label_seg

load_label_seg(file: Dict, get_objects: bool = False, n_classes: Optional[int] = None) -> Dict

Source code in SaigeToolkit/data/dataset/saige_vision_reader.py

def load_label_seg(file: Dict, get_objects: bool = False, n_classes: Optional[int] = None) -> Dict:
    if "labels" in file:
        labels = file["labels"]
    elif "labels_path" in file:
        with open(file["labels_path"], "rb") as f:
            labels = pickle.load(f)
    else:  # unlabeled
        return {}

    w, h = file["width"], file["height"]
    mask = np.zeros((h, w), dtype=np.uint8)

    if get_objects:
        objects = []

    for label in labels:
        class_index = label["class_index"]
        if class_index == 0:
            raise SegmentClassIndexError(f"class_index {class_index} for [path] {file['path']}")
        elif n_classes is not None and class_index >= n_classes:
            raise SegmentClassIndexRangeError(
                f"class_index {class_index} for [path] {file['path']} but n_classes is {n_classes}"
            )

        bitmap_nonzero = None

        if "contours" in label:
            if not isinstance(label["contours"][0], np.ndarray):
                # NOTE: this is inplace
                label["contours"] = [np.array(contour) for contour in label["contours"]]
            bbox, bitmap = convert_contours_to_box_and_bitmap(label["contours"])
            bbox_xywh = [bbox.left, bbox.top, bbox.width, bbox.height]
        else:
            bbox_xywh = label["bounding_box"]
            bitmap = label["bitmap"]
            if not isinstance(bitmap, np.ndarray):
                bitmap = Image.open(BytesIO(base64.b64decode(bitmap)))
                bitmap = np.array(bitmap)
            bitmap_nonzero = bitmap > 0
            bitmap[bitmap_nonzero] = BITMAP_PIXEL_VALUE  # NOTE: this is inplace

        bbox_x, bbox_y, bbox_w, bbox_h = bbox_xywh
        bbox_bottom = bbox_y + bbox_h
        bbox_right = bbox_x + bbox_w
        if bbox_x < 0 or bbox_y < 0 or bbox_bottom > h or bbox_right > w:
            raise SegmentLabelOutOfImageError(
                f"segment label is out of image: image_size={(w,h)}, bbox_xywh={(bbox_xywh)} in [path] {file['path']}"
            )
        mask_sliced = mask[bbox_y:bbox_bottom, bbox_x:bbox_right]
        if bitmap_nonzero is None:
            bitmap_nonzero = bitmap > 0
        mask_sliced[bitmap_nonzero] = class_index

        if get_objects:
            segment = {
                "bounding_box": bbox_xywh,
                "bitmap": bitmap,
                "class_index": label["class_index"],
            }
            if "contours" in label:
                segment["contours"] = label["contours"]
            objects.append(segment)

    label = {"mask": mask}
    if get_objects:
        label["objects"] = objects

    return label

load_label_ocr

load_label_ocr(file: Dict) -> Dict

Source code in SaigeToolkit/data/dataset/saige_vision_reader.py

def load_label_ocr(file: Dict) -> Dict:
    labels = load_labels(file)

    pols, strs, kies, verts = [], [], [], []
    for label in labels:
        pols.append(np.array(label["polygon"]))
        strs.append(label["text"])
        if "class_index" in label:
            kies.append(label["class_index"])
        if "is_vertical" in label:
            verts.append(label["is_vertical"])

    ignore = [s is None for s in strs]

    label = {"polygons": pols, "strings": strs, "ignore": ignore}
    if len(kies) == len(pols):
        label.update({"kie_labels": kies})
    if len(verts) == len(pols):
        label.update({"is_vertical": verts})

    return label

srproj_dataset

logger module-attribute

logger = getLogger('SaigeResearch')

SrprojDataset

SrprojDataset(code: Union[str, List[Union[str, dict]]], mode: Optional[str] = None, two_class: bool = False, split: str = 'Training', preload_data_on_memory: bool = False, srproj_params: Optional[dict] = None, **cfg_dataset: dict)

Bases: SaigeDataset

Dataset class with building data from srproj file. Basic structure of dataset class is defined in SaigeDataset. SrprojDataset class only defines data loading from srproj file.

Attributes:

• files (List[dict]) –

  meta_data dict list from srproj file.

• n_classes (int) –

  number of classes of data. fix to 2 if two_class mode is on.

• classes (List[str]) –

  list of class names for each class index. meta_data: only contains information about how to load data

• n_patch (int) –

  number of patches when crop_fn is used.

• data_on_memory (List[dict]) –

  list of loaded data dicts if preload_data_on_memory is activated

initializing SrprojDataset

Parameters:

• code (Union[str, List[Union[str, dict]]]) –

  dataset code(s) with srproj params. 여러 srproj를 사용하는 경우, 각 srproj_param을 dictionary 형태로 추가할 수 있습니다. 이 경우, "srproj_params" 파라미터를 각자의 srproj_param으로 업데이트 하여 사용합니다.

• mode (Optional[str], default: None ) –

  data label type. Defaults to None.

• two_class (bool, default: False ) –

  two_class mode selector. Defaults to False.

• split (str, default: 'Training' ) –

  ["Training", "Validation"]. Defaults to "Training".

• preload_data_on_memory (bool, default: False ) –

  whether pre-load all data and save on RAM memory. Defaults to False.

• srproj_params (dict, default: None ) –

  read_srproj 함수에 전달하는 추가적인 파라미터들 입니다. 현재는 이미지 경로 핸들링을 위한 파라미터들이 있으며, 지속적으로 추가될 수 있습니다.

• cfg_dataset (dict, default: {} ) –

  config dicts for mother class

Source code in SaigeToolkit/data/dataset/srproj_dataset.py

def __init__(
    self,
    code: Union[str, List[Union[str, dict]]],
    mode: Optional[str] = None,
    two_class: bool = False,
    split: str = "Training",
    preload_data_on_memory: bool = False,
    srproj_params: Optional[dict] = None,
    **cfg_dataset: dict,
) -> None:
    """initializing SrprojDataset

    Args:
        code (Union[str, List[Union[str, dict]]]): dataset code(s) with srproj params.
                                                여러 srproj를 사용하는 경우, 각 srproj_param을 dictionary 형태로 추가할 수 있습니다.
                                                이 경우, "srproj_params" 파라미터를 각자의 srproj_param으로 업데이트 하여 사용합니다.
        mode (Optional[str], optional): data label type. Defaults to None.
        two_class (bool, optional): two_class mode selector. Defaults to False.
        split (str, optional): ["Training", "Validation"]. Defaults to "Training".
        preload_data_on_memory (bool, optional):
            whether pre-load all data and save on RAM memory. Defaults to False.
        srproj_params (dict, optional): `read_srproj` 함수에 전달하는 추가적인 파라미터들 입니다.
                                        현재는 이미지 경로 핸들링을 위한 파라미터들이 있으며, 지속적으로 추가될 수 있습니다.
        cfg_dataset (dict): config dicts for mother class
    """
    super().__init__(**cfg_dataset)

    if srproj_params is None:
        srproj_params = {}

    if isinstance(code, list):
        self.files = []
        self.classes = []
        for c in code:
            if isinstance(c, dict):
                _c = c.pop("code")
                _srproj_params = deepcopy(srproj_params)
                _srproj_params.update(c)
                files, _, classes = read_srproj(_c, split, two_class, mode, **_srproj_params)
            elif isinstance(c, str):
                files, _, classes = read_srproj(c, split, two_class, mode, **srproj_params)
            else:
                raise NotImplementedError

            self.files.extend(files)
            self.classes = merge_srproj_classes(self.classes, classes)
        self.n_classes = len(self.classes)
    else:
        self.files, self.n_classes, self.classes = read_srproj(
            code, split, two_class, mode, **srproj_params
        )

    self.data_on_memory = self.stack_data_on_memory() if preload_data_on_memory else []

    logger.info(
        f'[{"DATA".center(9)}] [count {split}] {len(self)} [number of classes] {self.n_classes}'
    )

files instance-attribute

files = []

classes instance-attribute

classes = merge_srproj_classes(classes, classes)

n_classes instance-attribute

n_classes = len(classes)

data_on_memory instance-attribute

data_on_memory = stack_data_on_memory() if preload_data_on_memory else []

load_label

load_label(file: dict) -> dict

load label data from file-system

Parameters:

• file (dict) –

  meta data dict

Returns:

• dict ( dict ) –

  label data as dict

Source code in SaigeToolkit/data/dataset/srproj_dataset.py

def load_label(self, file: dict) -> dict:
    """load label data from file-system

    Args:
        file (dict): meta data dict

    Returns:
        dict: label data as dict
    """
    if self.crop_fn is not None:
        return load_srproj_label(file, use_crop_fn=True)
    else:
        return load_srproj_label(file, use_crop_fn=False)

srproj_reader

path_interpreter

path_interpreter(code: str) -> Tuple[str, str, str, str]

data path interpreter

Parameters:

• code (str) –

  dataset code (rule: {domain}-{source}-{category}) (ex) sample dataset for cls: "test_directory-sample-cls")

Raises:

• RuntimeError –

  invalid classifier keys (code) for srproj code.

• Exception –

  wrong type of input code

Returns:

• Tuple[str, str, str, str] –

  Tuple[str, str, str, str]: absolute path for dataset, domain, source, category

Source code in SaigeToolkit/data/dataset/srproj_reader.py

def path_interpreter(code: str) -> Tuple[str, str, str, str]:
    """data path interpreter

    Args:
        code (str): dataset code (rule: {domain}-{source}-{category})
            (ex) sample dataset for cls: "test_directory-sample-cls")

    Raises:
        RuntimeError: invalid classifier keys (code) for srproj code.
        Exception: wrong type of input code

    Returns:
        Tuple[str, str, str, str]: absolute path for dataset, domain, source, category
    """

    if os.path.isdir(code) or os.path.isfile(code):
        # correct directory/file
        path = code
        keys = os.path.dirname(os.path.dirname(path)).split(os.sep)
        keys = keys + ["dummy" for _ in range(3 - len(keys))]
        domain, source, category = keys[-3:]
    else:
        # wrong directory/file/code
        raise Exception(f"Wrong Directory / File / Code: {code}")

    return path, domain, source, category

read_srproj

read_srproj(code: str, split: str, two_class: bool = False, mode: Optional[str] = None, use_absolute_path: bool = False, srproj_image_directory: Optional[str] = None, system_image_directory: Optional[str] = None, get_class_colors: bool = False, skip_problematic_files: bool = False) -> Tuple[List[Dict], int, Union[List[str], List[Dict]]]

read srproj dataset from file-system.

Parameters:

• code (str) –

  dataset code.

• split (str) –

  split dataset, "Training" or "Validation".

• two_class (bool, default: False ) –

  load data as two class mode (ex) normal <-> abnormal)

• mode (Optional[str], default: None ) –

  select data label type.

• use_absolute_path (bool, default: False ) –

  use the image path in srproj as is.

• srproj_image_directory (Optional[str], default: None ) –

  srproj 파일에 작성된 이미지 경로를 현재 시스템 상의 이미지 경로로 변환하기 위해 치환해야 하는 이미지 폴더 경로. None이면 SaigeDatabase 규칙 사용.

• system_image_directory (Optional[str], default: None ) –

  코드가 구동되는 시스템 상의 이미지 폴더 경로. None이면 SaigeDatabase 규칙 사용.

• get_class_colors(bool) –

  srproj에 정의된 각 클래스의 color를 받을지 여부. True이면

• skip_problematic_files(bool) –

  srproj 파일에 작성된 이미지 경로로부터 이미지들을 읽을 때, 해당 파라미터가 True 라면, 파일이 실제 존재하지 않거나, 읽다가 문제가 발생할 경우 해당 파일을 제외하고 나머지 파일을 계속 읽음. 만약, False라면 Error를 raise 함.

Returns:

• Tuple[List[Dict], int, Union[List[str], List[Dict]]] –

  Tuple[List[Dict], int, Union[List[str], List[Dict]]]: list of meta data dicts, number of classes, and list of class names or dict

Note

.sproj 파일에는 각 이미지들의 경로와 라벨 정보 등이 포함되어 있습니다. 이때 이미지 경로는 해당 파일을 생성한 PC 기준의 절대 경로로 작성되어 있기 때문에, 현재 이 코드가 실행되는 시스템에서의 경로로 변환해 주어야 합니다. 현재 이미지 경로 변환 옵션은 3가지가 존재합니다. 1. sproj를 생성한 PC에서 코드를 실행하는 경우, use_absolute_path= True로 세팅하면 srproj에 작성된 이미지 경로를 그대로 사용합니다. 2. 이미지와 srproj가 SaigeDatabase 룰에 맞게 구성된 경우, 이미지는 images 폴더에 하위에 있고 (root/images/.../xxx.png) srproj는 images 폴더보다 한 단계 아래 경로에 있어야 합니다 (root/projects/xxx.srproj) 이때는srproj_image_directory= None,system_image_directory= None 으로 설정합니다. 3. 이미지와 srproj가 SaigeDatabase 룰을 따르지 않는 경우,srproj_image_directory와system_image_directory`를 설정하면 srproj에 작성된 이미지 경로를 다음 규칙으로 변환합니다: {srproj_image_directory}/.../xxx.png -> {system_image_directory}/.../xxx.png

Source code in SaigeToolkit/data/dataset/srproj_reader.py

def read_srproj(
    code: str,
    split: str,
    two_class: bool = False,
    mode: Optional[str] = None,
    use_absolute_path: bool = False,
    srproj_image_directory: Optional[str] = None,
    system_image_directory: Optional[str] = None,
    get_class_colors: bool = False,
    skip_problematic_files: bool = False,
) -> Tuple[List[Dict], int, Union[List[str], List[Dict]]]:
    """read srproj dataset from file-system.

    Args:
        code (str): dataset code.
        split (str): split dataset, "Training" or "Validation".
        two_class (bool): load data as two class mode (ex) normal <-> abnormal)
        mode (Optional[str]): select data label type.
        use_absolute_path (bool): use the image path in srproj as is.
        srproj_image_directory (Optional[str]): srproj 파일에 작성된 이미지 경로를 현재 시스템 상의 이미지 경로로 변환하기 위해
                                                치환해야 하는 이미지 폴더 경로. None이면 SaigeDatabase 규칙 사용.
        system_image_directory (Optional[str]): 코드가 구동되는 시스템 상의 이미지 폴더 경로. None이면 SaigeDatabase 규칙 사용.
        get_class_colors(bool): srproj에 정의된 각 클래스의 color를 받을지 여부. True이면
        skip_problematic_files(bool): srproj 파일에 작성된 이미지 경로로부터 이미지들을 읽을 때, 해당 파라미터가 True 라면,
                                      파일이 실제 존재하지 않거나, 읽다가 문제가 발생할 경우 해당 파일을 제외하고 나머지 파일을 계속 읽음.
                                      만약, False라면 Error를 raise 함.

    Returns:
        Tuple[List[Dict], int, Union[List[str], List[Dict]]]:
            list of meta data dicts, number of classes, and list of class names or dict

    Note:
        .sproj 파일에는 각 이미지들의 경로와 라벨 정보 등이 포함되어 있습니다.
        이때 이미지 경로는 해당 파일을 생성한 PC 기준의 절대 경로로 작성되어 있기 때문에, 현재 이 코드가 실행되는 시스템에서의 경로로 변환해 주어야 합니다.
        현재 이미지 경로 변환 옵션은 3가지가 존재합니다.
        1. sproj를 생성한 PC에서 코드를 실행하는 경우,
            use_absolute_path` = True로 세팅하면 srproj에 작성된 이미지 경로를 그대로 사용합니다.
        2. 이미지와 srproj가 SaigeDatabase 룰에 맞게 구성된 경우,
            이미지는 images 폴더에 하위에 있고 (root/images/.../xxx.png)
            srproj는 images 폴더보다 한 단계 아래 경로에 있어야 합니다 (root/projects/xxx.srproj)
            이때는 `srproj_image_directory` = None, `system_image_directory` = None 으로 설정합니다.
        3. 이미지와 srproj가 SaigeDatabase 룰을 따르지 않는 경우,
            `srproj_image_directory`와 `system_image_directory`를 설정하면
            srproj에 작성된 이미지 경로를 다음 규칙으로 변환합니다:
            {srproj_image_directory}/.../xxx.png -> {system_image_directory}/.../xxx.png

    """

    path, domain, source, category = path_interpreter(code)

    # Check if "path" is directory
    if os.path.isdir(path):
        srproj_list = [os.path.join(path, f) for f in os.listdir(path) if f.endswith(".srproj")]
        path = np.random.choice(srproj_list)

    # Read srproj
    with open(path, encoding="utf-8") as file:
        txt = file.read()
    srproj = xmltodict.parse(txt, dict_constructor=dict)["Project"]

    # Config mode
    if mode is None:
        mode = srproj["Type"]

    # Return empty list when dummy.srproj
    if srproj["ImageGroup"]["Image"] is None:
        return [], 0, []

    # Config files
    image_list = (
        srproj["ImageGroup"]["Image"]
        if isinstance(srproj["ImageGroup"]["Image"], list)
        else [srproj["ImageGroup"]["Image"]]
    )
    assert split in [
        "Training",
        "Validation",
        "All",
    ], "Split for *.srproj Should be 'Training' or 'Validation'"
    if split == "All":
        files = [l for l in image_list]
    else:
        files = [l for l in image_list if l["SplitState"] == split]
    files = check_label(files, mode)

    n_classes = 0

    image_path_converter = lambda srproj_image_path: srproj_image_path
    if not use_absolute_path:
        if srproj_image_directory is None:
            srproj_image_path = (
                files[0]["Path"] if "Path" in files[0] else files[0]["PathGroup"]["Path"][0]
            )
            srproj_image_path = srproj_image_path.replace("\\", os.sep)
            srproj_image_directory = srproj_image_path.split(os.sep + "images" + os.sep)[0]
        srproj_image_directory = srproj_image_directory.replace("\\", os.sep)

        if system_image_directory is None:
            system_image_directory = os.path.dirname(os.path.dirname(path))

        assert os.path.isdir(
            system_image_directory
        ), f"Image data should be in following directory: {system_image_directory}"

        image_path_converter = lambda srproj_image_path: os.path.join(
            system_image_directory,
            os.path.relpath(srproj_image_path.replace("\\", os.sep), srproj_image_directory),
        )

    _image_loader = ImageLoader()

    def _update_file_information(file: Dict) -> bool:
        image_path = file["image_path"]
        if isinstance(image_path, List):
            image_path = image_path[0]

        if os.path.isfile(image_path):
            try:
                _image = _image_loader(image_path)
                img_size = read_image_size(_image["image"])
            except (IOError, OSError):
                if skip_problematic_files:
                    return False
                raise error.InvalidImageFileError
        else:
            if skip_problematic_files:
                print(f"image not found: {file['image_path']}")
                return False
            raise error.FileNotFoundError

        file.update(
            {
                "domain": domain,
                "source": source,
                "category": category,
                "image_size": img_size,
                "label_extension": "srproj",
                "mode": mode,
            }
        )

        if mode == "Classification":
            lbl = int(file["ClassIndexOfLabel"])
            if two_class:
                lbl = 1 if (lbl > 0) else lbl

            file["ClassIndexOfLabel"] = lbl + n_classes

        elif mode == "MultiLabelClassification":
            lbl = list(file["ClassIndexOfLabel"])
            file["ClassIndexOfLabel"] = [int(l) + n_classes for l in lbl]

        elif mode == "Segmentation":
            if str(file["LabelGroup"]["IsNormal"]) in ["False", "false", "FALSE"]:
                lbl_list = (
                    file["LabelGroup"]["Label"]
                    if isinstance(file["LabelGroup"]["Label"], list)
                    else [file["LabelGroup"]["Label"]]
                )
                for l in lbl_list:
                    lbl = 1 if two_class else int(l["ClassIndex"]) + 1
                    l["ClassIndex"] = lbl + n_classes

        return True

    cleaned_files = []
    for f in files:
        if "Path" in f:
            f["image_path"] = image_path_converter(f["Path"])
        else:
            f["image_path"] = [image_path_converter(_path) for _path in f["PathGroup"]["Path"]]

        is_updated = _update_file_information(f)
        if is_updated:
            cleaned_files.append(f)

    # Config n_classes
    classes = []
    if "ClassGroup" in srproj:
        n_classes = int(srproj["ClassGroup"]["NumberOfClasses"])
        class_list = (
            srproj["ClassGroup"]["Class"]
            if isinstance(srproj["ClassGroup"]["Class"], list)
            else [srproj["ClassGroup"]["Class"]]
        )
        if get_class_colors:
            classes = [
                {
                    "name": class_["Name"],
                    "color": convert_srproj_class_color_to_rgb(int(class_["Color"])),
                }
                for class_ in class_list
            ]
        else:
            classes = [class_["Name"] for class_ in class_list]
    elif "CharacterGroup" in srproj:
        classes = srproj["CharacterGroup"]["Included"]["@xmlns"]
        n_classes = len(classes) + 1

    if mode in ["Segmentation"]:
        if isinstance(classes[0], dict):
            classes.insert(0, {"name": "background", "color": (0, 0, 0)})
        else:
            classes.insert(0, "background")
        n_classes = len(classes)

    if two_class:
        classes = ["Normal", "Defect"]
        n_classes = len(classes)

    return cleaned_files, n_classes, classes

check_label

check_label(files: List[Dict], mode: str) -> List[Dict]

check all data label is valid. If any data is corrupted, pop out from meta data list.

Parameters:

• files (List[Dict]) –

  meta data list to be checked

• mode (str) –

  label data type

Returns:

• List[Dict] –

  List[Dict]: cleaned meta data list

Source code in SaigeToolkit/data/dataset/srproj_reader.py

def check_label(files: List[Dict], mode: str) -> List[Dict]:
    """check all data label is valid.
    If any data is corrupted, pop out from meta data list.

    Args:
        files (List[Dict]): meta data list to be checked
        mode (str): label data type

    Returns:
        List[Dict]: cleaned meta data list
    """
    for i in range(len(files) - 1, -1, -1):
        if mode == "Classification":
            if int(files[i]["ClassIndexOfLabel"]) == -1:
                files.pop(i)
        elif mode == "MultiLabelClassification":
            if not isinstance(files[i]["ClassIndexOfLabel"], list):
                files[i]["ClassIndexOfLabel"] = [files[i]["ClassIndexOfLabel"]]

        elif mode == "OpticalCharacterRecognition":
            if int(files[i]["LabelGroup"].get("NumberOfLabels", 0)) == 0:
                continue
            labels = (
                files[i]["LabelGroup"]["Label"]
                if isinstance(files[i]["LabelGroup"]["Label"], list)
                else [files[i]["LabelGroup"]["Label"]]
            )
            for idx in range(len(labels) - 1, -1, -1):
                # XXX: Temporary implementation for memory issue
                for key, value in labels[idx]["Coordinate"].items():
                    labels[idx]["Coordinate"][key] = float(value)

                if labels[idx]["Characters"] == '"value4"':
                    labels.pop(idx)
                elif not labels[idx]["Characters"]:
                    labels.pop(idx)
                else:
                    continue

                files[i]["LabelGroup"]["Label"] = labels
                files[i]["LabelGroup"]["NumberOfLabels"] = len(labels)

        else:
            if str(files[i]["LabelGroup"]["IsNormal"]) in [
                "False",
                "false",
                "FALSE",
            ]:
                if int(files[i]["LabelGroup"].get("NumberOfLabels", 0)) == 0:
                    files.pop(i)
    return files

load_srproj_label

load_srproj_label(file: Dict, use_crop_fn: Optional[bool] = False, **kwargs) -> Dict

load single data dict from srproj meta data.

Parameters:

• file (Dict) –

  single meta data dict

• use_crop_fn (bool, default: False ) –

  Whether to use the crop function. Defaults to False.

Raises:

• NotImplementedError –

  only four modes are available. ["Classification", "Detection", "Segmentation", "OpticalCharacterRecognition"]

Returns:

• Dict ( Dict ) –

  label data as dict

Source code in SaigeToolkit/data/dataset/srproj_reader.py

def load_srproj_label(file: Dict, use_crop_fn: Optional[bool] = False, **kwargs) -> Dict:
    """load single data dict from srproj meta data.

    Args:
        file (Dict): single meta data dict
        use_crop_fn (bool, optional): Whether to use the crop function.
            Defaults to False.

    Raises:
        NotImplementedError: only four modes are available.
            ["Classification", "Detection", "Segmentation", "OpticalCharacterRecognition"]

    Returns:
        Dict: label data as dict
    """
    mode = file["mode"]

    # make label for the right mode
    if mode == "Classification":
        label = load_label_cls(file, **kwargs)

    elif mode == "Detection":
        label = load_label_det(file, **kwargs)

    elif mode == "Segmentation":
        # In Segmentation task, need polygon data to use the crop function.
        label = load_label_seg(file, return_polygon=use_crop_fn, **kwargs)

    elif mode == "OpticalCharacterRecognition":
        label = load_label_ocr(file, **kwargs)

    else:
        raise NotImplementedError(f"Mode {mode} not implemented")

    return label

load_label_cls

load_label_cls(file: Dict) -> Dict

Source code in SaigeToolkit/data/dataset/srproj_reader.py

def load_label_cls(file: Dict) -> Dict:
    return {"class": file["ClassIndexOfLabel"]}

load_label_det

load_label_det(file: Dict) -> Dict

load detection label

Parameters:

• file (Dict) –

  meta data for detection label

Returns:

• Dict ( Dict ) –

  detection label data dict

  {
      "bboxes": [
          [x, y, w, h],
          ...,
      ],
      "labels": List[int],
  }
  

Source code in SaigeToolkit/data/dataset/srproj_reader.py

def load_label_det(file: Dict) -> Dict:
    """load detection label

    Args:
        file (Dict): meta data for detection label

    Returns:
        Dict: detection label data dict
            ```python
            {
                "bboxes": [
                    [x, y, w, h],
                    ...,
                ],
                "labels": List[int],
            }
            ```
    """
    srproj_label_group = file["LabelGroup"]
    label = {"bboxes": [], "labels": []}

    if str(srproj_label_group["IsNormal"]).lower() == "false":
        srproj_labels = srproj_label_group["Label"]
        if not isinstance(srproj_labels, list):
            srproj_labels = [srproj_labels]
        for srproj_label in srproj_labels:
            x = int(srproj_label["Coordinate"]["@X"])
            y = int(srproj_label["Coordinate"]["@Y"])
            w = int(srproj_label["Coordinate"]["@Width"])
            h = int(srproj_label["Coordinate"]["@Height"])
            label["bboxes"].append([x, y, w, h])
            label["labels"].append(int(srproj_label["ClassIndex"]))

    return label

load_label_seg

load_label_seg(file: Dict, return_polygon: bool = True, mask_to_pil: bool = False) -> Dict

load segmentation label

Parameters:

• file (Dict) –

  meta data for segmentation label

• return_polygon (bool, default: True ) –

  whether return polygon data. Defaults to True.

• mask_to_pil (bool, default: False ) –

  whether return mask as PIL.Image. Defaults to False.

Returns:

• Dict ( Dict ) –

  segmentation label data dict

Source code in SaigeToolkit/data/dataset/srproj_reader.py

def load_label_seg(file: Dict, return_polygon: bool = True, mask_to_pil: bool = False) -> Dict:
    """load segmentation label

    Args:
        file (Dict): meta data for segmentation label
        return_polygon (bool, optional): whether return polygon data. Defaults to True.
        mask_to_pil (bool, optional): whether return mask as PIL.Image. Defaults to False.

    Returns:
        Dict: segmentation label data dict
    """
    w, h = file["image_size"]
    lbl = np.zeros((h, w), dtype=np.uint8)

    lbl_list = file["LabelGroup"]

    if str(lbl_list["IsNormal"]) in ["False", "false", "FALSE"]:
        lbl_list = lbl_list["Label"] if isinstance(lbl_list["Label"], list) else [lbl_list["Label"]]
        contour = []
        for l in lbl_list:
            canvas = np.zeros((h, w), dtype=np.uint8)
            color = l["ClassIndex"]

            contour_list = (
                l["ContourGroup"]["Contour"]
                if isinstance(l["ContourGroup"]["Contour"], list)
                else [l["ContourGroup"]["Contour"]]
            )

            # Contour functions in cv2 gets list type contours
            contours = {"Outer": [], "Inner": []}
            for c_l in contour_list:
                contours[c_l["@Type"]].append(
                    np.array([[int(p["@X"]), int(p["@Y"])] for p in c_l["Point"]])
                )

            # One Outer per One Label
            cv2.fillPoly(canvas, pts=contours["Outer"], color=color)

            # Many Inner could exist in One Label
            if contours["Inner"]:
                cv2.fillPoly(canvas, pts=contours["Inner"], color=0)

            nonzeros = canvas > 0
            lbl[nonzeros] = canvas[nonzeros]
            contour.append(contours["Outer"][0])
    else:
        contour = []

    if mask_to_pil:
        lbl = Image.fromarray(lbl)

    lbl = {"mask": lbl}

    if return_polygon:
        lbl.update({"polygons": contour})

    return lbl

load_label_ocr

load_label_ocr(file: Dict) -> Dict

load ocr label

Parameters:

• file (Dict) –

  meta data for ocr label

Returns:

• Dict ( Dict ) –

  ocr label data dict

Source code in SaigeToolkit/data/dataset/srproj_reader.py

def load_label_ocr(file: Dict) -> Dict:
    """load ocr label

    Args:
        file (Dict): meta data for ocr label

    Returns:
        Dict: ocr label data dict
    """
    lbl_list = file["LabelGroup"]

    if int(lbl_list.get("NumberOfLabels", 0)) == 0:
        return {"polygons": [], "strings": [], "ignore": []}

    pols, strs = [], []

    lbl_list = lbl_list["Label"] if isinstance(lbl_list["Label"], list) else [lbl_list["Label"]]
    for l in lbl_list:
        if not l["Characters"]:
            continue
        # if self.ignore_whitespace:
        #     l["Characters"] = l["Characters"].replace(" ", "")
        strs.append(l["Characters"])
        n_point = max(
            [
                int(key.replace("@Vertex", "").replace("X", "").replace("Y", ""))
                for key in l["Coordinate"].keys()
                if "@Vertex" in key
            ]
        )
        pols.append(
            np.array(
                [
                    [
                        float(l["Coordinate"][f"@Vertex{idx + 1}X"]),
                        float(l["Coordinate"][f"@Vertex{idx + 1}Y"]),
                    ]
                    for idx in range(n_point)
                ]
            )
        )

    strings, kie_labels = [], []
    for string in strs:
        string, kie_label = _refine_kie_labels(string)
        strings.append(string)
        kie_labels.append(kie_label)

    ignore = [s is None for s in strs]

    lbl = {"polygons": pols, "strings": strings, "kie_labels": kie_labels, "ignore": ignore}

    return lbl

_refine_kie_labels

_refine_kie_labels(string)

Source code in SaigeToolkit/data/dataset/srproj_reader.py

def _refine_kie_labels(string):
    # split kie_label and string_label
    if string.startswith('"key'):
        index = string[1:].find('"') + 1

        kie_label = 2 * int(string[4:index]) + 1 if index > 4 else 0
        string = string[index + 1 :]

    elif string.startswith('"value'):
        index = string[1:].find('"') + 1

        kie_label = 2 * int(string[6:index]) + 2 if index > 6 else 0
        string = string[index + 1 :]

    else:
        kie_label = 0

    if kie_label < 0:
        kie_label = 0

    return string, kie_label

merge_srproj_classes

merge_srproj_classes(classes1, classes2)

Source code in SaigeToolkit/data/dataset/srproj_reader.py

def merge_srproj_classes(classes1, classes2):
    if classes1 == classes2:
        return classes1

    if len(classes1) > len(classes2):
        classes1, classes2 = classes2, classes1

    if classes2[: len(classes1)] != classes1:
        raise RuntimeError("two srproj cannot be mixed")

    return classes2

convert_srproj_class_color_to_rgb

convert_srproj_class_color_to_rgb(argb: int) -> Tuple[int, int, int]

srproj의 각 클래스 색 코드를 rgb 값으로 변환합니다.

Source code in SaigeToolkit/data/dataset/srproj_reader.py

def convert_srproj_class_color_to_rgb(argb: int) -> Tuple[int, int, int]:
    """srproj의 각 클래스 색 코드를 rgb 값으로 변환합니다."""
    a = (argb >> 24) & 255
    r = (argb >> 16) & 255
    g = (argb >> 8) & 255
    b = argb & 255
    return (r, g, b)