Skip to content

util

Module diagram

classDiagram
  class util {
  }
  class color {
  }
  class config {
  }
  class cuda {
  }
  class file_handler {
  }
  class function_tree {
  }
  class lazy_loader {
  }
  class memory {
  }
  class pickle {
  }
  class registry {
  }
  class reproducibility {
  }
  class string {
  }
  class system {
  }
  class test {
  }
  class timer {
  }
  class version {
  }
  system --> reproducibility

util

color

sample_uniform_hue_colors 

sample_uniform_hue_colors(num_colors: int, saturation: float = 1.0, brightness: float = 1.0) -> List[List[int]]

Hue space에서 동일한 간격으로 서로 다른 색들을 샘플링합니다.

Parameters:

  • num_colors (int) –

    number of colors to sample

  • saturation (float, default: 1.0 ) –

    saturation. Defaults to 1.0.

  • brightness (float, default: 1.0 ) –

    value. Defaults to 1.0.

Returns:

  • List[List[int]] –

    List[List[int]]: [num_colors x [R, G, B]]

Source code in SaigeToolkit/util/color.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def sample_uniform_hue_colors(
    num_colors: int,
    saturation: float = 1.0,
    brightness: float = 1.0,
) -> List[List[int]]:
    """Hue space에서 동일한 간격으로 서로 다른 색들을 샘플링합니다.

    Args:
        num_colors (int): number of colors to sample
        saturation (float, optional): saturation. Defaults to 1.0.
        brightness (float, optional): value. Defaults to 1.0.

    Returns:
        List[List[int]]: [num_colors x [R, G, B]]

    """

    h = np.linspace(0.0, 1.0 * (num_colors - 1) / num_colors, num_colors)[:, np.newaxis]
    s = np.ones((num_colors, 1)) * saturation
    v = np.ones((num_colors, 1)) * brightness
    rgb = hsv_to_rgb(np.concatenate([h, s, v], axis=1))
    rgb = (rgb * 255).astype(int).tolist()
    return rgb

config

read_yml 

read_yml(config: str) -> Dict[str, Any]

load config dictionary

Parameters:

  • config (str) –

    file path

Returns:

  • Dict[str, Any] –

    Dict[str, Any]: result config dict

Source code in SaigeToolkit/util/config.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
def read_yml(config: str) -> Dict[str, Any]:
    """load config dictionary

    Args:
        config (str): file path

    Returns:
        Dict[str, Any]: result config dict
    """
    cfg = _load_yml(config)

    cfg = update_import(cfg)

    return cfg

_load_yml 

_load_yml(file: str) -> Dict[str, Any]

load config dictionary from *.yml file

Source code in SaigeToolkit/util/config.py 
24
25
26
27
28
29
30
31
32
def _load_yml(file: str) -> Dict[str, Any]:
    """load config dictionary from *.yml file"""

    assert os.path.isfile(file), f"string type data config should direct a file {file}"

    with open(file, encoding="utf-8") as fp:
        cfg = yaml.load(fp, Loader=yaml.FullLoader)

    return cfg

update_import 

update_import(cfg: Dict[str, Any]) -> Dict[str, Any]

recursively import and override base config, and there is two conditions of importing:

  • if dict has import key and its value pointing another config file (*.yml)
  • if dict value is pointing another config file (*.yml)

Parameters:

  • cfg (Dict[str, Any]) –

    base config

Returns:

  • Dict[str, Any] –

    Dict[str, Any]: result config (import conditions updated)

Source code in SaigeToolkit/util/config.py 
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
63
64
65
66
def update_import(cfg: Dict[str, Any]) -> Dict[str, Any]:
    """recursively import and override base config,
    and there is two conditions of importing:

    - if dict has `import` key and its value pointing another config file (*.yml)
    - if dict value is pointing another config file (*.yml)

    Args:
        cfg (Dict[str, Any]): base config

    Returns:
        Dict[str, Any]: result config (import conditions updated)
    """

    # 1. has `import` key and its value pointing another config file
    cfg_import = cfg.pop("import", None)
    if cfg_import:
        # update_import function is called for recursive update
        cfg_mother = update_import(_load_yml(cfg_import))
        cfg = override_dict(cfg_mother, cfg)

    for k, v in cfg.items():
        # if there's no import and value is dict type, update recursively
        if isinstance(v, dict):
            cfg[k] = update_import(v)

        # 2. value is pointing another config file (*.yml)
        elif isinstance(v, str) and v.endswith(".yml"):
            # update_import function is called for recursive update
            cfg[k] = update_import(_load_yml(v))

    return cfg

save_yml 

save_yml(config: Dict[str, Any], path: str)
Source code in SaigeToolkit/util/config.py 
69
70
71
72
73
74
75
76
77
78
79
def save_yml(config: Dict[str, Any], path: str):
    yaml.add_representer(
        list,
        lambda dumper, value: dumper.represent_sequence("tag:yaml.org,2002:seq", value, flow_style=True),
    )
    yaml.add_representer(
        type(None), lambda dumper, value: dumper.represent_scalar("tag:yaml.org,2002:null", "")
    )

    with open(path, "w") as f:
        yaml.dump(config, f, sort_keys=False, indent=4, allow_unicode=True, width=1000)

override_dict 

override_dict(target: Dict, source: Dict) -> Dict

recursively override dictionaries

Source code in SaigeToolkit/util/config.py 
82
83
84
85
86
87
88
89
90
91
def override_dict(target: Dict, source: Dict) -> Dict:
    """recursively override dictionaries"""
    if not source:
        return target
    for k, v in source.items():
        if k in target and isinstance(target[k], Dict) and isinstance(v, dict):
            target[k] = override_dict(target[k], v)
        else:
            target[k] = v
    return target

convert_keys_string_to_list 

convert_keys_string_to_list(function)

keys 입력값이 str인 경우 . 으로 분할해 함수에 전달해줍니다.

Source code in SaigeToolkit/util/config.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def convert_keys_string_to_list(function):
    """
    keys 입력값이 str인 경우 `.` 으로 분할해 함수에 전달해줍니다.
    """

    @wraps(function)
    def wrapper(obj, keys, *args, **kwargs):
        if isinstance(keys, str):
            keys = keys.split(".")
        return function(obj, keys, *args, **kwargs)

    return wrapper

check_key_exists 

check_key_exists(obj: Union[Dict, Tuple, List], key: Any)
Source code in SaigeToolkit/util/config.py 
108
109
110
111
112
113
def check_key_exists(obj: Union[Dict, Tuple, List], key: Any):
    if isinstance(obj, Dict) and key not in obj:
        return False
    elif isinstance(obj, (Tuple, List)) and len(obj) < key + 1:
        return False
    return True

get_tree_node 

get_tree_node(obj: Any, keys: List) -> Any

tree-like obj에 대해 keys의 값을 순서대로 각 레벨에서 탐색해 최종 노드의 값을 리턴, 최종 노드가 존재하지 않으면 error 발생

Source code in SaigeToolkit/util/config.py 
116
117
118
119
120
121
122
123
124
125
126
127
@convert_keys_string_to_list
def get_tree_node(obj: Any, keys: List) -> Any:
    """tree-like obj에 대해 keys의 값을 순서대로 각 레벨에서 탐색해 최종 노드의 값을 리턴, 최종 노드가 존재하지 않으면 error 발생"""
    for key in keys:
        if isinstance(obj, (Tuple, List)) and isinstance(key, str):
            key = int(key)
        if not check_key_exists(obj=obj, key=key):
            raise KeyError
        if not isinstance(obj, (Tuple, Dict, List)):
            raise TypeError
        obj = obj[key]
    return obj

set_tree_node 

set_tree_node(obj: Any, keys: List, value: Any) -> Any

tree-like obj에 대해 keys의 값을 순서대로 각 레벨에서 탐색해 최종 노드의 값을 수정, 최종 노드가 존재하지 않으면 error 발생

Source code in SaigeToolkit/util/config.py 
130
131
132
133
134
135
136
137
138
139
140
141
@convert_keys_string_to_list
def set_tree_node(obj: Any, keys: List, value: Any) -> Any:
    """tree-like obj에 대해 keys의 값을 순서대로 각 레벨에서 탐색해 최종 노드의 값을 수정, 최종 노드가 존재하지 않으면 error 발생"""
    obj = get_tree_node(obj=obj, keys=keys[:-1])
    key = keys[-1]
    if isinstance(obj, List) and isinstance(key, str):
        key = int(key)
    if not check_key_exists(obj=obj, key=key):
        raise KeyError
    if not isinstance(obj, (Tuple, Dict, List)):
        raise TypeError
    obj[key] = value

get_tree_node_with_default 

get_tree_node_with_default(obj: Any, keys: List, value: Optional[Any] = None, ignore_type_error: bool = False) -> Any

tree-like obj에 대해 keys의 값을 순서대로 각 레벨에서 탐색해 최종 노드의 값을 리턴, 최종 노드가 존재하지 않으면 value 리턴, ignore_type_error=True 일 경우 최종 value가 더이상 탐색하지 못할 때에도 error 대신 default value를 리턴.

Source code in SaigeToolkit/util/config.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@convert_keys_string_to_list
def get_tree_node_with_default(
    obj: Any, keys: List, value: Optional[Any] = None, ignore_type_error: bool = False
) -> Any:
    """tree-like obj에 대해 keys의 값을 순서대로 각 레벨에서 탐색해 최종 노드의 값을 리턴, 최종 노드가 존재하지 않으면 value 리턴,
    ignore_type_error=True 일 경우 최종 value가 더이상 탐색하지 못할 때에도 error 대신 default value를 리턴.
    """
    try:
        obj = get_tree_node(obj, keys)
    except KeyError:
        obj = value
    except TypeError:
        if not ignore_type_error:
            raise TypeError
        obj = value

    return obj

flatten_tree 

flatten_tree(obj, parent_key='', sep='.', depth=0, max_depth=None) -> Dict

tree-like obj를 1 레벨 dictionary로 변환

Source code in SaigeToolkit/util/config.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def flatten_tree(obj, parent_key="", sep=".", depth=0, max_depth=None) -> Dict:
    """tree-like obj를 1 레벨 dictionary로 변환"""
    flattened = dict()
    if isinstance(obj, Dict):
        iterator = obj.items()
    else:
        iterator = enumerate(obj)

    for key, value in iterator:
        flattened_key = f"{parent_key}{sep}{key}" if parent_key else key
        if isinstance(value, (Dict, List, Tuple)) and (max_depth is None or depth < max_depth):
            nested = flatten_tree(
                value,
                parent_key=flattened_key,
                sep=sep,
                depth=depth + 1,
                max_depth=max_depth,
            )
        else:
            nested = {flattened_key: value}

        for flattened_key, flattened_value in nested.items():
            if flattened_key in flattened:
                raise KeyError(f"multiple {flattened_key} exist in tree")
            flattened[flattened_key] = flattened_value

    return flattened

