Coverage for src/mafw/ui/rich_user_interface.py: 100%
68 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-26 09:13 +0000
1"""
2The rich user interface.
4The module provides an implementation of the abstract user interface that takes advantage from the `rich` library.
5Progress bars and spinners are shown during the processor execution along with log messages including markup language.
6In order for this logging message to appear properly rendered, the logger should be connected to a RichHandler.
7"""
9# Copyright 2025–2026 European Union
10# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
11# SPDX-License-Identifier: EUPL-1.2
12import logging
13from collections.abc import Generator
14from contextlib import contextmanager
15from types import TracebackType
16from typing import Any, Self
18import rich.prompt
19from rich.progress import Progress, SpinnerColumn, TaskID, TimeElapsedColumn
21from mafw.enumerators import ProcessorStatus
22from mafw.ui.abstract_user_interface import UserInterfaceBase
24log = logging.getLogger(__name__)
27class RichInterface(UserInterfaceBase):
28 """
29 Implementation of the interface for rich.
31 :param progress_kws: A dictionary of keywords passed to the `rich.Progress`. Defaults to None
32 :type progress_kws: dict, Optional
33 """
35 name = 'rich'
37 def __init__(self, progress_kws: dict[str, Any] | None = None) -> None:
38 if progress_kws is None:
39 progress_kws = dict(auto_refresh=True, expand=True)
41 self.progress = Progress(SpinnerColumn(), *Progress.get_default_columns(), TimeElapsedColumn(), **progress_kws)
42 self.task_dict: dict[str, TaskID] = {}
44 def __enter__(self) -> Self:
45 """
46 Context enter dunder.
48 It manually starts the progress extension and then return the class instance.
49 """
50 self.progress.start()
51 return self
53 def __exit__(
54 self, type_: type[BaseException] | None, value: BaseException | None, traceback: TracebackType | None
55 ) -> None:
56 """
57 Context exit dunder.
59 It manually stops the progress bar.
61 :param type_: Exception type.
62 :param value: Exception value.
63 :param traceback: Exception trace back.
64 """
65 self.progress.stop()
67 def create_task(
68 self,
69 task_name: str,
70 task_description: str = '',
71 completed: int = 0,
72 increment: int | None = None,
73 total: int | None = None,
74 **kwargs: Any,
75 ) -> None:
76 """
77 Create a new task.
79 :param task_name: A unique identifier for the task. You cannot have more than 1 task with the same name in
80 the whole execution. If you want to use the processor name, it is recommended to use the
81 :attr:`~mafw.processor.Processor.unique_name`.
82 :type task_name: str
83 :param task_description: A short description for the task. Defaults to ''.
84 :type task_description: str, Optional
85 :param completed: The amount of task already completed. Defaults to 0.
86 :type completed: int, Optional
87 :param increment: How much of the task has been done since last update. Defaults to None.
88 :type increment: int, Optional
89 :param total: The total amount of task. Defaults to None.
90 :type total: int, Optional
91 """
92 if task_name in self.task_dict:
93 log.warning('A task with this name (%s) already exists. Replacing it with the new one.' % task_name)
94 log.warning('Be sure to use unique names.')
96 self.task_dict[task_name] = self.progress.add_task(task_description, total=total, completed=completed)
98 def update_task(
99 self,
100 task_name: str,
101 completed: int | None = None,
102 increment: int | None = None,
103 total: int | None = None,
104 **kwargs: Any,
105 ) -> None:
106 """
107 Update an existing task.
109 :param task_name: A unique identifier for the task. You cannot have more than one task with the same name in
110 the whole execution. If you want to use the processor name, it is recommended to use the
111 :attr:`~~mafw.processor.Processor.replica_name`.
112 :type task_name: str
113 :param completed: The amount of task already completed. Defaults to 0.
114 :type completed: int, Optional
115 :param increment: How much of the task has been done since last update. Defaults to None.
116 :type increment: int, Optional
117 :param total: The total amount of task. Defaults to None.
118 :type total: int, Optional
119 """
120 if task_name not in self.task_dict:
121 log.warning('A task with this name (%s) does not exist.' % task_name)
122 log.warning('Skipping updates')
123 return
125 if completed is None and total is None:
126 visible = True
127 else:
128 visible = completed != total
130 self.progress.update(
131 self.task_dict[task_name], completed=completed, advance=increment, total=total, visible=visible
132 )
134 def display_progress_message(self, message: str, i_item: int, n_item: int | None, frequency: float) -> None:
135 if self._is_time_to_display_lopping_message(i_item, n_item, frequency):
136 if n_item is None:
137 n_item = max(1000, i_item)
138 width = len(str(n_item))
139 counter = f'[{i_item + 1:>{width}}/{n_item}] '
140 msg = counter + message
141 log.info(msg)
143 def change_of_processor_status(
144 self, processor_name: str, old_status: ProcessorStatus, new_status: ProcessorStatus
145 ) -> None:
146 """
147 Display a message when a processor status changes.
149 This method logs a debug message indicating that a processor has changed its status.
150 The message uses rich markup to highlight the processor name and new status.
152 :param processor_name: The name of the processor whose status has changed.
153 :type processor_name: str
154 :param old_status: The previous status of the processor.
155 :type old_status: ProcessorStatus
156 :param new_status: The new status of the processor.
157 :type new_status: ProcessorStatus
158 """
159 msg = f'[red]{processor_name}[/red] is [bold]{new_status}[/bold]'
160 log.debug(msg)
162 @contextmanager
163 def enter_interactive_mode(self) -> Generator[None, Any, None]:
164 """
165 Context manager to temporarily switch to interactive mode.
167 This method temporarily stops the progress display to allow for interactive input
168 while preserving the original transient state. After yielding control, it restores
169 the progress display with appropriate spacing to avoid overwriting previous output.
171 .. versionadded:: v2.0.0
173 .. note::
174 This method should be used within a ``with`` statement to ensure proper cleanup.
175 """
176 transient = self.progress.live.transient # save the old value
177 self.progress.live.transient = True
178 self.progress.stop()
179 self.progress.live.transient = transient # restore the old value
180 try:
181 yield
182 finally:
183 # make space for the progress to use so it doesn't overwrite any previous lines
184 visible_tasks = [task for task in self.progress.tasks if task.visible]
185 print('\n' * (len(visible_tasks) - 2))
186 self.progress.start()
188 def prompt_question(self, question: str, **kwargs: Any) -> Any:
189 """
190 Prompt the user with a question and return their response.
192 This method uses the rich library's prompt functionality to ask the user a question.
193 It supports various prompt types including confirmation, input, and choice prompts.
195 .. versionadded:: v2.0.0
197 :param question: The question to ask the user.
198 :type question: str
199 :param kwargs: Additional arguments to pass to the prompt function.
200 :return: The user's response based on the prompt type.
201 :rtype: Any
202 :param prompt_type: The type of prompt to use. Defaults to :class:`rich.prompt.Confirm`.
203 :param console: The console to use for the prompt. Defaults to None.
204 :param password: Whether to hide input when prompting for passwords. Defaults to False.
205 :param choices: List of valid choices for choice prompts. Defaults to None.
206 :param default: Default value for prompts that support it. Defaults to None.
207 :param show_default: Whether to show the default value. Defaults to True.
208 :param show_choices: Whether to show available choices. Defaults to True.
209 :param case_sensitive: Whether choices are case sensitive. Defaults to True.
210 """
211 prompt_type = kwargs.pop('prompt_type', rich.prompt.Confirm)
212 console = kwargs.pop('console', None)
213 password = kwargs.pop('password', False)
214 choices = kwargs.pop('choices', None)
215 default = kwargs.pop('default', None)
216 show_default = kwargs.pop('show_default', True)
217 show_choices = kwargs.pop('show_choices', True)
218 case_sensitive = kwargs.pop('case_sensitive', True)
220 return prompt_type.ask(
221 question,
222 console=console,
223 password=password,
224 choices=choices,
225 default=default,
226 case_sensitive=case_sensitive,
227 show_default=show_default,
228 show_choices=show_choices,
229 )