nirs4all.core.logging package

Submodules

Module contents

Unified logging system for nirs4all.

This module provides a structured, configurable logging infrastructure that replaces all print() statements with proper logging. It supports:

Human-readable console output optimized for scientists
Machine-parseable file logging for automation
Context tracking for runs, phases, branches, and sources
ASCII-safe output for HPC/cluster environments
Progress throttling to avoid terminal flooding
TTY-aware progress bars with multi-level support

Usage:

>>> from nirs4all.core.logging import get_logger, configure_logging, LogContext
>>>
>>> # Configure at application startup
>>> configure_logging(verbose=1, use_unicode=True)
>>>
>>> # Get logger in each module
>>> logger = get_logger(__name__)
>>>
>>> # Use context for run tracking
>>> with LogContext(run_id="my-experiment"):
...     logger.info("Starting analysis")
...     with LogContext.branch("snv", index=0, total=4):
...         logger.success("SNV preprocessing complete")

Progress Bars:

>>> from nirs4all.core.logging import ProgressBar, EvaluationProgress
>>>
>>> # Simple progress bar
>>> with ProgressBar(total=100, description="Processing") as pbar:
...     for i in range(100):
...         pbar.update(1)
>>>
>>> # ML-specific evaluation progress
>>> with EvaluationProgress(total_pipelines=42, metric_name="RMSE") as progress:
...     for pipeline in pipelines:
...         score = evaluate(pipeline)
...         progress.update(score=score)

See also

nirs4all.core.logging.config for configuration details
nirs4all.core.logging.context for context management
nirs4all.core.logging.formatters for output formatting
nirs4all.core.logging.progress for progress bars

class nirs4all.core.logging.BranchContext(name: str, path: list[str] = <factory>, index: int | None = None, total: int | None = None, parent: str | None = None, depth: int = 0)[source]

Bases: object

Context for a branch in the pipeline.

name

Branch name.

Type:: str

path

Full branch path (for nested branches).

Type:: list[str]

index

Branch index (0-based).

Type:: int | None

total

Total number of branches at this level.

Type:: int | None

parent

Parent branch name (for nested branches).

Type:: str | None

depth

Nesting depth (0 for top-level).

Type:: int

depth: int = 0

index: int | None = None

name: str

parent: str | None = None

path: list[str]

total: int | None = None

class nirs4all.core.logging.BufferedHandler(max_size: int = 1000)[source]

Bases: Handler

Handler that buffers log records for batch processing.

Useful for collecting logs during a phase and outputting them together, e.g., for branch comparison summaries.

clear() → None[source]: Clear the buffer.

emit(record: LogRecord) → None[source]

Buffer the log record.

Parameters:: record – Log record to buffer.

flush_to(handler: Handler) → None[source]

Flush buffered records to another handler.

Parameters:: handler – Handler to send buffered records to.

get_records() → list[LogRecord][source]

Get buffered records.

Returns:: List of buffered log records.

class nirs4all.core.logging.ConsoleFormatter(use_colors: bool = True, show_elapsed: bool = False, use_unicode: bool = True)[source]

Bases: Formatter

Human-readable console formatter for nirs4all logs.

Produces clean, scannable output optimized for terminal viewing with proper indentation for hierarchy, status symbols, and optional elapsed time.

COLORS = {'BLUE': '\x1b[34m', 'BOLD': '\x1b[1m', 'CYAN': '\x1b[36m', 'DIM': '\x1b[2m', 'GREEN': '\x1b[32m', 'MAGENTA': '\x1b[35m', 'RED': '\x1b[31m', 'RESET': '\x1b[0m', 'WHITE': '\x1b[37m', 'YELLOW': '\x1b[33m'}

LEVEL_COLORS = {'CRITICAL': 'RED', 'DEBUG': 'DIM', 'ERROR': 'RED', 'INFO': 'RESET', 'WARNING': 'YELLOW'}

format(record: LogRecord) → str[source]

Format a log record for console output.

Parameters:: record – Log record to format.
Returns:: Formatted log message.

set_run_start(start_time: datetime) → None[source]

Set the run start time for elapsed time calculation.

Parameters:: start_time – Datetime when the run started.

