Skip to main content

logging_utils

Utilities for logging and warning messages functionality.

Module​

Functions​

deprecated_class_name​

def deprecated_class_name(cls: Type[_T]) ‑> Callable[..., ~_T]:

Class decorator to log a DeprecationWarning for deprecated class names.

Used for cases where we have:

class

#### filter_stderr
<!-- prettier-ignore -->

```py showLineNumbers
def filter_stderr(
    to_filter: Union[str, 'codeBlockAnchor[re.Pattern](https://docs.python.org/3.12/library/re.html#re.Pattern)'],
) ‑> 'codeBlockAnchor[collections.abc.Generator](https://docs.python.org/3.12/library/collections.abc.html#collections.abc.Generator)'[None, None, None]:

Filter stderr messages emitted within this context manager.

Will remove any messages where the start matches the filter pattern whilst allowing any other messages to go through.

Arguments

  • to_filter: Regex pattern to match the start of messages to be filtered.

log_pytorch_env_info_if_available​

def log_pytorch_env_info_if_available() ‑> None:

Log PyTorch environment info if PyTorch is available.

setup_loggers​

def setup_loggers(    loggers: list[logging.Logger],    log_file_dir_name: Optional[str] = None,    log_level: Union[int, str] = 'INFO',    clear_existing_handlers: bool = True,    clear_existing_filters: bool = True,) ‑> list[logging.Logger]:

Set up supplied loggers with stdout and file handlers.

Creates a logfile in 'logs' directory with the current date and time and outputs all logs at the "DEBUG" level. Also outputs logs to stdout at the "INFO" level. A common scenario is to attach handlers only to the root logger, and to let propagation take care of the rest.

Arguments

  • loggers: The logger(s) to set up
  • log_file_dir_name: Creates a subdirectory inside config.settings.paths.logs_dir if provided. Defaults to None.
  • log_level: The log level to apply to the console logs
  • clear_existing_handlers: If True, clear existing handlers for each logger
  • clear_existing_filters: If True, clear existing filters for each logger

Returns A list of updated logger(s).

Classes​

DeduplicationFilter​

class DeduplicationFilter(    handler: Optional[logging.Handler] = None, *args: Any, **kwargs: Any,):

A logging filter that suppresses consecutive duplicate log messages.

When the same log message is emitted repeatedly from the same source location, this filter suppresses the duplicates and emits a summary message indicating how many times the message was repeated once a different message arrives.

Supports two levels of deduplication:

  • Exact: identical (logger name, source line, level, message) — summary says "repeated N times".
  • Normalized: same (logger name, source line, level) with messages that differ only in numeric values (e.g. "Column 123 missing" vs "Column 124 missing") — summary says "repeated N times (with varying values)".

Only applies to records under the bitfount logger namespace. Records from other namespaces pass through untouched.

Thread-safe: uses a lock to protect shared state across threads. Registered in a global WeakSet that is flushed via a single atexit hook.

Initialize a filter.

Initialize with the name of the logger which, together with its children, will have its events allowed through the filter. If no name is specified, allow every event.

Ancestors​

Methods​

filter​

def filter(self, record: logging.LogRecord) ‑> bool:

Filter consecutive duplicate log records.

Returns True if the record should be logged, False otherwise. When a new (different) record arrives after duplicates, a summary record is injected before the new record is emitted.

Records whose name does not start with "bitfount" pass through untouched — this prevents the filter from deduplicating orchestrator or third-party logs when a handler is shared across logger namespaces.

Records with truthy exc_info always pass through (never deduplicated) to avoid suppressing distinct tracebacks.

PytorchLightningWarningFilter​

class PytorchLightningWarningFilter(name=''):

Filter that suppresses PyTorch Lightning warnings.

warnings generates a log message of form "<file_path>:<line_no> ..." so we filter on lightning mentions in the file path.

Initialize a filter.

Initialize with the name of the logger which, together with its children, will have its events allowed through the filter. If no name is specified, allow every event.

Ancestors​

Methods​

filter​

def filter(self, record: logging.LogRecord) ‑> bool:

Returns True if the record should be logged, False otherwise.

SampleFilter​

class SampleFilter(interval: int = 1000, *args: Any, **kwargs: Any):

A logging filter that tracks counts for logs and only logs at intervals.

Useful for very frequent logs where we only care about logging "samples" of the logs, at some specific interval, for example, whilst iterating through a list of files.

Initialize a filter.

Initialize with the name of the logger which, together with its children, will have its events allowed through the filter. If no name is specified, allow every event.

Ancestors​

Methods​

filter​

def filter(self, record: logging.LogRecord) ‑> bool:

Returns True if the record should be logged, False otherwise.