Skip to content

logger

orchard.core.logger

Telemetry and Reporting Package.

This package centralizes experiment logging, environment reporting, and visual telemetry. It provides high-level utilities to initialize system-wide loggers and format experiment metadata for reproducibility.

Available Components:

  • Logger: Static utility for stream and file logging initialization.
  • Reporter: Metadata reporting engine for environment baseline status.
  • LogStyle: Unified logging style constants.
  • Progress functions: Optimization and training progress logging.

LogStyle

Unified logging style constants for consistent visual hierarchy.

Provides separators, symbols, indentation, and ANSI color codes used by all logging modules. Placed here (in paths.constants) rather than in logger.styles so that low-level packages (environment, config) can reference the constants without triggering circular imports.

Reporter

Bases: BaseModel

Centralized logging and reporting utility for experiment lifecycle events.

Transforms complex configuration states and hardware objects into human-readable logs. Called by Orchestrator during initialization.

log_phase_header(log, title, style=None) staticmethod

Log a centered phase header with separator lines.

Parameters:

Name Type Description Default
log Logger

Logger instance to write to.

 required
title str

Header text (will be uppercased and centered).

 required
style str | None

Separator string (defaults to LogStyle.HEAVY).

 None
Source code in orchard/core/logger/env_reporter.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@staticmethod
def log_phase_header(
    log: logging.Logger,
    title: str,
    style: str | None = None,
) -> None:
    """
    Log a centered phase header with separator lines.

    Args:
        log: Logger instance to write to.
        title: Header text (will be uppercased and centered).
        style: Separator string (defaults to ``LogStyle.HEAVY``).
    """
    sep = style if style is not None else LogStyle.HEAVY
    log.info("")
    log.info(sep)
    log.info(title.center(LogStyle.HEADER_WIDTH))
    log.info(sep)

log_initial_status(logger_instance, cfg, paths, device, applied_threads, num_workers)

Logs verified baseline environment configuration upon initialization.

Parameters:

Name Type Description Default
logger_instance Logger

Active experiment logger

 required
cfg 'Config'

Validated global configuration manifest

 required
paths 'RunPaths'

Dynamic path orchestrator for current session

 required
device 'torch.device'

Resolved PyTorch compute device

 required
applied_threads int

Number of intra-op threads assigned

 required
num_workers int

Number of DataLoader workers

 required
Source code in orchard/core/logger/env_reporter.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
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
def log_initial_status(
    self,
    logger_instance: logging.Logger,
    cfg: "Config",
    paths: "RunPaths",
    device: "torch.device",
    applied_threads: int,
    num_workers: int,
) -> None:
    """
    Logs verified baseline environment configuration upon initialization.

    Args:
        logger_instance: Active experiment logger
        cfg: Validated global configuration manifest
        paths: Dynamic path orchestrator for current session
        device: Resolved PyTorch compute device
        applied_threads: Number of intra-op threads assigned
        num_workers: Number of DataLoader workers
    """
    # Header Block
    Reporter.log_phase_header(
        logger_instance, "ENVIRONMENT INITIALIZATION"
    )  # pragma: no mutate

    I = LogStyle.INDENT  # noqa: E741
    A = LogStyle.ARROW

    # Experiment identifier
    logger_instance.info("%s%s %-18s: %s", I, A, "Experiment", cfg.run_slug)
    logger_instance.info("")

    # Task Section
    logger_instance.info("[TASK]")
    logger_instance.info("%s%s %-18s: %s", I, A, "Type", cfg.task_type.capitalize())
    logger_instance.info("")

    # Hardware Section
    self._log_hardware_section(logger_instance, cfg, device, applied_threads, num_workers)
    logger_instance.info("")

    # Dataset Section
    self._log_dataset_section(logger_instance, cfg)
    logger_instance.info("")

    # Strategy Section
    self._log_strategy_section(logger_instance, cfg, device)
    logger_instance.info("")

    # Hyperparameters Section
    logger_instance.info("[HYPERPARAMETERS]")
    logger_instance.info("%s%s %-18s: %s", I, A, "Epochs", cfg.training.epochs)
    logger_instance.info("%s%s %-18s: %s", I, A, "Batch Size", cfg.training.batch_size)
    lr = cfg.training.learning_rate
    lr_str = f"{lr:.2e}" if isinstance(lr, (float, int)) else str(lr)
    logger_instance.info("%s%s %-18s: %s", I, A, "Initial LR", lr_str)
    logger_instance.info("")

    # Tracking Section (only if configured)
    self._log_tracking_section(logger_instance, cfg)

    # Optimization Section (only if configured)
    self._log_optimization_section(logger_instance, cfg)

    # Export Section (only if configured)
    self._log_export_section(logger_instance, cfg)

    # Filesystem Section
    logger_instance.info("[FILESYSTEM]")
    logger_instance.info("%s%s %-18s: %s", I, A, "Run Root", paths.root.name)
    logger_instance.info(
        "%s%s %-18s: config.yaml, requirements.txt, git_info.txt", I, A, "Manifest"
    )

    # Closing separator
    logger_instance.info(LogStyle.HEAVY)
    logger_instance.info("")