class nirs4all.core.logging.EvaluationProgress(total_pipelines: int, metric_name: str = 'score', higher_is_better: bool = False, description: str = 'Evaluating pipelines', config: ProgressConfig | None = None, disable: bool = False)[source]

Bases: object

Specialized progress tracker for pipeline evaluation.

Provides ML-specific tracking with best score updates and automatic milestone reporting.

Example

>>> progress = EvaluationProgress(
...     total_pipelines=42,
...     metric_name="RMSE",
...     higher_is_better=False
... )
>>>
>>> with progress:
...     for pipeline in pipelines:
...         score = evaluate(pipeline)
...         progress.update(score=score, pipeline_name="SavGol+PLS")

__enter__() → EvaluationProgress[source]: Enter context manager.

__exit__(exc_type: Any, exc_val: Any, exc_tb: Any) → None[source]: Exit context manager.

property best_pipeline: str | None: Get the name of the best pipeline.

property best_score: float | None: Get the current best score.

update(score: float | None = None, pipeline_name: str | None = None, n: int = 1) → bool[source]

Update progress with optional score.

Parameters:

score – Score for the completed pipeline.
pipeline_name – Name of the pipeline.
n – Number of pipelines completed.

Returns:

True if this was a new best score.

class nirs4all.core.logging.EventType(value)[source]

Bases: str, Enum

Event types within phases.

ARTIFACT = 'artifact'

BRANCH_COMPARE = 'branch_compare'

BRANCH_ENTER = 'branch_enter'

BRANCH_EXIT = 'branch_exit'

BUILD = 'build'

COLLECT = 'collect'

COMPLETE = 'complete'

CONFIG_LOAD = 'config_load'

DEDUPLICATE = 'deduplicate'

ENVIRONMENT = 'environment'

ERROR = 'error'

EXPAND = 'expand'

LEAKAGE_CHECK = 'leakage_check'

LOAD = 'load'

METRIC = 'metric'

MODEL = 'model'

PREDICTIONS = 'predictions'

PROGRESS = 'progress'

PRUNE = 'prune'

REPORT = 'report'

SKIP = 'skip'

SOURCE_CONCAT = 'source_concat'

SOURCE_PROCESS = 'source_process'

START = 'start'

STEP = 'step'

TRAIN_META = 'train_meta'

TRANSFORM = 'transform'

VALIDATE = 'validate'

WARNING = 'warning'

class nirs4all.core.logging.FileFormatter[source]

Bases: Formatter

File formatter for human-readable log files.

Produces timestamped, leveled output suitable for file storage and later analysis.

format(record: LogRecord) → str[source]

Format a log record for file output.

Parameters:: record – Log record to format.
Returns:: Formatted log message with timestamp and level.

class nirs4all.core.logging.JsonFormatter(run_id: str | None = None)[source]

Bases: Formatter

JSON Lines formatter for machine-readable log files.

Produces one JSON object per line for easy parsing by log aggregation systems like ELK, Loki, etc.

format(record: LogRecord) → str[source]

Format a log record as JSON.

Parameters:: record – Log record to format.
Returns:: JSON string representation of the log record.

set_run_id(run_id: str) → None[source]

Set the run ID for log entries.

Parameters:: run_id – Unique run identifier.

class nirs4all.core.logging.LogContext(run_id: str | None = None, run_name: str | None = None, project: str | None = None, **extra: Any)[source]

Bases: object

Context manager for run-level logging context.

Provides a context manager that sets up run state and can be used to track the current run, phase, branches, sources, etc.

Example

>>> with LogContext(run_id="my-experiment", project="protein"):
...     logger.info("Starting analysis")
...     with LogContext.branch("snv", index=0, total=4):
...         logger.info("Processing SNV branch")

__enter__() → LogContext[source]: Enter the context, setting up run state.

__exit__(exc_type: Any, exc_val: Any, exc_tb: Any) → None[source]: Exit the context, restoring previous state.

static branch(name: str, index: int | None = None, total: int | None = None, parent: str | None = None) → Generator[BranchContext, None, None][source]

Context manager for tracking branch execution.

Parameters:

name – Branch name.
index – Branch index (0-based).
total – Total number of branches at this level.
parent – Parent branch name (for nested branches).

