Logging (rheojax.logging)

Comprehensive logging system for monitoring and debugging RheoJAX operations. Provides structured logging, JAX-safe utilities, and performance metrics.

Configuration

configure_logging

rheojax.logging.configure_logging(level='INFO', format='standard', file=None, colorize=True, **kwargs)

Configure the RheoJAX logging system.

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

Parameters:

• level (str) – Global log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)

• format (str) – Output format (standard, detailed, json, scientific)

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

• colorize (bool) – Enable colored console output

• **kwargs – Additional LogConfig parameters

Return type:

LogConfig

Returns:

The configured LogConfig instance.

Example

>>> from rheojax.logging import configure_logging
>>> configure_logging(level="DEBUG", file="rheojax.log")

Configure the RheoJAX logging system.

Quick Start:

from rheojax.logging import configure_logging

# Basic configuration
configure_logging(level="INFO")

# Verbose debugging
configure_logging(level="DEBUG")

# With file output
configure_logging(level="INFO", log_file="rheojax.log")

get_logger

rheojax.logging.get_logger(name, **context)

Get a RheoJAX logger for the given name.

Creates a new logger or returns a cached instance. The logger is automatically configured based on the current logging configuration.

Note

The cache is process-global: all modules share it, and clear_logger_cache() affects every caller.

Parameters:

• name (str) – Logger name (typically __name__).

• **context – Default context to bind to the logger. Values must be representable as strings for cache keying.

Return type:

RheoJAXLogger

Returns:

RheoJAXLogger instance.

Example

>>> from rheojax.logging import get_logger
>>> logger = get_logger(__name__)
>>> logger.info("Model fitted", R2=0.9987)

Get a logger instance for the specified name.

from rheojax.logging import get_logger

logger = get_logger(__name__)
logger.info("Starting model fitting", model="Maxwell")
logger.debug("Iteration 100", cost=1e-5)

LogConfig

class rheojax.logging.LogConfig(level='INFO', format=LogFormat.STANDARD, console=True, file=None, file_max_bytes=10000000, file_backup_count=5, subsystem_levels=<factory>, lazy_formatting=True, include_timestamps=True, include_thread=False, colorize=True)

Bases: object

RheoJAX logging configuration.

level

Global log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)

format

Output format (standard, detailed, json, scientific)

console

Enable console output

file

Path to log file (None disables file logging)

file_max_bytes

Maximum log file size before rotation (default 10MB)

file_backup_count

Number of backup files to keep (default 5)

subsystem_levels

Per-subsystem log level overrides

lazy_formatting

Enable lazy evaluation of log arguments

include_timestamps

Include timestamps in log output

include_thread

Include thread name in log output

colorize

Enable colored console output

level: str = 'INFO'

format: LogFormat | str = 'standard'

console: bool = True

file: Path | str | None = None

file_max_bytes: int = 10000000

file_backup_count: int = 5

subsystem_levels: dict[str, str]

lazy_formatting: bool = True

include_timestamps: bool = True

include_thread: bool = False

colorize: bool = True

__post_init__()

Validate configuration after initialization.

Return type:

None

classmethod from_env()

Create configuration from environment variables.

Reads the following environment variables:

• RHEOJAX_LOG_LEVEL: Global log level

• RHEOJAX_LOG_FILE: Path to log file

• RHEOJAX_LOG_FORMAT: Output format

• RHEOJAX_LOG_COLORIZE: Enable colors (true/false)

• RHEOJAX_LOG_<SUBSYSTEM>: Per-subsystem levels

Return type:

LogConfig

Returns:

LogConfig instance with environment-based settings.

get_level(logger_name)

Get the effective log level for a logger.

Parameters:

logger_name (str) – Full logger name (e.g., “rheojax.models.maxwell”)

Return type:

int

Returns:

Logging level as integer.

__init__(level='INFO', format=LogFormat.STANDARD, console=True, file=None, file_max_bytes=10000000, file_backup_count=5, subsystem_levels=<factory>, lazy_formatting=True, include_timestamps=True, include_thread=False, colorize=True)

LogFormat

class rheojax.logging.LogFormat(*values)

Bases: Enum

Available log output formats.

STANDARD = 'standard'

DETAILED = 'detailed'

JSON = 'json'

SCIENTIFIC = 'scientific'

Environment Variables

The logging system respects these environment variables:

Environment Variables