ReporterProtocol

Bases: Protocol

Protocol for environment reporting, allowing mocking in tests.

log_initial_status(logger_instance, cfg, paths, device, applied_threads, num_workers)

Logs the initial status of the environment.

Parameters:

Name Type Description Default
logger_instance Logger

The logger instance used to log the status.

 required
cfg 'Config'

The configuration object containing environment settings.

 required
paths 'RunPaths'

The paths object with directories for the run.

 required
device device

The device (e.g., CPU or GPU) to be used for processing.

 required
applied_threads int

The number of threads allocated for processing.

 required
num_workers int

The number of worker processes to use.

 required
Source code in orchard/core/logger/env_reporter.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def log_initial_status(
    self,
    logger_instance: logging.Logger,
    cfg: "Config",
    paths: "RunPaths",
    device: torch.device,
    applied_threads: int,
    num_workers: int,
) -> None:
    """
    Logs the initial status of the environment.

    Args:
        logger_instance: The logger instance used to log the status.
        cfg: The configuration object containing environment settings.
        paths: The paths object with directories for the run.
        device: The device (e.g., CPU or GPU) to be used for processing.
        applied_threads: The number of threads allocated for processing.
        num_workers: The number of worker processes to use.
    """
    ...  # pragma: no cover

Logger(name=LOGGER_NAME, log_dir=None, log_to_file=True, level=logging.INFO, max_bytes=5 * 1024 * 1024, backup_count=5)

Manages centralized logging configuration with singleton-like behavior.

Provides a unified logging interface for the entire framework with support for dynamic reconfiguration. Initially bootstraps with console-only output, then transitions to dual console+file logging when experiment directories become available.

The logger implements pseudo-singleton semantics via class-level tracking (_configured_names) to prevent duplicate handler registration while allowing intentional reconfiguration when log directories are provided.

Lifecycle
  1. Bootstrap Phase: Console-only logging (no log_dir specified)
  2. Orchestration Phase: RootOrchestrator calls setup() with log_dir
  3. Reconfiguration: Existing handlers removed, file handler added

Class Attributes: _configured_names (dict[str, bool]): Tracks which logger names have been configured

Attributes:

Name Type Description
name str

Logger identifier (typically LOGGER_NAME constant)
log_dir Path | None

Directory for log file storage
log_to_file bool

Enable file logging (requires log_dir)
level int

Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
max_bytes int

Maximum log file size before rotation (default: 5MB)
backup_count int

Number of rotated log files to retain (default: 5)
_log Logger

Underlying Python logger instance
Example

Bootstrap phase (console-only)

logger = Logger().get_logger() logger.info("Framework initializing...")

Orchestration phase (add file logging)

logger = Logger.setup( ... name=LOGGER_NAME, ... log_dir=Path("./outputs/run_123/logs"), ... level="INFO" ... ) logger.info("Logging to file now")

Notes:

  • Reconfiguration is idempotent: calling setup() multiple times is safe
  • All handlers are properly closed before reconfiguration
  • Log files use UTC timestamps for consistency across time zones
  • RotatingFileHandler prevents disk space exhaustion

Initializes the Logger with specified configuration.

Parameters:

Name Type Description Default
name str

Logger identifier (default: LOGGER_NAME constant)

 LOGGER_NAME
log_dir Path | None

Directory for log file storage (None = console-only)

 None
log_to_file bool

Enable file logging if log_dir provided (default: True)

 True
level int

Logging level as integer constant (default: logging.INFO)

 INFO
max_bytes int

Maximum log file size before rotation in bytes (default: 5MB)

 5 * 1024 * 1024
backup_count int

Number of rotated backup files to retain (default: 5)

 5
