Source code for catalyst.loggers.csv

from typing import Dict, TYPE_CHECKING
import os

from catalyst.core.logger import ILogger
from catalyst.utils.config import save_config

    from catalyst.core.runner import IRunner

[docs]class CSVLogger(ILogger): """CSV logger for the metrics storing under ``.csv`` file. Args: logdir: path to logdir for the logger use_logdir_postfix: boolean flag to use extra ``logs`` prefix in the logdir .. note:: This logger is used by default by ``dl.Runner`` and ``dl.SupervisedRunner`` in case of specified logdir during ``runner.train(..., logdir=/path/to/logdir)``. Examples: .. code-block:: python from catalyst import dl runner = dl.SupervisedRunner() runner.train( ..., loggers={"csv": dl.CSVLogger(logdir="./logdir/logs"} ) .. code-block:: python from catalyst import dl class CustomRunner(dl.IRunner): # ... def get_loggers(self): return { "console": dl.ConsoleLogger(), "csv": dl.CSVLogger(logdir="./logdir/logs") } # ... runner = CustomRunner().run() """ def __init__(self, logdir: str, use_logdir_postfix: bool = False): """Init.""" super().__init__(log_batch_metrics=False, log_epoch_metrics=True) if use_logdir_postfix: logdir = os.path.join(logdir, "csv_logger") self.logdir = logdir self.loggers = {} os.makedirs(self.logdir, exist_ok=True) @property def logger(self): """Internal logger/experiment/etc. from the monitoring system.""" return self.loggers def _make_header(self, metrics: Dict[str, float], loader_key: str): log_line_header = "step," for metric in sorted(metrics.keys()): log_line_header += metric + "," log_line_header = log_line_header[:-1] + "\n" # replace last "," with new line self.loggers[loader_key].write(log_line_header) def _log_metrics(self, metrics: Dict[str, float], step: int, loader_key: str): log_line_csv = f"{step}," for metric in sorted(metrics.keys()): log_line_csv += str(metrics[metric]) + "," log_line_csv = log_line_csv[:-1] + "\n" # replace last "," with new line self.loggers[loader_key].write(log_line_csv) def log_hparams(self, hparams: Dict, runner: "IRunner" = None) -> None: """Logs hyperparameters to the logger.""" save_config(config=hparams, path=os.path.join(self.logdir, "hparams.json")) def log_metrics( self, metrics: Dict[str, float], scope: str, runner: "IRunner", ) -> None: """Logs epoch metrics to csv file.""" if scope == "epoch": for loader_key, per_loader_metrics in metrics.items(): if loader_key not in self.loggers.keys(): self.loggers[loader_key] = open( os.path.join(self.logdir, f"{loader_key}.csv"), "a+" ) self._make_header(metrics=per_loader_metrics, loader_key=loader_key) self._log_metrics( metrics=per_loader_metrics, step=runner.epoch_step, loader_key=loader_key, ) def flush_log(self) -> None: """Flushes the logger.""" for logger in self.loggers.values(): logger.flush() def close_log(self) -> None: """Closes the logger.""" for logger in self.loggers.values(): logger.close()
__all__ = ["CSVLogger"]