Console

Provides the Console class and its supporting assets for writing messages and errors to terminal and log files.

class ataraxis_base_utilities.console.Console(log_directory=None, log_format=LogFormats.LOG, line_width=120, *, break_long_words=False, break_on_hyphens=False, debug=False, enqueue=False, show_progress=False)

Bases: object

Provides methods for printing and logging messages and errors.

This class wraps and extends the functionality of the ‘Loguru’ library to provide a centralized message handling interface.

Notes

After initialization, call the enable() method before calling other class methods.

Do not configure or enable the Console class from libraries that may be imported by other projects! To work as expected, the Console has to be enabled at the highest level of the call hierarchy.

This class conflicts with all libraries that make explicit calls to the loguru backend, as it reconfigures loguru handles to support its runtime.

Parameters:

  • log_directory (Path | None, default: None) – The path to the directory where the log files are saved. Setting this argument to None disables the logging functionality.

  • log_format (str | LogFormats, default: <LogFormats.LOG: '.log'>) – The format to use for log files. This is only used when log_directory is provided. Supported formats are LOG, TXT, and JSON.

  • line_width (int, default: 120) – The maximum length, in characters, for a single line of text printed to the terminal. Must exceed the 37 characters the loguru header reserves on the first line of each message. Lines written to log files carry loguru’s wider default header, so they exceed this limit.

  • break_long_words (bool, default: False) – Determines whether to break long words when formatting the text block to fit the width requirement.

  • break_on_hyphens (bool, default: False) – Determines whether to break sentences on hyphens when formatting the text block to fit the width requirement.

  • debug (bool, default: False) – Determines whether to print and log debug messages.

  • enqueue (bool, default: False) – Determines whether to pass logged messages through an asynchronous queue. Primarily, this is helpful when logging the messages from multiple producers running in parallel.

  • show_progress (bool, default: False) – Determines whether progress bars from track() and progress() are displayed. When False, progress bars are suppressed even if the console is enabled. This allows echo() output to remain active while hiding progress bar display.

_line_width

Stores the maximum allowed text block line width, in characters.

_break_long_words

Determines whether to break text on long words.

_break_on_hyphens

Determines whether to break text on hyphens.

_debug_log_path

Stores the path to the debug log file.

_message_log_path

Stores the path to the message log file.

_error_log_path

Stores the path to the error log file.

_is_enabled

Tracks whether logging through this class instance is enabled. When this tracker is False, the echo() method will have limited functionality.

_show_progress

Tracks whether progress bars are displayed. When False, track() and progress() suppress their tqdm bars regardless of the console’s enabled state.

Raises:

  • ValueError – If line_width is not valid, or if log_format is not a valid LogFormats member.

  • TypeError – If log_directory is not a valid Path object.

property debug_log_path: Path | None

Returns the path to the log file used to save messages at or below DEBUG level or None if the path is not set.

disable()

Disables processing messages and errors.

Return type:

None

Notes

When the console is disabled, the error() method raises exceptions, but does not log them to files or provide detailed traceback information.

disable_progress()

Disables progress bar display for track() and progress() calls.

Return type:

None

Notes

When progress is disabled, track() and progress() still yield items and accept updates normally, but no visual progress bar is rendered. Console echo() and error() methods are unaffected.

echo(message, level=LogLevel.INFO, *, raw=False)

Formats the input message according to the class configuration and outputs it to the terminal, file, or both.

When raw mode is enabled, the message bypasses format_message() and loguru’s format string, outputting the text without a timestamp header or level prefix. This is useful for pre-formatted content such as tables or DataFrames.

Parameters:

  • message (str) – The message to be processed.

  • level (str | LogLevel, default: <LogLevel.INFO: 'info'>) – The severity level of the message.

  • raw (bool, default: False) – Determines whether to bypass message formatting and loguru headers. When True, the message is output as-is without text wrapping or timestamp prefixes.

Raises:

ValueError – If the requested level is not one of the valid LogLevel members.

Return type:

None

enable()

Enables processing messages and errors.

Return type:

None

enable_progress()

Enables progress bar display for track() and progress() calls.

Return type:

None

property enabled: bool

Returns True if the instance is configured to process messages and errors.

error(message, error=<class 'RuntimeError'>)

Raises the requested error with integrated logging.

If the Console class is disabled, the method raises an exception without logging. Otherwise, if file-logging is enabled, the method logs the error to the file before raising an exception.

Parameters:

  • message (str) – The error message.

  • error (Callable[..., Exception], default: <class 'RuntimeError'>) – The exception class to raise.

Return type:

NoReturn

property error_log_path: Path | None

Returns the path to the log file used to save messages at or above ERROR level or None if the path is not set.

format_message(message, *, loguru=False)

Formats the input message string according to the instance configuration parameters.

Parameters:

  • message (str) – The text string to format.

  • loguru (bool, default: False) – Determines whether the message is subsequently processed by the loguru backend. When False, the message is formatted for another consumer, such as an Exception class.