cuda

init_cuda_when_available 

init_cuda_when_available()

init cuda when available. if cuda device has error and requires reboot, it will raise CudaDeviceError.

Raises:

  • CudaDeviceError –

    cuda device error.

Source code in SaigeToolkit/util/cuda.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def init_cuda_when_available():
    """init cuda when available.
    if cuda device has error and requires reboot, it will raise CudaDeviceError.

    Raises:
        CudaDeviceError: cuda device error.
    """
    try:
        # check cuda device is available
        _is_available = torch.cuda.is_available()
    except Exception as e:
        # if any error occurs while checking cuda.is_available, raise CudaDeviceError
        raise CudaDeviceError from e

    if _is_available:
        try:
            # init cuda
            torch.cuda.init()
        except Exception as exc:
            # if any error occurs while init cuda, raise CudaDeviceError
            raise CudaDeviceError from exc

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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@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 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@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

FileIOHandler 

FileIOHandler(max_workers: int = 16)

File save/load 및 동시성 프로그래밍을 지원하는 class 입니다. 이 class에서 save/load를 위해 지원하는 패키지는 zstandard, numpy, pickle이며 save/load 함수에서 각각 class 멤버변수 ZSTD, NUMPY, PICKLE로 접근할 수 있습니다.

Usage 
# generate file handler
file_handler = FileIOHandler(max_workers=16)

# image load
image_paths = [
    AnalysisParams.get_intermediate_file_path(image_info["save_dir"])
    for _, image_info in images.items()
]
load_results = file_handler.load_concurrent(image_paths, FileIOHandler.PICKLE)

# analyze
for (image_id, image_info), load_result in zip(images.items(), load_results):
    ... # analyze process
    file_handler.save_concurrent(analysis_path, analysis, FileIOHandler.PICKLE)

# wait for file save
file_handler.wait_save()
Source code in SaigeToolkit/util/file_handler.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def __init__(self, max_workers: int = 16):
    self._max_workers = max_workers
    self.concurrent_save_pool = ThreadPoolExecutor(max_workers=self._max_workers)
    self.concurrent_load_pool = ThreadPoolExecutor(max_workers=self._max_workers)

    self.save_futures = list()
    self.load_futures = list()

    self.save_func = {
        self.FileIOPackage.ZSTD: self._save_zstd,
        self.FileIOPackage.PICKLE: self._save_pickle,
        self.FileIOPackage.NUMPY: self._save_np,
    }
    self.load_func = {
        self.FileIOPackage.ZSTD: self._load_zstd,
        self.FileIOPackage.PICKLE: self._load_pickle,
        self.FileIOPackage.NUMPY: self._load_np,
    }
ZSTD class-attribute instance-attribute 
ZSTD = ZSTD
NUMPY class-attribute instance-attribute 
NUMPY = NUMPY
PICKLE class-attribute instance-attribute 
PICKLE = PICKLE
_max_workers instance-attribute 
_max_workers = max_workers
concurrent_save_pool instance-attribute 
concurrent_save_pool = ThreadPoolExecutor(max_workers=(_max_workers))
concurrent_load_pool instance-attribute 
concurrent_load_pool = ThreadPoolExecutor(max_workers=(_max_workers))
save_futures instance-attribute 
save_futures = list()
load_futures instance-attribute 
load_futures = list()
save_func instance-attribute 
save_func = {ZSTD: _save_zstd, PICKLE: _save_pickle, NUMPY: _save_np}
load_func instance-attribute 
load_func = {ZSTD: _load_zstd, PICKLE: _load_pickle, NUMPY: _load_np}
max_workers property writable 
max_workers: int
FileIOPackage

Bases: Flag

ZSTD class-attribute instance-attribute 
ZSTD = auto()
NUMPY class-attribute instance-attribute 
NUMPY = auto()
PICKLE class-attribute instance-attribute 
PICKLE = auto()
_save_pickle 
_save_pickle(file_path: str, data: Any) -> None
Source code in SaigeToolkit/util/file_handler.py 
233
234
235
def _save_pickle(self, file_path: str, data: Any) -> None:
    with open(file_path, "wb") as f:
        pickle.dump(data, f)
_load_pickle 
_load_pickle(file_path: str) -> Any
Source code in SaigeToolkit/util/file_handler.py 
237
238
239
240
def _load_pickle(self, file_path: str) -> Any:
    with open(file_path, "rb") as f:
        data = pickle.load(f)
    return data
_save_np 
_save_np(file_path: str, data: Any, allow_pickle: bool = False) -> None
Source code in SaigeToolkit/util/file_handler.py 
242
243
def _save_np(self, file_path: str, data: Any, allow_pickle: bool = False) -> None:
    np.savez(file_path, **data, allow_pickle=allow_pickle)
_load_np 
_load_np(file_path: str, allow_pickle: bool = False) -> Any
Source code in SaigeToolkit/util/file_handler.py 
245
246
def _load_np(self, file_path: str, allow_pickle: bool = False) -> Any:
    return np.load(file_path, allow_pickle=allow_pickle)
_save_zstd 
_save_zstd(file_path: str, data: Any) -> None
Source code in SaigeToolkit/util/file_handler.py 
248
249
def _save_zstd(self, file_path: str, data: Any) -> None:
    ZstdFileHandler.save_compressed_data(data, file_path)
_load_zstd 
_load_zstd(file_path: str) -> Any
Source code in SaigeToolkit/util/file_handler.py 
251
252
253
254
255
256
257
258
def _load_zstd(self, file_path: str) -> Any:
    try:
        result = ZstdFileHandler.load_compressed_data(file_path)
    except (zstd.ZstdError, pickle.UnpicklingError):
        # 하위호환성을 위해 기존 pickle 파일도 읽을 수 있도록 함
        result = self._load_pickle(file_path)

    return result
save 
save(file_path: str, data: Any, package_type: FileIOPackage = FileIOPackage.ZSTD, **kwargs)

Save any Python data object to a file.

This method saves the data using Zstd, numpy and pickle package. Zstd saves with ZstdFileHandler package, numpy saves with np.savez and pickle saves with pickle.dump.

Parameters:

  • file_path (str) –

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

  • data (Any) –

    The Python data object to be saved.

  • package_type (FileIOPackage, default: ZSTD ) –

    Package to save. package_type must be one of FileIOHandler.ZSTD, FileIOHandler.NUMPY or FileIOHandler.PICKLE. (default FileIOHandler.ZSTD)

Returns:

  • –

    None

Raises:

  • FileIOPackageError –

    If package_type is invalid I/O package.

Source code in SaigeToolkit/util/file_handler.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def save(
    self, file_path: str, data: Any, package_type: FileIOPackage = FileIOPackage.ZSTD, **kwargs
):
    """
    Save any Python data object to a file.

    This method saves the data using Zstd, numpy and pickle package. Zstd saves with
    `ZstdFileHandler` package, numpy saves with `np.savez` and pickle saves with `pickle.dump`.

    Args:
        file_path (str): The path of the file where the data will be stored.
                         If the file already exists, it will be overwritten.
        data (Any): The Python data object to be saved.
        package_type (FileIOPackage): Package to save. package_type must be one of
                                      `FileIOHandler.ZSTD`, `FileIOHandler.NUMPY` or `FileIOHandler.PICKLE`.
                                      (default `FileIOHandler.ZSTD`)

    Returns:
        None

    Raises:
        FileIOPackageError: If package_type is invalid I/O package.
    """
    if package_type not in self.FileIOPackage:
        raise FileIOPackageError

    os.makedirs(os.path.dirname(file_path), exist_ok=True)
    try:
        self.save_func[package_type](file_path, data, **kwargs)
    except Exception as e:
        raise FileIOHandlerSaveError(str(e))
save_concurrent 
save_concurrent(file_path: str, data: Any, package_type: FileIOPackage = FileIOPackage.ZSTD, **kwargs)

Save any Python data object to a file with concurrent method.

This method saves the data using Zstd, numpy and pickle package. Zstd saves with ZstdFileHandler package, numpy saves with np.savez and pickle saves with pickle.dump.

Parameters:

  • file_path (str) –

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

  • data (Any) –

    The Python data object to be saved.

  • package_type (FileIOPackage, default: ZSTD ) –

    Package to save. package_type must be one of FileIOHandler.ZSTD, FileIOHandler.NUMPY or FileIOHandler.PICKLE. (default FileIOHandler.ZSTD)

Returns:

  • –

    None

Raises:

  • FileIOPackageError –

    If package_type is invalid I/O package.

Source code in SaigeToolkit/util/file_handler.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def save_concurrent(
    self, file_path: str, data: Any, package_type: FileIOPackage = FileIOPackage.ZSTD, **kwargs
):
    """
    Save any Python data object to a file with concurrent method.

    This method saves the data using Zstd, numpy and pickle package. Zstd saves with
    `ZstdFileHandler` package, numpy saves with `np.savez` and pickle saves with `pickle.dump`.

    Args:
        file_path (str): The path of the file where the data will be stored.
                         If the file already exists, it will be overwritten.
        data (Any): The Python data object to be saved.
        package_type (FileIOPackage): Package to save. package_type must be one of
                                      `FileIOHandler.ZSTD`, `FileIOHandler.NUMPY` or `FileIOHandler.PICKLE`.
                                      (default `FileIOHandler.ZSTD`)

    Returns:
        None

    Raises:
        FileIOPackageError: If package_type is invalid I/O package.
    """
    if package_type not in self.FileIOPackage:
        raise FileIOPackageError

    os.makedirs(os.path.dirname(file_path), exist_ok=True)
    try:
        self.save_futures.append(
            self.concurrent_save_pool.submit(self.save_func[package_type], file_path, data, **kwargs)
        )
    except Exception as e:
        raise FileIOHandlerSaveError(str(e))
load 
load(file_path: str, package_type: FileIOPackage = FileIOPackage.ZSTD, **kwargs)

Loads data from a file.

This method reads data from a file using Zstd, numpy and pickle package. Zstd reads with ZstdFileHandler package, numpy reads with np.load and pickle reads with pickle.load.

Parameters:

  • file_path (str) –

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

  • package_type (FileIOPackage, default: ZSTD ) –

    Package to load. package_type must be one of FileIOHandler.ZSTD, FileIOHandler.NUMPY or FileIOHandler.PICKLE. (default FileIOHandler.ZSTD)

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.

  • FileIOPackageError –

    If package_type is invalid I/O package.

  • UnpicklingError –

    If the decompressed data cannot be deserialized.

  • ZstdError –

    If decompression fails.

