sage.core.logger

Filename : logger.py Description : Short description of the file

Created on 2025-11-07 18:53:42

__author__ = Narenraju Nagarajan __copyright__ = Copyright 2025, ProjectName __license__ = MIT Licence __version__ = 0.0.1 __maintainer__ = Narenraju Nagarajan __affiliation__ = N/A __email__ = N/A __status__ = [‘inProgress’, ‘Archived’, ‘inUsage’, ‘Debugging’]

GitHub Repository: NULL

Documentation: NULL

Classes

TensorRingBuffer	Fixed-capacity ring buffer that stores named tensor fields.
AsyncLogger	Non-blocking logger that offloads disk I/O to a background thread.
ChunkedTensorLogger	Accumulate tensors in memory and flush to disk in fixed-size chunks.
HDF5LossLogger	Persistent, epoch-indexed loss logger backed by an HDF5 file.

Functions

setup_logging([log_dir, level])	Configure global and per-module logging.
get_logger(module_name[, log_dir])	Get a logger for a specific module.

Module Contents

setup_logging(log_dir='logs', level=logging.INFO)

Configure global and per-module logging.

Parameters:

• log_dir (str) – Directory where log files are stored.

• level (int) – Minimum logging level.

get_logger(module_name, log_dir='logs')

Get a logger for a specific module. Each module has its own log file + logs also go to the main file.

Parameters:

• module_name (str) – Name of the module.

• log_dir (str) – Directory where log files are stored.

Returns:

Configured logger instance

Return type:

logging.Logger

class TensorRingBuffer(capacity, schema, device='cpu')

Fixed-capacity ring buffer that stores named tensor fields.

Pre-allocates a contiguous tensor for each named field and overwrites the oldest entries once the buffer is full. All data is kept on device (use "cpu" to avoid GPU memory pressure).

Parameters:

• capacity (int) – Maximum number of entries the buffer holds before wrapping.

• schema (dict[str, tuple]) –

  Mapping from field name to per-entry shape. Example:

  {
      "loss":   (1,),
      "params": (P,),
      "output": (C,),
      "target": (C,),
  }
  

• device (str) – Torch device string for the pre-allocated tensors.

Example

buffer = TensorRingBuffer(
    capacity=10000,
    schema={"loss": (1,), "params": (P,), "output": (C,), "target": (C,)},
    device="cpu",
)
# inside training loop
buffer.push(
    loss=loss.detach().cpu(),
    params=signal_targets.detach().cpu(),
    output=out.detach().cpu(),
    target=targets.detach().cpu(),
)

capacity

device = 'cpu'

ptr = 0

full = False

buffers

push(**kwargs)

Write one entry to the buffer, advancing the write pointer.

Parameters:

**kwargs (torch.Tensor) – One keyword per field defined in schema. Each tensor is detached before copying so no gradient is accidentally stored.

get()

Return all valid entries.

Returns:

If the buffer has not yet wrapped, returns only the filled prefix [0 : ptr]. Once full, returns all capacity entries (oldest-first order is not guaranteed after wrapping).

Return type:

dict[str, torch.Tensor]

class AsyncLogger(maxsize=1000, filepath='log.pt')

Non-blocking logger that offloads disk I/O to a background thread.

Incoming data dicts are placed on an in-memory queue; a daemon thread drains the queue in batches of 100 and serialises them to filepath with torch.save. Excess entries are silently dropped when the queue is full, so the training loop is never blocked.

Parameters:

• maxsize (int) – Maximum number of pending log entries before drops occur.

• filepath (str) – Path where batched entries are saved (overwritten each flush).

Example

logger = AsyncLogger()
logger.log({
    "loss":   loss.detach().cpu(),
    "params": signal_targets.detach().cpu(),
    "output": out.detach().cpu(),
    "target": targets.detach().cpu(),
})
logger.close()  # flush remaining entries and join the thread

q

filepath = 'log.pt'

running = True

thread

log(data)

Submit a data dict to the logging queue (non-blocking).

If the queue is full, the entry is silently discarded rather than blocking the caller.

Parameters:

data (dict) – Arbitrary dictionary of tensors or scalars to log.

close()

Flush remaining entries and join the background thread.

class ChunkedTensorLogger(chunk_size, path)

Accumulate tensors in memory and flush to disk in fixed-size chunks.

Each flush writes a Python list of tensors to {path}_{idx}.pt (via torch.save()) and increments the chunk index.

Parameters:

• chunk_size (int) – Number of items to accumulate before automatically flushing.

• path (str) – File-path prefix for the output files (suffix _<idx>.pt is appended automatically).

chunk_size

path

buffer = []

idx = 0

log(data)

Append data to the buffer and flush if the chunk is full.

Parameters:

data (any) – Tensor or other pickleable object to accumulate.

flush()

Write the current buffer to disk and reset state for the next chunk.

class HDF5LossLogger(path, num_epochs, num_components, dtype='float32')

Persistent, epoch-indexed loss logger backed by an HDF5 file.

Pre-allocates datasets of shape (num_epochs, num_components) for both the "training" and "validation" splits at construction time, then writes one row per epoch via log().

The resulting file can be read directly with h5py:

with h5py.File("losses.h5", "r") as f:
    train_loss = f["training"]["loss"][:]   # (E, C) float32
    val_loss   = f["validation"]["loss"][:]

Parameters:

• path (str) – File path for the HDF5 output (created fresh at init; any existing file at that path is overwritten).

• num_epochs (int) – Total number of training epochs (pre-allocates the dataset).

• num_components (int) – Number of scalar loss components logged per epoch (e.g. 4 for BCEWithPEsigmaLoss: total, BCE, reg, coupling).

• dtype (str) – NumPy dtype string for the stored values (default "float32").

path

num_epochs

num_components

dtype = 'float32'

log(loss_tensor, epoch, split)

Write one epoch’s loss vector to the HDF5 file.

Parameters:

• loss_tensor (torch.Tensor) – Shape (num_epochs, num_components). Only row epoch is written; the rest are ignored. This matches the loss_components tensor stored on training/validation objects.

• epoch (int) – Zero-based epoch index selecting the row to write.

• split (str) – Either "training" or "validation".