Yields:

BranchContext for the current branch.

static phase(phase: Phase) → Generator[None, None, None][source]

Context manager for tracking the current phase.

Parameters:: phase – The phase being entered.
Yields:: None

static source(name: str, index: int | None = None, total: int | None = None) → Generator[SourceContext, None, None][source]

Context manager for tracking source processing in multi-source pipelines.

Parameters:

name – Source name.
index – Source index (0-based).
total – Total number of sources.

Yields:

SourceContext for the current source.

static stack(n_branches: int, meta_model: str | None = None, branch_sources: list[str] | None = None) → Generator[StackContext, None, None][source]

Context manager for tracking stacking operations.

Parameters:

n_branches – Number of branches being stacked.
meta_model – Meta-model name/description.
branch_sources – List of branch names being stacked.

Yields:

StackContext for the stacking operation.

class nirs4all.core.logging.LogEvent(timestamp: ~datetime.datetime, level: str, phase: ~nirs4all.core.logging.events.Phase | None, event_type: ~nirs4all.core.logging.events.EventType, message: str, run_id: str | None = None, status: ~nirs4all.core.logging.events.Status | None = None, branch_name: str | None = None, branch_path: list[str] | None = None, branch_index: int | None = None, source_index: int | None = None, source_name: str | None = None, duration_ms: float | None = None, metrics: dict[str, ~typing.Any] = <factory>, extra: dict[str, ~typing.Any] = <factory>)[source]

Bases: object

Structured log event for machine-readable logging.

timestamp

Event timestamp.

Type:: datetime.datetime

level

Log level (INFO, DEBUG, WARNING, ERROR).

Type:: str

phase

Current workflow phase.

Type:: nirs4all.core.logging.events.Phase | None

event_type

Type of event within the phase.

Type:: nirs4all.core.logging.events.EventType

message

Human-readable message.

Type:: str

run_id

Unique run identifier.

Type:: str | None

status

Event status indicator.

Type:: nirs4all.core.logging.events.Status | None

branch_name

Current branch name (if in branch context).

Type:: str | None

branch_path

Full branch path for nested branches.

Type:: list[str] | None

branch_index

Index of current branch (0-based).

Type:: int | None

source_index

Index of current source (for multi-source).

Type:: int | None

source_name

Name of current source.

Type:: str | None

duration_ms

Duration in milliseconds (for completion events).

Type:: float | None

metrics

Dict of metric name to value.

Type:: dict[str, Any]

extra

Additional context fields.

Type:: dict[str, Any]

branch_index: int | None = None

branch_name: str | None = None

branch_path: list[str] | None = None

duration_ms: float | None = None

event_type: EventType

extra: dict[str, Any]

level: str

message: str

metrics: dict[str, Any]

phase: Phase | None

run_id: str | None = None

source_index: int | None = None

source_name: str | None = None

status: Status | None = None

timestamp: datetime

to_dict() → dict[str, Any][source]: Convert event to dictionary for JSON serialization.

to_json() → str[source]: Convert event to JSON string.

class nirs4all.core.logging.MultiLevelProgress(run_total: int | None = None, run_description: str = 'Run progress', config: ProgressConfig | None = None, disable: bool = False)[source]

Bases: object

Multi-level progress tracking for nested operations.

Supports tracking progress at multiple levels: - Run level (overall run progress) - Pipeline level (pipeline evaluation within run) - Fold level (CV folds within pipeline) - Training level (epochs/batches within training)

Example

>>> progress = MultiLevelProgress(
...     run_total=5,
...     run_description="Evaluating datasets"
... )
>>>
>>> with progress.run_level() as run_pbar:
...     for dataset in datasets:
...         with progress.pipeline_level(total=10) as pipe_pbar:
...             for pipeline in pipelines:
...                 # evaluate
...                 pipe_pbar.update(1)
...         run_pbar.update(1)

close_all() → None[source]: Close all active progress bars.

fold_level(total: int, description: str = 'Cross-validation') → ProgressBar[source]

Get fold-level progress bar.

Parameters:

total – Total number of folds.
description – Description text.

Returns:

ProgressBar for fold level.

