ahvn.utils.basic.progress_utils module¶

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

基类：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.

参数:

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

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

Initialize the progress bar.

参数:

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)[源代码]¶

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

参数:: payload (Optional[Dict[str, Any]]) -- Progress payload to interpret.
返回:: Subclass-specific return value.
返回类型:: Optional[Any]

write(s, level=logging.INFO, **kwargs)[源代码]¶

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

参数:

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

返回类型:

None

abstractmethod update_total(total)[源代码]¶

Update the total iterations for the progress bar.

参数:: total (Optional[int]) -- New total iterations; None for unknown.
返回类型:: None

abstractmethod update(n=1)[源代码]¶

Update the progress bar by n steps.

参数:: n (int) -- Number of steps to advance.
返回类型:: None

abstractmethod close()[源代码]¶

Close and cleanup the progress bar.

返回类型:: None

reset(total=None)[源代码]¶

Reset the progress bar to initial state.

参数:: total (Optional[int]) -- New total, or keep existing if None.
返回类型:: None

__enter__()[源代码]¶

Enter context manager.

返回类型:: Progress

__exit__(exc_type, exc_val, exc_tb)[源代码]¶

Exit context manager and close progress bar.

返回类型:: None

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

基类：Progress

A silent progress implementation that does nothing.

Used as the default when no progress bar is requested.

参数:

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

update(n=1)[源代码]¶

Update the progress bar by n steps.

参数:: n (int) -- Number of steps to advance.
返回类型:: None

emit(payload)[源代码]¶

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

参数:: payload (Optional[Dict[str, Any]]) -- Progress payload to interpret.
返回:: Subclass-specific return value.
返回类型:: Optional[Any]

update_total(total)[源代码]¶

Update the total iterations for the progress bar.

参数:: total (Optional[int]) -- New total iterations; None for unknown.
返回类型:: None

set_description(desc=None, refresh=True)[源代码]¶

返回类型:

None

参数:

desc (str | None)
refresh (bool)

set_postfix(ordered_dict=None, refresh=True, **kwargs)[源代码]¶

返回类型:

None

参数:

ordered_dict (dict | None)
refresh (bool)

close()[源代码]¶

Close and cleanup the progress bar.

返回类型:: 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)[源代码]¶

基类：Progress

Progress implementation using logging.

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

参数:

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)[源代码]¶

Initialize logger-based progress.

参数:

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)[源代码]¶

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

参数:: payload (Optional[Dict[str, Any]]) -- Progress payload to interpret.
返回:: Subclass-specific return value.
返回类型:: Optional[Any]

update_total(total)[源代码]¶

Update the total iterations for the progress bar.

参数:: total (Optional[int]) -- New total iterations; None for unknown.
返回类型:: None

update(n=1)[源代码]¶

Update the progress bar by n steps.

参数:: n (int) -- Number of steps to advance.
返回类型:: None

set_description(desc=None, refresh=True)[源代码]¶

返回类型:

None

参数:

desc (str | None)
refresh (bool)

set_postfix(ordered_dict=None, refresh=True, **kwargs)[源代码]¶

返回类型:

None

参数:

ordered_dict (dict | None)
refresh (bool)

set_prefix(prefix)[源代码]¶

Set prefix text (between description and percentage).

返回类型:: None
参数:: prefix (str)

write(s, level=None, **kwargs)[源代码]¶

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

参数:

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.

返回类型:

None

close()[源代码]¶

Close and cleanup the progress bar.

返回类型:: None

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

基类：Progress

Progress implementation using tqdm.

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

参数:

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)[源代码]¶

Initialize tqdm-based progress bar.

参数:

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)[源代码]¶

Update the total iterations for the progress bar.

参数:: total (Optional[int]) -- New total iterations; None for unknown.
返回类型:: None

update(n=1)[源代码]¶

Update the progress bar by n steps.

参数:: n (int) -- Number of steps to advance.
返回类型:: None

emit(payload)[源代码]¶

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

参数:: payload (Optional[Dict[str, Any]]) -- Progress payload to interpret.
返回:: Subclass-specific return value.
返回类型:: Optional[Any]

set_description(desc=None, refresh=True)[源代码]¶

返回类型:

None

参数:

desc (str | None)
refresh (bool)

set_postfix(ordered_dict=None, refresh=True, **kwargs)[源代码]¶

返回类型:

None

参数:

ordered_dict (dict | None)
refresh (bool)

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

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

参数:

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.

返回类型:

None

close()[源代码]¶

Close and cleanup the progress bar.

返回类型:: None

reset(total=None)[源代码]¶

Reset the progress bar to initial state.

参数:: total (Optional[int]) -- New total, or keep existing if None.
返回类型:: None

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

ahvn.utils.basic.progress_utils.get_progress()[源代码]¶

Get the currently active progress bar for this thread.

返回:: The active progress bar, or a NoProgress instance if none is active.
返回类型:: Progress

ahvn.utils.basic.progress_utils.progress(p)[源代码]¶

Context manager to set the active progress bar.

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

参数:: p (Progress) -- The progress bar to activate.
生成器:: Progress -- The activated progress bar.

示例

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

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