Source code in SaigeToolkit/util/file_handler.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
def load(self, file_path: str, package_type: FileIOPackage = FileIOPackage.ZSTD, **kwargs):
    """
    Loads data from a file.

    This method reads data from a file using Zstd, numpy and pickle package.
    Zstd reads with `ZstdFileHandler` package, numpy reads with `np.load` and
    pickle reads with `pickle.load`.

    Args:
        file_path (str): The path of the file from which the data is to be read.
                         The file must exist and contain valid data.
        package_type (FileIOPackage): Package to load. package_type must be one of
                                      `FileIOHandler.ZSTD`, `FileIOHandler.NUMPY` or `FileIOHandler.PICKLE`.
                                      (default `FileIOHandler.ZSTD`)

    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.
        FileIOPackageError: If package_type is invalid I/O package.
        pickle.UnpicklingError: If the decompressed data cannot be deserialized.
        zstd.ZstdError: If decompression fails.
    """
    if package_type not in self.FileIOPackage:
        raise FileIOPackageError

    try:
        return self.load_func[package_type](file_path, **kwargs)
    except FileNotFoundError as e:
        raise FileIOHandlerLoadError(str(e))
load_concurrent 
load_concurrent(file_paths: str | List[str], package_type: FileIOPackage = FileIOPackage.ZSTD, **kwargs)

Loads data from a multiple files with concurrent method.

This method reads data from a file using Zstd, numpy and pickle package. Zstd reads with ZstdFileHandler package, numpy reads with np.load and pickle reads with pickle.load.

Parameters:

  • file_paths (str | List[str]) –

    The paths of the files from which the data is to be read. The files must exist and contain valid data.

  • package_type (FileIOPackage, default: ZSTD ) –

    Package to load. package_type must be one of FileIOHandler.ZSTD, FileIOHandler.NUMPY or FileIOHandler.PICKLE. (default FileIOHandler.ZSTD)

Returns:

  • –

    List[Any]: The list of original Python data object that was saved in the files.

Raises:

  • FileNotFoundError –

    If the specified file path does not exist.

  • FileIOPackageError –

    If package_type is invalid I/O package.

  • UnpicklingError –

    If the decompressed data cannot be deserialized.

  • ZstdError –

    If decompression fails.

Source code in SaigeToolkit/util/file_handler.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
def load_concurrent(
    self, file_paths: str | List[str], package_type: FileIOPackage = FileIOPackage.ZSTD, **kwargs
):
    """
    Loads data from a multiple files with concurrent method.

    This method reads data from a file using Zstd, numpy and pickle package.
    Zstd reads with `ZstdFileHandler` package, numpy reads with `np.load` and
    pickle reads with `pickle.load`.

    Args:
        file_paths (str | List[str]): The paths of the files from which the data is to be read.
                                      The files must exist and contain valid data.
        package_type (FileIOPackage): Package to load. package_type must be one of
                                      `FileIOHandler.ZSTD`, `FileIOHandler.NUMPY` or `FileIOHandler.PICKLE`.
                                      (default `FileIOHandler.ZSTD`)

    Returns:
        List[Any]: The list of original Python data object that was saved in the files.

    Raises:
        FileNotFoundError: If the specified file path does not exist.
        FileIOPackageError: If package_type is invalid I/O package.
        pickle.UnpicklingError: If the decompressed data cannot be deserialized.
        zstd.ZstdError: If decompression fails.
    """
    if package_type not in self.FileIOPackage:
        raise FileIOPackageError

    if isinstance(file_paths, str):
        file_paths = [file_paths]

    try:
        for file_path in file_paths:
            self.load_futures.append(
                self.concurrent_load_pool.submit(self.load_func[package_type], file_path, **kwargs)
            )
    except FileNotFoundError as e:
        raise FileIOHandlerLoadError(str(e))

    return self.wait_load(return_when=ALL_COMPLETED)
wait_save 
wait_save(timeout=None, return_when=ALL_COMPLETED)

Wait for save files that concurrently processing.

This method must be called when save files concurrently. If saving files concurrently and not call this method, can save abnormaly or cannot finish to save.

Parameters:

  • timeout (int | float, default: None ) –

    Control the maximum number of seconds to wait before returning.

  • return_when (str, default: ALL_COMPLETED ) –

    Indicates when this function should return.

Returns:

  • –

    None

Source code in SaigeToolkit/util/file_handler.py 
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
def wait_save(self, timeout=None, return_when=ALL_COMPLETED):
    """
    Wait for save files that concurrently processing.

    This method must be called when save files concurrently.
    If saving files concurrently and not call this method, can save abnormaly or
    cannot finish to save.

    Args:
        timeout (int | float): Control the maximum number of seconds to wait before returning.
        return_when (str): Indicates when this function should return.

    Returns:
        None
    """
    futures.wait(self.save_futures, timeout=timeout, return_when=return_when)
    self.save_futures.clear()
wait_load 
wait_load(timeout=None, return_when=ALL_COMPLETED)
Source code in SaigeToolkit/util/file_handler.py 
418
419
420
421
422
def wait_load(self, timeout=None, return_when=ALL_COMPLETED):
    futures.wait(self.load_futures, timeout=timeout, return_when=return_when)
    load_result = [loaded_data.result() for loaded_data in self.load_futures]
    self.load_futures.clear()
    return load_result

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 
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
63
64
65
66
67
68
69
70
71
72
73
74
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

function_tree

FunctionBase

Bases: Callable

Base node definition for operation graph

Example 
class A(FunctionBase):
    def __call__(self):
        pass

class B(FunctionBase):
    prerequisites = [A]

    def __call__(self):
        pass

sort_functions([B])
# [A, B]
prerequisites class-attribute instance-attribute 
prerequisites: List[FunctionBase] = []

sort_functions 

sort_functions(functions: List[Type[FunctionBase]], roots: Optional[Type[FunctionBase]] = None) -> List[Type[FunctionBase]]

get sorted output functions with topological sort.

Parameters:

  • functions (List[Type[FunctionBase]]) –

    desired function classes

  • roots (Optional[Type[FunctionBase]], default: None ) –

    tree root function classes

Returns:

  • List[Type[FunctionBase]] –

    List[Type[FunctionBase]]: ordered list of all required function classes

Source code in SaigeToolkit/util/function_tree.py 
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
63
64
65
66
67
def sort_functions(
    functions: List[Type[FunctionBase]],
    roots: Optional[Type[FunctionBase]] = None,
) -> List[Type[FunctionBase]]:
    """get sorted output functions with topological sort.

    Args:
        functions (List[Type[FunctionBase]]): desired function classes
        roots (Optional[Type[FunctionBase]]): tree root function classes

    Returns:
        List[Type[FunctionBase]]: ordered list of all required function classes
    """

    if roots is None:
        roots = []

    call_order = []
    visited = []

    def _sort_tree(function=None, prerequisites=None):
        if function is not None:
            visited.append(function)

        prerequisites = prerequisites if prerequisites is not None else function.prerequisites
        prerequisites = [p for p in prerequisites if p not in roots]

        while prerequisites:
            p = prerequisites.pop(0)

            if p not in visited:
                _sort_tree(function=p)

        if function not in call_order and function is not None:
            call_order.append(function)

    _sort_tree(prerequisites=functions)

    return call_order

lazy_loader

모듈이나 객체를 실제 사용할 때까지 import를 지연시키는 클래스를 정의합니다.

_LazyModule 

_LazyModule(local_name: str, module_name: str, globals: Dict[str, Any])

Bases: ModuleType

Source code in SaigeToolkit/util/lazy_loader.py 
26
27
28
29
30
31
32
33
34
35
36
def __init__(
    self,
    local_name: str,
    module_name: str,
    globals: Dict[str, Any],
):
    super().__init__(name=module_name)

    self._local_name = local_name
    self._module_name = module_name
    self._globals = globals
_local_name instance-attribute 
_local_name = local_name
_module_name instance-attribute 
_module_name = module_name
_globals instance-attribute 
_globals = globals
_load 
_load() -> ModuleType
Source code in SaigeToolkit/util/lazy_loader.py 
38
39
40
41
42
def _load(self) -> ModuleType:
    module = import_module(name=self._module_name)
    self._globals[self._local_name] = module
    self.__dict__.update(module.__dict__)
    return module
__getattr__ 
__getattr__(name: str) -> Any
Source code in SaigeToolkit/util/lazy_loader.py 
44
45
46
def __getattr__(self, name: str) -> Any:
    module = self._load()
    return getattr(module, name)
__repr__ 
__repr__() -> str
Source code in SaigeToolkit/util/lazy_loader.py 
48
49
50
def __repr__(self) -> str:
    # For debugging
    return f"<lazy module '{self._module_name}'>"

_LazyClassMeta 

_LazyClassMeta(name, bases, namespace)

Bases: type

Source code in SaigeToolkit/util/lazy_loader.py 
54
55
56
57
58
59
60
def __init__(cls, name, bases, namespace):
    super().__init__(name, bases, namespace)
    cls._local_name = namespace["local_name"]
    cls._class_name = namespace["class_name"]
    cls._module_name = namespace["module_name"]
    cls._globals = namespace["globals"]
    cls._class = None
_load 
_load()
Source code in SaigeToolkit/util/lazy_loader.py 
62
63
64
65
66
67
def _load(cls):
    module = import_module(name=cls._module_name)
    _class = getattr(module, cls._class_name)
    cls._globals[cls._local_name] = _class
    cls._class = _class
    return _class
__call__ 
__call__(*args, **kwargs)
Source code in SaigeToolkit/util/lazy_loader.py 
69
70
71
def __call__(cls, *args, **kwargs):
    _class = cls._class or cls._load()
    return _class(*args, **kwargs)
__getattr__ 
__getattr__(name)
Source code in SaigeToolkit/util/lazy_loader.py 
73
74
75
def __getattr__(cls, name):
    _class = cls._class or cls._load()
    return getattr(_class, name)
__repr__ 
__repr__() -> str
Source code in SaigeToolkit/util/lazy_loader.py 
77
78
79
def __repr__(cls) -> str:
    # For debugging
    return f"<lazy class '{cls._module_name}.{cls._class_name}'>"

_resolve_package 

_resolve_package(package: Optional[str], parent_globals: Dict[str, Any]) -> Optional[str]
Source code in SaigeToolkit/util/lazy_loader.py 
13
14
15
16
17
18
19
20
21
22
def _resolve_package(package: Optional[str], parent_globals: Dict[str, Any]) -> Optional[str]:
    if package is None:
        return None
    elif package.startswith("."):
        base_package = parent_globals.get("__package__")
        if base_package is None:
            raise ImportError("attempted relative import with no known parent package")
        return base_package if package == "." else base_package + package
    else:
        return package

lazy_module 

lazy_module(name_as: str, /, import_from: Optional[str] = None, name: Optional[str] = None, *, globals: Optional[Dict[str, Any]] = None)

A wrapper for modules that delays the import until it is needed. After the module is imported, it is stored in the namespace.