pipeline_level(total: int, description: str = 'Evaluating pipelines') → ProgressBar[source]

Get pipeline-level progress bar.

Parameters:

total – Total number of pipelines.
description – Description text.

Returns:

ProgressBar for pipeline level.

run_level(total: int | None = None, description: str | None = None) → ProgressBar[source]

Get run-level progress bar.

Parameters:

total – Override total (uses init value if None).
description – Override description.

Returns:

ProgressBar for run level.

training_level(total: int, description: str = 'Training') → ProgressBar[source]

Get training-level progress bar.

Parameters:

total – Total epochs or batches.
description – Description text.

Returns:

ProgressBar for training level.

class nirs4all.core.logging.Nirs4allLogger(name: str, level: int = 0)[source]

Bases: Logger

Extended logger with nirs4all-specific methods.

Provides convenience methods for structured logging with status indicators, progress reporting, and context-aware logging.

artifact(artifact_type: str, path: str | Path, size_bytes: int | None = None, **kwargs: Any) → None[source]

Log an artifact save.

Parameters:

artifact_type – Type of artifact (“model”, “report”, “predictions”).
path – Path where artifact was saved.
size_bytes – Size of artifact in bytes.
**kwargs – Additional logging kwargs.

makeRecord(name: str, level: int, fn: str, lno: int, msg: object, args: tuple, exc_info: Any, func: str | None = None, extra: dict[str, Any] | None = None, sinfo: str | None = None) → LogRecord[source]

Create a LogRecord with context injection.

Overrides the standard makeRecord to inject run context.

metric(name: str, value: float, scope: str = 'cv', fold: int | None = None, pipeline: str | None = None, **kwargs: Any) → None[source]

Log a metric value.

Parameters:

name – Metric name (e.g., “RMSE”, “R2”).
value – Metric value.
scope – Metric scope (“cv”, “train”, “test”, “fold”).
fold – Fold number (for per-fold metrics).
pipeline – Pipeline identifier.
**kwargs – Additional logging kwargs.

phase_complete(phase: Phase, duration_seconds: float | None = None, **extra_fields: Any) → None[source]

Log the completion of a workflow phase.

Parameters:

phase – The phase that completed.
duration_seconds – Phase duration in seconds.
**extra_fields – Additional context fields.

phase_start(phase: Phase, **extra_fields: Any) → None[source]

Log the start of a workflow phase.

Parameters:

phase – The phase being started.
**extra_fields – Additional context fields.

progress(operation: str, current: int, total: int, best_score: float | None = None, is_new_best: bool = False, extra_fields: dict[str, Any] | None = None, **kwargs: Any) → None[source]

Log a progress update.

Progress messages are automatically throttled to avoid flooding.

Parameters:

operation – Name of the operation in progress.
current – Current step number.
total – Total number of steps.
best_score – Current best score (optional).
is_new_best – If True, this update reports a new best result.
extra_fields – Additional context fields.
**kwargs – Additional logging kwargs.

run_footer(status: Status, duration_seconds: float, best_pipeline: str | None = None, metrics: dict[str, float] | None = None) → None[source]

Log the run footer block.

Parameters:

status – Final run status.
duration_seconds – Total run duration.
best_pipeline – Best pipeline description.
metrics – Final metrics.

run_header(run_name: str, environment_info: dict[str, str] | None = None, reproducibility_info: dict[str, str] | None = None) → None[source]

Log the run header block.

Parameters:

run_name – Name of the run.
environment_info – Environment details (Python version, etc.).
reproducibility_info – Reproducibility info (seed, git hash, etc.).

starting(msg: str, *args: Any, extra_fields: dict[str, Any] | None = None, **kwargs: Any) → None[source]

Log a starting/beginning message.

Parameters:

msg – Log message.
*args – Message formatting args.
extra_fields – Additional context fields.
**kwargs – Additional logging kwargs.

success(msg: str, *args: Any, extra_fields: dict[str, Any] | None = None, **kwargs: Any) → None[source]

Log a success message.

Parameters:

msg – Log message.
*args – Message formatting args.
extra_fields – Additional context fields.
**kwargs – Additional logging kwargs.

trace(msg: str, *args: Any, **kwargs: Any) → None[source]