Return type:

str

Returns:

The formatted message string.

property message_log_path: Path | None

Returns the path to the log file used to save messages at INFO through WARNING levels or None if the path is not set.

progress(total, description='', unit='iteration')

Provides a context manager that yields a manually-updatable progress bar.

The progress bar is displayed only when both the console is enabled and progress display is enabled. The bar is automatically closed when the context manager exits.

Parameters:

  • total (float) – The total number of expected units for the progress bar.

  • description (str, default: '') – The label displayed next to the progress bar.

  • unit (str, default: 'iteration') – The string used to label each iteration unit.

Yields:

A ProgressBar instance that exposes update() and close() methods.

property progress_enabled: bool

Returns True if progress bar display is enabled.

temporarily_enabled()

Provides a context manager that temporarily enables the console.

Saves the current enabled state, enables the console for the duration of the context, and restores the original state on exit.

Yields:

None.

track(iterable, description='', total=None, unit='iteration')

Wraps an iterable with a tqdm progress bar tied to the console’s enabled and progress states.

The progress bar is displayed only when both the console is enabled and progress display is enabled. Items are always yielded regardless of the display state.

Parameters:

  • iterable (Iterable[TypeVar(T)]) – The iterable to wrap with a progress bar.

  • description (str, default: '') – The label displayed next to the progress bar.

  • total (float | None, default: None) – The expected total number of iterations. If None, tqdm attempts to infer it from the iterable.

  • unit (str, default: 'iteration') – The string used to label each iteration unit.

Return type:

Iterable[TypeVar(T)]

Returns:

An iterable that yields items from the input iterable while displaying a progress bar.

class ataraxis_base_utilities.console.LogFormats(*values)

Bases: StrEnum

Defines the log file formats supported by the Console class.

JSON = '.json'

JSON structured file format.

LOG = '.log'

Standard log file format.

TXT = '.txt'

Plain text file format.

class ataraxis_base_utilities.console.LogLevel(*values)

Bases: StrEnum

Defines the logging levels supported by the Console class.

CRITICAL = 'critical'

Logs messages indicating critical failures.

DEBUG = 'debug'

Logs diagnostic information for development and troubleshooting.

ERROR = 'error'

Logs messages indicating errors that need attention.

INFO = 'info'

Logs standard informational messages.

SUCCESS = 'success'

Logs messages indicating successful operations.

WARNING = 'warning'

Logs messages indicating potential issues.

class ataraxis_base_utilities.console.ProgressBar(tqdm_bar)

Bases: object

Wraps a tqdm progress bar to expose the update and close operations used for manual progress tracking.

Parameters:

tqdm_bar (tqdm[NoReturn]) – The tqdm progress bar instance to wrap.

_tqdm_bar

Stores the wrapped tqdm progress bar instance.

close()

Closes the progress bar and releases its resources.

Return type:

None

update(n=1)

Advances the progress bar by the specified amount.

Parameters:

n (float, default: 1) – The number of units to advance the progress bar.

Return type:

None

ataraxis_base_utilities.console.ensure_directory_exists(path, *, is_file=None)

Creates the directory portion of the input path if it does not already exist.

Parameters:

  • path (Path) – The path to be processed. Can be a file or a directory path.

  • is_file (bool | None, default: None) – Determines whether path points to a file, in which case the directory created is its parent. When None, the suffix of the final path component decides, so a directory whose own name carries a dot needs this argument set to False to be created rather than skipped.

Return type:

None

Standalone Methods

Provides standalone methods that abstract away common data manipulation and core-budget resolution tasks.

ataraxis_base_utilities.standalone_methods.chunk_iterable(iterable, chunk_size)

Yields successive chunks from the input ordered Python iterable or NumPy array.

Notes

For NumPy arrays, the function maintains the original data type and dimensionality, returning NumPy array chunks. For other iterables, it always returns chunks as tuples.

The last yielded chunk contains any leftover elements if the iterable’s length is not evenly divisible by chunk_size. This last chunk may be smaller than all other chunks.

Parameters:

  • iterable (NDArray[Any] | tuple[Any, ...] | list[Any]) – The Python iterable or NumPy array to split into chunks.

  • chunk_size (int) – The maximum number of elements in each chunk.

Yields:

Chunks of the input iterable (as a tuple) or NumPy array, containing at most chunk_size elements.

Raises:

  • TypeError – If iterable is not of a correct type.

  • ValueError – If chunk_size is below 1.

ataraxis_base_utilities.standalone_methods.convert_array_to_bytes(array)

Serializes a 1D numpy array of any supported dtype to a uint8 byte array.

The returned array owns its memory, so mutating it leaves the source array unchanged.

Parameters:

array (NDArray[Any]) – A 1D, non-empty numpy array to serialize.

Return type:

NDArray[uint8]

Returns:

A 1D numpy array of dtype uint8 containing the serialized bytes.

Raises:

  • TypeError – If array is not a numpy array.

  • ValueError – If array is not 1D or is empty.

ataraxis_base_utilities.standalone_methods.convert_bytes_to_array(data, dtype)

Deserializes a uint8 byte array to a typed numpy array.

Parameters:

  • data (NDArray[uint8]) – A 1D numpy array of dtype uint8 containing the serialized bytes.

  • dtype (dtype[Any]) – The numpy dtype specifying the target element type and byte order.

Return type:

NDArray[Any]

Returns:

A 1D numpy array of the specified dtype containing the deserialized values.

Raises:

  • TypeError – If data is not a uint8 numpy array.

  • ValueError – If data is not 1D or its byte count is not evenly divisible by the target dtype’s itemsize.

ataraxis_base_utilities.standalone_methods.convert_bytes_to_scalar(data, dtype=dtype('int64'))

Deserializes a uint8 byte array to a Python scalar.

Parameters:

  • data (NDArray[uint8]) – A 1D numpy array of dtype uint8 containing the serialized bytes.

  • dtype (dtype[Any], default: dtype('int64')) – The numpy dtype specifying the target type and byte order.

Return type:

int | float | bool

Returns:

The deserialized Python int, float, or bool value.

Raises:

  • TypeError – If data is not a uint8 numpy array.

  • ValueError – If data is not 1D or its byte count does not match the target dtype’s itemsize.

ataraxis_base_utilities.standalone_methods.convert_scalar_to_bytes(value, dtype=dtype('int64'))

Serializes a scalar value to a uint8 byte array.

The returned array length depends on the dtype: 1 byte for uint8/int8/bool, 2 for int16/uint16, 4 for int32/uint32/float32, 8 for int64/uint64/float64, etc.

Notes

Uses an internal LRU cache keyed on the (value, repr(value), dtype_str) tuple. The repr component separates entries that compare equal while serializing to different bytes, which covers 0.0 against -0.0 and every int, float, and bool that shares a numeric value. The cached raw bytes are converted to a new numpy array on each call, avoiding mutation issues. This benefits tight loops where the same value+dtype pair is serialized repeatedly.

Parameters:

  • value (int | float | bool | generic) – The scalar value to serialize.

  • dtype (dtype[Any], default: dtype('int64')) – The numpy dtype specifying the target type and byte order.

Return type:

NDArray[uint8]

Returns:

A 1D numpy array of dtype uint8 containing the serialized bytes.

Raises:

ValueError – If value carries a fractional part that the target integer dtype is unable to represent, or if value falls outside the range the target dtype is able to represent.

ataraxis_base_utilities.standalone_methods.ensure_list(input_item)

Ensures that the input object is returned as a list.

If the object is not already a list, attempts to convert it into a list. If the object is a list, returns the object unchanged.

Parameters:

input_item (Any) – The object to be converted into or preserved as a Python list.

Return type:

list[Any]

Returns:

The object converted to a Python list datatype.

Raises:

TypeError – If the input object cannot be converted to a list.

ataraxis_base_utilities.standalone_methods.error_format(message)

Formats the input message to match the default Console format and escapes it for regular expression matching.

Notes

The formatting parameters are read from the global console variable, so the output always matches the configuration used by the Console class.

Parameters:

message (str) – The message to format.

Return type:

str

Returns:

The formatted and escaped message.

ataraxis_base_utilities.standalone_methods.resolve_parallel_job_capacity(workers_per_job)

Determines how many jobs can run in parallel given the per-job core allocation.

Divides the available core count by workers_per_job, returning at least 1. If the core count cannot be auto-detected, returns 1.

Parameters:

workers_per_job (int) – The number of CPU cores each job requires. Must be >= 1.

Return type:

int

Returns:

The number of parallel jobs that can run concurrently, always >= 1.

Raises:

ValueError – If workers_per_job is less than 1.

ataraxis_base_utilities.standalone_methods.resolve_worker_count(requested_workers=0, reserved_cores=2)

Determines the number of CPU cores to allocate for a processing job.

A positive requested_workers is honored exactly, capped only by the logical core count, so an explicit request can claim every core on the machine. A non-positive requested_workers auto-resolves to every available core minus reserved_cores, clamped to at least 1, leaving headroom for the host system. If the core count cannot be auto-detected, the budget falls back to 1.

Parameters:

  • requested_workers (int, default: 0) – The number of workers to allocate. A positive value is honored up to the logical core count. Non-positive values auto-resolve to all available cores minus the reserved cores.

  • reserved_cores (int, default: 2) – The number of cores to reserve for host-system use during auto-resolution. Ignored for a positive requested_workers. Must be >= 0.

Return type:

int

Returns:

The number of CPU cores to use, always >= 1.

Raises:

ValueError – If reserved_cores is negative.