Variable	Description	Default
RHEOJAX_LOG_LEVEL	Global log level (DEBUG, INFO, WARNING, ERROR)	INFO
RHEOJAX_LOG_FILE	Path to log file (enables file logging)	None
RHEOJAX_LOG_FORMAT	Output format (standard, detailed, json)	standard

Context Managers

Operation Logging

Context managers for automatic timing and context tracking:

log_fit

rheojax.logging.log_fit(logger, model, data_shape=None, test_mode='unknown', level=20, **kwargs)

Context manager for model fitting operations.

Specialized wrapper around log_operation for model fitting.

Parameters:

• logger (Logger | RheoJAXLogger) – Logger instance to use.

• model (str) – Model name or class name.

• data_shape (tuple[int, ...] | None) – Shape of input data (optional).

• test_mode (str) – Test mode (relaxation, creep, oscillation, flow).

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

• **kwargs – Additional context.

Yields:

Dictionary for adding completion context (e.g., R2, parameters).

Example

>>> with log_fit(logger, "Maxwell", data_shape=(100,), test_mode="relaxation") as ctx:
...     result = model._fit(x, y)
...     ctx["R2"] = result.r_squared
...     ctx["n_iterations"] = result.iterations

Log model fitting operations with timing.

from rheojax.logging import log_fit, get_logger

logger = get_logger(__name__)

with log_fit(logger, model="Maxwell", data_shape=(100,)) as ctx:
    result = model.fit(x, y)
    ctx["R2"] = result.r_squared  # Add to completion log

log_bayesian

rheojax.logging.log_bayesian(logger, model, num_warmup, num_samples, num_chains=1, level=20, **kwargs)

Context manager for Bayesian inference operations.

Specialized wrapper for MCMC sampling operations.

Parameters:

• logger (Logger | RheoJAXLogger) – Logger instance to use.

• model (str) – Model name.

• num_warmup (int) – Number of warmup samples.

• num_samples (int) – Number of posterior samples.

• num_chains (int) – Number of MCMC chains.

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

• **kwargs – Additional context.

Yields:

Dictionary for adding completion context (e.g., R-hat, ESS).

Example

>>> with log_bayesian(logger, "Maxwell", num_warmup=1000, num_samples=2000) as ctx:
...     result = model.fit_bayesian(x, y)
...     ctx["r_hat_max"] = compute_rhat(result)
...     ctx["ess_min"] = compute_ess(result)
...     ctx["divergences"] = result.divergences

Log Bayesian inference operations.

from rheojax.logging import log_bayesian, get_logger

logger = get_logger(__name__)

with log_bayesian(logger, "Maxwell", num_warmup=1000, num_samples=2000) as ctx:
    result = model.fit_bayesian(x, y)
    ctx["r_hat"] = compute_rhat(result)
    ctx["divergences"] = result.divergences

log_transform

rheojax.logging.log_transform(logger, transform, input_shape=None, level=20, **kwargs)

Context manager for transform operations.

Specialized wrapper for data transformation operations.

Parameters:

• logger (Logger | RheoJAXLogger) – Logger instance to use.

• transform (str) – Transform name.

• input_shape (tuple[int, ...] | None) – Shape of input data.

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

• **kwargs – Additional context.

Yields:

Dictionary for adding completion context (e.g., output_shape).

Example

>>> with log_transform(logger, "mastercurve", input_shape=(10, 100)) as ctx:
...     result = transform.transform(datasets)
...     ctx["output_shape"] = result.shape
...     ctx["shift_factors"] = len(shift_factors)

Log data transformation operations.

log_io

rheojax.logging.log_io(logger, operation, filepath=None, level=20, **kwargs)

Context manager for I/O operations.

Specialized wrapper for file read/write operations.

Parameters:

• logger (Logger | RheoJAXLogger) – Logger instance to use.

• operation (str) – I/O operation type (read, write, load, save).

• filepath (str | None) – Path to file being accessed.

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

• **kwargs – Additional context.

Yields:

Dictionary for adding completion context (e.g., records, file_size).

Example

>>> with log_io(logger, "read", filepath="data.csv") as ctx:
...     data = read_csv(filepath)
...     ctx["records"] = len(data)
...     ctx["columns"] = list(data.columns)

Log I/O operations (file reading/writing).

log_pipeline_stage