Log at TRACE level (below DEBUG).

Parameters:

msg – Log message.
*args – Message formatting args.
**kwargs – Additional logging kwargs.

class nirs4all.core.logging.NullHandler(level=0)[source]

Bases: Handler

Handler that discards all log records.

Used when logging should be completely silent.

emit(record: LogRecord) → None[source]

Discard the log record.

Parameters:: record – Log record (ignored).

class nirs4all.core.logging.Phase(value)[source]

Bases: str, Enum

Major workflow phases for high-level tracking.

BRANCH = 'branch'

COMPLETE = 'complete'

DATA = 'data'

EVALUATE = 'evaluate'

EXPORT = 'export'

GENERATE = 'generate'

INIT = 'init'

PREDICT = 'predict'

SOURCE = 'source'

SPLIT = 'split'

STACK = 'stack'

TRAIN = 'train'

class nirs4all.core.logging.ProgressBar(total: int, description: str = '', config: ProgressConfig | None = None, leave: bool = True, disable: bool = False, unit: str = 'it', file: Any = None, initial: int = 0, ncols: int | None = None)[source]

Bases: object

TTY-aware progress bar for tracking iterations.

Provides a clean progress display that automatically adapts to terminal capabilities. Falls back to periodic line-based updates when not in a TTY.

Example

>>> with ProgressBar(total=100, description="Processing") as pbar:
...     for i in range(100):
...         # do work
...         pbar.update(1)

>>> # Or with iterator
>>> for item in ProgressBar.wrap(items, description="Processing"):
...     # do work with item

COLOR_CYAN = '\x1b[36m'

COLOR_DIM = '\x1b[2m'

COLOR_GREEN = '\x1b[32m'

COLOR_RESET = '\x1b[0m'

COLOR_YELLOW = '\x1b[33m'

EMPTY_ASCII = '-'

EMPTY_UNICODE = '░'

FILLED_ASCII = '#'

FILLED_UNICODE = '█'

__enter__() → ProgressBar[source]: Enter context manager.

__exit__(exc_type: Any, exc_val: Any, exc_tb: Any) → None[source]: Exit context manager.

close() → None[source]: Close the progress bar.

set_description(description: str) → None[source]

Update description text.

Parameters:: description – New description text.

set_postfix(**kwargs: Any) → None[source]

Set postfix values (shown after bar).

Parameters:: **kwargs – Key-value pairs to display.

start() → None[source]: Start the progress bar.

update(n: int = 1, best_score: float | None = None, best_label: str = 'best') → None[source]

Update progress by n items.

Parameters:

n – Number of items completed.
best_score – Current best score (for ML tracking).
best_label – Label for best score display.

classmethod wrap(iterable: Sequence[T], description: str = '', **kwargs: Any) → Iterator[T][source]

Wrap an iterable with a progress bar.

Parameters:

iterable – Iterable to wrap.
description – Progress bar description.
**kwargs – Additional ProgressBar arguments.

Yields:

Items from the iterable.

class nirs4all.core.logging.ProgressConfig(bar_width: int = 30, show_percentage: bool = True, show_count: bool = True, show_elapsed: bool = True, show_eta: bool = True, show_rate: bool = False, refresh_rate: float = 0.1, use_unicode: bool = True, use_colors: bool = True)[source]

Bases: object

Configuration for progress bar display.

bar_width

Width of the progress bar in characters.

Type:: int

show_percentage

Whether to show percentage.

Type:: bool

show_count

Whether to show current/total count.

Type:: bool

show_elapsed

Whether to show elapsed time.

Type:: bool

show_eta

Whether to show estimated time remaining.

Type:: bool

show_rate

Whether to show items per second.

Type:: bool

refresh_rate

Minimum seconds between updates.

Type:: float

use_unicode

Use Unicode box-drawing characters.

Type:: bool

use_colors

Use ANSI colors.

Type:: bool

bar_width: int = 30

refresh_rate: float = 0.1

show_count: bool = True

show_elapsed: bool = True

show_eta: bool = True

show_percentage: bool = True

show_rate: bool = False

use_colors: bool = True

use_unicode: bool = True

