nirs4all.core.logging.events module

Event types and structured logging for nirs4all.

This module defines event categories and structured event types used for consistent logging throughout the pipeline execution.

class nirs4all.core.logging.events.EventType(value)

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.events.LogEvent(timestamp: datetime, level: str, phase: Phase | None, event_type: EventType, message: str, run_id: str | None = None, status: 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>)

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]

Convert event to dictionary for JSON serialization.

to_json() → str

Convert event to JSON string.

class nirs4all.core.logging.events.Phase(value)

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.events.Status(value)

Bases: str, Enum

Status indicators for log messages.

ERROR = 'error'

IN_PROGRESS = 'in_progress'

SKIPPED = 'skipped'

STARTING = 'starting'

SUCCESS = 'success'

WARNING = 'warning'