rheojax.logging.log_pipeline_stage(logger, stage, pipeline_id=None, level=20, **kwargs)

Context manager for pipeline stage execution.

Parameters:

• logger (Logger | RheoJAXLogger) – Logger instance to use.

• stage (str) – Pipeline stage name.

• pipeline_id (str | None) – Optional pipeline identifier.

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

• **kwargs – Additional context.

Yields:

Dictionary for adding completion context.

Example

>>> with log_pipeline_stage(logger, "fit", pipeline_id="pipe_001") as ctx:
...     result = pipeline.fit()
...     ctx["model"] = result.model_name

Log pipeline stage execution.

log_operation

rheojax.logging.log_operation(logger, operation, level=20, **context)

Context manager for logging operation start/end with timing.

Automatically logs when an operation starts and completes, including elapsed time and any exceptions that occur.

Parameters:

• logger (Logger | RheoJAXLogger) – Logger instance to use.

• operation (str) – Name of the operation being performed.

• level (int) – Log level for start/end messages (default INFO).

• **context – Additional context to include in log messages.

Yields:

Dictionary that can be used to add additional context to the completion log message.

Example

>>> with log_operation(logger, "fitting", model="Maxwell"):
...     result = model.fit(x, y)
14:32:05 | INFO | rheojax.models | fitting started | model=Maxwell
14:32:07 | INFO | rheojax.models | fitting completed | model=Maxwell | elapsed_seconds=2.15

Example with additional context:

>>> with log_operation(logger, "fitting", model="Maxwell") as ctx:
...     result = model.fit(x, y)
...     ctx["R2"] = result.r_squared
14:32:05 | INFO | rheojax.models | fitting started | model=Maxwell
14:32:07 | INFO | rheojax.models | fitting completed | model=Maxwell | R2=0.9987 | elapsed_seconds=2.15

Generic operation logging context manager.

JAX-Safe Utilities

Utilities for logging JAX arrays without triggering device transfers:

log_array_info

rheojax.logging.log_array_info(arr, name='array', include_device=True)

Extract loggable info from JAX/NumPy array without device transfer.

This function extracts metadata from arrays (shape, dtype, device) without transferring array data from GPU to CPU, making it safe to use in performance-critical code.

Parameters:

• arr (Any) – JAX or NumPy array.

• name (str) – Name for the array in log output.

• include_device (bool) – Include device information for JAX arrays.

Return type:

dict[str, Any]

Returns:

Dictionary with array metadata.

Example

>>> import jax.numpy as jnp
>>> x = jnp.ones((100, 50))
>>> info = log_array_info(x, "input_data")
>>> logger.debug("Processing data", **info)
DEBUG | rheojax | Processing data | name=input_data | shape=(100, 50) | dtype=float32

Get array metadata without device transfer (shape, dtype only).

from rheojax.logging import log_array_info

# Safe at INFO level - no device transfer
info = log_array_info(jax_array, "residuals")
print(info)  # {"name": "residuals", "shape": (100,), "dtype": "float64"}

log_array_stats

rheojax.logging.log_array_stats(arr, name='array', logger=None, level=10)

Compute and log full array statistics.

WARNING: This function forces a device-to-host transfer for JAX arrays. Use only for debugging at DEBUG level.

Parameters:

• arr (Any) – JAX or NumPy array.

• name (str) – Name for the array.

• logger (Logger | None) – Logger to use (optional, for immediate logging).

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

Return type:

dict[str, Any]

Returns:

Dictionary with array metadata and statistics.

Example

>>> stats = log_array_stats(residuals, "residuals", logger)
DEBUG | rheojax | Array statistics | name=residuals | min=0.001 | max=0.234 | mean=0.045

Get array statistics (triggers device transfer - use at DEBUG level).

log_numerical_issue

rheojax.logging.log_numerical_issue(logger, arr, name='array', context='')

Check for and log numerical issues (NaN, Inf) in arrays.

Parameters:

• logger (Logger) – Logger instance.

• arr (Any) – Array to check.

• name (str) – Name for the array in log output.

• context (str) – Additional context about where the issue occurred.

Return type:

bool

Returns:

True if numerical issues were found, False otherwise.

Example

>>> if log_numerical_issue(logger, residuals, "residuals", "during fitting"):
...     raise ValueError("Numerical instability detected")

Check for and log NaN/Inf values.

from rheojax.logging import log_numerical_issue, get_logger