Parameters:

  • name_as (str) –

    The name of the module in the namespace.

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

    The base package of the module.

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

    The name of the module to import. Defaults to the value of name_as.

  • globals (dict, default: None ) –

    The current namespace (pass globals() here).

Usage 
# import numpy as np
np = lazy_module("np", name="numpy", globals=globals())
print(np)  # <lazy module 'numpy'>

np.random.rand(5)  # `numpy` is imported here
print(np)  # <module 'numpy' from '...'>

# import torch
torch = lazy_module("torch", globals=globals())

# from torchvision.models import resnet
resnet = lazy_module("resnet", "torchvision.models", globals=globals())

# from torch.nn import functional as F
F = lazy_module("F", "torch.nn", "functional", globals=globals())

# from .engine import api
api = lazy_module("api", ".engine", globals=globals())
Note

The name_as parameter should match the name of the variable in the namespace.

Source code in SaigeToolkit/util/lazy_loader.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def lazy_module(
    name_as: str,
    /,
    import_from: Optional[str] = None,
    name: Optional[str] = None,
    *,
    globals: Optional[Dict[str, Any]] = None,
):
    """
    A wrapper for modules that delays the import until it is needed.
    After the module is imported, it is stored in the namespace.

    Parameters:
        name_as (str): The name of the module in the namespace.
        import_from (Optional[str]): The base package of the module.
        name (Optional[str]): The name of the module to import. Defaults to the value of `name_as`.
        globals (dict): The current namespace (pass `globals()` here).

    Usage:
        ```python
        # import numpy as np
        np = lazy_module("np", name="numpy", globals=globals())
        print(np)  # <lazy module 'numpy'>

        np.random.rand(5)  # `numpy` is imported here
        print(np)  # <module 'numpy' from '...'>

        # import torch
        torch = lazy_module("torch", globals=globals())

        # from torchvision.models import resnet
        resnet = lazy_module("resnet", "torchvision.models", globals=globals())

        # from torch.nn import functional as F
        F = lazy_module("F", "torch.nn", "functional", globals=globals())

        # from .engine import api
        api = lazy_module("api", ".engine", globals=globals())
        ```

    Note:
        The `name_as` parameter should match the name of the variable in the namespace.
    """
    name = name or name_as
    globals = globals or sys._getframe(1).f_globals
    package = _resolve_package(import_from, globals)
    module_name = name if package is None else f"{package}.{name}"
    return _LazyModule(name_as, module_name, globals)

lazy_class 

lazy_class(name_as: str, /, import_from: str, name: Optional[str] = None, *, globals: Optional[Dict[str, Any]] = None)

A wrapper for classes that delays the import until it is needed. After the class is imported, it is stored in the namespace.

Parameters:

  • name_as (str) –

    The name of the class in the namespace.

  • import_from (str) –

    The module path of the class.

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

    The name of the class to import. Defaults to the value of name_as.

  • globals (dict, default: None ) –

    The current namespace (pass globals() here).

Usage 
# from torch.nn import Linear
Linear = lazy_class("Linear", "torch.nn", globals=globals())
print(Linear)  # <lazy class 'torch.nn.Linear'> (cannot resolve the actual path)

linear = Linear(3, 4)  # `Linear` is imported here
print(linear)  # Linear(in_features=3, out_features=4, bias=True)
print(Linear)  # <class 'torch.nn.modules.linear.Linear'>

# from torchvision.models.resnet import ResNet as _ResNet
_ResNet = lazy_module("_ResNet", "torchvision.models.resnet", "ResNet", globals=globals())

# from .engine.api import Trainer
Trainer = lazy_class("Trainer", ".engine.api", globals=globals())
trainer = Trainer.build(...)
Note

The name_as parameter should match the name of the variable in the namespace.

Warning

The only safe way to use this is class attribute access or instantiation. Do not use this with class-level methods like isinstance or issubclass.

Source code in SaigeToolkit/util/lazy_loader.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def lazy_class(
    name_as: str,
    /,
    import_from: str,
    name: Optional[str] = None,
    *,
    globals: Optional[Dict[str, Any]] = None,
):
    """
    A wrapper for classes that delays the import until it is needed.
    After the class is imported, it is stored in the namespace.

    Parameters:
        name_as (str): The name of the class in the namespace.
        import_from (str): The module path of the class.
        name (Optional[str]): The name of the class to import. Defaults to the value of `name_as`.
        globals (dict): The current namespace (pass `globals()` here).

    Usage:
        ```python
        # from torch.nn import Linear
        Linear = lazy_class("Linear", "torch.nn", globals=globals())
        print(Linear)  # <lazy class 'torch.nn.Linear'> (cannot resolve the actual path)

        linear = Linear(3, 4)  # `Linear` is imported here
        print(linear)  # Linear(in_features=3, out_features=4, bias=True)
        print(Linear)  # <class 'torch.nn.modules.linear.Linear'>

        # from torchvision.models.resnet import ResNet as _ResNet
        _ResNet = lazy_module("_ResNet", "torchvision.models.resnet", "ResNet", globals=globals())

        # from .engine.api import Trainer
        Trainer = lazy_class("Trainer", ".engine.api", globals=globals())
        trainer = Trainer.build(...)
        ```

    Note:
        The `name_as` parameter should match the name of the variable in the namespace.

    Warning:
        The only safe way to use this is class attribute access or instantiation.
        Do not use this with class-level methods like `isinstance` or `issubclass`.
    """
    name = name or name_as
    globals = globals or sys._getframe(1).f_globals
    module_name = _resolve_package(import_from, globals)
    namespace = {
        "local_name": name_as,
        "class_name": name,
        "module_name": module_name,
        "globals": globals,
    }
    return _LazyClassMeta(name_as, (object,), namespace)

memory

CudaMemoryTracker 

CudaMemoryTracker(device: Union[device, str])

특정 구간에서 추가적으로 사용된 최대 GPU 메모리를 측정합니다. 예를 들어 모델을 먼저 GPU에 올린 뒤 __enter__에 진입하면 모델이 점유한 메모리는 제외됩니다. 각 연산의 메모리를 정확하게 트래킹하려면 torch.profiler를 사용하세요.

Example
  1. Context Manager로 사용하는 경우 with CudaMemoryTracker() as tracker: results = model.inference(batch)
  2. class로 사용하는 경우 tracker = CudaMemoryTracker() tracker.start() results = model.inference(batch) tracker.end() ...

print(tracker.get_results())

Source code in SaigeToolkit/util/memory.py 
89
90
91
92
93
94
95
96
def __init__(self, device: Union[torch.device, str]):
    self.device = torch.device(device)
    self.enabled = all(
        [self.device.type == "cuda", platform.system() == "Linux", torch.cuda.is_available()]
    )
    self.cuda_context_memory = 0
    self.max_memory_allocated = 0
    self.max_memory_reserved = 0
device instance-attribute 
device = device(device)
enabled instance-attribute 
enabled = all([type == 'cuda', system() == 'Linux', is_available()])
cuda_context_memory instance-attribute 
cuda_context_memory = 0
max_memory_allocated instance-attribute 
max_memory_allocated = 0
max_memory_reserved instance-attribute 
max_memory_reserved = 0
__enter__ 
__enter__()
Source code in SaigeToolkit/util/memory.py 
 98
 99
100
101
102
103
104
105
def __enter__(self):
    if not self.enabled:
        return self
    cuda_empty_cache(self.device)
    torch.cuda.reset_peak_memory_stats(self.device)
    self.memory_already_allocated = torch.cuda.max_memory_allocated(self.device)
    self.memory_already_reserved = torch.cuda.max_memory_reserved(self.device)
    return self
__exit__ 
__exit__(*args, **kwargs)
Source code in SaigeToolkit/util/memory.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def __exit__(self, *args, **kwargs):
    if not self.enabled:
        return
    self.max_memory_allocated = max(
        self.max_memory_allocated,
        torch.cuda.max_memory_allocated(self.device) - self.memory_already_allocated,
    )
    self.max_memory_reserved = max(
        self.max_memory_reserved,
        torch.cuda.max_memory_reserved(self.device) - self.memory_already_reserved,
    )
    self.cuda_context_memory = max(
        0, get_process_gpu_memory() - torch.cuda.memory_reserved(self.device)
    )
start 
start()
Source code in SaigeToolkit/util/memory.py 
122
123
def start(self):
    return self.__enter__()
end 
end(*args, **kwargs)
Source code in SaigeToolkit/util/memory.py 
125
126
def end(self, *args, **kwargs):
    return self.__exit__(*args, **kwargs)
get_results 
get_results() -> dict
Source code in SaigeToolkit/util/memory.py 
128
129
130
131
132
133
134
135
136
def get_results(self) -> dict:
    reserved_and_context = (
        f"{(self.max_memory_reserved + self.cuda_context_memory)/ 1024 ** 2:.0f} MiB"
    )
    return {
        "GPU memory": f"{self.max_memory_allocated / 1024 ** 2:.0f} MiB",
        "GPU memory (+ cache)": f"{self.max_memory_reserved / 1024 ** 2:.0f} MiB",
        "GPU memory (+ cache + CUDA context)": reserved_and_context,
    }

get_process_cpu_memory 

get_process_cpu_memory(pid: int = None, include_virtual: bool = False) -> int

프로세스가 사용하는 RAM 크기를 바이트 단위로 읽어옵니다.

Parameters:

  • pid (int, default: None ) –

    If None, use current PID.

  • include_virtual (bool, default: False ) –

    to include virtual memory. Defaults to False.

Returns:

  • int ( int ) –

    CPU memory in bytes

Note

Use --pid=host flag with Docker.

Source code in SaigeToolkit/util/memory.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def get_process_cpu_memory(pid: int = None, include_virtual: bool = False) -> int:
    """프로세스가 사용하는 RAM 크기를 바이트 단위로 읽어옵니다.

    Args:
        pid (int, optional):  If None, use current PID.
        include_virtual (bool, optional): to include virtual memory. Defaults to False.

    Returns:
        int: CPU memory in bytes

    Note:
        Use --pid=host flag with Docker.
    """
    if pid is None:
        pid = os.getpid()
    memory_info = psutil.Process(pid).memory_info()
    return memory_info.vms if include_virtual else memory_info.rss

get_process_gpu_memory 

get_process_gpu_memory(pid: int = None) -> int

프로세스가 사용하는 GPU메모리를 NVML 모듈로부터 바이트 단위로 읽어옵니다.

Parameters:

  • pid (int, default: None ) –

    If None, use current PID.

Returns:

  • int ( int ) –

    GPU memory in bytes

Note

Use --pid=host flag with Docker.

