DataMorphProgress#

class data_morph.progress.DataMorphProgress(auto_refresh: bool = True)[source]#

Bases: Progress

Progress tracker for Data Morph.

Note

Both the Python interface and CLI provide progress tracking using this class automatically. It is unlikely you will need to use this class yourself.

Parameters:

auto_refresh (bool, default True) – Whether to automatically refresh the progress bar. This should be set to False for Jupyter Notebooks per the Rich progress documentation.

See also

rich.progress.Progress

The base class from which all progress bar functionality derives.

add_task(description: str, start: bool = True, total: float | None = 100.0, completed: int = 0, visible: bool = True, **fields: Any) → TaskID#

Add a new ‘task’ to the Progress display.

Parameters:

  • description (str) – A description of the task.

  • start (bool, optional) – Start the task immediately (to calculate elapsed time). If set to False, you will need to call start manually. Defaults to True.

  • total (float, optional) – Number of total steps in the progress if known. Set to None to render a pulsing animation. Defaults to 100.

  • completed (int, optional) – Number of steps completed so far. Defaults to 0.

  • visible (bool, optional) – Enable display of the task. Defaults to True.

  • **fields (str) – Additional data fields required for rendering.

Returns:

An ID you can use when calling update.

Return type:

TaskID

advance(task_id: TaskID, advance: float = 1) → None#

Advance task by a number of steps.

Parameters:

  • task_id (TaskID) – ID of task.

  • advance (float) – Number of steps to advance. Default is 1.

property console: Console#
property finished: bool#

Check if all tasks have been completed.

classmethod get_default_columns() → Tuple[ProgressColumn, ...]#
Get the default columns used for a new Progress instance:

  • a text column for the description (TextColumn)

  • the bar itself (BarColumn)

  • a text column showing completion percentage (TextColumn)

  • an estimated-time-remaining column (TimeRemainingColumn)

If the Progress instance is created without passing a columns argument, the default columns defined here will be used.

You can also create a Progress instance using custom columns before and/or after the defaults, as in this example:

progress = Progress(

SpinnerColumn(), *Progress.get_default_columns(), “Elapsed:”, TimeElapsedColumn(),

)

This code shows the creation of a Progress display, containing a spinner to the left, the default columns, and a labeled elapsed time column.

get_renderable() → ConsoleRenderable | RichCast | str#

Get a renderable for the progress display.

get_renderables() → Iterable[ConsoleRenderable | RichCast | str]#

Get a number of renderables for the progress display.

make_tasks_table(tasks: Iterable[Task]) → Table#

Get a table to render the Progress display.

Parameters:

tasks (Iterable[Task]) – An iterable of Task instances, one per row of the table.

Returns:

A table instance.

Return type:

Table

open(file: str | PathLike[str] | bytes, mode: Literal['rb', 'rt', 'r'] = 'r', buffering: int = -1, encoding: str | None = None, errors: str | None = None, newline: str | None = None, *, total: int | None = None, task_id: TaskID | None = None, description: str = 'Reading...') → BinaryIO | TextIO#

Track progress while reading from a binary file.

Parameters:

  • path (Union[str, PathLike[str]]) – The path to the file to read.

  • mode (str) – The mode to use to open the file. Only supports “r”, “rb” or “rt”.

  • buffering (int) – The buffering strategy to use, see io.open().

  • encoding (str, optional) – The encoding to use when reading in text mode, see io.open().

  • errors (str, optional) – The error handling strategy for decoding errors, see io.open().

  • newline (str, optional) – The strategy for handling newlines in text mode, see io.open().

  • total (int, optional) – Total number of bytes to read. If none given, os.stat(path).st_size is used.

  • task_id (TaskID) – Task to track. Default is new task.

  • description (str, optional) – Description of task, if new task is created.

Returns:

A readable file-like object in binary mode.

Return type:

BinaryIO

Raises:

ValueError – When an invalid mode is given.

refresh() → None#

Refresh (render) the progress information.

remove_task(task_id: TaskID) → None#

Delete a task if it exists.

Parameters:

task_id (TaskID) – A task ID.

reset(task_id: TaskID, *, start: bool = True, total: float | None = None, completed: int = 0, visible: bool | None = None, description: str | None = None, **fields: Any) → None#

Reset a task so completed is 0 and the clock is reset.

Parameters:

  • task_id (TaskID) – ID of task.

  • start (bool, optional) – Start the task after reset. Defaults to True.

  • total (float, optional) – New total steps in task, or None to use current total. Defaults to None.

  • completed (int, optional) – Number of steps completed. Defaults to 0.

  • visible (bool, optional) – Enable display of the task. Defaults to True.

  • description (str, optional) – Change task description if not None. Defaults to None.

  • **fields (str) – Additional data fields required for rendering.

start() → None#

Start the progress display.

start_task(task_id: TaskID) → None#

Start a task.

Starts a task (used when calculating elapsed time). You may need to call this manually, if you called add_task with start=False.

Parameters:

task_id (TaskID) – ID of task.

stop() → None#

Stop the progress display.

stop_task(task_id: TaskID) → None#

Stop a task.

This will freeze the elapsed time on the task.

Parameters:

task_id (TaskID) – ID of task.

property task_ids: List[TaskID]#

A list of task IDs.

property tasks: List[Task]#

Get a list of Task instances.

track(sequence: Iterable[ProgressType] | Sequence[ProgressType], total: float | None = None, completed: int = 0, task_id: TaskID | None = None, description: str = 'Working...', update_period: float = 0.1) → Iterable[ProgressType]#

Track progress by iterating over a sequence.

Parameters:

  • sequence (Sequence[ProgressType]) – A sequence of values you want to iterate over and track progress.

  • total – (float, optional): Total number of steps. Default is len(sequence).

  • completed (int, optional) – Number of steps completed so far. Defaults to 0.

  • task_id – (TaskID): Task to track. Default is new task.

  • description – (str, optional): Description of task, if new task is created.

  • update_period (float, optional) – Minimum time (in seconds) between calls to update(). Defaults to 0.1.

Returns:

An iterable of values taken from the provided sequence.

Return type:

Iterable[ProgressType]

update(task_id: TaskID, *, total: float | None = None, completed: float | None = None, advance: float | None = None, description: str | None = None, visible: bool | None = None, refresh: bool = False, **fields: Any) → None#

Update information associated with a task.

Parameters:

  • task_id (TaskID) – Task id (returned by add_task).

  • total (float, optional) – Updates task.total if not None.

  • completed (float, optional) – Updates task.completed if not None.

  • advance (float, optional) – Add a value to task.completed if not None.

  • description (str, optional) – Change task description if not None.

  • visible (bool, optional) – Set visible flag if not None.

  • refresh (bool) – Force a refresh of progress information. Default is False.

  • **fields (Any) – Additional data fields required for rendering.

wrap_file(file: BinaryIO, total: int | None = None, *, task_id: TaskID | None = None, description: str = 'Reading...') → BinaryIO#

Track progress file reading from a binary file.

Parameters:

  • file (BinaryIO) – A file-like object opened in binary mode.

  • total (int, optional) – Total number of bytes to read. This must be provided unless a task with a total is also given.

  • task_id (TaskID) – Task to track. Default is new task.

  • description (str, optional) – Description of task, if new task is created.

Returns:

A readable file-like object in binary mode.

Return type:

BinaryIO

Raises:

ValueError – When no total value can be extracted from the arguments or the task.