logger = get_logger(__name__)

if log_numerical_issue(logger, residuals, "residuals", "during fitting"):
    raise ValueError("Numerical instability detected")

log_jax_config

rheojax.logging.log_jax_config(logger=None)

Log JAX configuration state.

Logs JAX version, available devices, default backend, and float64 configuration.

Parameters:

logger (Logger | None) – Logger to use. If provided, logs immediately.

Return type:

dict[str, Any]

Returns:

Dictionary with JAX configuration.

Example

>>> log_jax_config(logger)
INFO | rheojax | JAX Configuration | jax_version=0.8.0 | devices=['gpu:0'] | float64_enabled=True

Log JAX configuration (version, devices, float64 status).

from rheojax.logging import log_jax_config, get_logger

logger = get_logger(__name__)
log_jax_config(logger)  # Logs JAX version, devices, precision

jax_safe_log

rheojax.logging.jax_safe_log(logger, level, msg, **kwargs)

Log only if not inside JAX JIT tracing.

This function checks if we’re currently being traced by JAX JIT and skips logging if so, preventing tracing issues.

Parameters:

• logger (Logger) – Logger instance.

• level (int) – Log level.

• msg (str) – Log message.

• **kwargs – Extra context to log.

Return type:

None

Example

>>> @jax.jit
... def my_function(x):
...     jax_safe_log(logger, logging.DEBUG, "Inside JIT", value=x.shape)
...     return x * 2

Safely log a value that may be a JAX array.

jax_debug_log

rheojax.logging.jax_debug_log(logger, msg, *values, level=10)

Use jax.debug.callback for logging inside JIT-compiled functions.

This allows logging from within JIT-compiled code using JAX’s debug callback mechanism.

Parameters:

• logger (Logger) – Logger instance.

• msg (str) – Log message (can include {} placeholders for values).

• *values (Any) – Values to log (will be passed through debug.callback).

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

Return type:

None

Example

>>> @jax.jit
... def my_function(x):
...     y = x * 2
...     jax_debug_log(logger, "Computed y with shape {}", y.shape)
...     return y

Log JAX values only at DEBUG level (with device transfer).

Performance Tracking

timed

rheojax.logging.timed(logger=None, level=10, include_args=False)

Decorator to log function execution time.

Parameters:

• logger (Logger | RheoJAXLogger | None) – Logger to use. If None, uses function’s module logger.

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

• include_args (bool) – Include function arguments in log output.

Return type:

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

Returns:

Decorator function.

Example

>>> @timed()
... def compute_something(x, y):
...     return x + y

>>> @timed(level=logging.INFO, include_args=True)
... def fit_model(data):
...     return model.fit(data)

Decorator for timing function execution.

from rheojax.logging import timed
import logging

@timed(level=logging.INFO)
def expensive_operation():
    # ... computation ...
    pass

log_memory

rheojax.logging.log_memory(logger=None, operation='operation', level=10, trace_lines=False)

Context manager for tracking memory usage.

Uses tracemalloc to measure memory allocation during an operation.

Parameters:

• logger (Logger | RheoJAXLogger | None) – Logger to use.

• operation (str) – Name of operation being measured.

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

• trace_lines (bool) – Include top memory-allocating lines.

Yields:

None

Example

>>> with log_memory(logger, "large_computation"):
...     result = compute_large_array()
DEBUG | rheojax.core | large_computation memory | current_mb=45.2 | peak_mb=128.5

Log current memory usage.

IterationLogger

class rheojax.logging.IterationLogger(logger=None, log_every=10, level=10, operation='optimization')

Bases: object

Logger for optimization iterations with rate limiting.

Logs iteration progress at configurable intervals to avoid flooding logs during long-running optimizations.

logger

Logger instance.

log_every

Log every N iterations.

level

Log level.

iteration

Current iteration count.

start_time

Time when logging started.

Example

>>> iter_logger = IterationLogger(logger, log_every=100)
>>> for i in range(1000):
...     cost = optimizer.step()
...     iter_logger.log(cost=cost, grad_norm=grad_norm)
DEBUG | rheojax.opt | Iteration 100 | cost=0.0234 | grad_norm=0.001
DEBUG | rheojax.opt | Iteration 200 | cost=0.0189 | grad_norm=0.0008

Log optimization iterations at configurable frequency.

from rheojax.logging import IterationLogger, get_logger

