log

Module diagram

classDiagram
  class log {
  }
  class dashboard {
  }
  class log_handler {
  }
  class logger {
  }
  class string_formatter {
  }
  log_handler --> dashboard
  log_handler --> logger

log

결과 로그 및 실험 환경 정보를 저장하기 위한 LogHandler 클래스를 제공합니다.

dashboard

Dashboard

Dashboard(logdir: Optional[str] = None, log_image: bool = False)

Bases: Registerable

Source code in SaigeToolkit/log/dashboard.py

def __init__(self, logdir: Optional[str] = None, log_image: bool = False):
    self.logdir = logdir
    self._enabled = self.logdir is not None
    self.log_image = log_image

registry instance-attribute

registry: Dict[str, Dashboard]

logdir instance-attribute

logdir = logdir

_enabled instance-attribute

_enabled = logdir is not None

log_image instance-attribute

log_image = log_image

enabled property

enabled: bool

log abstractmethod

log(data: Dict, step: int, prefix: str) -> None

Source code in SaigeToolkit/log/dashboard.py

@abstractmethod
def log(self, data: Dict, step: int, prefix: str) -> None:
    pass

Tensorboard

Tensorboard(logdir: Optional[str] = None, log_image: bool = False)

Bases: Dashboard

Source code in SaigeToolkit/log/dashboard.py

def __init__(self, logdir: Optional[str] = None, log_image: bool = False):
    super().__init__(logdir, log_image)
    self.tensorboard_writer = SummaryWriter(log_dir=self.logdir) if self.enabled else None

tensorboard_writer instance-attribute

tensorboard_writer = SummaryWriter(log_dir=logdir) if enabled else None

log

log(data: dict, step: int, prefix: str = 'train') -> None

Source code in SaigeToolkit/log/dashboard.py

def log(self, data: dict, step: int, prefix: str = "train") -> None:
    if not self.enabled:
        return

    for key, value in data.items():
        if isinstance(value, numbers.Number):
            self.tensorboard_writer.add_scalar(f"{prefix}/{key}", value, step)
        elif isinstance(value, torch.Tensor):
            if not self.log_image:
                continue
            if len(value.shape) == 4:  # batched images
                self.tensorboard_writer.add_images(f"{prefix}/{key}", value, step)
        else:
            pass

build_dashboard

build_dashboard(_target_: str = 'Tensorboard', **kwargs) -> Dashboard

Source code in SaigeToolkit/log/dashboard.py

def build_dashboard(_target_: str = "Tensorboard", **kwargs) -> Dashboard:
    return Dashboard.registry[_target_](**kwargs)

log_handler

LogHandler

LogHandler(name: str = 'SaigeResearch', logdir: Optional[str] = None, add_time_to_logdir: bool = False, logfile: Optional[str] = None, stdout: bool = False, level: str = 'INFO', formatter: str = f'%(asctime)s %(levelname)s [%(filename)s:%(lineno)d] %(message)s', dashboard: Optional[Dict] = None)

Source code in SaigeToolkit/log/log_handler.py

def __init__(
    self,
    name: str = "SaigeResearch",
    logdir: Optional[str] = None,
    add_time_to_logdir: bool = False,
    logfile: Optional[str] = None,
    stdout: bool = False,
    level: str = "INFO",
    formatter: str = f"%(asctime)s %(levelname)s [%(filename)s:%(lineno)d] %(message)s",
    dashboard: Optional[Dict] = None,
):
    logfile_path = None
    if logdir is not None:
        if add_time_to_logdir:
            num_trial = 10
            for trial_index in range(num_trial):
                cur_logdir = os.path.join(logdir, self.get_time_string())
                try:
                    os.makedirs(cur_logdir, exist_ok=False)
                except FileExistsError:
                    if trial_index == (num_trial - 1):
                        traceback.print_exc()
                        sys.exit()
                    time.sleep(1.1)
                    continue
                else:
                    logdir = cur_logdir
                    break
        else:
            os.makedirs(logdir, exist_ok=False)

        if logfile is not None:
            logfile_path = os.path.join(logdir, logfile)

    self.logdir = logdir

    self.logger = configure_logger(
        name=name,
        level=level,
        formatter=formatter,
        path=logfile_path,
        stdout=stdout,
    )

    dashboard = dashboard or {}
    self.dashboard = build_dashboard(logdir=self.logdir, **dashboard)

    self.logger.info(f"{type(self).__name__} logdir {self.logdir}")
    self.logger.info(
        f"{type(self).__name__} dashboard {'ENABLED' if self.dashboard.enabled else 'DISABLED'}"
    )

logdir instance-attribute

logdir: Optional[str] = logdir

logger instance-attribute

logger = configure_logger(name=name, level=level, formatter=formatter, path=logfile_path, stdout=stdout)