Source code in SaigeToolkit/util/memory.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def get_process_gpu_memory(pid: int = None) -> int:
    """프로세스가 사용하는 GPU메모리를 NVML 모듈로부터 바이트 단위로 읽어옵니다.

    Args:
        pid (int, optional): If None, use current PID.

    Returns:
        int: GPU memory in bytes

    Note:
        Use --pid=host flag with Docker.
    """
    if pid is None:
        pid = os.getpid()
    pynvml.nvmlInit()
    for device_id in range(pynvml.nvmlDeviceGetCount()):
        handle = pynvml.nvmlDeviceGetHandleByIndex(device_id)
        for process in pynvml.nvmlDeviceGetComputeRunningProcesses(handle):
            if process.pid == pid:
                return 0 if process.usedGpuMemory is None else process.usedGpuMemory
    return 0

get_memory_stats 

get_memory_stats() -> Dict[str, int]

현재 프로세스의 메모리 정보를 MiB 단위로 읽어옵니다. torch_max_allocated gpu mem의 경우 해당 함수 호출 직전까지의 최댓값이며, 해당 함수 호출 시 리셋됩니다.

Returns:

  • Dict[str, int] –

    Dict[str, int]: description

Source code in SaigeToolkit/util/memory.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def get_memory_stats() -> Dict[str, int]:
    """현재 프로세스의 메모리 정보를 MiB 단위로 읽어옵니다.
    torch_max_allocated gpu mem의 경우 해당 함수 호출 직전까지의 최댓값이며, 해당 함수 호출 시 리셋됩니다.

    Returns:
        Dict[str, int]: _description_
    """
    memory_stats = {
        "cpu_memory(MiB)": get_process_cpu_memory() // 1024**2,
        "gpu_memory(MiB)": get_process_gpu_memory() // 1024**2,
        "gpu_memory(torch_max_allocated, MiB)": torch.cuda.max_memory_allocated() // 1024**2,
    }
    torch.cuda.reset_peak_memory_stats()
    return memory_stats

pickle

handle_unpickling_error 

handle_unpickling_error(function)
Source code in SaigeToolkit/util/pickle.py 
 7
 8
 9
10
11
12
13
14
15
def handle_unpickling_error(function):
    @wraps(function)
    def wrapper(*args, **kwargs):
        try:
            return function(*args, **kwargs)
        except pickle.UnpicklingError:
            raise InvalidModelFileError

    return wrapper

registry

클래스 레지스트리를 가질 수 있도록 하는 추상 클래스를 정의합니다.

Registerable

Bases: ABC

Abstract class for base module classes which have class registry.

Usage 
from __future__ import annotations
from typing import Dict
from .SaigeToolkit.SaigeToolkit.util.registry import Registerable


class BaseModule(Registerable):
    registry: Dict[str, BaseModule]


class ImplementedModuleA(BaseModule):
    pass


class ImplementedModuleB(BaseModule):
    pass


print(BaseModule.registry)
# output
# {'ImplementedModuleA': __main__.ImplementedModuleA,
#  'ImplementedModuleB': __main__.ImplementedModuleB}

If you want use arbitrary key for registering, 

class BaseModule2(Registerable):
    registry: Dict[str, BaseModule2]


class ImplementedModule2A(BaseModule2, key="module_2_a"):
    pass


print(BaseModule2.registry)
# output
# {'module_2_a': __main__.ImplementedModule2A}

or, define "init_subclass" again. 

class BaseModule3(Registerable):
    registry: Dict[str, BaseModule3]

   def __init_subclass__(cls):
       super().__init_subclass__(key=cls.__name__.lower())

class ImplementedModule3A(BaseModule3):
    pass


print(BaseModule3.registry)
# output
# {'implementedmodule3a': __main__.ImplementedModule3A}

registry abstractmethod property 
registry: Dict[str, Registerable]
__init_subclass__ 
__init_subclass__(key=None, **kwargs) -> None
Source code in SaigeToolkit/util/registry.py 
77
78
79
80
81
82
83
84
85
86
def __init_subclass__(cls, key=None, **kwargs) -> None:
    if isinstance(cls.registry, property):
        cls.registry = {}
    else:
        if key is None:
            key = cls.__name__

        if key in cls.registry:
            raise AlreadlyRegisteredClassError
        cls.registry[key] = cls

reproducibility

실험 재현성을 위한 유틸리티 함수를 제공합니다.

TIME_FORMAT module-attribute 

TIME_FORMAT = '%Y-%m-%d %H:%M:%S KST'

set_randomness 