Source code in orchard/core/logger/logger.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def __init__(
    self,
    name: str = LOGGER_NAME,
    log_dir: Path | None = None,
    log_to_file: bool = True,
    level: int = logging.INFO,
    max_bytes: int = 5 * 1024 * 1024,
    backup_count: int = 5,
) -> None:
    """
    Initializes the Logger with specified configuration.

    Args:
        name: Logger identifier (default: LOGGER_NAME constant)
        log_dir: Directory for log file storage (None = console-only)
        log_to_file: Enable file logging if log_dir provided (default: True)
        level: Logging level as integer constant (default: logging.INFO)
        max_bytes: Maximum log file size before rotation in bytes (default: 5MB)
        backup_count: Number of rotated backup files to retain (default: 5)
    """
    self.name = name
    self.log_dir = log_dir
    self.log_to_file = log_to_file and (log_dir is not None)
    self.level = level
    self.max_bytes = max_bytes
    self.backup_count = backup_count

    self._log = logging.getLogger(name)

    if name not in Logger._configured_names or log_dir is not None:
        self._setup_logger()
        Logger._configured_names[name] = True

get_logger()

Returns the configured logging.Logger instance.

Returns:

Type Description
Logger

The underlying Python logging.Logger instance with configured handlers
Source code in orchard/core/logger/logger.py 
284
285
286
287
288
289
290
291
def get_logger(self) -> logging.Logger:
    """
    Returns the configured logging.Logger instance.

    Returns:
        The underlying Python logging.Logger instance with configured handlers
    """
    return self._log

setup(name, log_dir=None, level='INFO', **kwargs) classmethod

Main entry point for configuring the logger, called by RootOrchestrator.

Bridges semantic LogLevel strings (INFO, DEBUG, WARNING) to Python logging constants. Provides convenient string-based level specification while internally using numeric logging constants.

Parameters:

Name Type Description Default
name str

Logger identifier (typically LOGGER_NAME constant)

 required
log_dir Path | None

Directory for log file storage (None = console-only mode)

 None
level str

Logging level as string (DEBUG, INFO, WARNING, ERROR, CRITICAL)

 'INFO'
**kwargs Any

Additional arguments passed to Logger constructor

 {}

Returns:

Type Description
Logger

Configured logging.Logger instance ready for use
Environment Variables

DEBUG: If set to "1", overrides level to DEBUG regardless of level parameter

Example

logger = Logger.setup( ... name="OrchardML", ... log_dir=Path("./outputs/run_123/logs"), ... level="INFO" ... ) logger.info("Training started")

Source code in orchard/core/logger/logger.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
@classmethod
def setup(
    cls, name: str, log_dir: Path | None = None, level: str = "INFO", **kwargs: Any
) -> logging.Logger:
    """
    Main entry point for configuring the logger, called by RootOrchestrator.

    Bridges semantic LogLevel strings (INFO, DEBUG, WARNING) to Python logging
    constants. Provides convenient string-based level specification while internally
    using numeric logging constants.

    Args:
        name: Logger identifier (typically LOGGER_NAME constant)
        log_dir: Directory for log file storage (None = console-only mode)
        level: Logging level as string (DEBUG, INFO, WARNING, ERROR, CRITICAL)
        **kwargs (Any): Additional arguments passed to Logger constructor

    Returns:
        Configured logging.Logger instance ready for use

    Environment Variables:
        DEBUG: If set to "1", overrides level to DEBUG regardless of level parameter

    Example:
        >>> logger = Logger.setup(
        ...     name="OrchardML",
        ...     log_dir=Path("./outputs/run_123/logs"),
        ...     level="INFO"
        ... )
        >>> logger.info("Training started")
    """
    if os.getenv("DEBUG") == "1":
        numeric_level = logging.DEBUG
    else:
        numeric_level = getattr(logging, level.upper(), logging.INFO)

    return cls(name=name, log_dir=log_dir, level=numeric_level, **kwargs).get_logger()

log_optimization_header(cfg, logger_instance=None)

Log Optuna optimization configuration details.

Logs search-specific parameters only (dataset/model already shown in environment).

Parameters:

Name Type Description Default
cfg 'Config'

Configuration with optuna settings

 required
logger_instance Logger | None

Logger instance to use (defaults to module logger)

 None