dashboard instance-attribute

dashboard = build_dashboard(logdir=logdir, **dashboard)

log_dashboard

log_dashboard(*args, **kwargs) -> None

Source code in SaigeToolkit/log/log_handler.py

def log_dashboard(self, *args, **kwargs) -> None:
    return self.dashboard.log(*args, **kwargs)

skip_if_logdir_is_none

skip_if_logdir_is_none(method)

Decorator to skip the method if logdir is None.

Source code in SaigeToolkit/log/log_handler.py

def skip_if_logdir_is_none(method):
    """Decorator to skip the method if logdir is None."""

    @wraps(method)
    def decorator(self, *args, **kwargs):
        if getattr(self, "logdir", None) is None:
            if getattr(self, "logger", None) is not None:
                self.logger.info(f"{type(self).__name__} {method.__name__} skipped: logdir is None")
            return
        return method(self, *args, **kwargs)

    return decorator

save_json

save_json(data: Dict, json_path: str = 'log.json') -> None

Save the data in the {self.logdir}/{json_path}.

Parameters:

• data (Dict) –

  data to save

• json_path (str, default: 'log.json' ) –

  json path. Defaults to "log.json".

Source code in SaigeToolkit/log/log_handler.py

@skip_if_logdir_is_none
def save_json(self, data: Dict, json_path: str = "log.json") -> None:
    """Save the data in the {self.logdir}/{json_path}.

    Args:
        data (Dict): data to save
        json_path (str, optional): json path. Defaults to "log.json".
    """
    save_path = os.path.join(self.logdir, json_path)
    os.makedirs(os.path.dirname(save_path), exist_ok=True)
    with open(save_path, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=4, sort_keys=False)

save_image

save_image(image: Image, file_path: str) -> None

Save the image in the {self.logdir}/{file_path}.

Source code in SaigeToolkit/log/log_handler.py

@skip_if_logdir_is_none
def save_image(self, image: Image.Image, file_path: str) -> None:
    """Save the image in the {self.logdir}/{file_path}."""
    save_path = os.path.join(self.logdir, file_path)
    os.makedirs(os.path.dirname(save_path), exist_ok=True)
    image.save(save_path)

save_code

save_code(target_file: str, directory: str = 'source_code', relative_path: Optional[str] = None) -> None

target_file을 실행하는데 필요한 모든 코드를 "source_code" 디렉토리에 저장합니다.

Parameters:

• target_file (str) –

  target_file path

• directory (str, default: 'source_code' ) –

  save directory. Defaults to "source_code".

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

  relative path of the target file from the module(main.py). Defaults to None.

Caution

target_file의 경로에 "."이 포함되어 있으면 제대로 작동하지 않습니다.

Usage

1) module과 같은 레벨의 실행 파일에서 소스 코드 저장을 위해 다음과 같이 사용할 수 있습니다. ex) python run.py 로 실행하고 run.py에서 ./script/train.py의 train 함수를 호출하는 경우

``` python
import __main__
from ..SaigeToolkit.SaigeToolkit.util.log_handler import LogHandler

def train(..., log_handler, ...):
    ...
    log_handler = LogHandler(**log_handler)
    log_handler.save_code(__main__.__file__)
```

2) module보다 하위 디렉토리에 있는 실행 파일에서 소스 코드 저장을 위해 다음과 같이 사용할 수 있습니다. ex) python experiment/test.py로 실행하는 경우.

``` python
import __main__

def test(...):
...
log_handler = LogHandler(**log_handler)
log_handler.save_code(__main__.__file__, relative_path="experiment")
```

Source code in SaigeToolkit/log/log_handler.py

