Coverage for src/mafw/ui/abstract_user_interface.py: 100%
51 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-26 09:13 +0000
1"""
2An abstract generic user interface.
4The module provides a generic user interface that can be implemented to allow MAFw to communicate with different
5user interfaces.
7MAFw is designed to operate seamlessly without a user interface; however, users often appreciate the added benefit of communication between the process execution and themselves.
9There are several different interfaces and different interface types (Command Line, Textual, Graphical...) and
10everyone has its own preferences. In order to be as generic as possible, MAFw is allowing for an abstract
11interface layer so that the user can either decide to use one of the few coming with MAFw or to implement the
12interface to their favorite interface.
13"""
15# Copyright 2025–2026 European Union
16# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
17# SPDX-License-Identifier: EUPL-1.2
18from __future__ import annotations
20from collections.abc import Generator
21from contextlib import contextmanager
22from types import TracebackType
23from typing import Any, Self
25from mafw.enumerators import ProcessorStatus
28class UserInterfaceMeta(type):
29 """
30 A metaclass used for the creation of user interface
31 """
33 __required_members__ = (
34 'create_task',
35 'update_task',
36 'display_progress_message',
37 'change_of_processor_status',
38 'prompt_question',
39 'enter_interactive_mode',
40 )
41 __required_callable__ = (
42 'create_task',
43 'update_task',
44 'display_progress_message',
45 'change_of_processor_status',
46 'prompt_question',
47 'enter_interactive_mode',
48 )
50 def __instancecheck__(cls, instance: Any) -> bool:
51 return cls.__subclasscheck__(type(instance))
53 def __subclasscheck__(cls, subclass: Any) -> bool:
54 members = all([hasattr(subclass, m) for m in cls.__required_members__])
55 if not members:
56 return False
57 callables = all([callable(getattr(subclass, c)) for c in cls.__required_callable__])
58 return callables
61class UserInterfaceBase(metaclass=UserInterfaceMeta):
62 """
63 An abstract base user interface class that defines the interface for communicating with different user interfaces.
65 This class provides a standardized way for MAFw to interact with various user interfaces including command-line,
66 textual, and graphical interfaces. It defines the required methods that any concrete implementation must provide.
68 The interface allows for task management, progress reporting, status updates, and user interaction capabilities.
70 .. versionadded:: v1.0
71 """
73 always_display_progress_message = 10
74 """
75 Threshold for displaying progress messages.
77 If the total number of events is below this value, then the progress message is always displayed, otherwise
78 follow the standard update frequency.
79 """
80 name = 'base'
81 """The name of the interface"""
83 def create_task(
84 self,
85 task_name: str,
86 task_description: str = '',
87 completed: int = 0,
88 increment: int | None = None,
89 total: int | None = None,
90 **kwargs: Any,
91 ) -> None:
92 """
93 Create a new task.
95 :param task_name: A unique identifier for the task. You cannot have more than 1 task with the same name in
96 the whole execution. If you want to use the processor name, it is recommended to use the
97 :attr:`~mafw.processor.Processor.unique_name`.
98 :type task_name: str
99 :param task_description: A short description for the task. Defaults to None.
100 :type task_description: str, Optional
101 :param completed: The amount of task already completed. Defaults to None.
102 :type completed: int, Optional
103 :param increment: How much of the task has been done since last update. Defaults to None.
104 :type increment: int, Optional
105 :param total: The total amount of task. Defaults to None.
106 :type total: int, Optional
107 """
108 pass
110 def update_task(
111 self, task_name: str, completed: int = 0, increment: int | None = None, total: int | None = None, **kwargs: Any
112 ) -> None:
113 """
114 Update an existing task.
116 :param task_name: A unique identifier for the task. You cannot have more than 1 task with the same name in
117 the whole execution. If you want to use the processor name, it is recommended to use the
118 :attr:`~mafw.processor.Processor.replica_name`.
119 :type task_name: str
120 :param completed: The amount of task already completed. Defaults to None.
121 :type completed: int, Optional
122 :param increment: How much of the task has been done since last update. Defaults to None.
123 :type increment: int, Optional
124 :param total: The total amount of task. Defaults to None.
125 :type total: int, Optional
126 """
127 pass
129 def __enter__(self) -> Self:
130 """Context enter dunder."""
131 return self
133 def __exit__(
134 self, type_: type[BaseException] | None, value: BaseException | None, traceback: TracebackType | None
135 ) -> None:
136 """
137 Context exit dunder.
139 :param type_: Exception type.
140 :param value: Exception value.
141 :param traceback: Exception trace back.
142 """
143 pass
145 def display_progress_message(self, message: str, i_item: int, n_item: int | None, frequency: float) -> None:
146 """
147 Display a message during the process execution.
149 :param message: The message to be displayed.
150 :type message: str
151 :param i_item: The current item enumerator.
152 :type i_item: int
153 :param n_item: The total number of items or None for an indeterminate progress (while loop).
154 :type n_item: int | None
155 :param frequency: How often (in percentage of n_item) to display the message.
156 :type frequency: float
157 """
158 pass
160 def _is_time_to_display_lopping_message(self, i_item: int, n_item: int | None, frequency: float) -> bool:
161 if n_item is None:
162 # let's print the message everytime i_item is a multiple of 10.
163 return i_item % 10 == 0
165 if n_item > self.always_display_progress_message:
166 always = False
167 else:
168 always = True
169 if always:
170 do_display = True
171 else:
172 mod = max([round(frequency * n_item), 1])
173 do_display = i_item == 0 or i_item % mod == 0 or i_item == n_item - 1
174 return do_display
176 def change_of_processor_status(
177 self, processor_name: str, old_status: ProcessorStatus, new_status: ProcessorStatus
178 ) -> None:
179 pass
181 @contextmanager
182 def enter_interactive_mode(self) -> Generator[None, Any, None]:
183 """
184 A context manager for entering interactive mode.
186 This method provides a way to temporarily switch to interactive mode, allowing for direct user interaction
187 during processing. It should be used as a context manager with a ``with`` statement.
189 .. versionadded:: v2.0.0
191 :returns: A context manager that yields control to the interactive section
192 :rtype: Generator
193 """
194 try:
195 yield
196 finally:
197 pass
199 def prompt_question(self, question: str, **kwargs: Any) -> Any:
200 """
201 Prompt the user with a question and return their response.
203 This method should display a question to the user and wait for their input. The implementation may vary
204 depending on the specific user interface being used.
206 .. versionadded:: v2.0.0
208 :param question: The question to be asked to the user.
209 :type question: str
210 :param kwargs: Additional keyword arguments that might be used by specific implementations.
211 :returns: The user's response to the question.
212 :rtype: Any
213 """
214 pass # pragma: no cover