Source code in orchard/core/logger/progress.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def log_optimization_header(cfg: "Config", logger_instance: logging.Logger | None = None) -> None:
    """
    Log Optuna optimization configuration details.

    Logs search-specific parameters only (dataset/model already shown in environment).

    Args:
        cfg: Configuration with optuna settings
        logger_instance: Logger instance to use (defaults to module logger)
    """
    log = logger_instance or logger

    # Search configuration (no duplicate header - phase header already shown)
    log.info("")
    I = LogStyle.INDENT  # noqa: E741  # pragma: no mutate
    A = LogStyle.ARROW  # pragma: no mutate
    log.info("%s%s Dataset      : %s", I, A, cfg.dataset.dataset_name)
    model_search = "Enabled" if cfg.optuna.enable_model_search else "Disabled"  # pragma: no mutate
    log.info("%s%s Model Search : %s", I, A, model_search)
    if cfg.optuna.model_pool is not None:
        log.info("%s%s Model Pool   : %s", I, A, ", ".join(cfg.optuna.model_pool))
    log.info("%s%s Search Space : %s", I, A, cfg.optuna.search_space_preset)
    log.info("%s%s Trials       : %s", I, A, cfg.optuna.n_trials)
    log.info("%s%s Epochs/Trial : %s", I, A, cfg.optuna.epochs)
    log.info("%s%s Metric       : %s", I, A, cfg.training.monitor_metric)
    pruning = "Enabled" if cfg.optuna.enable_pruning else "Disabled"  # pragma: no mutate
    log.info("%s%s Pruning      : %s", I, A, pruning)

    if cfg.optuna.enable_early_stopping:
        threshold = cfg.optuna.early_stopping_threshold or "auto"  # pragma: no mutate
        log.info(
            "%s%s Early Stop   : Enabled (threshold=%s, patience=%s)",
            I,
            A,
            threshold,
            cfg.optuna.early_stopping_patience,
        )

    log.info("")

log_optimization_summary(study, cfg, device, paths, logger_instance=None)

Log optimization study completion summary.

Parameters:

Name Type Description Default
study 'optuna.Study'

Completed Optuna study

 required
cfg 'Config'

Configuration object

 required
device 'torch.device'

PyTorch device used

 required
paths 'RunPaths'

Run paths for artifacts

 required
logger_instance Logger | None

Logger instance to use (defaults to module logger)

 None
Source code in orchard/core/logger/progress.py 
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
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def log_optimization_summary(
    study: "optuna.Study",
    cfg: "Config",
    device: "torch.device",
    paths: "RunPaths",
    logger_instance: logging.Logger | None = None,
) -> None:
    """
    Log optimization study completion summary.

    Args:
        study: Completed Optuna study
        cfg: Configuration object
        device: PyTorch device used
        paths: Run paths for artifacts
        logger_instance: Logger instance to use (defaults to module logger)
    """
    log = logger_instance or logger
    completed, pruned, failed = _count_trial_states(study)

    I = LogStyle.INDENT  # noqa: E741  # pragma: no mutate
    A = LogStyle.ARROW  # pragma: no mutate
    S = LogStyle.SUCCESS  # pragma: no mutate
    W = LogStyle.WARNING  # pragma: no mutate

    Reporter.log_phase_header(log, "OPTIMIZATION SUMMARY", LogStyle.DOUBLE)  # pragma: no mutate
    log.info("%s%s Dataset        : %s", I, A, cfg.dataset.dataset_name)
    log.info("%s%s Search Space   : %s", I, A, cfg.optuna.search_space_preset)
    log.info("%s%s Total Trials   : %d", I, A, len(study.trials))
    log.info("%s%s Completed      : %d", I, S, len(completed))
    log.info("%s%s Pruned         : %d", I, A, len(pruned))

    if failed:
        log.info("%s%s Failed         : %d", I, W, len(failed))

    if completed:
        try:
            log.info(
                "%s%s Best %-9s : %.6f",
                I,
                S,
                cfg.training.monitor_metric.upper(),
                study.best_value,
            )
            log.info("%s%s Best Trial     : %d", I, S, study.best_trial.number)
        except ValueError:  # pragma: no cover
            # fmt: off
            log.error("%s%s Best trial lookup failed (check study integrity)", I, W)  # pragma: no mutate
            # fmt: on
    else:
        log.warning("%s%s No trials completed", I, W)

    log.info("%s%s Device         : %s", I, A, str(device).upper())
    log.info("%s%s Artifacts      : %s", I, A, Path(paths.root).name)
    log.info(LogStyle.DOUBLE)
    log.info("")

