Utilities

Structured logging configuration and filesystem path validation helpers.

Logging

Structured logging utilities for the heterodyne package.

Provides a lightweight but flexible logging system that matches the CMC reimplementation requirements: contextual log prefixes, configurable console and rotating file handlers, and helpers for performance monitoring.

class heterodyne.utils.logging.LogConfiguration

Bases: object

Programmatic logging configuration.

Alternative to configure_logging() for programmatic control over logging settings.

console_level

Console log level (default “INFO”).

console_format

Console format (“simple” or “detailed”).

console_colors

Enable ANSI colors in console (default False).

file_enabled

Enable file logging (default True).

file_path

Log file path (None = auto-generate).

file_level

File log level (default “DEBUG”).

file_format

File format (“simple” or “detailed”).

file_rotation_mb

Max file size before rotation (default 10).

file_backup_count

Number of backup files to keep (default 5).

module_overrides

Per-module log level overrides.

Example

>>> config = LogConfiguration.from_cli_args(verbose=True, log_file="analysis.log")
>>> config.apply()

>>> config = LogConfiguration(
...     console_level="INFO",
...     file_level="DEBUG",
...     module_overrides={"jax": "WARNING", "heterodyne.optimization": "DEBUG"}
... )
>>> config.apply()

console_level: str = 'INFO'

console_format: str = 'simple'

console_colors: bool = False

file_enabled: bool = True

file_path: str | Path | None = None

file_level: str = 'DEBUG'

file_format: str = 'detailed'

file_rotation_mb: int = 10

file_backup_count: int = 5

module_overrides: dict[str, str]

apply()

Apply this configuration to the logging system.

Return type:

Path | None

Returns:

Path to log file if file logging is enabled, None otherwise.

classmethod from_dict(config)

Create configuration from dictionary.

Parameters:

config (dict[str, Any]) – Dictionary with configuration values.

Return type:

LogConfiguration

Returns:

LogConfiguration instance.

classmethod from_cli_args(verbose=False, quiet=False, log_file=None)

Create configuration from CLI flags.

Parameters:

• verbose (bool) – Enable DEBUG level console logging.

• quiet (bool) – Enable ERROR-only console logging.

• log_file (str | None) – Path to log file (None = auto-generate if file logging enabled).

Return type:

LogConfiguration

Returns:

LogConfiguration instance.

__init__(console_level='INFO', console_format='simple', console_colors=False, file_enabled=True, file_path=None, file_level='DEBUG', file_format='detailed', file_rotation_mb=10, file_backup_count=5, module_overrides=<factory>)

class heterodyne.utils.logging.AnalysisSummaryLogger

Bases: object

Structured logging for analysis completion summaries.

Tracks phase timings, metrics, output files, and convergence status for logging a structured summary at analysis completion.

Example

>>> summary = AnalysisSummaryLogger(run_id="analysis_001", analysis_mode="two_component")
>>> summary.start_phase("loading")
>>> data = load_data(config)
>>> summary.end_phase("loading", memory_peak_gb=2.1)
>>> summary.record_metric("chi_squared", result.chi_squared)
>>> summary.set_convergence_status("converged")
>>> summary.log_summary(logger)

__init__(run_id, analysis_mode)

Initialize summary logger for an analysis run.

Parameters:

• run_id (str) – Unique identifier for this analysis run.

• analysis_mode (str) – Analysis mode (e.g., “two_component”, “single_component”).

start_phase(name)

Mark phase start for timing.

Parameters:

name (str) – Phase name (e.g., “loading”, “optimization”).

Return type:

None

end_phase(name, memory_peak_gb=None)

Mark phase completion.

Parameters:

• name (str) – Phase name that was started.

• memory_peak_gb (float | None) – Optional peak memory usage during phase.

Return type:

None

record_metric(name, value)

Record a named metric (e.g., chi_squared).

Parameters:

• name (str) – Metric name.

• value (float) – Metric value.

Return type:

None

add_output_file(path)

Record an output file path.

Parameters:

path (Path | str) – Path to output file.

Return type:

None

set_convergence_status(status)

Set final convergence status (failure is sticky).

Once a failure status (“failed”, “not_converged”, “max_iter”) has been recorded, subsequent calls with a non-failure status are ignored. Failure-to-failure transitions and success-to-failure transitions are always allowed.

Parameters:

status (str) – Convergence status (e.g., “converged”, “max_iter”, “failed”).

Return type:

None

property convergence_status: str | None

Final convergence status recorded for this run (read-only).

is_failure()

Return True when the recorded status indicates a failed run.

Return type:

bool

increment_warning_count()

Increment warning counter.

Return type:

None

increment_error_count()

Increment error counter.

Return type:

None

set_config_summary(optimizer=None, n_params=None, n_data_points=None, n_phi_angles=None, data_file=None, **kwargs)

Set configuration summary for logging.

Parameters:

• optimizer (str | None) – Optimizer used (e.g., “nlsq”, “cmc”).

• n_params (int | None) – Number of parameters being optimized.

• n_data_points (int | None) – Total number of data points.

• n_phi_angles (int | None) – Number of phi angles.

• data_file (str | None) – Path to data file.

• **kwargs (Any) – Additional key-value pairs to include.

Return type:

None

log_summary(logger)

Log the complete analysis summary.

Parameters:

logger (Logger | LoggerAdapter[Any]) – Logger to use for output.

Return type:

None

as_dict()

Export summary as dictionary for JSON serialization.

Return type:

dict[str, Any]

Returns:

Dictionary with all summary data.

class heterodyne.utils.logging.MinimalLogger

Bases: object

Configurable logger manager for the heterodyne package.

Thread-safe singleton for managing heterodyne logging configuration.

static __new__(cls)

Return type:

MinimalLogger

__init__()

configure(level='INFO', *, console_level=None, console_format='detailed', console_colors=False, file_path=None, file_level=None, max_size_mb=10, backup_count=5, module_levels=None, force=False)

Configure heterodyne logging.

Thread-safe configuration of the logging system.

Parameters:

• level (str | int) – Root logger level.

• console_level (str | int | None) – Console handler level. Pass False to suppress the console handler entirely.

• console_format (str) – “simple” or “detailed”.

• console_colors (bool) – Enable ANSI color output on the console.

• file_path (str | Path | None) – Path to log file. None disables file logging.

• file_level (str | int | None) – File handler level. Pass False to suppress the file handler even when file_path is set.

• max_size_mb (int) – Rotating log file maximum size in MB.

• backup_count (int) – Number of backup files to keep.

• module_levels (Mapping[str, str | int] | None) – Per-module level overrides.

• force (bool) – Remove existing managed handlers before reconfiguring.

Return type:

Path | None

Returns:

Path to the log file if a file handler was created, else None.

configure_from_dict(logging_config, *, verbose=False, quiet=False, output_dir=None, run_id=None)

Configure logging from a logging: config section.

Parameters:

• logging_config (Mapping[str, Any] | None) – Mapping with logging configuration (may be None or contain enabled: false to skip configuration).

• verbose (bool) – Override console level to DEBUG.

• quiet (bool) – Override console level to ERROR.

• output_dir (Path | str | None) – Base directory for auto-generated log file paths.

• run_id (str | None) – Run identifier used for auto-generated log filenames.

Return type:

Path | None

Returns:

Path to the log file if a file handler was created, else None.

get_logger(name)

Get or create a logger with hierarchical naming.

Return type:

Logger

heterodyne.utils.logging.get_logger(name=None, *, context=None)

Get a logger instance with automatic naming and optional context.

Delegates to logging.getLogger via MinimalLogger, which caches logger instances by name. Callers that previously used:

get_logger("heterodyne")

or:

get_logger(__name__)

continue to work unchanged. The optional context keyword adds structured key-value prefixes to every message emitted through the returned adapter.

Parameters:

• name (str | None) – Logger name, typically __name__. When None the caller’s __name__ is inferred automatically via frame inspection.

