Source code for avalanche.logging.text_logging

################################################################################
# Copyright (c) 2021 ContinualAI.                                              #
# Copyrights licensed under the MIT License.                                   #
# See the accompanying LICENSE file for terms.                                 #
#                                                                              #
# Date: 2020-01-25                                                             #
# Author(s): Antonio Carta                                                     #
# E-mail: contact@continualai.org                                              #
# Website: avalanche.continualai.org                                           #
################################################################################
import sys
from typing import List, TYPE_CHECKING, Tuple, Type

import torch

from avalanche.evaluation.metric_results import MetricValue, TensorImage
from avalanche.logging import StrategyLogger
from avalanche.evaluation.metric_utils import stream_type, phase_and_task

if TYPE_CHECKING:
    from avalanche.training.strategies import BaseStrategy


UNSUPPORTED_TYPES: Tuple[Type] = (TensorImage, )


[docs]class TextLogger(StrategyLogger): """ The `TextLogger` class provides logging facilities printed to a user specified file. The logger writes metric results after each training epoch, evaluation experience and at the end of the entire evaluation stream. .. note:: To avoid an excessive amount of printed lines, this logger will **not** print results after each iteration. If the user is monitoring metrics which emit results after each minibatch (e.g., `MinibatchAccuracy`), only the last recorded value of such metrics will be reported at the end of the epoch. .. note:: Since this logger works on the standard output, metrics producing images or more complex visualizations will be converted to a textual format suitable for console printing. You may want to add more loggers to your `EvaluationPlugin` to better support different formats. """
[docs] def __init__(self, file=sys.stdout): """ Creates an instance of `TextLogger` class. :param file: destination file to which print metrics (default=sys.stdout). """ super().__init__() self.file = file self.metric_vals = {}
def log_single_metric(self, name, value, x_plot) -> None: self.metric_vals[name] = (name, x_plot, value) def _val_to_str(self, m_val): if isinstance(m_val, torch.Tensor): return '\n' + str(m_val) elif isinstance(m_val, float): return f'{m_val:.4f}' else: return str(m_val) def print_current_metrics(self): sorted_vals = sorted(self.metric_vals.values(), key=lambda x: x[0]) for name, x, val in sorted_vals: if isinstance(val, UNSUPPORTED_TYPES): continue val = self._val_to_str(val) print(f'\t{name} = {val}', file=self.file, flush=True) def before_training_exp(self, strategy: 'BaseStrategy', metric_values: List['MetricValue'], **kwargs): super().before_training_exp(strategy, metric_values, **kwargs) self._on_exp_start(strategy) def before_eval_exp(self, strategy: 'BaseStrategy', metric_values: List['MetricValue'], **kwargs): super().before_eval_exp(strategy, metric_values, **kwargs) self._on_exp_start(strategy) def after_training_epoch(self, strategy: 'BaseStrategy', metric_values: List['MetricValue'], **kwargs): super().after_training_epoch(strategy, metric_values, **kwargs) print(f'Epoch {strategy.clock.train_exp_epochs} ended.', file=self.file, flush=True) self.print_current_metrics() self.metric_vals = {} def after_eval_exp(self, strategy: 'BaseStrategy', metric_values: List['MetricValue'], **kwargs): super().after_eval_exp(strategy, metric_values, **kwargs) exp_id = strategy.experience.current_experience task_id = phase_and_task(strategy)[1] if task_id is None: print(f'> Eval on experience {exp_id} ' f'from {stream_type(strategy.experience)} stream ended.', file=self.file, flush=True) else: print(f'> Eval on experience {exp_id} (Task ' f'{task_id}) ' f'from {stream_type(strategy.experience)} stream ended.', file=self.file, flush=True) self.print_current_metrics() self.metric_vals = {} def before_training(self, strategy: 'BaseStrategy', metric_values: List['MetricValue'], **kwargs): super().before_training(strategy, metric_values, **kwargs) print('-- >> Start of training phase << --', file=self.file, flush=True) def before_eval(self, strategy: 'BaseStrategy', metric_values: List['MetricValue'], **kwargs): super().before_eval(strategy, metric_values, **kwargs) print('-- >> Start of eval phase << --', file=self.file, flush=True) def after_training(self, strategy: 'BaseStrategy', metric_values: List['MetricValue'], **kwargs): super().after_training(strategy, metric_values, **kwargs) print('-- >> End of training phase << --', file=self.file, flush=True) def after_eval(self, strategy: 'BaseStrategy', metric_values: List['MetricValue'], **kwargs): super().after_eval(strategy, metric_values, **kwargs) print('-- >> End of eval phase << --', file=self.file, flush=True) self.print_current_metrics() self.metric_vals = {} def _on_exp_start(self, strategy: 'BaseStrategy'): action_name = 'training' if strategy.is_training else 'eval' exp_id = strategy.experience.current_experience task_id = phase_and_task(strategy)[1] stream = stream_type(strategy.experience) if task_id is None: print('-- Starting {} on experience {} from {} stream --' .format(action_name, exp_id, stream), file=self.file, flush=True) else: print('-- Starting {} on experience {} (Task {}) from {} stream --' .format(action_name, exp_id, task_id, stream), file=self.file, flush=True)