Skip to content

log_handler

log.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}"

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

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)