• context (Mapping[str, Any] | None) – Optional mapping of key-value pairs to include as a prefix on every log message (e.g. {"run_id": "abc", "q_bin": 5}).

Return type:

Logger | LoggerAdapter[Logger]

Returns:

A plain logging.Logger when context is None, or a _ContextAdapter that prepends [key=value ...] to every message when context is provided.

heterodyne.utils.logging.configure_logging(level='INFO', log_file=None, format_string=None)

Configure logging for the heterodyne package.

Simple interface for configuring console (and optionally file) logging. For richer control — rotating files, per-module overrides, CLI flag integration — use LogConfiguration instead.

Parameters:

• level (Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']) – Logging level.

• log_file (Path | str | None) – Optional path to log file.

• format_string (str | None) – Custom format string, or None for the default detailed format.

Return type:

None

heterodyne.utils.logging.with_context(logger, **context)

Create a contextual logger with key-value prefixes.

Context is formatted as [key=value key2=value2] message. Nested calls merge contexts (inner overrides outer on key conflicts). Thread-safe for use in multiprocessing.

Parameters:

• logger (Logger | LoggerAdapter[Logger]) – Base logger or existing contextual adapter to wrap.

• **context (Any) – Key-value pairs to include as prefix.

Return type:

LoggerAdapter[Logger]

Returns:

A logger adapter that prefixes all messages with context.

Example

>>> logger = get_logger(__name__)
>>> ctx_logger = with_context(logger, run_id="abc123", mode="two_component")
>>> ctx_logger.info("Starting analysis")
# Output: [run_id=abc123 mode=two_component] Starting analysis

>>> # Nested context
>>> shard_logger = with_context(ctx_logger, q_bin=5)
>>> shard_logger.info("Processing bin")
# Output: [run_id=abc123 mode=two_component q_bin=5] Processing bin

class heterodyne.utils.logging.PhaseContext

Bases: object

Context object returned by log_phase() with timing and memory info.

name: str

duration: float = 0.0

memory_peak_gb: float | None = None

memory_delta_gb: float | None = None

__init__(name, duration=0.0, memory_peak_gb=None, memory_delta_gb=None)

heterodyne.utils.logging.log_phase(name, logger=None, level=20, track_memory=False, threshold_s=0.0)

Context manager for phase-level timing with optional memory tracking.

Parameters:

• name (str) – Phase name for logging.

• logger (Logger | LoggerAdapter[Logger] | None) – Logger to use. If None, uses the caller’s module logger.

• level (int) – Log level for phase messages.

• track_memory (bool) – Track memory usage during phase.

• threshold_s (float) – Only log if duration > threshold (0 = always log).

Yields:

PhaseContext with name, duration, memory_peak_gb, and memory_delta_gb. Duration and memory values are populated after the context exits.

Example

>>> with log_phase("optimization", track_memory=True) as phase:
...     result = run_optimization(data)
>>> print(f"Took {phase.duration:.1f}s")
# Logs: Phase 'optimization' completed in 45.3s (peak memory: 12.4 GB)

heterodyne.utils.logging.log_exception(logger, exc, context=None, level=logging.ERROR, include_traceback=True)

Log an exception with full context for debugging.

Extracts module, function, and line number from exception traceback. Formats context as key-value pairs in the message.

Parameters:

• logger (Logger | LoggerAdapter[Logger]) – Logger to use.

• exc (BaseException) – Exception to log.

• context (dict[str, Any] | None) – Additional context (e.g., parameter values).

• level (int) – Log level (default ERROR).

• include_traceback (bool) – Include full traceback (default True).

Return type:

None

Example

>>> try:
...     result = compute_jacobian(params)
... except ValueError as e:
...     log_exception(logger, e, context={
...         "iteration": 45,
...         "params": params.tolist()[:5]
...     })
...     raise
# Logs:
# ERROR | heterodyne.optimization.nlsq.core | Exception in compute_jacobian:
# ValueError: invalid value
# Context: iteration=45, params=[1.2e-11, 0.85, ...]
# Traceback (most recent call last):
#   ...

heterodyne.utils.logging.log_calls(logger=None, level=logging.DEBUG, include_args=False, include_result=False)

Decorator to log function calls.

Parameters:

• logger (Logger | LoggerAdapter[Logger] | None) – Logger to use. If None, creates one for the decorated function’s module.

• level (int) – Logging level to use.

• include_args (bool) – Whether to log function arguments.

• include_result (bool) – Whether to log the function return value.

Return type:

Callable[[TypeVar(F, bound= Callable[..., Any])], TypeVar(F, bound= Callable[..., Any])]

Returns:

Decorator that wraps func with entry/exit logging.

heterodyne.utils.logging.log_performance(logger=None, level=logging.INFO, threshold=0.1)

Decorator to log function performance.

Only emits a log record when the function’s wall-clock time meets or exceeds threshold seconds.

Parameters:

• logger (Logger | LoggerAdapter[Logger] | None) – Logger to use. If None, creates one for the decorated function’s module.

• level (int) – Logging level to use.

• threshold (float) – Minimum duration in seconds required to emit a log record.

Return type:

Callable[[TypeVar(F, bound= Callable[..., Any])], TypeVar(F, bound= Callable[..., Any])]

Returns:

Decorator that wraps func with performance logging.

heterodyne.utils.logging.log_operation(operation_name, logger=None, level=20)

Context manager for logging named operations with timing.

Parameters:

• operation_name (str) – Human-readable name of the operation.

• logger (Logger | LoggerAdapter[Logger] | None) – Logger to use. If None, uses the caller’s module logger.

• level (int) – Logging level to use.

Yields:

The resolved logger so callers can emit additional messages inside the with block without holding a separate reference.

Example

>>> with log_operation("jacobian computation") as log:
...     J = compute_jacobian(params)
...     log.debug("Jacobian shape: %s", J.shape)
# Logs: Starting operation: jacobian computation
# Logs: Completed operation: jacobian computation in 0.042s

class heterodyne.utils.logging.ConvergenceLogger

Bases: object

Structured logger for optimization convergence diagnostics.

__init__(logger=None)

log_iteration(iteration, loss, gradient_norm=None, step_size=None)

Log optimization iteration metrics.

Return type:

None

log_convergence(reason, final_loss)

Log convergence result.

Return type:

None

log_diagnostic(metric_name, value, threshold, higher_is_better=True)

Log diagnostic metric with pass/fail status.

Parameters:

• metric_name (str) – Name of the diagnostic metric.

• value (float) – Observed value.

• threshold (float) – Threshold for pass/fail.

• higher_is_better (bool) – If True, values above threshold pass (e.g. ESS). If False, values below threshold pass (e.g. R-hat).

Return type:

None

Path Validation

Path validation and filesystem utilities.

exception heterodyne.utils.path_validation.PathValidationError

Bases: Exception

Raised when path validation fails.

heterodyne.utils.path_validation.resolve_path(path)

Resolve path to absolute, expanding user and symlinks.

Parameters:

path (str | Path) – Path string or Path object

Return type:

Path

Returns:

Resolved absolute Path

heterodyne.utils.path_validation.validate_file_exists(path, description='File')

Validate that a file exists and is readable.

Parameters:

• path (str | Path) – Path to validate

• description (str) – Description for error messages

Return type:

Path

Returns:

Resolved Path object

Raises:

PathValidationError – If file doesn’t exist or isn’t readable

heterodyne.utils.path_validation.validate_output_path(path, create_parents=True)

Validate and prepare output path.

Parameters:

• path (str | Path) – Output path to validate

• create_parents (bool) – Whether to create parent directories

Return type:

Path

Returns:

Resolved Path object

Raises:

PathValidationError – If path is invalid

heterodyne.utils.path_validation.ensure_directory(path)

Ensure directory exists, creating if necessary.

Parameters:

path (str | Path) – Directory path

Return type:

Path

Returns:

Resolved Path object