Skip to content

file_handler

util.file_handler

IMAGE_EXTENSION module-attribute 

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

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 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@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 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@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

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 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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