logger = get_logger(__name__)
iter_logger = IterationLogger(logger, log_every=100)

for i in range(1000):
    cost = optimizer.step()
    iter_logger.log(cost=cost)

iter_logger.log_final()

__init__(logger=None, log_every=10, level=10, operation='optimization')

Initialize the iteration logger.

Parameters:

• logger (Logger | RheoJAXLogger | None) – Logger instance (creates default if None).

• log_every (int) – Log every N iterations (default 10).

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

• operation (str) – Operation name for log messages.

log(cost=None, force=False, **metrics)

Log iteration if at logging interval.

Parameters:

• cost (float | None) – Current cost/loss value.

• force (bool) – Force logging regardless of interval.

• **metrics – Additional metrics to log.

Return type:

None

log_final(**metrics)

Log final iteration summary.

Parameters:

**metrics – Final metrics to include.

Return type:

None

reset()

Reset iteration counter and timer.

Return type:

None

ConvergenceTracker

class rheojax.logging.ConvergenceTracker(logger=None, tolerance=1e-06, patience=5, min_iterations=10)

Bases: object

Track and log convergence metrics for optimization.

Monitors cost progression and determines when convergence criteria are met.

Example

>>> tracker = ConvergenceTracker(logger, tolerance=1e-6)
>>> for i in range(1000):
...     cost = optimizer.step()
...     if tracker.update(cost):
...         print("Converged!")
...         break

Track convergence metrics over iterations.

__init__(logger=None, tolerance=1e-06, patience=5, min_iterations=10)

Initialize the convergence tracker.

Parameters:

• logger (Logger | RheoJAXLogger | None) – Logger instance.

• tolerance (float) – Convergence tolerance for cost improvement.

• patience (int) – Number of iterations with small improvement before converged.

• min_iterations (int) – Minimum iterations before convergence check.

update(cost)

Update with new cost and check for convergence.

Parameters:

cost (float) – Current cost/loss value.

Return type:

bool

Returns:

True if convergence criteria met.

reset()

Reset the tracker.

Return type:

None

property improvement_rate: float | None

Calculate average improvement rate.

Returns:

Average cost reduction per iteration, or None if insufficient data.

Formatters

StandardFormatter

class rheojax.logging.StandardFormatter(colorize=True)

Bases: Formatter

Human-readable format for console output.

Format: HH:MM:SS | LEVEL | logger.name | message

Supports optional colorization for terminal output.

Standard log format: LEVEL - message [key=value ...]

FORMAT = '%(asctime)s | %(levelname)-8s | %(name)s | %(message)s'

DATE_FORMAT = '%H:%M:%S'

__init__(colorize=True)

Initialize the formatter.

Parameters:

colorize (bool) – Enable ANSI color codes in output.

format(record)

Format the log record.

Parameters:

record (LogRecord) – LogRecord instance to format.

Return type:

str

Returns:

Formatted log string.

DetailedFormatter

class rheojax.logging.DetailedFormatter(colorize=False)

Bases: Formatter

Detailed format with file/line info for debugging.

Format: YYYY-MM-DD HH:MM:SS.ffffff | LEVEL | logger:line | func | message

Detailed format with timestamp, logger name, file location.

FORMAT = '%(asctime)s | %(levelname)-8s | %(name)s:%(lineno)d | %(funcName)s | %(message)s'

DATE_FORMAT = '%Y-%m-%d %H:%M:%S'

__init__(colorize=False)

Initialize the formatter.

Parameters:

colorize (bool) – Enable ANSI color codes (disabled by default for files).

formatTime(record, datefmt=None)

Format timestamp with true microsecond precision.

Parameters:

• record (LogRecord) – LogRecord instance.

• datefmt (str | None) – Date format string (unused, uses DATE_FORMAT).

Return type:

str

Returns:

Timestamp string with microseconds.

format(record)

Format the log record with microseconds.

Parameters:

record (LogRecord) – LogRecord instance to format.

Return type:

str

Returns:

Formatted log string.

JSONFormatter

class rheojax.logging.JSONFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)

Bases: Formatter

JSON format for machine parsing and log aggregation.

Output: {“timestamp”: “…”, “level”: “…”, “logger”: “…”, …}

JSON output for machine parsing and log aggregation.

format(record)

Format the log record as JSON.

Parameters:

record (LogRecord) – LogRecord instance to format.