set_randomness(seed=0, deterministic=False)
Source code in SaigeToolkit/util/reproducibility.py 
24
25
26
27
28
29
30
31
32
33
def set_randomness(seed=0, deterministic=False):
    random.seed(seed)
    os.environ["PYTHONHASHSEED"] = str(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    if deterministic:
        torch.backends.cudnn.deterministic = True
        torch.backends.cudnn.benchmark = False

store_config 

store_config(attr: Optional[str] = None)

클래스 메소드 호출 시 파라미터를 self.attr 변수에 dict로 저장하는 데코레이터를 리턴합니다.

Usage

class MyClass: @store_config(attr="_config") def init(self, param1, param2): pass

MyClass 인스턴스 생성 시 self._config에 파라미터가 저장됩니다.

Source code in SaigeToolkit/util/reproducibility.py 
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
63
64
65
def store_config(attr: Optional[str] = None):
    """클래스 메소드 호출 시 파라미터를 `self.attr` 변수에 dict로 저장하는 데코레이터를 리턴합니다.

    Usage:
        class MyClass:
            @store_config(attr="_config")
            def __init__(self, param1, param2):
                pass

        MyClass 인스턴스 생성 시 self._config에 파라미터가 저장됩니다.

    """

    def _store_config(method, attr=attr):
        attr = attr or f"{method.__name__}_params"

        @wraps(method)
        def decorator(self, *args, **kwargs):
            signature = inspect.signature(method)
            bound_args = signature.bind(self, *args, **kwargs)
            bound_args.apply_defaults()
            config = dict(bound_args.arguments)
            config.pop("self")
            config = copy.deepcopy(config)
            setattr(self, attr, config)
            return method(self, *args, **kwargs)

        return decorator

    return _store_config

get_function_args 

get_function_args() -> Dict

Returns the argument received by the parent function as a dictionary.

Returns:

  • Dict ( Dict ) –

    keyword arguments dictionary

Example

code: def func(A=1, B=2, C=3): print(get_function_args()) func(C=5)

results: {'A': 1, 'B': 2, 'C': 5}

Note

해당 함수를 호출하기 전에 argument를 수정하는 경우, 수정된 argument가 반환됩니다. Example: code: def func(A=1, B=2, C=3): B = 7 print(get_function_args()) func(C=5)

results:
    {'A': 1, 'B': 7, 'C': 5}
Source code in SaigeToolkit/util/reproducibility.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def get_function_args() -> Dict:
    """Returns the argument received by the parent function as a dictionary.

    Returns:
        Dict: keyword arguments dictionary

    Example:
        code:
            def func(A=1, B=2, C=3):
                print(get_function_args())
            func(C=5)

        results:
            {'A': 1, 'B': 2, 'C': 5}

    Note:
        해당 함수를 호출하기 전에 argument를 수정하는 경우, 수정된 argument가 반환됩니다.
        Example:
            code:
                def func(A=1, B=2, C=3):
                    B = 7
                    print(get_function_args())
                func(C=5)

            results:
                {'A': 1, 'B': 7, 'C': 5}
    """
    frame = sys._getframe(1)
    args, _, _, values = inspect.getargvalues(frame)
    return copy.deepcopy({arg: values[arg] for arg in args})

get_source_git_info 

get_source_git_info(git_path: Optional[str] = None) -> Dict

Parse git info

Parameters:

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

    git directory path. Defaults to None.

Returns:

  • dict ( Dict ) –

    { "directory": str, "commit": str, "modified files": List[str], "untracked files": List[str], "submodules_info": List[Dict],

  • Dict –

    }

Source code in SaigeToolkit/util/reproducibility.py 
100
101
102
103
104
105
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
def get_source_git_info(git_path: Optional[str] = None) -> Dict:
    """Parse git info

    Args:
        git_path (Optional[str], optional): git directory path. Defaults to None.

    Returns:
        dict: {
            "directory": str,
            "commit": str,
            "modified files": List[str],
            "untracked files": List[str],
            "submodules_info": List[Dict],
        }
    """
    if git is None:
        return {}

    def _get_source_git_info_by_repo(repo: git.Repo) -> Dict:
        info = {
            "directory": repo.working_dir,
            "commit": repo.head.commit.hexsha,
            "modified files": [item.a_path for item in repo.index.diff(None)],
            "untracked files": repo.untracked_files,
            "submodules_info": [
                _get_source_git_info_by_repo(submodule.module()) for submodule in repo.submodules
            ],
        }
        return info

    try:
        repo = git.Repo(git_path, search_parent_directories=git_path is None)
        return_value = _get_source_git_info_by_repo(repo)
    except git.GitError:
        return_value = {}

    return return_value

get_source_info 

get_source_info(git_path: Optional[str] = None) -> Dict

Returns current time (KST) + git info + run command

Parameters:

  • git_path (str, default: None ) –

    git directory path. Defaults to GIT_PATH.

Returns:

  • dict ( Dict ) –

    {"time": str, "command": str, **git_info}

Source code in SaigeToolkit/util/reproducibility.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def get_source_info(git_path: Optional[str] = None) -> Dict:
    """Returns current time (KST) + git info + run command

    Args:
        git_path (str, optional): git directory path. Defaults to GIT_PATH.

    Returns:
        dict: {"time": str, "command": str, **git_info}
    """
    return {
        "time": datetime.datetime.now(pytz.timezone("Asia/Seoul")).strftime(TIME_FORMAT),
        **get_source_git_info(git_path),
        "command": "python " + " ".join(sys.argv),
    }

string

pascal_to_snake 

pascal_to_snake(name: str) -> str
Source code in SaigeToolkit/util/string.py 
6
7
def pascal_to_snake(name: str) -> str:
    return "_".join(re.sub(r"([A-Z])", r" \1", name).split()).lower()

get_time_string 

get_time_string(timezone: str = 'Asia/Seoul') -> str
Source code in SaigeToolkit/util/string.py 
10
11
12
def get_time_string(timezone: str = "Asia/Seoul") -> str:
    x = datetime.datetime.now(pytz.timezone(timezone))
    return f"{(x.year - 2000):02d}{x.month:02d}{x.day:02d}-{x.hour:02d}{x.minute:02d}{x.second:02d}"

system

get_cpu_name 

get_cpu_name() -> str

Get CPU name in string

Returns:

  • str ( str ) –

    CPU name (ex: "AMD EPYC 7502 32-Core Processor")

Source code in SaigeToolkit/util/system.py 
13
14
15
16
17
18
19
def get_cpu_name() -> str:
    """Get CPU name in string

    Returns:
        str: CPU name (ex: "AMD EPYC 7502 32-Core Processor")
    """
    return cpuinfo.get_cpu_info()["brand_raw"]

get_torch_device_name 

get_torch_device_name(device: Union[device, str, int]) -> str

Get device name in string

Parameters:

  • device (Union[device, str, int]) –

    torch device

Returns:

  • str ( str ) –

    device name (ex: "A100-PCIE-40GB", "AMD EPYC 7502 32-Core Processor")

Source code in SaigeToolkit/util/system.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def get_torch_device_name(device: Union[torch.device, str, int]) -> str:
    """Get device name in string

    Args:
        device (Union[torch.device, str, int]): torch device

    Returns:
        str: device name (ex: "A100-PCIE-40GB", "AMD EPYC 7502 32-Core Processor")
    """
    device = torch.device(device)
    if device.type == "cpu":
        return get_cpu_name()
    else:
        return torch.cuda.get_device_name(device)

get_system_info 

get_system_info() -> Dict[str, str]
Source code in SaigeToolkit/util/system.py 
38
39
40
41
42
43
44
45
def get_system_info() -> Dict[str, str]:
    return {
        "time": datetime.now(pytz.timezone("Asia/Seoul")).strftime(TIME_FORMAT),
        "pc": platform.node(),
        "os": platform.platform(),
        "python": platform.python_version(),
        "cpu": get_cpu_name(),
    }

get_torch_info 

get_torch_info() -> Dict
Source code in SaigeToolkit/util/system.py 
48
49
50
51
52
53
54
55
56
57
58
def get_torch_info() -> Dict:
    return {
        "torch": torch.__version__,
        "cuda": torch.version.cuda,
        "cudnn": torch.backends.cudnn.version(),
        "torch.backends.cuda.matmul.allow_tf32": torch.backends.cuda.matmul.allow_tf32,
        "torch.backends.cudnn.allow_tf32": torch.backends.cudnn.allow_tf32,
        "torch.backends.cudnn.deterministic": torch.backends.cudnn.deterministic,
        "torch.backends.cudnn.benchmark": torch.backends.cudnn.benchmark,
        "os.environ.NVIDIA_TF32_OVERRIDE": os.environ.get("NVIDIA_TF32_OVERRIDE", None),
    }

test

API test에 유용하게 사용할 수 있는 함수들을 정의한 파일입니다.

thick_divider module-attribute 

thick_divider = '=' * 80

thin_divider module-attribute 

thin_divider = '-' * 80

tab module-attribute 

tab = ' ' * 2

to_string 

to_string(obj, indent=4, depth=0)

Recursively convert object to string.

Source code in SaigeToolkit/util/test.py 
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
def to_string(obj, indent=4, depth=0):
    """Recursively convert object to string."""

    if isinstance(obj, Dict) and obj:
        string = "{" + "\n"
        for k, v in obj.items():
            if k in ["polygon"]:
                k = f"{k} (mid point)"
                v = list(np.mean(np.array(v), axis=-2))
            elif k in ["topk_sparse_matrix"]:
                k = f"{k} (mean value)"
                v = np.mean(v)
            string += " " * indent * (depth + 1) + f'"{k}": ' + to_string(v, depth=depth + 1) + ",\n"
        string += " " * indent * depth + "}"
    elif isinstance(obj, List) and obj:
        string = "[" + "\n"
        string += " " * indent * (depth + 1) + to_string(obj[0], depth=depth + 1) + ",\n"
        string += " " * indent * (depth + 1) + f"...  # {len(obj) - 1} results,\n"
        string += " " * indent * depth + "]"
    elif isinstance(obj, Tuple) and obj:
        string = "(" + "\n"
        for v in obj:
            string += " " * indent * (depth + 1) + to_string(v, depth=depth + 1) + ",\n"
        string += " " * indent * depth + ")"
    elif isinstance(obj, (np.ndarray, torch.Tensor)):
        string = f"{type(obj).__name__}({obj.dtype}, shape={obj.shape}, min={obj.min().item()}, max={obj.max().item()})"
    elif isinstance(obj, str):
        string = f'"{obj}"' + f" ({type(obj).__name__})"
    else:
        string = f"{str(obj)} ({type(obj).__name__})"

    return string

type_checker 

type_checker(output, style, key=None)
Source code in SaigeToolkit/util/test.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def type_checker(output, style, key=None):
    if key is not None:
        output = output[key]
        style = style[key]

    if not isinstance(output, type(style)):
        raise RuntimeError(f'value type different for "{key}": {type(output)} / {type(style)}')

    if isinstance(output, dict):
        if len(style.keys()) == 0:
            if len(output.keys()) != 0:
                added = list(output.keys())
                raise RuntimeError(f"key error: {added} Added")

        elif list(style.keys())[0] == "0":
            type_checker(list(output.values())[0], list(style.values())[0])

        else:
            if set(output.keys()) != set(style.keys()):
                added = list(set(output.keys()) - set(style.keys()))
                deleted = list(set(style.keys()) - set(output.keys()))
                raise RuntimeError(f"key error: {added} Added / {deleted} Deleted")

            for k in output.keys():
                type_checker(output, style, k)

    elif isinstance(output, list):
        if len(output) == 0:
            if key == "generated_labels":
                return
            raise RuntimeError("list should have at least one value")
        if len(style) == 0:
            raise RuntimeError("list should have at least one value")

        type_checker(output[0], style[0])

check_and_print 

check_and_print(message: str, result: Tuple[int, Any], output_style: Optional[Any] = None, verbose: bool = False)
Source code in SaigeToolkit/util/test.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def check_and_print(
    message: str, result: Tuple[int, Any], output_style: Optional[Any] = None, verbose: bool = False
):
    error, error_message, output = result

    if output_style:
        type_checker(output, output_style)

    if verbose:
        print(thin_divider)
        print(f"{message:60}  |  {'PASSED' if error == 0 else 'FAILED'}")
        print(f"error code: {error}, {error_message}")
        print(f"return value: {to_string(output)}")

    else:
        print(f"{message:60}  |  {'PASSED' if error == 0 else 'FAILED'}  ({error}, {error_message})")

    if error < 0:
        exit(-1)

    return output

sweep_options 

sweep_options(options: Dict[str, List[Any]])
Source code in SaigeToolkit/util/test.py 
110
111
112
def sweep_options(options: Dict[str, List[Any]]):
    option_tuples = [[(name, value) for value in values] for name, values in options.items()]
    return [dict(option) for option in itertools.product(*option_tuples)]

load_numpy_image 

load_numpy_image(image_path) -> ndarray
Source code in SaigeToolkit/util/test.py 
115
116
117
118
119
120
def load_numpy_image(image_path) -> np.ndarray:
    # 8bit, 16bit 이미지를 그대로 로드하기 위해 PIL.open 대신 cv2.imread 사용
    image = cv2.imread(filename=image_path, flags=cv2.IMREAD_UNCHANGED)
    if image.ndim == 3:
        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
    return image

timer

timer_timer module-attribute 

timer_timer = Timer(devices='cpu', print_elapsed=False, is_enabled=True)

cpu_timer module-attribute 

cpu_timer = Timer(devices='cpu', print_elapsed=False, is_enabled=True)

cuda_timer module-attribute 

cuda_timer = Timer(devices='cuda', print_elapsed=False, is_enabled=True)

test_time module-attribute 

test_time = 1000

BaseTimer 

BaseTimer()

Bases: ABC

Source code in SaigeToolkit/util/timer.py 
13
14
def __init__(self) -> None:
    self._accumulated_time = 0.0
_accumulated_time instance-attribute 
_accumulated_time = 0.0
device_name property 
device_name
accumulated_time property 
accumulated_time
_start_record abstractmethod 
_start_record() -> None

시간 측정을 시작하는 함수를 구현하면 됩니다.

Source code in SaigeToolkit/util/timer.py 
16
17
18
19
@abstractmethod
def _start_record(self) -> None:
    """시간 측정을 시작하는 함수를 구현하면 됩니다."""
    pass
_end_record abstractmethod 
_end_record() -> Union[int, float]

시간 측정을 마치고 걸린 시간을 return 하는 함수를 구현하면 됩니다.

Source code in SaigeToolkit/util/timer.py 
21
22
23
24
@abstractmethod
def _end_record(self) -> Union[int, float]:
    """시간 측정을 마치고 걸린 시간을 return 하는 함수를 구현하면 됩니다."""
    pass
reset 
reset()
Source code in SaigeToolkit/util/timer.py 
34
35
def reset(self):
    self._accumulated_time = 0.0

_CpuTimer 

_CpuTimer()

Bases: BaseTimer

Source code in SaigeToolkit/util/timer.py 
39
40
41
42
43
def __init__(self) -> None:
    super().__init__()
    self._device_name = "CPU"
    self.start = self._get_time_milliseconds()
    self.end = self._get_time_milliseconds()
_device_name instance-attribute 
_device_name = 'CPU'
start instance-attribute 
start = _get_time_milliseconds()
end instance-attribute 
end = _get_time_milliseconds()
_start_record 
_start_record() -> None
Source code in SaigeToolkit/util/timer.py 
45
46
def _start_record(self) -> None:
    self.start = self._get_time_milliseconds()
_end_record 
_end_record() -> Union[int, float]
Source code in SaigeToolkit/util/timer.py 
48
49
50
def _end_record(self) -> Union[int, float]:
    self.end = self._get_time_milliseconds()
    return self.end - self.start
_get_time_milliseconds staticmethod 
_get_time_milliseconds() -> float
Source code in SaigeToolkit/util/timer.py 
52
53
54
@staticmethod
def _get_time_milliseconds() -> float:
    return time.perf_counter() * 1000

_CudaTimer 

_CudaTimer(devices: List[device])

Bases: BaseTimer

Source code in SaigeToolkit/util/timer.py 
58
59
60
61
62
63
64
65
66
def __init__(self, devices: List[torch.device]) -> None:
    super().__init__()
    self._device_name = "CUDA"
    self.devices = devices
    self.streams = [
        torch.cuda.current_stream(device) for device in self.devices if device.type == "cuda"
    ]
    self.start = [torch.cuda.Event(enable_timing=True) for _ in range(len(self.streams))]
    self.end = [torch.cuda.Event(enable_timing=True) for _ in range(len(self.streams))]
_device_name instance-attribute 
_device_name = 'CUDA'
devices instance-attribute 
devices = devices
streams instance-attribute 
streams = [(current_stream(device)) for device in (devices) if type == 'cuda']
start instance-attribute 
start = [(Event(enable_timing=True)) for _ in (range(len(streams)))]
end instance-attribute 
end = [(Event(enable_timing=True)) for _ in (range(len(streams)))]
_start_record 
_start_record() -> None
Source code in SaigeToolkit/util/timer.py 
68
69
70
def _start_record(self) -> None:
    for start, stream in zip(self.start, self.streams):
        start.record(stream=stream)
_end_record 
_end_record() -> Union[int, float]
Source code in SaigeToolkit/util/timer.py 
72
73
74
75
76
77
78
79
80
def _end_record(self) -> Union[int, float]:
    elapsed = 0
    for start, end, stream in zip(self.start, self.end, self.streams):
        with torch.cuda.device(stream.device):
            end.record(stream=stream)
            end.synchronize()
            elapsed = max(elapsed, start.elapsed_time(end))

    return elapsed

Timer 

Timer(devices: Union[device, str, list] = 'cuda:0', print_elapsed: bool = False, is_enabled: bool = True)

Bases: ContextDecorator

argument로 받은 device에 따라 _CudaTimer, _CpuTimer 중 하나를 선택해서 시간을 측정합니다. _CpuTimer의 경우 time.perf_counter()으로 시작과 끝 시간을 측정하며 _CudaTimer의 경우 CUDA synchronization을 포함해서 시간을 측정합니다.

_CudaTimer의 경우 CUDA synchronization을 사용하기 때문에, GPU time 측정시 보다 정확하지만, _CpuTimer보다 속도가 느립니다. (1000번 실행시 약 100ms 정도 차이남)

Context Manager로 사용하거나 decorator 형태로 사용할 수 있습니다.

enable, disable 함수를 이용해서 Timer를 켜고 끌수 있으며, is_enable함수를 통해 현재 Timer의 작동여부를 확인할 수 있습니다.

reset 함수를 이용해서 현재까지 측정한 시간을 초기화 할 수 있습니다.

측정 된 시간은 accumulated_time과 time_stack property를 통해 가져올 수 있습니다. Timer를 사용해서 여러번 시간을 측정할 경우 accumulated_time은 측정한 총 시간의 합을 나타내고, time_stack은 List안에 측정된 시간이 각각 들어있습니다.

Example

  1. Context Manager로 사용하는 경우 timer = CudaTimer() for batch in dataloader: with timer: results = model.inference(batch) print(f"{timer.accumulated_time} ms")

  2. Decorator로 사용하는 경우 @CudaTimer(print_elapsed=True) # 함수 실행시 마다 프린트 def my_function(): ...

    my_function_timer = CudaTimer() # 해당 함수 호출시 시간 축적 @my_function_timer def my_function(): ...

    print(f"{my_function_timer.accumulated_time} ms")

Source code in SaigeToolkit/util/timer.py 
125
126
127
128
129
130
131
132
133
134
135
136
def __init__(
    self,
    devices: Union[torch.device, str, list] = "cuda:0",
    print_elapsed: bool = False,
    is_enabled: bool = True,
) -> None:
    self._devices = devices
    self._print_elapsed = print_elapsed
    self._is_enabled = is_enabled
    self._timer = self._build_timer(self._devices)
    self._accumulated_time = 0.0
    self._time_stack = []
_devices instance-attribute 
_devices = devices
_print_elapsed instance-attribute 
_print_elapsed = print_elapsed
_is_enabled instance-attribute 
_is_enabled = is_enabled
_timer instance-attribute 
_timer = _build_timer(_devices)
_accumulated_time instance-attribute 
_accumulated_time = 0.0
_time_stack instance-attribute 
_time_stack = []
accumulated_time property 
accumulated_time: int
time_stack property 
time_stack: List
to_device 
to_device(device: Union[device, str, list])
Source code in SaigeToolkit/util/timer.py 
138
139
140
def to_device(self, device: Union[torch.device, str, list]):
    self._devices = device
    self._timer = self._build_timer(self._devices)
__enter__ 
__enter__()
Source code in SaigeToolkit/util/timer.py 
142
143
144
145
def __enter__(self):
    if self.is_enabled():
        self._timer._start_record()
    return self
__exit__ 
__exit__(*args, **kwargs)
Source code in SaigeToolkit/util/timer.py 
147
148
149
150
151
152
153
def __exit__(self, *args, **kwargs):
    if self.is_enabled():
        elapsed = self._timer._end_record()
        if self._print_elapsed:
            print(f"{self._timer.device_name} Timer: {elapsed} ms")
        self._accumulated_time += elapsed
        self._time_stack.append(elapsed)
reset 
reset() -> None
Source code in SaigeToolkit/util/timer.py 
163
164
165
def reset(self) -> None:
    self._accumulated_time = 0.0
    self._time_stack.clear()
is_enabled 
is_enabled() -> bool
Source code in SaigeToolkit/util/timer.py 
167
168
def is_enabled(self) -> bool:
    return self._is_enabled
enable 
enable() -> None
Source code in SaigeToolkit/util/timer.py 
170
171
def enable(self) -> None:
    self._is_enabled = True
disable 
disable() -> None
Source code in SaigeToolkit/util/timer.py 
173
174
def disable(self) -> None:
    self._is_enabled = False
_build_timer staticmethod 
_build_timer(devices: Union[device, str, list]) -> Union[_CudaTimer, _CpuTimer]
Source code in SaigeToolkit/util/timer.py 
176
177
178
179
180
181
182
183
184
185
@staticmethod
def _build_timer(devices: Union[torch.device, str, list]) -> Union[_CudaTimer, _CpuTimer]:
    if not isinstance(devices, list):
        devices = [devices]
    devices = [torch.device(device) for device in devices]
    is_cuda = all([any([device.type == "cuda" for device in devices]), torch.cuda.is_available()])

    timer = _CudaTimer(devices) if is_cuda else _CpuTimer()

    return timer

InferenceHandlerTimer 

InferenceHandlerTimer()

IAD/SEG/CLS/DET의 InferenceHandler에서 사용되며, 다른 곳에서 사용하려면 get_time, add_time_to_result를 수정해야합니다. 시간 측정을 원하는 함수를 등록하면 해당 함수가 호출 될 때 시간을 측정해주는 class 입니다. TODO: 범용적으로 사용할 수 있는 부분 분리

Warning 1

instance의 함수를 등록하는 경우 해당 instance의 함수에만 timer가 적용되기 때문에 주의가 필요합니다. ex) inference_handler.postprocess.call_method를 등록했을 때, inference_handler.postprocess 변수가 다른 instance로 변경되는 경우 시간이 측정되지 않습니다.

