NEDAS.utils.progress module

NEDAS.utils.progress.print_with_cache(msg: str, prev_msg: str) → str[source]
NEDAS.utils.progress.watch_files(files, timeout=1000, check_dt=1)[source]
NEDAS.utils.progress.watch_log(logfile: str, keyword: str, timeout: int = 1000, check_dt: int = 1) → None[source]
NEDAS.utils.progress.find_keyword_in_file(file: str, keyword: str) → bool[source]
NEDAS.utils.progress.count_lines_in_file(file: str) → int[source]
class NEDAS.utils.progress.Formatter(interactive: bool = True, is_notebook: bool = False, cols=80, anchor=50, tabspace=4, progress_bar_width=10)[source]

Bases: object

Formatter of the progress display.

Parameters:

  • interactive (bool, optional) – Whether the output is interactive (supports ansi escape code). Defaults to True.

  • anchor (int, optional) – Characters to anchor the left part of status line. Defaults to 50.

  • tabspace (int, optional) – Number of spaces for one call stack level indentation. Defaults to 4.

  • progress_bar_width (int, optional) – Width of the progress bar in characters. Defaults to 10.

strip_escape_code(text: str) → str[source]
truncate(text: str) → str[source]
indent(level: int, branch: bool = True) → str[source]

Generate the indent string to form call stack tree structure in log.

Parameters:

  • level (int) – The current call stack level.

  • branch (bool) – Whether a branch is needed at the end.

Returns:

The indent string

Return type:

str

padding(level: int, name: str) → str[source]

Generate the padding string to align the status line.

Parameters:

  • level (int) – The current call stack level.

  • name (str) – The name of the current function or task.

Returns:

The padding string

Return type:

str

dimmer(msg)[source]
progress_bar(task_id: int, ntask: int) → str[source]

Generate a progress bar based on task_id and ntask.

Parameters:

  • task_id (int) – Current task index, from 0 to ntask-1

  • ntask (int) – Total number of tasks

Returns:

The progress bar msg to be shown.

Return type:

str

Note: Will require the print command with end=”” option so that new line updated is overwritting the old line.

class NEDAS.utils.progress.Progress(interactive: bool = True, is_notebook: bool = False, cols: int = 80, debug: bool = False, call_stack: list[dict] | None = None, call_stack_max_level: int | None = None, anchor: int = 50, tabspace: int = 4, progress_bar_width: int = 10, io_interval: float = 0.1)[source]

Bases: object

Progress tracker and displayer. Used by Context.logger to show runtime progress.

formatter: Formatter
interactive: bool
debug: bool
call_stack: list[dict[str, Any]]
call_stack_max_level: int | None
new_node(func_name: str | None = None) → dict[source]
property node: dict
within_max_level(level: int) → bool[source]
is_leaf(node) → bool[source]
property level: int
set_flag(flag: str)[source]
get_timer_msg(node)[source]
push(func_name: str)[source]
pop()[source]
update() → str[source]
log(msg: str, flag: str) → str[source]

Safely injects a global message without breaking the tree.