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

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

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

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

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

override_dict

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

recursively override dictionaries

Source code in SaigeToolkit/util/config.py

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

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

get_tree_node

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

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

Source code in SaigeToolkit/util/config.py

@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

@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

@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

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

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:
        # if any error occurs while checking cuda.is_available, raise CudaDeviceError
        raise CudaDeviceError

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

file_handler

ZstdFileHandler

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

save_compressed_data `staticmethod`

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

Compresses and saves any Python data object to a file.

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

Parameters:

data (Any) –

The Python data object to be compressed and saved.
file_path (str) –

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

Returns:

None –

None

Raises:

FileNotFoundError –

If the specified file path is not valid.
PicklingError –

If the data object cannot be serialized.
ZstdError –

If compression fails.

Source code in SaigeToolkit/util/file_handler.py

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

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

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

    Returns:
        None

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

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

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

load_compressed_data `staticmethod`

load_compressed_data(file_path: str) -> Any

Loads and decompresses data from a file.

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

Parameters:

file_path (str) –

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

Returns:

Any ( Any ) –

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

Raises:

FileNotFoundError –

If the specified file path does not exist.
UnpicklingError –

If the decompressed data cannot be deserialized.
ZstdError –

If decompression fails.

Source code in SaigeToolkit/util/file_handler.py

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

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

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

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

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

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

    data = pickle.loads(bytes_data)

    return data

glob_files

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

glob files with specified extensions

Parameters:

root_path (str) –

description
extensions (Tuple[str]) –

description
recursive (bool, default: True ) –

description. Defaults to True.
skip_hidden_directories (bool, default: True ) –

description. Defaults to True.
max_directories (Optional[int], default: None ) –

max number of directories to search. Defaults to None.
max_files (Optional[int], default: None ) –

max file number limit. Defaults to None.
relative_path (bool, default: False ) –

description. Defaults to False.

Returns:

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

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

Source code in SaigeToolkit/util/file_handler.py

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

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

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

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

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

        if not recursive:
            break

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

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

    return paths, hit_max_directories, hit_max_files

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]

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

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

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

print(tracker.get_results())

Source code in SaigeToolkit/util/memory.py

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

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

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()
    if include_virtual:
        return memory_info.vms
    else:
        return 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

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:
                if process.usedGpuMemory is None:
                    return 0
                else:
                    return 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

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}

reproducibility

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

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 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

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)
    args_dict = copy.deepcopy({arg: values[arg] for arg in args})

    return args_dict

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

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=True if git_path is None else False)
        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

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}
    """
    source_info = {
        "time": datetime.datetime.now(pytz.timezone("Asia/Seoul")).strftime(TIME_FORMAT),
        **get_source_git_info(git_path),
        "command": "python " + " ".join(sys.argv),
    }
    return source_info

string

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

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

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)

test

timer

BaseTimer

BaseTimer()

Bases: ABC

Source code in SaigeToolkit/util/timer.py

def __init__(self) -> None:
    self._accumulated_time = 0.0

_start_record `abstractmethod`

_start_record() -> None

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

Source code in SaigeToolkit/util/timer.py

@abstractmethod
def _start_record(self) -> None:
    """시간 측정을 시작하는 함수를 구현하면 됩니다."""
    pass

_end_record `abstractmethod`

_end_record() -> Union[int, float]

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

Source code in SaigeToolkit/util/timer.py

@abstractmethod
def _end_record(self) -> Union[int, float]:
    """시간 측정을 마치고 걸린 시간을 return 하는 함수를 구현하면 됩니다."""
    pass

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

Context Manager로 사용하는 경우 timer = CudaTimer() for batch in dataloader: with timer: results = model.inference(batch) print(f"{timer.accumulated_time} ms")
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

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 = []

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

def __init__(self) -> None:
    self._check_list = {}
    self._is_enable = self.default_config["outputs"]["time"]

_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

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

version

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

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]

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

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)

util

Module diagram

util

color

sample_uniform_hue_colors

config

read_yml

_load_yml

update_import

override_dict

convert_keys_string_to_list

get_tree_node

set_tree_node

get_tree_node_with_default

flatten_tree

cuda

init_cuda_when_available

file_handler

ZstdFileHandler

save_compressed_data staticmethod

load_compressed_data staticmethod

glob_files

function_tree

FunctionBase

sort_functions

memory

CudaMemoryTracker

get_process_cpu_memory

get_process_gpu_memory

get_memory_stats

registry

Registerable

reproducibility

store_config

get_function_args

get_source_git_info

get_source_info

string

system

get_cpu_name

get_torch_device_name

test

timer

BaseTimer

_start_record abstractmethod

_end_record abstractmethod

Timer

InferenceHandlerTimer

_get_time

version

Postfix

Version

save_compressed_data `staticmethod`

load_compressed_data `staticmethod`

_start_record `abstractmethod`

_end_record `abstractmethod`