@skip_if_logdir_is_none
def save_code(
    self, target_file: str, directory: str = "source_code", relative_path: Optional[str] = None
) -> None:
    """`target_file`을 실행하는데 필요한 모든 코드를 "source_code" 디렉토리에 저장합니다.

    Args:
        target_file (str): target_file path
        directory (str, optional): save directory. Defaults to "source_code".
        relative_path (Optional[str], optional): relative path of the target file from the module(main.py). Defaults to None.

    Caution:
        target_file의 경로에 "."이 포함되어 있으면 제대로 작동하지 않습니다.

    Usage:
        1) module과 같은 레벨의 실행 파일에서 소스 코드 저장을 위해 다음과 같이 사용할 수 있습니다.
            ex) `python run.py` 로 실행하고 run.py에서 `./script/train.py`의 train 함수를 호출하는 경우

            ``` python
            import __main__
            from ..SaigeToolkit.SaigeToolkit.util.log_handler import LogHandler

            def train(..., log_handler, ...):
                ...
                log_handler = LogHandler(**log_handler)
                log_handler.save_code(__main__.__file__)
            ```


        2) module보다 하위 디렉토리에 있는 실행 파일에서 소스 코드 저장을 위해 다음과 같이 사용할 수 있습니다.
            ex) `python experiment/test.py`로 실행하는 경우.

            ``` python
            import __main__

            def test(...):
            ...
            log_handler = LogHandler(**log_handler)
            log_handler.save_code(__main__.__file__, relative_path="experiment")
            ```

    """
    code_save_dir = os.path.join(self.logdir, directory)

    if relative_path is None:
        code_original_dir = os.path.dirname(target_file)
        code_path_list = [os.path.basename(target_file)]

    if relative_path:
        code_original_dir = os.path.relpath(os.path.dirname(target_file), relative_path)
        code_save_path = os.path.join(code_save_dir, target_file)
        os.makedirs(os.path.dirname(code_save_path), exist_ok=True)
        shutil.copy(target_file, code_save_path)

        code_path_list = []

    module_list = sorted(parse_compile_list(target_file))

    for module in module_list:
        code_path = module.replace(".", os.path.sep) + ".py"
        if os.path.isfile(code_path):
            code_path_list.append(code_path)

    for code_path in code_path_list:
        code_original_path = os.path.join(code_original_dir, code_path)
        code_save_path = os.path.join(code_save_dir, code_path)
        os.makedirs(os.path.dirname(code_save_path), exist_ok=True)
        shutil.copy(code_original_path, code_save_path)

    self.logger.info(f"{type(self).__name__} All codes are saved in {code_save_dir}.")

save_yaml

save_yaml(data: Dict, filename: str) -> None

Save the dict in yaml format.

Parameters:

• data (Dict) –

  Dict to save

• filename (str) –

  filename

Source code in SaigeToolkit/log/log_handler.py

@skip_if_logdir_is_none
def save_yaml(self, data: Dict, filename: str) -> None:
    """Save the dict in yaml format.

    Args:
        data (Dict): Dict to save
        filename (str): filename
    """
    save_path = os.path.join(self.logdir, filename)
    os.makedirs(os.path.dirname(save_path), exist_ok=True)

    OmegaConf.save(data, save_path)

    self.logger.info(f"{type(self).__name__} yaml file is saved in {save_path}.")

save_source_info

save_source_info(git_root_dir: str) -> None

Save current time (KST) + git info + run command in "source_info.json".

Parameters:

• git_root_dir (str) –

  .git 디렉토리가 위치한 상위 디렉토리 경로.

Source code in SaigeToolkit/log/log_handler.py

@skip_if_logdir_is_none
def save_source_info(self, git_root_dir: str) -> None:
    """Save current time (KST) + git info + run command in "source_info.json".

    Args:
        git_root_dir (str): `.git` 디렉토리가 위치한 상위 디렉토리 경로.
    """
    save_dir = self.logdir
    filename = "source_info.json"

    save_path = os.path.join(save_dir, filename)
    source_info = get_source_info(git_root_dir)
    self.save_json(source_info, filename)

    self.logger.info(f"{type(self).__name__} source info is saved in {save_path}.")

save_checkpoint

save_checkpoint(trainer: BaseTrainer, filename: str, password: Optional[str] = None) -> None

Save checkpoint in the "checkpoint" directory.

Parameters:

• trainer (Trainer) –

  An instance containing information to be saved in the checkpoint.

• filename (str) –

  filename

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

  Password for checkpoint file. Defaults to None.

Source code in SaigeToolkit/log/log_handler.py

@skip_if_logdir_is_none
def save_checkpoint(
    self, trainer: BaseTrainer, filename: str, password: Optional[str] = None
) -> None:
    """Save checkpoint in the "checkpoint" directory.

    Args:
        trainer (Trainer): An instance containing information to be saved in the checkpoint.
        filename (str): filename
        password (Optional[str], optional): Password for checkpoint file. Defaults to None.
    """
    save_dir = os.path.join(self.logdir, "checkpoint")
    save_path = os.path.join(save_dir, filename)
    os.makedirs(os.path.dirname(save_path), exist_ok=True)

    trainer.save_checkpoint(save_path, password=password)

get_time_string staticmethod

get_time_string() -> str

Source code in SaigeToolkit/log/log_handler.py

@staticmethod
def get_time_string() -> str:
    x = datetime.datetime.now(pytz.timezone("Asia/Seoul"))
    return f"{(x.year - 2000):02d}{x.month:02d}{x.day:02d}-{x.hour:02d}{x.minute:02d}{x.second:02d}"

logger

configure_logger

configure_logger(name: Optional[str] = 'SaigeResearch', level: str = 'INFO', formatter: str = f'%(asctime)s %(levelname)s [%(filename)s:%(lineno)d] %(message)s', path: Optional[str] = None, stdout: bool = False)

Source code in SaigeToolkit/log/logger.py