class nirs4all.core.logging.RotatingRunFileHandler(log_dir: Path, run_id: str, max_runs: int = 100, max_age_days: int | None = 30, max_bytes: int | None = None, compress_rotated: bool = True, json_output: bool = False)[source]

Bases: Handler

Handler that writes logs to run-specific files with rotation.

Creates a new log file for each run, with optional rotation to limit total log storage. Supports both count-based and age-based rotation policies.

Features:

Separate log files per run
Count-based rotation (max_runs)
Age-based rotation (max_age_days)
Size-based rotation (max_bytes)
Optional gzip compression for rotated logs
Optional JSON Lines output

close() → None[source]: Close file handles.

emit(record: LogRecord) → None[source]

Write log record to file(s).

Parameters:: record – Log record to write.

get_json_file_path() → Path | None[source]

Get the path to the current JSON log file.

Returns:: Path to JSON log file, or None if not enabled.

get_log_file_path() → Path[source]

Get the path to the current log file.

Returns:: Path to the current log file.

class nirs4all.core.logging.RunState(run_id: str, run_name: str | None = None, project: str | None = None, start_time: ~datetime.datetime = <factory>, current_phase: ~nirs4all.core.logging.events.Phase | None = None, branch_stack: list[~nirs4all.core.logging.context.BranchContext] = <factory>, source_context: ~nirs4all.core.logging.context.SourceContext | None = None, stack_context: ~nirs4all.core.logging.context.StackContext | None = None, extra: dict[str, ~typing.Any] = <factory>)[source]

Bases: object

State for a single run.

run_id

Unique run identifier.

Type:: str

run_name

Human-readable run name.

Type:: str | None

project

Project name (optional).

Type:: str | None

start_time

Run start timestamp.

Type:: datetime.datetime

current_phase

Current workflow phase.

Type:: nirs4all.core.logging.events.Phase | None

branch_stack

Stack of branch contexts (for nesting).

Type:: list[nirs4all.core.logging.context.BranchContext]

source_context

Current source context (if any).

Type:: nirs4all.core.logging.context.SourceContext | None

stack_context

Current stacking context (if any).

Type:: nirs4all.core.logging.context.StackContext | None

extra

Additional run-level metadata.

Type:: dict[str, Any]

property branch_depth: int: Get the current branch nesting depth.

property branch_path: list[str]: Get the full branch path.

branch_stack: list[BranchContext]

property current_branch: BranchContext | None: Get the current (innermost) branch context.

current_phase: Phase | None = None

extra: dict[str, Any]

project: str | None = None

run_id: str

run_name: str | None = None

source_context: SourceContext | None = None

stack_context: StackContext | None = None

start_time: datetime

class nirs4all.core.logging.SourceContext(name: str, index: int | None = None, total: int | None = None)[source]

Bases: object

Context for a source in multi-source pipelines.

name

Source name.

Type:: str

index

Source index (0-based).

Type:: int | None

total

Total number of sources.

Type:: int | None

index: int | None = None

name: str

total: int | None = None

class nirs4all.core.logging.SpinnerProgress(description: str = 'Processing', use_unicode: bool = True, use_colors: bool = True, disable: bool = False)[source]

Bases: object

Spinner for indeterminate progress indication.

Use when the total number of items is unknown.

Example

>>> with SpinnerProgress("Loading data") as spinner:
...     data = load_large_dataset()
...     spinner.update("Parsing...")
...     parsed = parse(data)

FRAMES_ASCII = ['|', '/', '-', '\\']

FRAMES_UNICODE = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏']

__enter__() → SpinnerProgress[source]: Enter context manager.

__exit__(exc_type: Any, exc_val: Any, exc_tb: Any) → None[source]: Exit context manager.

start() → None[source]: Start the spinner.

stop(success: bool = True) → None[source]

Stop the spinner.

Parameters:: success – Whether operation completed successfully.

update(description: str | None = None) → None[source]

Update spinner with new description.

Parameters:: description – New description text.

class nirs4all.core.logging.StackContext(n_branches: int, meta_model: str | None = None, branch_sources: list[str] = <factory>)[source]

Bases: object

Context for stacking operations.

n_branches

Number of branches being stacked.

Type:: int

meta_model

Meta-model name/description.

Type:: str | None

branch_sources