Return type:

str

Returns:

JSON-formatted log string.

ScientificFormatter

class rheojax.logging.ScientificFormatter(colorize=False)

Bases: DetailedFormatter

Format optimized for scientific computing output.

Provides consistent scientific notation for numerical values and special handling for array shapes and dtypes.

Scientific notation for numerical values.

Handlers

RheoJAXStreamHandler

class rheojax.logging.RheoJAXStreamHandler(stream=None, immediate_flush=True)

Bases: StreamHandler

Enhanced stream handler with flush control.

Provides immediate flushing for interactive use and buffered output for batch processing.

Console output handler with optional color support.

__init__(stream=None, immediate_flush=True)

Initialize the handler.

Parameters:

• stream – Output stream (default: sys.stderr)

• immediate_flush (bool) – Flush after each log message

emit(record)

Emit a log record.

Parameters:

record (LogRecord) – LogRecord to emit.

Return type:

None

RheoJAXRotatingFileHandler

class rheojax.logging.RheoJAXRotatingFileHandler(filename, max_bytes=10000000, backup_count=5, encoding='utf-8')

Bases: RotatingFileHandler

Enhanced rotating file handler with UTF-8 encoding.

Automatically handles log rotation and maintains backup files.

Rotating file handler for log rotation.

__init__(filename, max_bytes=10000000, backup_count=5, encoding='utf-8')

Initialize the rotating file handler.

Parameters:

• filename (Path | str) – Path to log file.

• max_bytes (int) – Maximum file size before rotation (default 10MB).

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

• encoding (str) – File encoding (default UTF-8).

RheoJAXMemoryHandler

class rheojax.logging.RheoJAXMemoryHandler(capacity=1000, flush_level=40, target=None)

Bases: MemoryHandler

Memory handler for buffered logging.

Useful for batch operations where you want to collect logs and flush them periodically or at the end of an operation.

In-memory buffer for log capture and testing.

__init__(capacity=1000, flush_level=40, target=None)

Initialize the memory handler.

Parameters:

• capacity (int) – Number of log records to buffer.

• flush_level (int) – Level that triggers immediate flush.

• target (Handler | None) – Target handler to flush to.

shouldFlush(record)

Check if buffer should be flushed.

Extends stdlib behavior: when no target is set, caps the buffer at capacity by dropping the oldest records to prevent unbounded memory growth. With a target, delegates to stdlib MemoryHandler.

Parameters:

record (LogRecord) – Current log record.

Return type:

bool

Returns:

True if buffer should be flushed.

Exporters (OpenTelemetry)

LogExporter

class rheojax.logging.LogExporter

Bases: ABC

Abstract base class for log exporters.

Exporters transform and send log entries to external systems like OpenTelemetry collectors, Datadog, or custom backends.

Base class for log exporters.

abstractmethod export(entries)

Export log entries to the backend.

Parameters:

entries (list[LogEntry]) – List of LogEntry objects to export.

Return type:

bool

Returns:

True if export succeeded, False otherwise.

abstractmethod shutdown()

Shutdown the exporter and flush any pending entries.

Return type:

None

OpenTelemetryLogExporter

class rheojax.logging.OpenTelemetryLogExporter(endpoint='http://localhost:4317', service_name='rheojax', service_version=None, insecure=True, headers=None)

Bases: LogExporter

OpenTelemetry-compatible log exporter.

Exports logs in OTLP (OpenTelemetry Protocol) format to an OTLP collector endpoint. Falls back to console output if opentelemetry-api is not installed.

Example

>>> from rheojax.logging.exporters import OpenTelemetryLogExporter
>>> exporter = OpenTelemetryLogExporter(
...     endpoint="http://localhost:4317",
...     service_name="rheojax-app"
... )
>>> handler = ExportingHandler(exporter)
>>> logger.addHandler(handler)

Export logs to OpenTelemetry-compatible backends.

__init__(endpoint='http://localhost:4317', service_name='rheojax', service_version=None, insecure=True, headers=None)

Initialize the OpenTelemetry exporter.

Parameters:

• endpoint (str) – OTLP collector endpoint URL.

• service_name (str) – Service name for resource attributes.

• service_version (str | None) – Service version (auto-detected if None).

• insecure (bool) – Use insecure connection (no TLS).

• headers (dict[str, str] | None) – Additional headers for OTLP requests.

