Console

Provides the Console class that exposes methods 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 / or 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:

• line_width (int, default: 120) – The maximum length, in characters, for a single line of displayed text. This is used to limit the width of the text block as it is displayed in the terminal and written to log files.

• log_directory (Path | None, default: None) – The path to the directory where to save the log files. 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 the ‘log_directory’ is provided. Supported formats are LOG, TXT, and JSON.

• 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 the input line_width number is not valid.

• TypeError – If the input 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 log_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.

This method is primarily intended to be used internally as part of the echo() or error() method runtimes.

Parameters:

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

• loguru (bool, default: False) – Determines if the message is intended to be subsequently processed via loguru backend or another method or backend (e.g.: 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.

Return type:

Iterator[ProgressBar]

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. This is useful for code that needs to produce output even when the console is normally disabled.

Yields:

None.

Return type:

Iterator[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 valid logging levels for Console.echo() method calls.

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 provide a simplified update interface.

This class is yielded by the Console.progress() context manager and exposes only the update and close operations needed for manual progress tracking.

Parameters:

tqdm_bar – 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)

Determines if the directory portion of the input path exists and, if not, creates it.

Parameters:

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

Return type:

None

Standalone Methods

Provides standalone methods that abstract away common data manipulation 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[tuple[Any, ...], dtype[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’ value is below 1.

Return type:

Generator[tuple[Any, ...] | ndarray[tuple[Any, ...], dtype[Any]], None, None]

ataraxis_base_utilities.standalone_methods.convert_array_to_bytes(array)

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

The .copy() ensures the returned array is independent of the source array’s memory.

Parameters:

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

Return type:

ndarray[tuple[Any, ...], dtype[uint8]]

Returns:

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

Raises:

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

ataraxis_base_utilities.standalone_methods.convert_bytes_to_array(data, dtype)

Deserializes an uint8 byte array to a typed numpy array.

Parameters:

• data (ndarray[tuple[Any, ...], dtype[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[tuple[Any, ...], dtype[Any]]

Returns:

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

Raises:

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 an uint8 byte array to a Python scalar.

Parameters:

• data (ndarray[tuple[Any, ...], dtype[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 an 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 an 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, dtype_str) tuple. 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[tuple[Any, ...], dtype[uint8]]

Returns:

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

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 using re, so that it can be used to verify raised exceptions.

Notes

This method is primarily designed to help developers writing test functions for the Ataraxis codebase.

This method directly accesses the global console variable to retrieve the formatting parameters. Therefore, it always matches the configuration used by the Console class.

Parameters:

message (str) – The message to format.

Return type:

str

Returns:

The formatted 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.

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.

Auto-detects available cores, subtracts reserved_cores, and clamps the result to at least 1. A positive requested_workers further caps the result. Any non-positive value requests all available cores.

Parameters:

• requested_workers (int, default: 0) – The maximum number of workers to allocate. Non-positive values request all available cores.

• reserved_cores (int, default: 2) – The number of cores to reserve for system use. Must be >= 0.

Return type:

int

Returns:

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

Raises:

ValueError – If reserved_cores is negative.