List of branch names being stacked.

Type:: list[str]

branch_sources: list[str]

meta_model: str | None = None

n_branches: int

class nirs4all.core.logging.Status(value)[source]

Bases: str, Enum

Status indicators for log messages.

ERROR = 'error'

IN_PROGRESS = 'in_progress'

SKIPPED = 'skipped'

STARTING = 'starting'

SUCCESS = 'success'

WARNING = 'warning'

class nirs4all.core.logging.Symbols(use_unicode: bool = True)[source]

Bases: object

Symbol system for log output with ASCII/Unicode modes.

Provides consistent symbols for status indicators, hierarchy markers, and other visual elements in log output.

property arrow: str: Flow/result arrow.

property branch: str: Branch indicator.

property branch_item: str: Branch item prefix.

property error: str: Fatal error.

get_status_symbol(status: Status | None) → str[source]

Get symbol for a status indicator.

Parameters:: status – Status to get symbol for.
Returns:: Appropriate symbol string.

property indent: str: Hierarchy level indent.

property progress: str: In progress.

property separator_heavy: str: Heavy separator line (80 chars).

property separator_light: str: Light separator line (76 chars).

property skipped: str: Skipped (cached).

property starting: str: Starting/beginning of a phase.

property sub_item: str: Sub-item prefix.

property success: str: Successful completion.

property warning: str: Non-fatal issue.

class nirs4all.core.logging.ThrottledHandler(base_handler: Handler, min_interval: float = 5.0)[source]

Bases: Handler

Handler that throttles progress messages to avoid flooding.

Uses time-based and percentage-based throttling to limit progress updates while still reporting important milestones.

MILESTONES = {10, 25, 50, 75, 90, 100}

close() → None[source]: Close this handler and the base handler.

emit(record: LogRecord) → None[source]

Emit a log record, throttling progress messages.

Parameters:: record – Log record to potentially emit.

reset() → None[source]: Reset throttle state for a new operation.

nirs4all.core.logging.configure_logging(verbose: int = 1, log_file: bool = False, log_dir: str | Path | None = None, log_format: str = 'pretty', use_unicode: bool | None = None, use_colors: bool | None = None, show_elapsed: bool = False, show_progress: bool = True, show_progress_bar: bool = True, json_output: bool = False, run_id: str | None = None, max_log_runs: int = 100, max_log_age_days: int | None = 30, max_log_bytes: int | None = None, compress_logs: bool = True) → None[source]

Configure the nirs4all logging system.

This function should be called once at application startup to set up logging. Subsequent calls will reconfigure the logging system.

Parameters:

verbose – Verbosity level (0=WARNING, 1=INFO, 2=DEBUG, 3=TRACE).
log_file – If True, write logs to workspace/logs/ directory.
log_dir – Directory for log files (used if log_file=True).
log_format – Output format: “pretty”, “minimal”, or “json”.
use_unicode – Use Unicode symbols (auto-detected if None).
use_colors – Use ANSI colors (auto-detected if None).
show_elapsed – Show elapsed time since run start.
show_progress – Show progress updates for long operations.
show_progress_bar – Show TTY-aware progress bars.
json_output – Also write JSON Lines log file.
run_id – Override run ID (auto-generated if not provided).
max_log_runs – Maximum number of run logs to keep (count-based rotation).
max_log_age_days – Maximum age of logs in days (None to disable).
max_log_bytes – Maximum log file size before rotation (None to disable).
compress_logs – Whether to gzip rotated log files.

nirs4all.core.logging.configure_progress(bar_width: int = 30, show_percentage: bool = True, show_count: bool = True, show_elapsed: bool = True, show_eta: bool = True, show_rate: bool = False, refresh_rate: float = 0.1, use_unicode: bool = True, use_colors: bool = True) → None[source]

Configure global progress bar settings.

Parameters:

bar_width – Width of progress bar in characters.
show_percentage – Show percentage completion.
show_count – Show current/total count.
show_elapsed – Show elapsed time.
show_eta – Show estimated time remaining.
show_rate – Show items per second.
refresh_rate – Minimum seconds between display updates.
use_unicode – Use Unicode characters for bar.
use_colors – Use ANSI colors.

