Skip to content

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 
16
17
18
19
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 
25
26
27
@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 
31
32
33
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 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
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 
51
52
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 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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 
77
78
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 
80
81
82
83
84
85
86
87
88
89
90
91
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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@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 
106
107
108
109
110
111
@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 
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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
@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 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
@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 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
@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 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
@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 
232
233
234
235
@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 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
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 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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