def configure_logger(
    name: Optional[str] = "SaigeResearch",
    level: str = "INFO",
    formatter: str = f"%(asctime)s %(levelname)s [%(filename)s:%(lineno)d] %(message)s",
    path: Optional[str] = None,
    stdout: bool = False,
):
    logger = logging.getLogger(name)

    logger.setLevel(level)
    logger.propagate = False
    logger.handlers = []

    if path is not None:
        handler = logging.FileHandler(path, mode="a")
        handler.setFormatter(logging.Formatter(formatter))
        logger.addHandler(handler)

    if stdout:
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(logging.Formatter(formatter))
        logger.addHandler(handler)

    return logger

string_formatter

list_to_pretty_string

list_to_pretty_string(list_of_number: List[Union[int, float]], row_name: str = None, each_length: int = 10, indent: int = 12) -> str

list of Union[int, float] string으로 바꾸는 함수입니다. each_length의 길이가 되도록 " "를 늘려줍니다. log 예시:

     class 0|         0|         1|         2|

Args: list_of_number (List[Union[int, float]]): 숫자가 들어있는 list입니다. row_name (str, optional): 각 row의 이름을 넣어줍니다. Defaults to None. each_length (int, optional): 각 원소마다 가지는 길이. Defaults to 10. indent (int, optional): indentation. Defaults to 12.

Returns: str

Source code in SaigeToolkit/log/string_formatter.py

def list_to_pretty_string(
    list_of_number: List[Union[int, float]],
    row_name: str = None,
    each_length: int = 10,
    indent: int = 12,
) -> str:
    """
    list of Union[int, float] string으로 바꾸는 함수입니다.
    each_length의 길이가 되도록 " "를 늘려줍니다.
    log 예시:
    ```
         class 0|         0|         1|         2|
    ```
    Args:
        list_of_number (List[Union[int, float]]): 숫자가 들어있는 list입니다.
        row_name (str, optional): 각 row의 이름을 넣어줍니다. Defaults to None.
        each_length (int, optional): 각 원소마다 가지는 길이. Defaults to 10.
        indent (int, optional): indentation. Defaults to 12.

    Returns: str
    """

    if row_name is None:
        row_name = ""
    stdout = "\n" + row_name.rjust(indent, " ") + "|"
    for integer in list_of_number:
        stdout += str(integer).rjust(each_length, " ") + "|"
    return stdout

confusion_matrix_to_pretty_string

confusion_matrix_to_pretty_string(confusion_matrix: List[List[Union[int, float]]], each_length: int = 10, indent: int = 12) -> str

confusion matrix를 logging하기 위해 string으로 바꾸는 함수입니다. each_length의 길이가 되도록 " "를 늘려줍니다. log 예시:

...
2023-xx-xx xx:yy:zz,000 INFO [train.py:143]
    object/confusion_matrix:
               |   class 0|   class 1|   class 2|
        class 0|         0|       132|       123|
        class 1|         1|       458|      8394|
        class 2|        10|         1|         2|
    object/accuracy: 0.0000
    ...

Args: confusion_matrix (List[List[Union[int, float]]]): confusion matrix입니다. each_length (int, optional): 각 원소마다 가지는 길이. Defaults to 10. indent (int, optional): indentation. Defaults to 12.

Returns: str

Source code in SaigeToolkit/log/string_formatter.py

def confusion_matrix_to_pretty_string(
    confusion_matrix: List[List[Union[int, float]]],
    each_length: int = 10,
    indent: int = 12,
) -> str:
    """
    confusion matrix를 logging하기 위해 string으로 바꾸는 함수입니다.
    each_length의 길이가 되도록 " "를 늘려줍니다.
    log 예시:
    ```
    ...
    2023-xx-xx xx:yy:zz,000 INFO [train.py:143]
        object/confusion_matrix:
                   |   class 0|   class 1|   class 2|
            class 0|         0|       132|       123|
            class 1|         1|       458|      8394|
            class 2|        10|         1|         2|
        object/accuracy: 0.0000
        ...
    ```
    Args:
        confusion_matrix (List[List[Union[int, float]]]): confusion matrix입니다.
        each_length (int, optional): 각 원소마다 가지는 길이. Defaults to 10.
        indent (int, optional): indentation. Defaults to 12.

    Returns: str
    """

    stdout = "\n" + " " * indent + "|"
    assert len(confusion_matrix) > 0, "confusion matrix is empty"
    nclasses = len(confusion_matrix[0])
    for idx in range(nclasses):
        stdout += f"class {idx}".rjust(each_length, " ") + "|"
    for i, row in enumerate(confusion_matrix):
        row_name = f"class {i}"
        stdout += list_to_pretty_string(
            list_of_number=row,
            row_name=row_name,
            each_length=each_length,
            indent=indent,
        )
    return stdout