ahvn.utils.basic.progress_utils module¶

class ahvn.utils.basic.progress_utils.Progress(total=None, desc=None, initial=0, unit='it', leave=True)[source]¶

Bases: ABC

Abstract base class for progress reporting.

Provides a unified interface for progress tracking that can be extended for different backends (tqdm, custom frontends, etc.).

Subclasses must implement: emit, update_total, update, close. Optional helpers: set_description, set_postfix, write.

Parameters:

total (int | None)
desc (str | None)
initial (int)
unit (str)
leave (bool)

__init__(total=None, desc=None, initial=0, unit='it', leave=True)[source]¶

Initialize the progress bar.

Parameters:

total (Optional[int]) – Total number of iterations. None for unknown.
desc (Optional[str]) – Description prefix for the progress bar.
initial (int) – Initial counter value.
unit (str) – Unit of iteration.
leave (bool) – Whether to leave the progress bar after completion.

property total: int | None¶: Get total iterations.

property n: int¶: Get current iteration count.

property desc: str | None¶: Get description.

emit(payload)[source]¶

Emit a structured progress update.

The base implementation maps standardized keys: - total: calls update_total - update/advance: calls update - refresh: forwarded to subclasses when they override emit

Parameters:: payload (Optional[Dict[str, Any]]) – Progress payload to interpret.
Returns:: Subclass-specific return value.
Return type:: Optional[Any]

write(s, level=logging.INFO, **kwargs)[source]¶

Write a message through the progress bar’s output mechanism. Default to using the AgentHeaven logger using the progress bar’s class name.

Parameters:

s (str) – The string to write.
level (int) – Logging level to use. Defaults to logging.INFO.
**kwargs – Additional keyword arguments for logging.

Return type:

None

abstractmethod update_total(total)[source]¶

Update the total iterations for the progress bar.

Parameters:: total (Optional[int]) – New total iterations; None for unknown.
Return type:: None

abstractmethod update(n=1)[source]¶

Update the progress bar by n steps.

Parameters:: n (int) – Number of steps to advance.
Return type:: None

abstractmethod close()[source]¶

Close and cleanup the progress bar.

Return type:: None

reset(total=None)[source]¶

Reset the progress bar to initial state.

Parameters:: total (Optional[int]) – New total, or keep existing if None.
Return type:: None

__enter__()[source]¶

Enter context manager.

Return type:: Progress

__exit__(exc_type, exc_val, exc_tb)[source]¶

Exit context manager and close progress bar.

Return type:: None

class ahvn.utils.basic.progress_utils.NoProgress(total=None, desc=None, initial=0, unit='it', leave=True)[source]¶

Bases: Progress

A silent progress implementation that does nothing.

Used as the default when no progress bar is requested.

Parameters:

total (int | None)
desc (str | None)
initial (int)
unit (str)
leave (bool)

update(n=1)[source]¶

Update the progress bar by n steps.

Parameters:: n (int) – Number of steps to advance.
Return type:: None

emit(payload)[source]¶

Emit a structured progress update.

The base implementation maps standardized keys: - total: calls update_total - update/advance: calls update - refresh: forwarded to subclasses when they override emit

Parameters:: payload (Optional[Dict[str, Any]]) – Progress payload to interpret.
Returns:: Subclass-specific return value.
Return type:: Optional[Any]

update_total(total)[source]¶

Update the total iterations for the progress bar.

Parameters:: total (Optional[int]) – New total iterations; None for unknown.
Return type:: None

set_description(desc=None, refresh=True)[source]¶

Return type:

None

Parameters:

desc (str | None)
refresh (bool)

set_postfix(ordered_dict=None, refresh=True, **kwargs)[source]¶

Return type:

None

Parameters:

ordered_dict (dict | None)
refresh (bool)

close()[source]¶

Close and cleanup the progress bar.

Return type:: None

class ahvn.utils.basic.progress_utils.LogProgress(total=None, desc=None, initial=0, unit='it', leave=True, logger=None, level=logging.INFO, interval=10)[source]¶

Bases: Progress

Progress implementation using logging.

Outputs progress as log messages in format: [INFO] <desc>: <prefix> [xx%] <suffix> Logs at configurable intervals to avoid spam.

Parameters:

total (int | None)
desc (str | None)
initial (int)
unit (str)
leave (bool)
logger (Logger | None)
level (int)
interval (int)

__init__(total=None, desc=None, initial=0, unit='it', leave=True, logger=None, level=logging.INFO, interval=10)[source]¶

Initialize logger-based progress.

Parameters:

total (Optional[int]) – Total number of iterations.
desc (Optional[str]) – Description prefix.
initial (int) – Initial counter value.
unit (str) – Unit of iteration.
leave (bool) – Whether to log completion message.
logger (Optional[logging.Logger]) – Logger to use. Defaults to ahvn logger.
level (int) – Log level. Defaults to INFO.
interval (int) – Log every N percent change. Defaults to 10.

emit(payload)[source]¶

Emit a structured progress update.

The base implementation maps standardized keys: - total: calls update_total - update/advance: calls update - refresh: forwarded to subclasses when they override emit

Parameters:: payload (Optional[Dict[str, Any]]) – Progress payload to interpret.
Returns:: Subclass-specific return value.
Return type:: Optional[Any]

update_total(total)[source]¶

Update the total iterations for the progress bar.

Parameters:: total (Optional[int]) – New total iterations; None for unknown.
Return type:: None

update(n=1)[source]¶

Update the progress bar by n steps.

Parameters:: n (int) – Number of steps to advance.
Return type:: None

set_description(desc=None, refresh=True)[source]¶

Return type:

None

Parameters:

desc (str | None)
refresh (bool)

set_postfix(ordered_dict=None, refresh=True, **kwargs)[source]¶

Return type:

None

Parameters:

ordered_dict (dict | None)
refresh (bool)

set_prefix(prefix)[source]¶

Set prefix text (between description and percentage).

Return type:: None
Parameters:: prefix (str)

write(s, level=None, **kwargs)[source]¶

Write a message through the progress bar’s logging mechanism.

Parameters:

s (str) – The string to write.
level (Optional[int]) – Logging level to use. Defaults to the progress bar’s level.
**kwargs – Additional keyword arguments for logging.

Return type:

None

close()[source]¶

Close and cleanup the progress bar.

Return type:: None

class ahvn.utils.basic.progress_utils.TqdmProgress(total=None, desc=None, initial=0, unit='it', leave=True, **tqdm_kwargs)[source]¶

Bases: Progress

Progress implementation using tqdm.

Wraps tqdm to provide a consistent interface with other Progress implementations.

Parameters:

total (int | None)
desc (str | None)
initial (int)
unit (str)
leave (bool)

__init__(total=None, desc=None, initial=0, unit='it', leave=True, **tqdm_kwargs)[source]¶

Initialize tqdm-based progress bar.

Parameters:

total (Optional[int]) – Total number of iterations.
desc (Optional[str]) – Description prefix.
initial (int) – Initial counter value.
unit (str) – Unit of iteration.
leave (bool) – Whether to leave progress bar after completion.
**tqdm_kwargs – Additional arguments passed to tqdm.

update_total(total)[source]¶

Update the total iterations for the progress bar.

Parameters:: total (Optional[int]) – New total iterations; None for unknown.
Return type:: None

update(n=1)[source]¶

Update the progress bar by n steps.

Parameters:: n (int) – Number of steps to advance.
Return type:: None

emit(payload)[source]¶

Emit a structured progress update.

The base implementation maps standardized keys: - total: calls update_total - update/advance: calls update - refresh: forwarded to subclasses when they override emit

Parameters:: payload (Optional[Dict[str, Any]]) – Progress payload to interpret.
Returns:: Subclass-specific return value.
Return type:: Optional[Any]

set_description(desc=None, refresh=True)[source]¶

Return type:

None

Parameters:

desc (str | None)
refresh (bool)

set_postfix(ordered_dict=None, refresh=True, **kwargs)[source]¶

Return type:

None

Parameters:

ordered_dict (dict | None)
refresh (bool)

write(s, file=None, end='\\n', **kwargs)[source]¶

Write a message through the tqdm progress bar’s output mechanism.

Parameters:

s (str) – The string to write.
file (Any) – File-like object to write to. Defaults to sys.stderr.
end (str) – End character. Defaults to newline.
**kwargs – Additional keyword arguments for logging.

Return type:

None

close()[source]¶

Close and cleanup the progress bar.

Return type:: None

reset(total=None)[source]¶

Reset the progress bar to initial state.

Parameters:: total (Optional[int]) – New total, or keep existing if None.
Return type:: None

property pbar: Any¶: Access the underlying tqdm progress bar for advanced operations.

ahvn.utils.basic.progress_utils.get_progress()[source]¶

Get the currently active progress bar for this thread.

Returns:: The active progress bar, or a NoProgress instance if none is active.
Return type:: Progress

ahvn.utils.basic.progress_utils.progress(p)[source]¶

Context manager to set the active progress bar.

Supports nesting - each context pushes onto a stack and pops on exit.

Parameters:: p (Progress) – The progress bar to activate.
Yields:: Progress – The activated progress bar.

Example

with progress(TqdmProgress(total=100)) as pbar:

for i in range(100):: pbar.update(1)