file_handler

util.file_handler

IMAGE_EXTENSION module-attribute

IMAGE_EXTENSION = ('.jpg', '.jpeg', '.png', '.ppm', '.bmp', '.pgm', '.tif', '.tiff', '.webp')

VIDEO_EXTENSION module-attribute

VIDEO_EXTENSION = ('.mp4', '.webm')

get_image_list module-attribute

get_image_list = partial(glob_files, extensions=IMAGE_EXTENSION)

ZstdFileHandler

ZstdFileHandler is a utility class for handling the compression and decompression of data using the Zstandard (Zstd) compression algorithm. This class provides static methods to save and load data in a compressed format using Zstd and pickle for serialization.

save_compressed_data staticmethod

save_compressed_data(data: Any, file_path: str) -> None

Compresses and saves any Python data object to a file.

This method serializes the given data object using pickle, compresses the serialized data using Zstd, and writes the compressed data to a file.

Parameters:

• data (Any) –

  The Python data object to be compressed and saved.

• file_path (str) –

  The path of the file where the compressed data will be stored. If the file already exists, it will be overwritten.

Returns:

• None –

  None

Raises:

• FileNotFoundError –

  If the specified file path is not valid.

• PicklingError –

  If the data object cannot be serialized.

• ZstdError –

  If compression fails.

Source code in SaigeToolkit/util/file_handler.py

@staticmethod
def save_compressed_data(data: Any, file_path: str) -> None:
    """
    Compresses and saves any Python data object to a file.

    This method serializes the given data object using pickle, compresses
    the serialized data using Zstd, and writes the compressed data to a file.

    Args:
        data (Any): The Python data object to be compressed and saved.
        file_path (str): The path of the file where the compressed data will
                         be stored. If the file already exists, it will be
                         overwritten.

    Returns:
        None

    Raises:
        FileNotFoundError: If the specified file path is not valid.
        pickle.PicklingError: If the data object cannot be serialized.
        zstd.ZstdError: If compression fails.
    """
    bytes_data = pickle.dumps(data)

    cctx = zstd.ZstdCompressor()
    compressed_data = cctx.compress(bytes_data)

    with open(file_path, "wb") as f:
        f.write(compressed_data)

load_compressed_data staticmethod

load_compressed_data(file_path: str) -> Any

Loads and decompresses data from a file.

This method reads compressed data from a file, decompresses it using Zstd, and then deserializes it using pickle to convert it back to the original Python data object.

Parameters:

• file_path (str) –

  The path of the file from which the compressed data is to be read. The file must exist and contain valid compressed data.

Returns:

• Any ( Any ) –

  The original Python data object that was compressed and saved in the file.

Raises:

• FileNotFoundError –

  If the specified file path does not exist.

• UnpicklingError –

  If the decompressed data cannot be deserialized.

• ZstdError –

  If decompression fails.

Source code in SaigeToolkit/util/file_handler.py

@staticmethod
def load_compressed_data(file_path: str) -> Any:
    """
    Loads and decompresses data from a file.

    This method reads compressed data from a file, decompresses it using Zstd,
    and then deserializes it using pickle to convert it back to the original
    Python data object.

    Args:
        file_path (str): The path of the file from which the compressed data
                         is to be read. The file must exist and contain valid
                         compressed data.

    Returns:
        Any: The original Python data object that was compressed and saved
             in the file.

    Raises:
        FileNotFoundError: If the specified file path does not exist.
        pickle.UnpicklingError: If the decompressed data cannot be deserialized.
        zstd.ZstdError: If decompression fails.
    """
    with open(file_path, "rb") as f:
        compressed_data = f.read()

    dctx = zstd.ZstdDecompressor()
    bytes_data = dctx.decompress(compressed_data)

    data = pickle.loads(bytes_data)

    return data

load_files

load_files(path)

Source code in SaigeToolkit/util/file_handler.py

def load_files(path):
    images = []

    def get_image_format(image_path):
        with Image.open(image_path) as image_file:
            try:
                # HELP: .bmp 확장자를 처리할 때 원인을 알수없는 에러 발생..
                transposed_img = ImageOps.exif_transpose(image_file)
            except ValueError as e:
                if str(e) == "unknown raw mode for given image mode":
                    transposed_img = image_file
                else:
                    raise e

            image_size = list(transposed_img.size)

        keys = os.path.dirname(image_path).split(os.sep)
        keys = keys + ["dummy" for _ in range(3 - len(keys))]
        domain, source, category = keys[-3:]

        return {
            "image_path": image_path,
            "domain": domain,
            "source": source,
            "category": category,
            "image_size": image_size,
            "label_extension": "srproj",
            "mode": "OpticalCharacterRecognition",
            "LabelGroup": {"NumberOfLabels": 0},
        }

    try:
        # default case: input (path) is str-type directing file or directory
        if os.path.isfile(path):
            if path.lower().endswith(IMAGE_EXTENSION):
                images.append(get_image_format(path))

            elif path.lower().endswith(VIDEO_EXTENSION):
                for img in extract_img_from_video(path):
                    images.append(get_image_format(img))

            else:
                with open(path, "r") as file:
                    data = json.load(file)
                if isinstance(data, list):
                    for d in data:
                        images.extend(load_files(d))
                else:
                    images.extend(load_files(data))

        elif os.path.isdir(path):
            for root, subdirs, files in os.walk(path):
                for f in files:
                    if not f.lower().endswith(IMAGE_EXTENSION):
                        continue

                    img = os.path.join(root, f)
                    images.append(get_image_format(img))

    except TypeError:
        raise NotImplementedError(f"no type implemented {type(path)}")

    except:
        raise RuntimeError(f"no images applied from: {path}")

    return images

