From c8135a95d846f178ce66fa896fadf6bf9c411d07 Mon Sep 17 00:00:00 2001 From: fehret Date: Mon, 22 Jun 2026 14:53:24 +0300 Subject: [PATCH 1/4] Add a logging level to only show training progress - added METRICS logging level to only log training status, not applicable to HPO for now - added early argument parsing for level as it is needed before the regular parsing --- dpdl/callbacks/epoch_stats.py | 24 ++++++++++++------------ dpdl/cli.py | 13 +++++++++++++ dpdl/experimentmanager.py | 32 +++++++++++++++++++++++++++----- dpdl/logger_config.py | 27 +++++++++++++++++++++++++-- run.py | 16 ++++++++++++++-- 5 files changed, 91 insertions(+), 21 deletions(-) diff --git a/dpdl/callbacks/epoch_stats.py b/dpdl/callbacks/epoch_stats.py index 40ec195f..d6fd60f7 100644 --- a/dpdl/callbacks/epoch_stats.py +++ b/dpdl/callbacks/epoch_stats.py @@ -36,34 +36,34 @@ def on_train_start(self, trainer): steps_per_epoch = data_size / batch_size epochs = math.ceil(trainer.total_steps / steps_per_epoch) - log.info( + log.metrics( f"!!! Starting training for approximately {epochs} epochs ({trainer.total_steps} steps)." ) else: - log.info(f"!!! Starting training for {trainer.epochs} epochs.") + log.metrics(f"!!! Starting training for {trainer.epochs} epochs.") def on_train_end(self, trainer): if self._is_global_zero(): - log.info("!!! Training finished.") + log.metrics("!!! Training finished.") def on_train_epoch_start(self, trainer, epoch): self.train_loss.reset() if self._is_global_zero(): - log.info(f"--------------------------------------------------") + log.metrics(f"--------------------------------------------------") if not self.use_steps: - log.info(f"Starting training epoch {epoch+1}.") + log.metrics(f"Starting training epoch {epoch+1}.") else: - log.info(f"Starting training approximate epoch {epoch+1}.") + log.metrics(f"Starting training approximate epoch {epoch+1}.") def on_train_epoch_end(self, trainer, epoch, metrics): loss = self.train_loss.compute() if self._is_global_zero(): if not self.use_steps: - log.info(f"Epoch {epoch+1} finished. Loss: {loss:.4f}.") + log.metrics(f"Epoch {epoch+1} finished. Loss: {loss:.4f}.") else: - log.info(f"Approximate epoch {epoch+1} finished. Loss: {loss:.4f}.") + log.metrics(f"Approximate epoch {epoch+1} finished. Loss: {loss:.4f}.") self._log_metrics(metrics, "Train metrics") @@ -75,7 +75,7 @@ def on_validation_epoch_end(self, trainer, epoch, metrics): self.evaluation_loss.reset() if self._is_global_zero(): - log.info(f"Validation finished. Loss: {loss:.4f}.") + log.metrics(f"Validation finished. Loss: {loss:.4f}.") self._log_metrics(metrics, "Validation metrics") def on_validation_batch_end(self, trainer, batch_idx, batch, loss): @@ -86,7 +86,7 @@ def on_test_epoch_end(self, trainer, epoch, metrics): self.evaluation_loss.reset() if self._is_global_zero(): - log.info(f"Test finished. Loss: {loss:.4f}.") + log.metrics(f"Test finished. Loss: {loss:.4f}.") self._log_metrics(metrics, "Test metrics") def on_test_batch_end(self, trainer, batch_idx, batch, loss): @@ -98,7 +98,7 @@ def _log_metrics(self, metrics, annotation="Metrics"): metrics = tensor_to_python_type(metrics) - log.info(annotation + ":") + log.metrics(annotation + ":") for key, value in metrics.items(): if not key in self._metrics_to_ignore: - log.info(f" - {key}: {value}.") + log.metrics(f" - {key}: {value}.") diff --git a/dpdl/cli.py b/dpdl/cli.py index 969535a7..222404b2 100644 --- a/dpdl/cli.py +++ b/dpdl/cli.py @@ -607,6 +607,19 @@ def cli( rich_help_panel='Prediction options', ) ] = 'test', + log_level: Annotated[ + str, + typer.Option( + '--log-level', + help=( + 'Console log verbosity: "info" (default, all messages), ' + '"quiet" (metrics and loss only), ' + '"warning" (warnings and errors only), ' + '"debug" (verbose).' + ), + rich_help_panel='Logging options', + ) + ] = 'info', ): # Map from commands to functions diff --git a/dpdl/experimentmanager.py b/dpdl/experimentmanager.py index d9849562..9085c12e 100644 --- a/dpdl/experimentmanager.py +++ b/dpdl/experimentmanager.py @@ -5,6 +5,7 @@ import pickle import shutil import subprocess +import sys import optuna import pandas as pd @@ -322,6 +323,28 @@ def _create_experiment_directory( return full_log_dir +class _TeeStream: + """Writes to two streams simultaneously (here: original stderr + a log file).""" + + def __init__(self, primary, secondary): + self._primary = primary + self._secondary = secondary + + def write(self, data): + self._primary.write(data) + self._secondary.write(data) + + def flush(self): + self._primary.flush() + self._secondary.flush() + + def fileno(self): + return self._primary.fileno() + + def isatty(self): + return self._primary.isatty() + + def _start_logging_to_files(log: logging.Logger, log_path: pathlib.Path): formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') @@ -331,8 +354,7 @@ def _start_logging_to_files(log: logging.Logger, log_path: pathlib.Path): stdout_file_handler.setFormatter(formatter) log.addHandler(stdout_file_handler) - # create a file handler for saving stderr logs to a file - stderr_file_handler = logging.FileHandler(log_path / 'stderr.txt') - stderr_file_handler.setLevel(logging.INFO) - stderr_file_handler.setFormatter(formatter) - log.addHandler(stderr_file_handler) + # Tee sys.stderr to a file while keeping the original stream open as well, + # so that warnings and tracebacks are still visible in the terminal. + stderr_file = open(log_path / 'stderr.txt', 'w', buffering=1) + sys.stderr = _TeeStream(sys.stderr, stderr_file) diff --git a/dpdl/logger_config.py b/dpdl/logger_config.py index 1f0c9b6a..ef9b7270 100644 --- a/dpdl/logger_config.py +++ b/dpdl/logger_config.py @@ -1,13 +1,33 @@ import logging import sys -def configure_logger() -> logging.Logger: +# Custom level between INFO (20) and WARNING (30): always shown in quiet mode. +METRICS = 25 +logging.addLevelName(METRICS, 'METRICS') + +def _metrics(self, message, *args, **kwargs): + if self.isEnabledFor(METRICS): + self._log(METRICS, message, args, **kwargs) + +logging.Logger.metrics = _metrics + +LOG_LEVELS: dict[str, int] = { + 'debug': logging.DEBUG, + 'info': logging.INFO, + 'quiet': METRICS, + 'warning': logging.WARNING, +} + + +def configure_logger(level: int = logging.INFO) -> logging.Logger: log = logging.getLogger('dpdl') + + # Keep the logger itself at INFO so that the file handler always receives the full INFO stream log.setLevel(logging.INFO) # create a stream handler for stdout handler = logging.StreamHandler(sys.stdout) - handler.setLevel(logging.INFO) + handler.setLevel(level) # create a formatter and set it for the handler formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') @@ -16,4 +36,7 @@ def configure_logger() -> logging.Logger: # add the new handler log.addHandler(handler) + # Prevent messages from propagating to the root logger, which causes double logging + log.propagate = False + return log diff --git a/run.py b/run.py index 4df6336c..5d4f6b10 100644 --- a/run.py +++ b/run.py @@ -9,7 +9,7 @@ from dpdl.cli import cli from dpdl.device import distributed_backend, resolve_device, set_cuda_device -from dpdl.logger_config import configure_logger +from dpdl.logger_config import LOG_LEVELS, configure_logger def setup_torch(): @@ -29,6 +29,18 @@ def setup_torch(): multiprocess.set_start_method('spawn', force=True) +def _parse_log_level_arg(argv) -> int: + import logging + for i, arg in enumerate(argv): + if arg == '--log-level' and i + 1 < len(argv): + return LOG_LEVELS.get(argv[i + 1], logging.INFO) + + if arg.startswith('--log-level='): + return LOG_LEVELS.get(arg.split('=', 1)[1], logging.INFO) + + return logging.INFO + + def _parse_device_arg(argv): for i, arg in enumerate(argv): if arg == '--device' and i + 1 < len(argv): @@ -99,7 +111,7 @@ def main(): return 0 - log = configure_logger() + log = configure_logger(level=_parse_log_level_arg(sys.argv)) setup_torch() device_arg = os.getenv('DPDL_DEVICE') or _parse_device_arg(sys.argv) From 0178855d07021d376b3bddafe5cdea4052cb3673 Mon Sep 17 00:00:00 2001 From: fehret Date: Tue, 23 Jun 2026 10:53:10 +0300 Subject: [PATCH 2/4] Changed relevant logger calls in HPO as well - marked all trial/metric related logging infos to also appear in quiet/metric mode --- dpdl/hyperparameteroptimizer.py | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/dpdl/hyperparameteroptimizer.py b/dpdl/hyperparameteroptimizer.py index 7e64aede..e9bf3f28 100644 --- a/dpdl/hyperparameteroptimizer.py +++ b/dpdl/hyperparameteroptimizer.py @@ -152,18 +152,18 @@ def optimize_hypers(config_manager: ConfigurationManager) -> None: # need to also keep track of how many manual trials we are # enqueuing that have not been completed yet if not study._should_skip_enqueue(trial): - log.info(f'Enqueuing trial: {trial}') + log.metrics(f'Enqueuing trial: {trial}') study.enqueue_trial(trial, skip_if_exists=True) enqueued_trial_count += 1 - log.info(f'Enqueued {enqueued_trial_count} manual trials.') + log.metrics(f'Enqueued {enqueued_trial_count} manual trials.') if enqueued_trial_count >= configuration.n_trials: raise ValueError('The number of enqueued trials exceeds or matches the total number of trials. Please reduce the number of manual trials or increase `n_trials` to allow Optuna to perform additional trials.') # Adjust number of random trials to account for enqueued trials remaining_random_trials = max(configuration.optuna_random_trials - enqueued_trial_count, 0) - log.info(f'Setting n_startup_trials to {remaining_random_trials} to account for enqueued trials.') + log.metrics(f'Setting n_startup_trials to {remaining_random_trials} to account for enqueued trials.') sampler = sampler_cls( n_startup_trials=remaining_random_trials, seed=configuration.seed, @@ -183,10 +183,10 @@ def optimize_hypers(config_manager: ConfigurationManager) -> None: # log the results of the best trial if torch.distributed.get_rank() == 0: trial = study.best_trial - log.info(f'Best objective ralue: {trial.value}') - log.info('Params: ') + log.metrics(f'Best objective ralue: {trial.value}') + log.metrics('Params: ') for key, value in trial.params.items(): - log.info(f' - {key}: {value}') + log.metrics(f' - {key}: {value}') # first we need to broadcast the best parameters to all ranks, # so we pack them into a list for sending @@ -254,8 +254,8 @@ def _final_evaluation_round(best_params, config_manager): trainer = TrainerFactory.get_trainer(config_manager) if torch.distributed.get_rank() == 0: - log.info('!! Final training round on the full training dataset (train + valid) and evaluating on test.') - log.info('--------------------------------------------------------------------------------------------') + log.metrics('!! Final training round on the full training dataset (train + valid) and evaluating on test.') + log.metrics('--------------------------------------------------------------------------------------------') # fit model using training AND validation data. we also use the test # set for validation here. @@ -263,11 +263,11 @@ def _final_evaluation_round(best_params, config_manager): # now we can evaluate the final performance of the best model if torch.distributed.get_rank() == 0: - log.info('Evaluating final model on the test set.') + log.metrics('Evaluating final model on the test set.') if torch.distributed.get_rank() == 0: loss, metrics = trainer.test() - log.info(f'Final loss: {loss:.4f}') + log.metrics(f'Final loss: {loss:.4f}') # let's share the loss and metrics with other ranks # rank 0 is the source @@ -284,9 +284,9 @@ def _final_evaluation_round(best_params, config_manager): if torch.distributed.get_rank() == 0: if metrics: - log.info('Final metrics:') + log.metrics('Final metrics:') for key, value in metrics.items(): - log.info(f' - {key}: {value}.') + log.metrics(f' - {key}: {value}.') log_final_epsilon(config_manager, trainer) @@ -370,7 +370,7 @@ def objective( trainer = TrainerFactory.get_trainer(config_manager) if torch.distributed.get_rank() == 0: - log.info(f'Starting trial {trial.number}.') + log.metrics(f'Starting trial {trial.number}.') log.info(config_manager.hyperparams) trainer.fit() From c8092f155266a0944a0fffd068741ed477cf654a Mon Sep 17 00:00:00 2001 From: fehret Date: Mon, 6 Jul 2026 15:11:50 +0300 Subject: [PATCH 3/4] Simplify logging to builtin types and a single command switch - remove custom logging level, move all infologs to debug level and all metrics logs to info level - add single 'quiet' command line switch instead of logging argument --- dpdl/callbacks/body_head_gradient.py | 2 +- dpdl/callbacks/checkpoint.py | 4 +- dpdl/callbacks/clipping_bias.py | 2 +- dpdl/callbacks/cosine_similarity.py | 4 +- dpdl/callbacks/debug.py | 34 ++++++------ dpdl/callbacks/epoch_stats.py | 24 ++++----- dpdl/callbacks/gradient_proportion.py | 2 +- dpdl/callbacks/gradient_stats.py | 2 +- dpdl/callbacks/per_class_accuracy.py | 2 +- dpdl/callbacks/record_accuracy.py | 2 +- dpdl/callbacks/record_losses.py | 4 +- dpdl/callbacks/record_snr.py | 2 +- dpdl/cli.py | 41 +++++++-------- dpdl/configurationmanager.py | 4 +- dpdl/datamodules.py | 76 +++++++++++++-------------- dpdl/experimentmanager.py | 36 +++---------- dpdl/hyperparameteroptimizer.py | 38 +++++++------- dpdl/logger_config.py | 30 ++--------- dpdl/metrics_factory.py | 4 +- dpdl/models/model_base.py | 4 +- dpdl/peft.py | 10 ++-- dpdl/predictor.py | 4 +- dpdl/trainer.py | 10 ++-- run.py | 26 ++++----- 24 files changed, 157 insertions(+), 210 deletions(-) diff --git a/dpdl/callbacks/body_head_gradient.py b/dpdl/callbacks/body_head_gradient.py index dfef0376..2b60f8b6 100644 --- a/dpdl/callbacks/body_head_gradient.py +++ b/dpdl/callbacks/body_head_gradient.py @@ -143,4 +143,4 @@ def _save_to_csv(self, file_path): writer.writeheader() writer.writerows(self.grad_history) - log.info(f'Gradient norms (body & head per class) saved to {file_path}') + log.debug(f'Gradient norms (body & head per class) saved to {file_path}') diff --git a/dpdl/callbacks/checkpoint.py b/dpdl/callbacks/checkpoint.py index 37fbd73d..d905c0ee 100644 --- a/dpdl/callbacks/checkpoint.py +++ b/dpdl/callbacks/checkpoint.py @@ -112,7 +112,7 @@ def on_train_end(self, trainer, *args, **kwargs): def save_checkpoint(self, trainer, checkpoint_path: str): trainer.save_model(checkpoint_path) - log.info(f'Model checkpoint saved at {checkpoint_path}') + log.debug(f'Model checkpoint saved at {checkpoint_path}') def save_metrics(self, metrics, metrics_path: str): metrics = tensor_to_python_type(metrics) @@ -120,4 +120,4 @@ def save_metrics(self, metrics, metrics_path: str): with open(metrics_path, 'w') as fh: json.dump(metrics, fh) - log.info(f'Model checkpoint metrics saved at {metrics_path}') + log.debug(f'Model checkpoint metrics saved at {metrics_path}') diff --git a/dpdl/callbacks/clipping_bias.py b/dpdl/callbacks/clipping_bias.py index e0a37d43..94fc0bc0 100644 --- a/dpdl/callbacks/clipping_bias.py +++ b/dpdl/callbacks/clipping_bias.py @@ -322,4 +322,4 @@ def on_train_end(self, trainer, *args, **kwargs) -> None: writer.writerow(header) writer.writerows(self._rows) - log.info(f'Clipping MSE decomposition written to {out_path}') + log.debug(f'Clipping MSE decomposition written to {out_path}') diff --git a/dpdl/callbacks/cosine_similarity.py b/dpdl/callbacks/cosine_similarity.py index 896a9a3a..3ad87ff8 100644 --- a/dpdl/callbacks/cosine_similarity.py +++ b/dpdl/callbacks/cosine_similarity.py @@ -136,7 +136,7 @@ def on_train_end(self, trainer, *args, **kwargs): writer.writerow(row) - log.info(f'Cosine similarity data saved at {file_path}') + log.debug(f'Cosine similarity data saved at {file_path}') class RecordPerClassCosineSimilarityCallback(Callback): @@ -230,4 +230,4 @@ def _save_to_csv(self, file_path, history, header): for record in history: writer.writerow([record.get(col, None) for col in header]) - log.info(f'Cosine Similarities data saved at {file_path}') + log.debug(f'Cosine Similarities data saved at {file_path}') diff --git a/dpdl/callbacks/debug.py b/dpdl/callbacks/debug.py index f94de79b..de8b9ac4 100644 --- a/dpdl/callbacks/debug.py +++ b/dpdl/callbacks/debug.py @@ -10,53 +10,53 @@ def __init__(self): super().__init__() def _is_global_zero(self): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] Calling _is_global_zero") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] Calling _is_global_zero") super().__is_global_zero() def on_train_start(self, trainer): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_start") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_start") def on_train_end(self, trainer): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_end") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_end") def on_train_epoch_start(self, trainer, epoch): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_epoch_start") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_epoch_start") def on_train_epoch_end(self, trainer, epoch, epoch_loss): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_epoch_end") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_epoch_end") def on_train_batch_start(self, trainer, batch_idx, batch): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_batch_start") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_batch_start") def on_train_physical_batch_start(self, trainer, batch_idx, batch): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_physical_batch_start") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_physical_batch_start") def on_train_batch_end(self, trainer, batch_idx, batch, loss): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_batch_end") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_batch_end") def on_train_physical_batch_end(self, trainer, batch_idx, batch, loss): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_physical_batch_end") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_train_physical_batch_end") def on_validation_epoch_start(self, trainer, epoch): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_validation_epoch_start") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_validation_epoch_start") def on_validation_epoch_end(self, trainer, epoch, metrics): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_validation_epoch_end") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_validation_epoch_end") def on_validation_batch_start(self, trainer, batch_idx, batch): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_validation_batch_start") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_validation_batch_start") def on_validation_batch_end(self, trainer, batch_idx, batch, loss): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_validation_batch_end") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_validation_batch_end") def on_test_epoch_start(self, trainer, epoch): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_test_epoch_start") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_test_epoch_start") def on_test_epoch_end(self, trainer, epoch, metrics): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_test_epoch_end") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_test_epoch_end") def on_test_batch_start(self, trainer, batch_idx, batch): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_test_batch_start") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_test_batch_start") def on_test_batch_end(self, trainer, batch_idx, batch, loss): - log.info(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_test_batch_end") + log.debug(f"[DEBUG - RANK {torch.distributed.get_rank()}] on_test_batch_end") diff --git a/dpdl/callbacks/epoch_stats.py b/dpdl/callbacks/epoch_stats.py index d6fd60f7..40ec195f 100644 --- a/dpdl/callbacks/epoch_stats.py +++ b/dpdl/callbacks/epoch_stats.py @@ -36,34 +36,34 @@ def on_train_start(self, trainer): steps_per_epoch = data_size / batch_size epochs = math.ceil(trainer.total_steps / steps_per_epoch) - log.metrics( + log.info( f"!!! Starting training for approximately {epochs} epochs ({trainer.total_steps} steps)." ) else: - log.metrics(f"!!! Starting training for {trainer.epochs} epochs.") + log.info(f"!!! Starting training for {trainer.epochs} epochs.") def on_train_end(self, trainer): if self._is_global_zero(): - log.metrics("!!! Training finished.") + log.info("!!! Training finished.") def on_train_epoch_start(self, trainer, epoch): self.train_loss.reset() if self._is_global_zero(): - log.metrics(f"--------------------------------------------------") + log.info(f"--------------------------------------------------") if not self.use_steps: - log.metrics(f"Starting training epoch {epoch+1}.") + log.info(f"Starting training epoch {epoch+1}.") else: - log.metrics(f"Starting training approximate epoch {epoch+1}.") + log.info(f"Starting training approximate epoch {epoch+1}.") def on_train_epoch_end(self, trainer, epoch, metrics): loss = self.train_loss.compute() if self._is_global_zero(): if not self.use_steps: - log.metrics(f"Epoch {epoch+1} finished. Loss: {loss:.4f}.") + log.info(f"Epoch {epoch+1} finished. Loss: {loss:.4f}.") else: - log.metrics(f"Approximate epoch {epoch+1} finished. Loss: {loss:.4f}.") + log.info(f"Approximate epoch {epoch+1} finished. Loss: {loss:.4f}.") self._log_metrics(metrics, "Train metrics") @@ -75,7 +75,7 @@ def on_validation_epoch_end(self, trainer, epoch, metrics): self.evaluation_loss.reset() if self._is_global_zero(): - log.metrics(f"Validation finished. Loss: {loss:.4f}.") + log.info(f"Validation finished. Loss: {loss:.4f}.") self._log_metrics(metrics, "Validation metrics") def on_validation_batch_end(self, trainer, batch_idx, batch, loss): @@ -86,7 +86,7 @@ def on_test_epoch_end(self, trainer, epoch, metrics): self.evaluation_loss.reset() if self._is_global_zero(): - log.metrics(f"Test finished. Loss: {loss:.4f}.") + log.info(f"Test finished. Loss: {loss:.4f}.") self._log_metrics(metrics, "Test metrics") def on_test_batch_end(self, trainer, batch_idx, batch, loss): @@ -98,7 +98,7 @@ def _log_metrics(self, metrics, annotation="Metrics"): metrics = tensor_to_python_type(metrics) - log.metrics(annotation + ":") + log.info(annotation + ":") for key, value in metrics.items(): if not key in self._metrics_to_ignore: - log.metrics(f" - {key}: {value}.") + log.info(f" - {key}: {value}.") diff --git a/dpdl/callbacks/gradient_proportion.py b/dpdl/callbacks/gradient_proportion.py index cb8e3e9c..69368927 100644 --- a/dpdl/callbacks/gradient_proportion.py +++ b/dpdl/callbacks/gradient_proportion.py @@ -96,4 +96,4 @@ def _save_to_csv(self, file_path, history): for step, proportions in enumerate(history): writer.writerow([step] + proportions) - log.info(f'Clipped proportions data saved at {file_path}') + log.debug(f'Clipped proportions data saved at {file_path}') diff --git a/dpdl/callbacks/gradient_stats.py b/dpdl/callbacks/gradient_stats.py index 65faebd2..99e6996c 100644 --- a/dpdl/callbacks/gradient_stats.py +++ b/dpdl/callbacks/gradient_stats.py @@ -112,4 +112,4 @@ def _save_to_csv(self, file_name, grad_history): writer.writeheader() writer.writerows(grad_history) - log.info(f'{file_name} saved successfully at {file_path}.') + log.debug(f'{file_name} saved successfully at {file_path}.') diff --git a/dpdl/callbacks/per_class_accuracy.py b/dpdl/callbacks/per_class_accuracy.py index a1899c3f..4250894d 100644 --- a/dpdl/callbacks/per_class_accuracy.py +++ b/dpdl/callbacks/per_class_accuracy.py @@ -47,4 +47,4 @@ def on_train_end(self, trainer, *args, **kwargs): for i, accuracies in enumerate(self.per_class_accuracies_history): writer.writerow([i] + accuracies) - log.info(f'Per-class accuracy data saved at {file_path}') + log.debug(f'Per-class accuracy data saved at {file_path}') diff --git a/dpdl/callbacks/record_accuracy.py b/dpdl/callbacks/record_accuracy.py index 3f1015f6..e2bee392 100644 --- a/dpdl/callbacks/record_accuracy.py +++ b/dpdl/callbacks/record_accuracy.py @@ -58,4 +58,4 @@ def on_train_end(self, trainer): writer.writerow([i + 1, train_acc, val_acc]) - log.info(f'Per epoch accuracies written to {self.csv_path}') + log.debug(f'Per epoch accuracies written to {self.csv_path}') diff --git a/dpdl/callbacks/record_losses.py b/dpdl/callbacks/record_losses.py index 8b0b28c1..274a715c 100644 --- a/dpdl/callbacks/record_losses.py +++ b/dpdl/callbacks/record_losses.py @@ -33,7 +33,7 @@ def on_train_end(self, trainer, *args, **kwargs): writer.writeheader() writer.writerows(self.train_losses) - log.info(f'Training losses (by step) saved to {train_loss_path}') + log.debug(f'Training losses (by step) saved to {train_loss_path}') class RecordLossesByEpochCallback(Callback): @@ -88,4 +88,4 @@ def on_train_end(self, trainer): val_loss_val = self.val_losses[i] if i < len(self.val_losses) else '' writer.writerow([i+1, train_loss_val, val_loss_val]) - log.info('Training finished and all epoch losses have been logged to CSV.') + log.debug('Training finished and all epoch losses have been logged to CSV.') diff --git a/dpdl/callbacks/record_snr.py b/dpdl/callbacks/record_snr.py index 58def83b..c9e80cb7 100644 --- a/dpdl/callbacks/record_snr.py +++ b/dpdl/callbacks/record_snr.py @@ -146,4 +146,4 @@ def on_train_end(self, trainer, *args, **kwargs): writer.writerow(header) writer.writerows(self._rows) - log.info("SNR log written to %s", out_path) + log.debug("SNR log written to %s", out_path) diff --git a/dpdl/cli.py b/dpdl/cli.py index 222404b2..b15ad333 100644 --- a/dpdl/cli.py +++ b/dpdl/cli.py @@ -607,19 +607,14 @@ def cli( rich_help_panel='Prediction options', ) ] = 'test', - log_level: Annotated[ - str, + quiet: Annotated[ + bool, typer.Option( - '--log-level', - help=( - 'Console log verbosity: "info" (default, all messages), ' - '"quiet" (metrics and loss only), ' - '"warning" (warnings and errors only), ' - '"debug" (verbose).' - ), + '--quiet', + help='Only show training metrics, suppressing other console output.', rich_help_panel='Logging options', ) - ] = 'info', + ] = False, ): # Map from commands to functions @@ -659,8 +654,8 @@ def cli( def run_show_layers(config_manager: ConfigurationManager) -> None: - log.info(config_manager.configuration) - log.info('Showing model layers.') + log.debug(config_manager.configuration) + log.debug('Showing model layers.') model, _, _ = ModelFactory.get_model( config_manager.configuration, @@ -676,9 +671,9 @@ def run_train(config_manager: ConfigurationManager) -> Optional[Path]: rank_zero = torch.distributed.get_rank() == 0 if rank_zero: - log.info('Starting training.') - log.info(config_manager.hyperparams) - log.info(config_manager.configuration) + log.debug('Starting training.') + log.debug(config_manager.hyperparams) + log.debug(config_manager.configuration) seed_everything(config_manager.configuration.seed) @@ -691,7 +686,7 @@ def run_train(config_manager: ConfigurationManager) -> Optional[Path]: # log final train accuracy if needed if config_manager.configuration.record_final_train_accuracy: if rank_zero: - log.info('Evaluating on train set..') + log.debug('Evaluating on train set..') train_loss, train_metrics = trainer._evaluate('train', enable_callbacks=False) @@ -700,7 +695,7 @@ def run_train(config_manager: ConfigurationManager) -> Optional[Path]: # log test accuracy and run time, and save model if asked if rank_zero: - log.info('Evaluating on test set..') + log.debug('Evaluating on test set..') test_loss, test_metrics = trainer.test() log_test_metrics(config_manager, test_metrics, test_loss) @@ -726,9 +721,9 @@ def run_train(config_manager: ConfigurationManager) -> Optional[Path]: config_manager.configuration.model_weights_path = str(save_path) if rank_zero: - log.info(f'Saving model to "{save_path}"...') + log.debug(f'Saving model to "{save_path}"...') trainer.save_model(save_path) - log.info('Saving model done.') + log.debug('Saving model done.') saved_model_path = save_path torch.distributed.barrier() @@ -738,8 +733,8 @@ def run_train(config_manager: ConfigurationManager) -> Optional[Path]: def run_optimize(config_manager: ConfigurationManager) -> None: if torch.distributed.get_rank() == 0: - log.info('Starting hyperparameter optimization.') - log.info(config_manager.configuration) + log.debug('Starting hyperparameter optimization.') + log.debug(config_manager.configuration) seed_everything(config_manager.configuration.seed) @@ -754,8 +749,8 @@ def run_optimize(config_manager: ConfigurationManager) -> None: def run_predict(config_manager: ConfigurationManager) -> None: if torch.distributed.get_rank() == 0: - log.info('Starting prediction.') - log.info(config_manager.configuration) + log.debug('Starting prediction.') + log.debug(config_manager.configuration) seed_everything(config_manager.configuration.seed) diff --git a/dpdl/configurationmanager.py b/dpdl/configurationmanager.py index 153830c5..be18fd70 100644 --- a/dpdl/configurationmanager.py +++ b/dpdl/configurationmanager.py @@ -342,7 +342,7 @@ def save_configuration(self, directory: pathlib.Path): with open(directory / 'configuration.json', 'w') as fh: fh.write(self.configuration.json()) - log.info(f'Configuration saved to {directory}.') + log.debug(f'Configuration saved to {directory}.') def save_hyperparameters(self, directory: pathlib.Path): if torch.distributed.get_rank() == 0: @@ -352,7 +352,7 @@ def save_hyperparameters(self, directory: pathlib.Path): with open(directory / 'hyperparameters.json', 'w') as fh: fh.write(self.hyperparams.json()) - log.info(f'Hyperparameters saved to {directory}/.') + log.debug(f'Hyperparameters saved to {directory}/.') def clone_with_overrides(self, **overrides) -> 'ConfigurationManager': params = dict(self._cli_params) diff --git a/dpdl/datamodules.py b/dpdl/datamodules.py index 800627b2..f1acec9b 100644 --- a/dpdl/datamodules.py +++ b/dpdl/datamodules.py @@ -126,7 +126,7 @@ def initialize(self, transforms: torchvision.transforms.transforms.Compose): batch_size = int(self.sample_rate * len(self.train_dataset)) if torch.distributed.get_rank() == 0: - log.info(f'Sample rate is {self.sample_rate}, setting batch size to: {batch_size}.') + log.debug(f'Sample rate is {self.sample_rate}, setting batch size to: {batch_size}.') self.batch_size = batch_size @@ -154,7 +154,7 @@ def initialize_datasets_only(self): batch_size = int(self.sample_rate * len(self.train_dataset)) if torch.distributed.get_rank() == 0: - log.info(f'Sample rate is {self.sample_rate}, setting batch size to: {batch_size}.') + log.debug(f'Sample rate is {self.sample_rate}, setting batch size to: {batch_size}.') self.batch_size = batch_size @@ -187,17 +187,17 @@ def _initialize_datasets(self): # exponential distribution, not Fairness-style imbalance if self._imbalance_factor and not self._fairness_imbalance_class: if torch.distributed.get_rank() == 0: - log.info('Creating imbalanced train set..') + log.debug('Creating imbalanced train set..') self.train_dataset = self._get_imbalanced_subset(self.train_dataset, self._imbalance_reverse) if torch.distributed.get_rank() == 0: - log.info('Creating imbalanced validation set..') + log.debug('Creating imbalanced validation set..') self.val_dataset = self._get_imbalanced_subset(self.val_dataset, self._imbalance_reverse) if torch.distributed.get_rank() == 0: - log.info('Creating imbalanced test set..') + log.debug('Creating imbalanced test set..') self.test_dataset = self._get_imbalanced_subset(self.test_dataset, self._imbalance_reverse) @@ -208,19 +208,19 @@ def _initialize_datasets(self): raise ValueError('Cannot reverse imbalance for fairness style imbalanced dataset.') if torch.distributed.get_rank() == 0: - log.info('Creating fairness imbalanced train set..') + log.debug('Creating fairness imbalanced train set..') self.train_dataset = self._get_fairness_imbalanced_subset( self.train_dataset ) if torch.distributed.get_rank() == 0: - log.info('Creating fairness imbalanced validation set..') + log.debug('Creating fairness imbalanced validation set..') self.val_dataset = self._get_fairness_imbalanced_subset( self.val_dataset ) if torch.distributed.get_rank() == 0: - log.info( + log.debug( f'We will not create fairness imbalanced test sets. Size of test set: {len(self.test_dataset)}' ) @@ -230,13 +230,13 @@ def _initialize_datasets(self): if torch.distributed.get_rank() == 0: train_distribution = Counter(self.train_dataset[self._label_field]) - log.info(f'Training set (size: {len(self.train_dataset)}) class distribution after taking subset of size {self.subset_size}: {sorted(train_distribution.items())}') + log.debug(f'Training set (size: {len(self.train_dataset)}) class distribution after taking subset of size {self.subset_size}: {sorted(train_distribution.items())}') self.val_dataset = self._get_stratified_subset(self.val_dataset) if torch.distributed.get_rank() == 0: val_distribution = Counter(self.val_dataset[self._label_field]) - log.info(f'Validation set (size: {len(self.val_dataset)}) class distribution after taking subset of size {self.subset_size}: {sorted(val_distribution.items())}') + log.debug(f'Validation set (size: {len(self.val_dataset)}) class distribution after taking subset of size {self.subset_size}: {sorted(val_distribution.items())}') if self.shots is not None: self.train_dataset = self._get_few_shot_subset(self.train_dataset) @@ -244,7 +244,7 @@ def _initialize_datasets(self): if self.max_test_examples: if len(self.val_dataset) > self.max_test_examples: if torch.distributed.get_rank() == 0: - log.info(f'Validation dataset has {len(self.val_dataset)} examples which is more than the configured maximum ({self.max_test_examples}). Limiting dataset size.') + log.debug(f'Validation dataset has {len(self.val_dataset)} examples which is more than the configured maximum ({self.max_test_examples}). Limiting dataset size.') _, self.val_dataset = self.val_dataset.train_test_split( test_size=self.max_test_examples, @@ -255,7 +255,7 @@ def _initialize_datasets(self): if len(self.test_dataset) > self.max_test_examples: if torch.distributed.get_rank() == 0: - log.info(f'Test dataset has {len(self.test_dataset)} examples which is more than the configured maximum ({self.max_test_examples}). Limiting dataset size.') + log.debug(f'Test dataset has {len(self.test_dataset)} examples which is more than the configured maximum ({self.max_test_examples}). Limiting dataset size.') _, self.test_dataset = self.test_dataset.train_test_split( test_size=self.max_test_examples, @@ -273,7 +273,7 @@ def _initialize_datasets(self): def _load_datasets(self): '''Load the datasets to memory.''' if torch.distributed.get_rank() == 0: - log.info(f'Loading dataset {self.dataset_name} from Huggingface datasets.') + log.debug(f'Loading dataset {self.dataset_name} from Huggingface datasets.') if self.dataset_path: dataset_splits = datasets.load_from_disk(self.dataset_path) @@ -291,7 +291,7 @@ def _load_datasets(self): self.output_dim = dataset_splits['train'].features[self._label_field].num_classes if torch.distributed.get_rank() == 0: - log.info(f'Determined the output dimension to be {self.output_dim}.') + log.debug(f'Determined the output dimension to be {self.output_dim}.') def _create_dataset_splits(self): # Check if there's a validation split available @@ -372,7 +372,7 @@ def _enforce_label_field_type(self, dataset_splits): def _set_dataset_label_fields(self, dataset_splits): # extract the keys that contain the labels and images if torch.distributed.get_rank() == 0: - log.info('Setting dataset fields.') + log.debug('Setting dataset fields.') self._set_image_field(dataset_splits['train']) self._set_label_field(dataset_splits['train']) @@ -386,7 +386,7 @@ def _set_image_field(self, dataset): if self._image_field: if torch.distributed.get_rank() == 0: - log.info(f' - Determined image field: {self._image_field}') + log.debug(f' - Determined image field: {self._image_field}') else: features = dataset.features.keys() raise ValueError('Could not determine image field for dataset.') @@ -400,7 +400,7 @@ def _set_label_field(self, dataset): if self._label_field: if torch.distributed.get_rank() == 0: - log.info(f' - Determined label field: {self._label_field}') + log.debug(f' - Determined label field: {self._label_field}') else: features = dataset.features.keys() raise ValueError(f'Could not determine label field for dataset. Available features: {features}') @@ -430,7 +430,7 @@ def _create_dataloaders(self): # from the dataset length. if not self.batch_size: if torch.distributed.get_rank() == 0: - log.info('Batch size not yet initialized, skipping dataloader creation.') + log.debug('Batch size not yet initialized, skipping dataloader creation.') return self._set_samplers_and_batch_size() @@ -575,7 +575,7 @@ def _get_few_shot_subset(self, dataset): if torch.distributed.get_rank() == 0: c = Counter(subset[self._label_field]) n_examples = sum(c.values()) - log.info(f'Collected few shot dataset with {n_examples} examples: {c}') + log.debug(f'Collected few shot dataset with {n_examples} examples: {c}') return subset @@ -629,7 +629,7 @@ def _get_imbalanced_subset(self, dataset, reverse=False): if torch.distributed.get_rank() == 0: distribution = Counter(sampled_dataset[self._label_field]) - log.info(f'Created imbalanced dataset (size: {len(sampled_dataset)}) with class distribution: {sorted(distribution.items())}') + log.debug(f'Created imbalanced dataset (size: {len(sampled_dataset)}) with class distribution: {sorted(distribution.items())}') return sampled_dataset @@ -678,7 +678,7 @@ def _get_fairness_imbalanced_subset(self, dataset): if torch.distributed.get_rank() == 0: distribution = Counter(sampled_dataset[self._label_field]) - log.info(f'Created fairness imbalanced dataset (size: {len(sampled_dataset)}) with class distribution: {sorted(distribution.items())}') + log.debug(f'Created fairness imbalanced dataset (size: {len(sampled_dataset)}) with class distribution: {sorted(distribution.items())}') return sampled_dataset @@ -694,7 +694,7 @@ def cache_features(self, model): '''Cache features for the train, validation, and test datasets using the provided model.''' if torch.distributed.get_rank() == 0: - log.info('Feature caching enabled, caching features.') + log.debug('Feature caching enabled, caching features.') def _extract_features(model, image_field, transforms, target_device, target_dtype, examples): inputs = examples[image_field] @@ -733,7 +733,7 @@ def _extract_features(model, image_field, transforms, target_device, target_dtyp ) if torch.distributed.get_rank() == 0: - log.info(f' - Processing {len(self.train_dataset)} examples in the train dataset.') + log.debug(f' - Processing {len(self.train_dataset)} examples in the train dataset.') datasets_map_bs = 512 @@ -748,7 +748,7 @@ def _extract_features(model, image_field, transforms, target_device, target_dtyp ) if torch.distributed.get_rank() == 0: - log.info(f' - Processing {len(self.val_dataset)} examples in the validation dataset.') + log.debug(f' - Processing {len(self.val_dataset)} examples in the validation dataset.') self.val_dataset = self.val_dataset.with_format('torch', device=device_str).map( _extract_features_fn, @@ -760,7 +760,7 @@ def _extract_features(model, image_field, transforms, target_device, target_dtyp if self.test_dataset: if torch.distributed.get_rank() == 0: - log.info(f' - Processing {len(self.test_dataset)} examples in the test dataset.') + log.debug(f' - Processing {len(self.test_dataset)} examples in the test dataset.') self.test_dataset = self.test_dataset.with_format('torch', device=device_str).map( _extract_features_fn, @@ -775,14 +775,14 @@ def _extract_features(model, image_field, transforms, target_device, target_dtyp self._create_dataloaders() if torch.distributed.get_rank() == 0: - log.info('Feature caching finished.') + log.debug('Feature caching finished.') def _apply_transforms_to_datasets(self): if torch.distributed.get_rank() == 0: - log.info('Applying transformations to dataset.') + log.debug('Applying transformations to dataset.') def _apply_transforms(transforms, label_field, image_field, examples): - log.info('.') + log.debug('.') examples[image_field] = [transforms(image) for image in examples[image_field]] return examples @@ -795,7 +795,7 @@ def _apply_transforms(transforms, label_field, image_field, examples): ) if torch.distributed.get_rank() == 0: - log.info(f' - Processing {len(self.train_dataset)} examples in the train dataset.') + log.debug(f' - Processing {len(self.train_dataset)} examples in the train dataset.') self.train_dataset = self.train_dataset.map( transforms_func, @@ -805,7 +805,7 @@ def _apply_transforms(transforms, label_field, image_field, examples): ) if torch.distributed.get_rank() == 0: - log.info(f' - Processing {len(self.val_dataset)} examples in the validation dataset.') + log.debug(f' - Processing {len(self.val_dataset)} examples in the validation dataset.') self.val_dataset = self.val_dataset.map( transforms_func, @@ -816,7 +816,7 @@ def _apply_transforms(transforms, label_field, image_field, examples): if self.test_dataset: if torch.distributed.get_rank() == 0: - log.info(f' - Processing {len(self.test_dataset)} examples in the test dataset.') + log.debug(f' - Processing {len(self.test_dataset)} examples in the test dataset.') self.test_dataset = self.test_dataset.map( transforms_func, @@ -989,7 +989,7 @@ def __init__( def _load_datasets(self): '''Load the datasets to memory.''' if torch.distributed.get_rank() == 0: - log.info( + log.debug( f'Loading dataset {self.dataset_name} from Huggingface datasets.' ) @@ -1020,7 +1020,7 @@ def _load_datasets(self): ) if torch.distributed.get_rank() == 0: - log.info(f'Determined the output dimension to be {self.output_dim}.') + log.debug(f'Determined the output dimension to be {self.output_dim}.') else: self._detect_text_fields( @@ -1030,13 +1030,13 @@ def _load_datasets(self): def _set_dataset_label_fields(self, dataset_splits): if torch.distributed.get_rank() == 0: - log.info('Setting dataset label field (LLM mode).') + log.debug('Setting dataset label field (LLM mode).') self._set_label_field(dataset_splits['train']) # find the label column def _detect_text_fields(self, dataset): if self._text_fields: if torch.distributed.get_rank() == 0: - log.info(f'Using manually provided text fields: {self._text_fields}') + log.debug(f'Using manually provided text fields: {self._text_fields}') return @@ -1045,7 +1045,7 @@ def _detect_text_fields(self, dataset): self._text_fields = keys[:1] if torch.distributed.get_rank() == 0: - log.info(f'Detected text fields: {self._text_fields}') + log.debug(f'Detected text fields: {self._text_fields}') if not self._text_fields: raise ValueError('Could not determine any text field for NLP dataset.') @@ -1061,7 +1061,7 @@ def initialize(self, tokenizer): ) if torch.distributed.get_rank() == 0: - log.info('Initializing NLPDataModule datasets...') + log.debug('Initializing NLPDataModule datasets...') self._initialize_datasets() torch.distributed.barrier() else: @@ -1077,7 +1077,7 @@ def initialize(self, tokenizer): batch_size = int(self.sample_rate * len(self.train_dataset)) if torch.distributed.get_rank() == 0: - log.info( + log.debug( f'Sample rate is {self.sample_rate}, setting batch size to: {batch_size}.' ) diff --git a/dpdl/experimentmanager.py b/dpdl/experimentmanager.py index 9085c12e..cecbc4a5 100644 --- a/dpdl/experimentmanager.py +++ b/dpdl/experimentmanager.py @@ -311,39 +311,18 @@ def _create_experiment_directory( full_log_dir = pathlib.Path(f'{log_dir}/{experiment_name}') if full_log_dir.exists() and overwrite: - log.info(f'Experiment directory "{full_log_dir}" exists, removing it and restarting experiment.') + log.debug(f'Experiment directory "{full_log_dir}" exists, removing it and restarting experiment.') shutil.rmtree(full_log_dir) if full_log_dir.exists() and not overwrite: - log.info(f'Experiment directory "{full_log_dir}" exists, resuming experiment.') + log.debug(f'Experiment directory "{full_log_dir}" exists, resuming experiment.') if not full_log_dir.exists(): - log.info(f'Creating experiment directory "{full_log_dir}".') + log.debug(f'Creating experiment directory "{full_log_dir}".') full_log_dir.mkdir(parents=True) return full_log_dir -class _TeeStream: - """Writes to two streams simultaneously (here: original stderr + a log file).""" - - def __init__(self, primary, secondary): - self._primary = primary - self._secondary = secondary - - def write(self, data): - self._primary.write(data) - self._secondary.write(data) - - def flush(self): - self._primary.flush() - self._secondary.flush() - - def fileno(self): - return self._primary.fileno() - - def isatty(self): - return self._primary.isatty() - def _start_logging_to_files(log: logging.Logger, log_path: pathlib.Path): formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') @@ -354,7 +333,8 @@ def _start_logging_to_files(log: logging.Logger, log_path: pathlib.Path): stdout_file_handler.setFormatter(formatter) log.addHandler(stdout_file_handler) - # Tee sys.stderr to a file while keeping the original stream open as well, - # so that warnings and tracebacks are still visible in the terminal. - stderr_file = open(log_path / 'stderr.txt', 'w', buffering=1) - sys.stderr = _TeeStream(sys.stderr, stderr_file) + # create a file handler for saving stderr logs to a file + stderr_file_handler = logging.FileHandler(log_path / 'stderr.txt') + stderr_file_handler.setLevel(logging.INFO) + stderr_file_handler.setFormatter(formatter) + log.addHandler(stderr_file_handler) diff --git a/dpdl/hyperparameteroptimizer.py b/dpdl/hyperparameteroptimizer.py index e9bf3f28..ee100ab1 100644 --- a/dpdl/hyperparameteroptimizer.py +++ b/dpdl/hyperparameteroptimizer.py @@ -91,12 +91,12 @@ def optimize_hypers(config_manager: ConfigurationManager) -> None: ) if torch.distributed.get_rank() == 0: - log.info('Determining maximum batch size for optimization.') + log.debug('Determining maximum batch size for optimization.') max_batch_size = HyperparameterOptimizer.get_max_batch_size(config_manager) if torch.distributed.get_rank() == 0: - log.info(f'- Maximum batch size for optimization: {max_batch_size}.') + log.debug(f'- Maximum batch size for optimization: {max_batch_size}.') # the optimization objective objective = partial( @@ -152,18 +152,18 @@ def optimize_hypers(config_manager: ConfigurationManager) -> None: # need to also keep track of how many manual trials we are # enqueuing that have not been completed yet if not study._should_skip_enqueue(trial): - log.metrics(f'Enqueuing trial: {trial}') + log.info(f'Enqueuing trial: {trial}') study.enqueue_trial(trial, skip_if_exists=True) enqueued_trial_count += 1 - log.metrics(f'Enqueued {enqueued_trial_count} manual trials.') + log.info(f'Enqueued {enqueued_trial_count} manual trials.') if enqueued_trial_count >= configuration.n_trials: raise ValueError('The number of enqueued trials exceeds or matches the total number of trials. Please reduce the number of manual trials or increase `n_trials` to allow Optuna to perform additional trials.') # Adjust number of random trials to account for enqueued trials remaining_random_trials = max(configuration.optuna_random_trials - enqueued_trial_count, 0) - log.metrics(f'Setting n_startup_trials to {remaining_random_trials} to account for enqueued trials.') + log.info(f'Setting n_startup_trials to {remaining_random_trials} to account for enqueued trials.') sampler = sampler_cls( n_startup_trials=remaining_random_trials, seed=configuration.seed, @@ -183,10 +183,10 @@ def optimize_hypers(config_manager: ConfigurationManager) -> None: # log the results of the best trial if torch.distributed.get_rank() == 0: trial = study.best_trial - log.metrics(f'Best objective ralue: {trial.value}') - log.metrics('Params: ') + log.info(f'Best objective ralue: {trial.value}') + log.info('Params: ') for key, value in trial.params.items(): - log.metrics(f' - {key}: {value}') + log.info(f' - {key}: {value}') # first we need to broadcast the best parameters to all ranks, # so we pack them into a list for sending @@ -238,7 +238,7 @@ def _final_evaluation_round(best_params, config_manager): # using the best params to get a model that we can evaluate on the test # for the final accuracy set if torch.distributed.get_rank() == 0: - log.info('Training final model with best hyperparameters for evaluation.') + log.debug('Training final model with best hyperparameters for evaluation.') # update the training hypers to the best values from the optimization for hyper, best_value in best_params.items(): @@ -254,8 +254,8 @@ def _final_evaluation_round(best_params, config_manager): trainer = TrainerFactory.get_trainer(config_manager) if torch.distributed.get_rank() == 0: - log.metrics('!! Final training round on the full training dataset (train + valid) and evaluating on test.') - log.metrics('--------------------------------------------------------------------------------------------') + log.info('!! Final training round on the full training dataset (train + valid) and evaluating on test.') + log.info('--------------------------------------------------------------------------------------------') # fit model using training AND validation data. we also use the test # set for validation here. @@ -263,11 +263,11 @@ def _final_evaluation_round(best_params, config_manager): # now we can evaluate the final performance of the best model if torch.distributed.get_rank() == 0: - log.metrics('Evaluating final model on the test set.') + log.info('Evaluating final model on the test set.') if torch.distributed.get_rank() == 0: loss, metrics = trainer.test() - log.metrics(f'Final loss: {loss:.4f}') + log.info(f'Final loss: {loss:.4f}') # let's share the loss and metrics with other ranks # rank 0 is the source @@ -284,9 +284,9 @@ def _final_evaluation_round(best_params, config_manager): if torch.distributed.get_rank() == 0: if metrics: - log.metrics('Final metrics:') + log.info('Final metrics:') for key, value in metrics.items(): - log.metrics(f' - {key}: {value}.') + log.info(f' - {key}: {value}.') log_final_epsilon(config_manager, trainer) @@ -296,7 +296,7 @@ def _final_evaluation_round(best_params, config_manager): save_path = config_manager.configuration.model_weights_path else: save_path = Path(config_manager.configuration.log_dir, config_manager.configuration.experiment_name, 'final_model.pt') - log.info(f'Saving final model after HPO to "{save_path}"...') + log.debug(f'Saving final model after HPO to "{save_path}"...') trainer.save_model(save_path) return metrics @@ -370,8 +370,8 @@ def objective( trainer = TrainerFactory.get_trainer(config_manager) if torch.distributed.get_rank() == 0: - log.metrics(f'Starting trial {trial.number}.') - log.info(config_manager.hyperparams) + log.info(f'Starting trial {trial.number}.') + log.debug(config_manager.hyperparams) trainer.fit() @@ -379,7 +379,7 @@ def objective( if torch.distributed.get_rank() == 0: loss, metrics = trainer.validate() - log.info('Writing the loss and metrics of current trial into file.') + log.debug('Writing the loss and metrics of current trial into file.') save_hpo_metrics( config_manager, loss, diff --git a/dpdl/logger_config.py b/dpdl/logger_config.py index ef9b7270..d559a8fd 100644 --- a/dpdl/logger_config.py +++ b/dpdl/logger_config.py @@ -1,33 +1,16 @@ import logging import sys -# Custom level between INFO (20) and WARNING (30): always shown in quiet mode. -METRICS = 25 -logging.addLevelName(METRICS, 'METRICS') -def _metrics(self, message, *args, **kwargs): - if self.isEnabledFor(METRICS): - self._log(METRICS, message, args, **kwargs) - -logging.Logger.metrics = _metrics - -LOG_LEVELS: dict[str, int] = { - 'debug': logging.DEBUG, - 'info': logging.INFO, - 'quiet': METRICS, - 'warning': logging.WARNING, -} - - -def configure_logger(level: int = logging.INFO) -> logging.Logger: +def configure_logger(quiet: bool = False) -> logging.Logger: log = logging.getLogger('dpdl') - - # Keep the logger itself at INFO so that the file handler always receives the full INFO stream - log.setLevel(logging.INFO) + + # Keep the logger itself at DEBUG so the stream handler can be given any level + log.setLevel(logging.DEBUG) # create a stream handler for stdout handler = logging.StreamHandler(sys.stdout) - handler.setLevel(level) + handler.setLevel(logging.INFO if quiet else logging.DEBUG) # create a formatter and set it for the handler formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') @@ -36,7 +19,4 @@ def configure_logger(level: int = logging.INFO) -> logging.Logger: # add the new handler log.addHandler(handler) - # Prevent messages from propagating to the root logger, which causes double logging - log.propagate = False - return log diff --git a/dpdl/metrics_factory.py b/dpdl/metrics_factory.py index b6a9f355..2bf63d48 100644 --- a/dpdl/metrics_factory.py +++ b/dpdl/metrics_factory.py @@ -114,7 +114,7 @@ def get_metrics( if task in ('ImageClassification', 'SequenceClassification'): if torch.distributed.get_rank() == 0: - log.info(f'Task is "{configuration.task}", initializing classification metrics.') + log.debug(f'Task is "{configuration.task}", initializing classification metrics.') if not output_dim or output_dim < 1: raise ValueError('output_dim required for classification tasks') @@ -137,7 +137,7 @@ def get_metrics( elif task in ('CausalLM', 'InstructLM'): if torch.distributed.get_rank() == 0: - log.info(f'Task is "{configuration.task}", initializing language model metrics.') + log.debug(f'Task is "{configuration.task}", initializing language model metrics.') vocab_size = int(output_dim) ignore_index = -100 diff --git a/dpdl/models/model_base.py b/dpdl/models/model_base.py index f6dc2e96..6afdf297 100644 --- a/dpdl/models/model_base.py +++ b/dpdl/models/model_base.py @@ -144,7 +144,7 @@ def load_model( self.model.load_model(fpath) if torch.distributed.get_rank() == 0: - log.info(f'Loaded model from {fpath}') + log.debug(f'Loaded model from {fpath}') return @@ -234,7 +234,7 @@ def overlap(sd: Dict[str, torch.Tensor]) -> Tuple[int, int]: if unexpected: log.warning(f'load_model: unexpected keys ({len(unexpected)}): {unexpected[:20]}') - log.info( + log.debug( f'Loaded weights from {fpath} using candidate={best_name} ' f'(match {matched}/{len(tgt_keys)}, ~{pct:.1f}% of target).' ) diff --git a/dpdl/peft.py b/dpdl/peft.py index 41d2f407..665a2cbb 100644 --- a/dpdl/peft.py +++ b/dpdl/peft.py @@ -25,10 +25,10 @@ def get_nb_trainable_parameters(model: torch.nn.Module): return trainable_params, all_param def print_trainable_modules(model: torch.nn.Module): - log.info('Trainable modules:') + log.debug('Trainable modules:') for module_name, module in model.named_modules(): if any(p.requires_grad for p in module.parameters()): - log.info(module_name) + log.debug(module_name) class PeftFactory: @@ -78,7 +78,7 @@ def get_peft_model(model: torch.nn.Module, model_name: str): if torch.distributed.get_rank() == 0: print_trainable_modules(model) - log.info(f'Finetuning head only - trainable params: {trainable_params:,d} || all params: {all_params:,d} || trainable%: {100 * trainable_params / all_params}') + log.debug(f'Finetuning head only - trainable params: {trainable_params:,d} || all params: {all_params:,d} || trainable%: {100 * trainable_params / all_params}') return model @@ -119,7 +119,7 @@ def get_peft_model(model: torch.nn.Module, model_name: str): if torch.distributed.get_rank() == 0: print_trainable_modules(model) - log.info(f'FiLM setup done - trainable params: {trainable_params:,d} || all params: {all_params:,d} || trainable%: {100 * trainable_params / all_params}') + log.debug(f'FiLM setup done - trainable params: {trainable_params:,d} || all params: {all_params:,d} || trainable%: {100 * trainable_params / all_params}') return model @@ -180,7 +180,7 @@ def get_peft_model( if torch.distributed.get_rank() == 0: print_trainable_modules(model) - log.info( + log.debug( f'LoRA setup done - trainable params: {trainable_params:,d} || all params: {all_params:,d} || trainable%: {100 * trainable_params / all_params}' ) diff --git a/dpdl/predictor.py b/dpdl/predictor.py index 33d84c1d..5cdc2a0e 100644 --- a/dpdl/predictor.py +++ b/dpdl/predictor.py @@ -89,7 +89,7 @@ def predict(self, configuration): for i, (X, y) in enumerate(dataloader): if torch.distributed.get_rank() == 0: - log.info(f' - Predicting on batch {i}') + log.debug(f' - Predicting on batch {i}') # Move inputs to device X, y = self.trainer.adapter.move_to_device(X, y) @@ -159,7 +159,7 @@ def load_model( model.load_model(fpath) if torch.distributed.get_rank() == 0: - log.info(f'Loaded weights from: {fpath}') + log.debug(f'Loaded weights from: {fpath}') def _get_model_params_and_buffers(self): model = self.trainer._unwrap_model() diff --git a/dpdl/trainer.py b/dpdl/trainer.py index bd688f6f..58b6b0ec 100644 --- a/dpdl/trainer.py +++ b/dpdl/trainer.py @@ -331,18 +331,18 @@ def unwrap_model_for_saving(m): # PeftModel knows to save the adapters only model.save_pretrained(fpath) - log.info(f'Saved merged HF PEFT adapters to {fpath}') + log.debug(f'Saved merged HF PEFT adapters to {fpath}') else: # Merge PEFT into model and save the whole model merged = model.merge_and_unload() - log.info(f'GOT A NEW MODEL FROM MERGE_AND_UNLOAD: {merged}') + log.debug(f'GOT A NEW MODEL FROM MERGE_AND_UNLOAD: {merged}') # The `merge_and_unload` will incorporate the LoRA layers in # the model. Then it will return as ModelBase. merged.save_model(fpath) if torch.distributed.get_rank() == 0: - log.info(f'Saved merged HF PEFT model to {fpath}') + log.debug(f'Saved merged HF PEFT model to {fpath}') return @@ -384,7 +384,7 @@ def _sample_impl(self): eos_token_id=self.datamodule.tokenizer.eos_token_id, ) - log.info('Sampled text decoded', self.datamodule.decode(generated_ids)) + log.debug('Sampled text decoded', self.datamodule.decode(generated_ids)) self.model.train() @@ -945,7 +945,7 @@ def _get_target_privacy_params(hyperparams): target_delta = _calculate_target_delta(N) if torch.distributed.get_rank() == 0: - log.info(f'Dataset size is {N}, setting target delta to: {target_delta}.') + log.debug(f'Dataset size is {N}, setting target delta to: {target_delta}.') # are we given a target epsilon? if hyperparams.target_epsilon is not None: diff --git a/run.py b/run.py index 5d4f6b10..6370343b 100644 --- a/run.py +++ b/run.py @@ -9,7 +9,7 @@ from dpdl.cli import cli from dpdl.device import distributed_backend, resolve_device, set_cuda_device -from dpdl.logger_config import LOG_LEVELS, configure_logger +from dpdl.logger_config import configure_logger def setup_torch(): @@ -29,16 +29,8 @@ def setup_torch(): multiprocess.set_start_method('spawn', force=True) -def _parse_log_level_arg(argv) -> int: - import logging - for i, arg in enumerate(argv): - if arg == '--log-level' and i + 1 < len(argv): - return LOG_LEVELS.get(argv[i + 1], logging.INFO) - - if arg.startswith('--log-level='): - return LOG_LEVELS.get(arg.split('=', 1)[1], logging.INFO) - - return logging.INFO +def _parse_quiet_arg(argv) -> bool: + return '--quiet' in argv def _parse_device_arg(argv): @@ -65,7 +57,7 @@ def _resolve_distributed_env(log) -> tuple[int, int, int, str | None, str | None init_method = None if world_size is None or local_rank is None or rank is None: - log.info( + log.debug( "Distributed env vars 'WORLD_SIZE', 'RANK', and 'LOCAL_RANK' not set; " "defaulting to single-process distributed mode." ) @@ -111,7 +103,7 @@ def main(): return 0 - log = configure_logger(level=_parse_log_level_arg(sys.argv)) + log = configure_logger(quiet=_parse_quiet_arg(sys.argv)) setup_torch() device_arg = os.getenv('DPDL_DEVICE') or _parse_device_arg(sys.argv) @@ -119,7 +111,7 @@ def main(): world_size, rank, local_rank, init_method, dist_init_file = _resolve_distributed_env(log) - log.info( + log.debug( f'Rank {rank} initializing - our world size is {world_size} and local rank is {local_rank}.' ) @@ -129,10 +121,10 @@ def main(): # Initialize the process group _init_process_group(device, world_size, rank, init_method) - log.info(f'Rank {rank} initialized.') + log.debug(f'Rank {rank} initialized.') if torch.distributed.get_rank() == 0: - log.info('All ranks initialized.') + log.debug('All ranks initialized.') exit_code = 0 @@ -152,7 +144,7 @@ def main(): torch.distributed.destroy_process_group() - log.info(f'Rank {rank} done!') + log.debug(f'Rank {rank} done!') if dist_init_file is not None: try: From 5915aa56a9ad95e2af453fcd283b026c28feee65 Mon Sep 17 00:00:00 2001 From: fehret Date: Mon, 6 Jul 2026 15:45:29 +0300 Subject: [PATCH 4/4] Added propagate back to ensure functionality of the feature, simplified early flag checking --- dpdl/logger_config.py | 3 +++ run.py | 2 +- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/dpdl/logger_config.py b/dpdl/logger_config.py index d559a8fd..a127dcb5 100644 --- a/dpdl/logger_config.py +++ b/dpdl/logger_config.py @@ -19,4 +19,7 @@ def configure_logger(quiet: bool = False) -> logging.Logger: # add the new handler log.addHandler(handler) + # Prevent messages from propagating to the root logger, which causes double logging + log.propagate = False + return log diff --git a/run.py b/run.py index 6370343b..b76196ee 100644 --- a/run.py +++ b/run.py @@ -103,7 +103,7 @@ def main(): return 0 - log = configure_logger(quiet=_parse_quiet_arg(sys.argv)) + log = configure_logger(quiet='--quiet' in sys.argv) setup_torch() device_arg = os.getenv('DPDL_DEVICE') or _parse_device_arg(sys.argv)