Warning 2

현재 get_time, add_time_to_result의 경우 아래 api 함수에 매우 specific하게 구현되어 있기 때문에, 코드 변경시 주의가 필요합니다. - API.InferenceHandler.infer_and_postprocess - API.InferenceHandler.step_analysis

Warning 3

빠르게 필요한 기능만 구현하느라 등록한 함수를 제거하는 기능은 아직 추가되어 있지 않습니다. 다만 inference_options를 통해 시간 측정 자체를 끄거나 켤 수 있습니다.

Warning 4

현재는 시간 측정 구간에 cuda synchronize가 필요없어서 overhead를 줄이기 위해 CpuTimer를 사용합니다. 이후에 시간을 측정하는 구간이 늘어난다면 이 부분을 고려한 코드 수정이 필요합니다.

Source code in SaigeToolkit/util/timer.py 
214
215
216
def __init__(self) -> None:
    self._check_list = {}
    self._is_enable = self.default_config["outputs"]["time"]
_check_list instance-attribute 
_check_list = {}
_is_enable instance-attribute 
_is_enable = default_config['outputs']['time']
config property 
config: Dict
default_config property 
default_config: Dict
register 
register(key: str, object: Any, func_name: str, device: Union[device, str, list] = 'cpu')
Source code in SaigeToolkit/util/timer.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
def register(
    self,
    key: str,
    object: Any,
    func_name: str,
    device: Union[torch.device, str, list] = "cpu",
):
    if key in self._check_list:
        raise InferenceHandlerTimerDuplicateKeyError
    timer = Timer(devices=device, is_enabled=False)
    func = getattr(object, func_name)
    setattr(object, func_name, timer(func))
    self._check_list[key] = {"key": key, "object": object, "func_name": func_name, "timer": timer}
to_device 
to_device(key: str, device: Union[device, str, list])
Source code in SaigeToolkit/util/timer.py 
232
233
234
235
236
def to_device(self, key: str, device: Union[torch.device, str, list]):
    if key not in self._check_list:
        raise InferenceHandlerTimerKeyNotFoundError
    timer: Timer = self._check_list[key]["timer"]
    timer.to_device(device)
to_device_all 
to_device_all(device: Union[device, str, list])
Source code in SaigeToolkit/util/timer.py 
238
239
240
def to_device_all(self, device: Union[torch.device, str, list]):
    for key in self._check_list.keys():
        self.to_device(key, device)
is_enable 
is_enable() -> None
Source code in SaigeToolkit/util/timer.py 
242
243
def is_enable(self) -> None:
    return self._is_enable
set 
set(outputs: Dict[str, bool], params: Dict)
Source code in SaigeToolkit/util/timer.py 
245
246
247
def set(self, outputs: Dict[str, bool], params: Dict):
    self._is_enable = outputs.pop("time")
    self._enable() if self._is_enable else self._disable()
reset 
reset()
Source code in SaigeToolkit/util/timer.py 
257
258
259
260
def reset(self):
    for check_dict in self._check_list.values():
        timer: Timer = check_dict["timer"]
        timer.reset()
add_time_to_result 
add_time_to_result(predictions: List[Dict]) -> List[Dict]
Source code in SaigeToolkit/util/timer.py 
262
263
264
265
266
267
268
269
270
271
272
273
def add_time_to_result(self, predictions: List[Dict]) -> List[Dict]:
    if self.is_enable():
        times = self._get_time()
        updated_predictions = []
        for prediction, time in zip(predictions, times):
            if "time" in prediction:
                raise KeyError
            updated_predictions.append({**prediction, "time": time})
    else:
        updated_predictions = predictions

    return updated_predictions
_get_time 
_get_time()

현재 등록되어 있는 key 및 특징 - imread_time - ImageLoader class의 call_method 시간을 측정. - 여러 image가 들어오는 경우 각각 측정됨. - step_analysis에서 이전 prediction 결과를 재사용 하는 경우 시간 측정이 안됨. - preprocess_time - PreProcess class의 call_method 시간을 측정. - ImageLoader 시간이 포함되어 있기 때문에 ImageLoader에 들어간 시간을 제외해야함. - 여러 image가 들어오는 경우 한번만 측정됨. (batch 연산 처리) - step_analysis에서 이전 prediction 결과를 재사용 하는 경우 시간 측정이 안됨. - inference_time - infer 함수 시간을 측정. - 여러 image가 들어오는 경우 한번만 측정됨. (batch 연산 처리) - step_analysis에서 이전 prediction 결과를 재사용 하는 경우 시간 측정이 안됨. - post_processing_time - PostProcess의 _apply 시간을 측정. (각 이미지별로 측정하기 위함) - 여러 image가 들어오는 경우 각각 측정됨.