export(entries)

Export log entries via OpenTelemetry.

Parameters:

entries (list[LogEntry]) – List of LogEntry objects.

Return type:

bool

Returns:

True if export succeeded.

shutdown()

Shutdown the OpenTelemetry logger provider.

Return type:

None

ConsoleExporter

class rheojax.logging.ConsoleExporter(format='json', output='stderr', include_resource=False)

Bases: LogExporter

Console exporter for structured logs.

Outputs log entries in structured format (JSON or key-value) to stdout/stderr for debugging or piping to other tools.

Export logs to console (for debugging).

__init__(format='json', output='stderr', include_resource=False)

Initialize the console exporter.

Parameters:

• format (str) – Output format (“json” or “keyvalue”).

• output (str) – Output destination (“stdout” or “stderr”).

• include_resource (bool) – Include resource attributes in output.

export(entries)

Export log entries to console.

Parameters:

entries (list[LogEntry]) – List of LogEntry objects.

Return type:

bool

Returns:

Always True.

shutdown()

Flush the output stream.

Return type:

None

BatchingExporter

class rheojax.logging.BatchingExporter(inner_exporter, batch_size=100, flush_interval=5.0, max_queue_size=1000)

Bases: LogExporter

Batching wrapper for log exporters.

Collects log entries and exports them in batches for efficiency. Supports configurable batch size and flush intervals.

Batch logs before exporting for efficiency.

__init__(inner_exporter, batch_size=100, flush_interval=5.0, max_queue_size=1000)

Initialize the batching exporter.

Parameters:

• inner_exporter (LogExporter) – Underlying exporter to use.

• batch_size (int) – Maximum entries per batch.

• flush_interval (float) – Seconds between automatic flushes.

• max_queue_size (int) – Maximum queue size before blocking.

export(entries)

Add entries to the batch queue.

Parameters:

entries (list[LogEntry]) – List of LogEntry objects.

Return type:

bool

Returns:

True if entries were queued.

shutdown()

Shutdown the batching exporter.

Return type:

None

Examples

Basic Setup

from rheojax.logging import configure_logging, get_logger

# Configure once at startup
configure_logging(level="INFO")

# Get logger in each module
logger = get_logger(__name__)

# Log with structured data
logger.info("Model fitted", model="Maxwell", R2=0.9987, time=1.23)
logger.debug("Parameter values", G0=1e5, eta=1000)

Production Configuration

from rheojax.logging import configure_logging

configure_logging(
    level="INFO",
    log_file="/var/log/rheojax/app.log",
    format="json",  # Machine-readable
    max_bytes=10_000_000,  # 10 MB rotation
    backup_count=5
)

Debugging Workflow

import os

# Enable debug logging via environment
os.environ["RHEOJAX_LOG_LEVEL"] = "DEBUG"

from rheojax.logging import configure_logging, get_logger, log_fit

configure_logging()  # Uses environment variable
logger = get_logger(__name__)

# All operations now logged at debug level
with log_fit(logger, "FractionalMaxwell", data_shape=(500,)) as ctx:
    result = model.fit(x, y)
    ctx["iterations"] = result.nit
    ctx["final_cost"] = result.fun

Integration with Model Fitting

from rheojax.logging import (
    configure_logging,
    get_logger,
    log_fit,
    log_bayesian,
    log_numerical_issue
)
from rheojax.models import Maxwell

configure_logging(level="INFO")
logger = get_logger(__name__)

model = Maxwell()

# NLSQ fitting with logging
with log_fit(logger, "Maxwell", data_shape=x.shape) as ctx:
    model.fit(x, y)
    ctx["R2"] = model.score(x, y)

# Bayesian inference with logging
with log_bayesian(logger, "Maxwell", num_samples=2000) as ctx:
    result = model.fit_bayesian(x, y, num_samples=2000)

    # Check for numerical issues
    if log_numerical_issue(logger, result.posterior_samples["G0"], "G0", "posterior"):
        logger.warning("Posterior samples contain numerical issues")

    ctx["r_hat"] = max(result.diagnostics["r_hat"].values())
    ctx["divergences"] = result.diagnostics.get("divergences", 0)

See Also

• Troubleshooting Guide - Debugging with logs

• Core Module (rheojax.core) - Core module with BaseModel and BayesianMixin

• Pipeline API - Pipeline API with built-in logging