Skip to content

util

Module diagram

classDiagram
  class util {
  }
  class color {
  }
  class config {
  }
  class cuda {
  }
  class file_handler {
  }
  class function_tree {
  }
  class memory {
  }
  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 
 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

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

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

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

VERBOSE module-attribute 

VERBOSE = False

to_string 

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

Recursively convert object to string.

Source code in SaigeToolkit/util/test.py 
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
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

get_print_function 

get_print_function(error_message_function=lambda x: '')
Source code in SaigeToolkit/util/test.py 
 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
 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
def get_print_function(error_message_function=lambda x: ""):
    def type_checker(output, style, key=None):
        if key is not None:
            output = output[key]
            style = style[key]
        # print(output, 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])

    def check_and_print(message: str, result: Tuple[int, Any], output_style: Optional[Any] = None):
        error, output = result
        _, error_message = error_message_function(error)

        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

    return check_and_print

sweep_options 

sweep_options(options: Dict[str, List[Any]])
Source code in SaigeToolkit/util/test.py 
113
114
115
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 
118
119
120
121
122
123
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