Source code in SaigeToolkit/util/timer.py 
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
def _get_time(self):
    """
    현재 등록되어 있는 key 및 특징
    - imread_time
        - ImageLoader class의 call_method 시간을 측정.
        - 여러 image가 들어오는 경우 각각 측정됨.
        - step_analysis에서 이전 prediction 결과를 재사용 하는 경우 시간 측정이 안됨.
    - preprocess_time
        - PreProcess class의 call_method 시간을 측정.
        - ImageLoader 시간이 포함되어 있기 때문에 ImageLoader에 들어간 시간을 제외해야함.
        - 여러 image가 들어오는 경우 한번만 측정됨. (batch 연산 처리)
        - step_analysis에서 이전 prediction 결과를 재사용 하는 경우 시간 측정이 안됨.
    - inference_time
        - infer 함수 시간을 측정.
        - 여러 image가 들어오는 경우 한번만 측정됨. (batch 연산 처리)
        - step_analysis에서 이전 prediction 결과를 재사용 하는 경우 시간 측정이 안됨.
    - post_processing_time
        - PostProcess의 _apply 시간을 측정. (각 이미지별로 측정하기 위함)
        - 여러 image가 들어오는 경우 각각 측정됨.
    """

    # key별로 time_stack 가져오기
    time_dict = {k: v["timer"].time_stack for k, v in self._check_list.items()}

    # prediction 결과를 재사용 하여 시간 측정이 안되는 경우 (즉 리스트가 비어있는 경우) 0을 추가해줌.
    # 현재는 이런 경우가 step_analysis 함수에서만 발생하며,
    # step_analysis 함수의 경우 image를 한장씩 처리하기 때문에 하나의 0만 추가
    for k, v in time_dict.items():
        if len(v) == 0:
            time_dict[k] = [0.0]

    # 여기까지 처리했을 때, imread_time과 post_processing_time의 time_stack의 길이가 같아야하며,
    # preprocess_time과 inference_time의 time_stack은 길이가 1이어야 함.
    assert len(time_dict["imread_time"]) == len(time_dict["post_processing_time"])
    assert len(time_dict["preprocess_time"]) == len(time_dict["inference_time"]) == 1

    # preprocess_time은 imread_time이 포함되어 있기 때문에 imread_time의 전체 합을 빼줌.
    time_dict["preprocess_time"][0] -= sum(time_dict["imread_time"])

    # preprocess_time과 inference_time을 실제 image 개수에 맞게 변경 (1/n)
    num_of_image = len(time_dict["imread_time"])
    for key, time_stack in time_dict.items():
        if key in ["preprocess_time", "inference_time"]:
            new_time_stack = [time_stack[0] / num_of_image for _ in range(num_of_image)]
            time_dict[key] = new_time_stack

    # preprocess_time은 따로 저장하지 않고 inference_time에 포함시키기로 기획에서 결정.
    time_dict["inference_time"] = [
        preprocess_time + inference_time
        for preprocess_time, inference_time in zip(
            time_dict["preprocess_time"],
            time_dict["inference_time"],
        )
    ]

    # 사용되지 않는 preprocess_time 제거
    time_dict.pop("preprocess_time")

    # data를 취합하기 쉽도록 구조 변경
    time_list = [{k: v[idx] for k, v in time_dict.items()} for idx in range(num_of_image)]

    return time_list
_enable 
_enable()
Source code in SaigeToolkit/util/timer.py 
338
339
340
341
def _enable(self):
    for check_dict in self._check_list.values():
        timer: Timer = check_dict["timer"]
        timer.enable()
_disable 
_disable()
Source code in SaigeToolkit/util/timer.py 
343
344
345
346
def _disable(self):
    for check_dict in self._check_list.values():
        timer: Timer = check_dict["timer"]
        timer.disable()
_make_config staticmethod 
_make_config(measure_time)
Source code in SaigeToolkit/util/timer.py 
348
349
350
@staticmethod
def _make_config(measure_time):
    return {"outputs": {"time": measure_time}}

with_cpu_timer 

with_cpu_timer()
Source code in SaigeToolkit/util/timer.py 
358
359
360
@cpu_timer
def with_cpu_timer():
    time.sleep(0.001)

with_cuda_timer 

with_cuda_timer()
Source code in SaigeToolkit/util/timer.py 
362
363
364
@cuda_timer
def with_cuda_timer():
    time.sleep(0.001)

timer_test 

timer_test(func, test_time)
Source code in SaigeToolkit/util/timer.py 
366
367
368
369
@timer_timer
def timer_test(func, test_time):
    for _ in range(test_time):
        func()

version

POSTFIXES module-attribute 

POSTFIXES = {None: -9999, 'rc': 1, 'beta': 2, 'alpha': 3}

Postfix 

Postfix(postfix: Optional[str] = None)

Defines pre-release version type. (None > "rc" > "beta" > "alpha")

Usage 
version = Version(0, 0, 0, "rc1")
print(version.postfix.name)
# rc
print(version.postfix.version)
# 1

Raises:

  • VersionFormatError –

    postfix must be one of "rc", "beta", "alpha" or None

Source code in SaigeToolkit/util/version.py 
35
36
37
38
39
40
41
42
43
44
45
def __init__(self, postfix: Optional[str] = None):
    if postfix is None:
        self.name, self.version = None, None
    else:
        match = re.match(r"^([a-zA-Z]+)(\d+)$", postfix)
        if match:
            self.name = match.groups()[0]
            self.version = int(match.groups()[-1])
        else:
            raise VersionFormatError
    self.priority = POSTFIXES[self.name]
name class-attribute instance-attribute 
name: Optional[str] = None
version class-attribute instance-attribute 
version: Optional[int] = None
priority class-attribute instance-attribute 
priority: int = POSTFIXES[name]
__bool__ 
__bool__()
Source code in SaigeToolkit/util/version.py 
47
48
def __bool__(self):
    return bool(self.name)
__str__ 
__str__()
Source code in SaigeToolkit/util/version.py 
50
51
52
53
54
55
def __str__(self):
    if self:
        postfix_string = f"{self.name}{self.version}"
    else:
        postfix_string = ""
    return postfix_string
__repr__ 
__repr__() -> str
Source code in SaigeToolkit/util/version.py 
57
58
def __repr__(self) -> str:
    return str(self)
__eq__ 
__eq__(other)
Source code in SaigeToolkit/util/version.py 
60
61
def __eq__(self, other):
    return self.priority == other.priority and self.version == other.version
__gt__ 
__gt__(other)
Source code in SaigeToolkit/util/version.py 
63
64
65
66
67
68
69
def __gt__(self, other):
    if self.version or other.version:
        return -self.priority > -other.priority or (
            self.priority == other.priority and self.version > other.version
        )
    # self.version == other.version == None인 경우 False 반환
    return False
__ge__ 
__ge__(other)
Source code in SaigeToolkit/util/version.py 
71
72
def __ge__(self, other):
    return self == other or self > other
__lt__ 
__lt__(other)
Source code in SaigeToolkit/util/version.py 
74
75
def __lt__(self, other):
    return not self >= other
__le__ 
__le__(other)
Source code in SaigeToolkit/util/version.py 
77
78
def __le__(self, other):
    return not self > other

Version 

Version(major: Union[int, str], minor: Union[int, str], patch: Union[int, str], postfix: Optional[str] = None)

Defines Version type: https://semver.org/

Usage 
version = Version(0, 0, 0)
print(version)
# 0.0.0
version_rc = Version(0, 0, 0, "rc1")
print(version_rc)
# 0.0.0-rc1
version2 = Version.from_string("0.0.1")
print(version2 > version)
# True
print(Version.from_string("0.0.1") > Version(0, 0, 1, "rc1"))
# True
print(Version.from_string("0.0.1-rc1") > Version(0, 0, 1, "beta2"))
# True

Raises:

  • VersionFormatError –

    version string format should be f"{self.major}.{self.minor}.{self.patch}-{self.postfix}"

Source code in SaigeToolkit/util/version.py 
111
112
113
114
115
116
117
118
119
120
121
122
def __init__(
    self,
    major: Union[int, str],
    minor: Union[int, str],
    patch: Union[int, str],
    postfix: Optional[str] = None,
):

    self.major = int(major)
    self.minor = int(minor)
    self.patch = int(patch)
    self.postfix = Postfix(postfix)
major instance-attribute 
major: int = int(major)
minor instance-attribute 
minor: int = int(minor)
patch instance-attribute 
patch: int = int(patch)
postfix instance-attribute 
postfix: Postfix = Postfix(postfix)
__str__ 
__str__() -> str
Source code in SaigeToolkit/util/version.py 
124
125
126
127
128
def __str__(self) -> str:
    version_string = f"{self.major}.{self.minor}.{self.patch}"
    if self.postfix:
        version_string = f"{version_string}-{self.postfix}"
    return version_string
__repr__ 
__repr__() -> str
Source code in SaigeToolkit/util/version.py 
130
131
def __repr__(self) -> str:
    return str(self)
from_string classmethod 
from_string(version_string: str)
Source code in SaigeToolkit/util/version.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
@classmethod
def from_string(cls, version_string: str):
    try:
        major, minor, patch_postfix = version_string.split(".")

        if "-" in patch_postfix:
            patch = patch_postfix.split("-")[0]
            postfix = patch_postfix.split("-")[-1]
        else:
            patch = patch_postfix
            postfix = None

        return cls(major, minor, patch, postfix)
    except Exception as e:
        raise VersionFormatError from e
__eq__ 
__eq__(other)
Source code in SaigeToolkit/util/version.py 
149
150
151
152
153
154
155
def __eq__(self, other):
    return (
        self.major == other.major
        and self.minor == other.minor
        and self.patch == other.patch
        and self.postfix == other.postfix
    )
__gt__ 
__gt__(other)
Source code in SaigeToolkit/util/version.py 
157
158
159
160
161
162
163
164
165
166
167
168
def __gt__(self, other):
    return (
        self.major > other.major
        or (self.major == other.major and self.minor > other.minor)
        or (self.major == other.major and self.minor == other.minor and self.patch > other.patch)
        or (
            self.major == other.major
            and self.minor == other.minor
            and self.patch == other.patch
            and self.postfix > other.postfix
        )
    )
__ge__ 
__ge__(other)
Source code in SaigeToolkit/util/version.py 
170
171
def __ge__(self, other):
    return self == other or self > other
__lt__ 
__lt__(other)
Source code in SaigeToolkit/util/version.py 
173
174
def __lt__(self, other):
    return not self >= other
__le__ 
__le__(other)
Source code in SaigeToolkit/util/version.py 
176
177
def __le__(self, other):
    return not self > other