nirs4all.core.logging.evaluation_progress(total_pipelines: int, metric_name: str = 'score', higher_is_better: bool = False, **kwargs: Any) → EvaluationProgress[source]

Create an evaluation progress tracker.

Parameters:

total_pipelines – Total pipelines to evaluate.
metric_name – Name of optimization metric.
higher_is_better – True if higher is better.
**kwargs – Additional arguments.

Returns:

EvaluationProgress instance.

nirs4all.core.logging.format_duration(seconds: float) → str[source]

Format a duration in seconds to human-readable string.

Parameters:: seconds – Duration in seconds.
Returns:: Human-readable duration string (e.g., “2m 5.9s”, “1h 23m 45s”).

nirs4all.core.logging.format_number(value: float | int, precision: int = 3) → str[source]

Format a number for display.

Uses thousands separators for large integers and appropriate precision for floats.

Parameters:

value – Number to format.
precision – Decimal precision for floats.

Returns:

Formatted number string.

nirs4all.core.logging.format_run_footer(status: Status, duration_seconds: float, best_pipeline: str | None = None, metrics: dict[str, float] | None = None, use_unicode: bool = True) → str[source]

Format the run footer block.

Parameters:

status – Final run status.
duration_seconds – Total run duration in seconds.
best_pipeline – Description of best pipeline (optional).
metrics – Final metrics dict (optional).
use_unicode – If True, use Unicode symbols.

Returns:

Formatted footer block string.

nirs4all.core.logging.format_run_header(run_name: str, start_time: datetime, environment_info: dict[str, str] | None = None, reproducibility_info: dict[str, str] | None = None, use_unicode: bool = True) → str[source]

Format the run header block.

Parameters:

run_name – Name of the run.
start_time – Run start timestamp.
environment_info – Optional environment details (Python version, etc.).
reproducibility_info – Optional reproducibility info (seed, git hash, etc.).
use_unicode – If True, use Unicode symbols.

Returns:

Formatted header block string.

nirs4all.core.logging.format_table(headers: list[str], rows: list[list[str]], use_unicode: bool = True) → str[source]

Format a simple ASCII table.

Parameters:

headers – Column headers.
rows – Table rows (list of lists).
use_unicode – Unused, kept for API consistency.

Returns:

Formatted table string.

nirs4all.core.logging.get_config() → LogConfig[source]

Get the current logging configuration.

Returns:: Current LogConfig instance.

nirs4all.core.logging.get_current_state() → RunState | None[source]

Get the current run state.

Returns:: Current RunState or None if not in a run context.

nirs4all.core.logging.get_logger(name: str) → Nirs4allLogger[source]

Get a logger for the specified module.

This is the primary interface for obtaining a logger in nirs4all modules. The logger is automatically configured with context injection and nirs4all-specific methods.

Parameters:: name – Logger name, typically __name__.
Returns:: Configured Nirs4allLogger instance.

Example

>>> from nirs4all.core.logging import get_logger
>>> logger = get_logger(__name__)
>>> logger.info("Processing data")

nirs4all.core.logging.get_run_id() → str | None[source]

Get the current run ID.

Returns:: Current run ID or None if not in a run context.

nirs4all.core.logging.get_symbols() → Symbols[source]: Get the global symbols instance.

nirs4all.core.logging.is_configured() → bool[source]

Check if logging has been configured.

Returns:: True if configure_logging() has been called.

nirs4all.core.logging.progress_bar(total: int, description: str = '', **kwargs: Any) → ProgressBar[source]

Create a progress bar.

Parameters:

total – Total number of items.
description – Description text.
**kwargs – Additional ProgressBar arguments.

Returns:

ProgressBar instance.

nirs4all.core.logging.reset_logging() → None[source]

Reset logging configuration to defaults.

This clears all handlers and resets the configuration. Useful for testing. Ensures file handlers are properly closed to avoid issues on Windows where open file handles prevent file deletion.

nirs4all.core.logging.spinner(description: str = 'Processing', **kwargs: Any) → SpinnerProgress[source]

Create a spinner for indeterminate progress.

Parameters:

description – Description text.
**kwargs – Additional SpinnerProgress arguments.

Returns:

SpinnerProgress instance.