extract_img_from_video

extract_img_from_video(video_path, frame_rate=10) -> List[str]

Source code in SaigeToolkit/util/file_handler.py

def extract_img_from_video(video_path, frame_rate=10) -> List[str]:
    video_name = os.path.basename(video_path)
    video_dir = os.path.dirname(video_path)
    vidcap = cv2.VideoCapture(video_path)
    temporary_img_folder_path = os.path.join(video_dir, video_name + "_cached_images")
    os.makedirs(temporary_img_folder_path, exist_ok=True)

    fps = vidcap.get(cv2.CAP_PROP_FPS)
    frame_count = int(vidcap.get(cv2.CAP_PROP_FRAME_COUNT))
    duration = frame_count / fps

    image_count = 0
    sampling_time_list = np.arange(0, duration, 1 / frame_rate)

    images = []

    def get_and_save_frame(sec):
        vidcap.set(cv2.CAP_PROP_POS_MSEC, sec * 1000)
        has_frame, image = vidcap.read()
        if has_frame:
            img_path = os.path.join(temporary_img_folder_path, format(image_count, "05d") + ".jpg")

            if os.path.isfile(img_path):
                images.append(img_path)
            else:
                cv2.imwrite(img_path, image)
                images.append(img_path)

        return has_frame

    print(f'[{"DATA".center(9)}] Processing video... target directory: {temporary_img_folder_path}')
    for time in tqdm(sampling_time_list):
        if get_and_save_frame(time):
            image_count += 1

    return images

glob_files

glob_files(root_path: str, extensions: Tuple[str], recursive: bool = True, skip_hidden_directories: bool = True, max_directories: Optional[int] = None, max_files: Optional[int] = None, relative_path: bool = False) -> Tuple[List[str], bool, bool]

glob files with specified extensions

Parameters:

• root_path (str) –

  description

• extensions (Tuple[str]) –

  description

• recursive (bool, default: True ) –

  description. Defaults to True.

• skip_hidden_directories (bool, default: True ) –

  description. Defaults to True.

• max_directories (Optional[int], default: None ) –

  max number of directories to search. Defaults to None.

• max_files (Optional[int], default: None ) –

  max file number limit. Defaults to None.

• relative_path (bool, default: False ) –

  description. Defaults to False.

Returns:

• Tuple[List[str], bool, bool] –

  Tuple[List[str], bool, bool]: description

Source code in SaigeToolkit/util/file_handler.py

def glob_files(
    root_path: str,
    extensions: Tuple[str],
    recursive: bool = True,
    skip_hidden_directories: bool = True,
    max_directories: Optional[int] = None,
    max_files: Optional[int] = None,
    relative_path: bool = False,
) -> Tuple[List[str], bool, bool]:
    """glob files with specified extensions

    Args:
        root_path (str): _description_
        extensions (Tuple[str]): _description_
        recursive (bool, optional): _description_. Defaults to True.
        skip_hidden_directories (bool, optional): _description_. Defaults to True.
        max_directories (Optional[int], optional): max number of directories to search. Defaults to None.
        max_files (Optional[int], optional): max file number limit. Defaults to None.
        relative_path (bool, optional): _description_. Defaults to False.

    Returns:
        Tuple[List[str], bool, bool]: _description_
    """
    paths = []
    hit_max_directories = False
    hit_max_files = False
    for directory_idx, (directory, _, fnames) in enumerate(os.walk(root_path, followlinks=True)):
        if skip_hidden_directories and os.path.basename(directory).startswith("."):
            continue

        if max_directories is not None and directory_idx >= max_directories:
            hit_max_directories = True
            break

        paths += [
            os.path.join(directory, fname)
            for fname in sorted(fnames)
            if fname.lower().endswith(extensions)
        ]

        if not recursive:
            break

        if max_files is not None and len(paths) > max_files:
            hit_max_files = True
            paths = paths[:max_files]
            break

    if relative_path:
        paths = [os.path.relpath(p, root_path) for p in paths]

    return paths, hit_max_directories, hit_max_files

uniquify_path

uniquify_path(path: str, extension: Optional[str] = None)

Source code in SaigeToolkit/util/file_handler.py

def uniquify_path(path: str, extension: Optional[str] = None):
    if extension is None:
        extension = os.path.splitext(path)[-1]

    filename = path
    if len(extension) > 0:
        filename = filename[: -len(extension)]

    counter = 1
    while os.path.exists(path):
        path = filename + " (" + str(counter) + ")" + extension
        counter += 1

    return path