mafw.ui.rich_user_interface

The rich user interface.

The module provides an implementation of the abstract user interface that takes advantage from the rich library. Progress bars and spinners are shown during the processor execution along with log messages including markup language. In order for this logging message to appear properly rendered, the logger should be connected to a RichHandler.

Classes

RichInterface([progress_kws])

Implementation of the interface for rich.

class mafw.ui.rich_user_interface.RichInterface(progress_kws: dict[str, Any] | None = None)[source]

Bases: UserInterfaceBase

Implementation of the interface for rich.

Parameters:

progress_kws (dict, Optional) – A dictionary of keywords passed to the rich.Progress. Defaults to None

change_of_processor_status(processor_name: str, old_status: ProcessorStatus, new_status: ProcessorStatus) None[source]

Display a message when a processor status changes.

This method logs a debug message indicating that a processor has changed its status. The message uses rich markup to highlight the processor name and new status.

Parameters:
  • processor_name (str) – The name of the processor whose status has changed.

  • old_status (ProcessorStatus) – The previous status of the processor.

  • new_status (ProcessorStatus) – The new status of the processor.

create_task(task_name: str, task_description: str = '', completed: int = 0, increment: int | None = None, total: int | None = None, **kwargs: Any) None[source]

Create a new task.

Parameters:
  • task_name (str) – A unique identifier for the task. You cannot have more than 1 task with the same name in the whole execution. If you want to use the processor name, it is recommended to use the unique_name.

  • task_description (str, Optional) – A short description for the task. Defaults to ‘’.

  • completed (int, Optional) – The amount of task already completed. Defaults to 0.

  • increment (int, Optional) – How much of the task has been done since last update. Defaults to None.

  • total (int, Optional) – The total amount of task. Defaults to None.

display_progress_message(message: str, i_item: int, n_item: int | None, frequency: float) None[source]

Display a message during the process execution.

Parameters:
  • message (str) – The message to be displayed.

  • i_item (int) – The current item enumerator.

  • n_item (int | None) – The total number of items or None for an indeterminate progress (while loop).

  • frequency (float) – How often (in percentage of n_item) to display the message.

enter_interactive_mode() Generator[None, Any, None][source]

Context manager to temporarily switch to interactive mode.

This method temporarily stops the progress display to allow for interactive input while preserving the original transient state. After yielding control, it restores the progress display with appropriate spacing to avoid overwriting previous output.

Added in version v2.0.0.

Note

This method should be used within a with statement to ensure proper cleanup.

prompt_question(question: str, **kwargs: Any) Any[source]

Prompt the user with a question and return their response.

This method uses the rich library’s prompt functionality to ask the user a question. It supports various prompt types including confirmation, input, and choice prompts.

Added in version v2.0.0.

Parameters:
  • question (str) – The question to ask the user.

  • kwargs – Additional arguments to pass to the prompt function.

  • prompt_type – The type of prompt to use. Defaults to rich.prompt.Confirm.

  • console – The console to use for the prompt. Defaults to None.

  • password – Whether to hide input when prompting for passwords. Defaults to False.

  • choices – List of valid choices for choice prompts. Defaults to None.

  • default – Default value for prompts that support it. Defaults to None.

  • show_default – Whether to show the default value. Defaults to True.

  • show_choices – Whether to show available choices. Defaults to True.

  • case_sensitive – Whether choices are case sensitive. Defaults to True.

Returns:

The user’s response based on the prompt type.

Return type:

Any

update_task(task_name: str, completed: int | None = None, increment: int | None = None, total: int | None = None, **kwargs: Any) None[source]

Update an existing task.

Parameters:
  • task_name (str) – A unique identifier for the task. You cannot have more than one task with the same name in the whole execution. If you want to use the processor name, it is recommended to use the replica_name.

  • completed (int, Optional) – The amount of task already completed. Defaults to 0.

  • increment (int, Optional) – How much of the task has been done since last update. Defaults to None.

  • total (int, Optional) – The total amount of task. Defaults to None.

name = 'rich'

The name of the interface