log_pipeline_summary(test_acc, macro_f1, best_model_path, run_dir, duration, test_auc=None, onnx_path=None, logger_instance=None)

Log final pipeline completion summary.

Called at the end of the pipeline after all phases complete. Consolidates key metrics and artifact locations.

Parameters:

Name Type Description Default
test_acc float

Final test accuracy

 required
macro_f1 float

Final macro F1 score

 required
best_model_path Path

Path to best model checkpoint

 required
run_dir Path

Root directory for this run

 required
duration str

Human-readable duration string

 required
test_auc float | None

Final test AUC (if available)

 None
onnx_path Path | None

Path to ONNX export (if performed)

 None
logger_instance Logger | None

Logger instance to use (defaults to module logger)

 None
Source code in orchard/core/logger/progress.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def log_pipeline_summary(
    test_acc: float,
    macro_f1: float,
    best_model_path: Path,
    run_dir: Path,
    duration: str,
    test_auc: float | None = None,
    onnx_path: Path | None = None,
    logger_instance: logging.Logger | None = None,
) -> None:
    """
    Log final pipeline completion summary.

    Called at the end of the pipeline after all phases complete.
    Consolidates key metrics and artifact locations.

    Args:
        test_acc: Final test accuracy
        macro_f1: Final macro F1 score
        best_model_path: Path to best model checkpoint
        run_dir: Root directory for this run
        duration: Human-readable duration string
        test_auc: Final test AUC (if available)
        onnx_path: Path to ONNX export (if performed)
        logger_instance: Logger instance to use (defaults to module logger)
    """
    log = logger_instance or logger

    I = LogStyle.INDENT  # noqa: E741  # pragma: no mutate
    A = LogStyle.ARROW  # pragma: no mutate
    S = LogStyle.SUCCESS  # pragma: no mutate

    Reporter.log_phase_header(log, "PIPELINE COMPLETE", LogStyle.DOUBLE)  # pragma: no mutate
    log.info("%s%s Test Accuracy  : %7.2f%%", I, S, test_acc * 100)
    log.info("%s%s Macro F1       : %8.4f", I, S, macro_f1)
    if test_auc is not None:
        log.info("%s%s Test AUC       : %8.4f", I, S, test_auc)
    log.info("%s%s Best Model     : %s", I, A, Path(best_model_path).name)
    if onnx_path:
        log.info("%s%s ONNX Export    : %s", I, A, Path(onnx_path).name)
    log.info("%s%s Run Directory  : %s", I, A, Path(run_dir).name)
    log.info("%s%s Duration       : %s", I, A, duration)
    log.info(LogStyle.DOUBLE)

log_trial_start(trial_number, params, logger_instance=None)

Log trial start with formatted parameters (grouped by category).

Parameters:

Name Type Description Default
trial_number int

Trial index

 required
params dict[str, Any]

Sampled hyperparameters

 required
logger_instance Logger | None

Logger instance to use (defaults to module logger)

 None
Source code in orchard/core/logger/progress.py 
101
102
103
104
105
106
107
108
109
110
111
112
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
def log_trial_start(
    trial_number: int, params: dict[str, Any], logger_instance: logging.Logger | None = None
) -> None:
    """
    Log trial start with formatted parameters (grouped by category).

    Args:
        trial_number: Trial index
        params: Sampled hyperparameters
        logger_instance: Logger instance to use (defaults to module logger)
    """
    log = logger_instance or logger

    log.info(LogStyle.LIGHT)
    log.info("[Trial %d Hyperparameters]", trial_number)

    categories = {
        "Optimization": ["learning_rate", "weight_decay", "momentum", "min_lr"],
        "Loss": ["criterion_type", "focal_gamma", "label_smoothing"],
        "Regularization": ["mixup_alpha", "dropout"],
        "Scheduling": ["scheduler_type", "scheduler_patience", "batch_size"],
        "Augmentation": ["rotation_angle", "jitter_val", "min_scale"],
        "Architecture": ["model_name", "pretrained", "weight_variant"],
    }

    for category_name, keys in categories.items():
        category_params = {k: params[k] for k in keys if k in params}
        if category_params:
            log.info("%s[%s]", LogStyle.INDENT, category_name)
            for key, value in category_params.items():
                log.info(
                    "%s%s %-20s : %s",
                    LogStyle.DOUBLE_INDENT,
                    LogStyle.BULLET,
                    key,
                    _format_param_value(value),
                )

    log.info(LogStyle.LIGHT)