Coverage for src / mafw / ui / abstract_user_interface.py: 100%
50 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-09 09:08 +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 European Union
16# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
17# SPDX-License-Identifier: EUPL-1.2
18from __future__ import annotations
20from contextlib import contextmanager
21from types import TracebackType
22from typing import Any, Generator, Self
24from mafw.enumerators import ProcessorStatus
27class UserInterfaceMeta(type):
28 """
29 A metaclass used for the creation of user interface
30 """
32 __required_members__ = (
33 'create_task',
34 'update_task',
35 'display_progress_message',
36 'change_of_processor_status',
37 'prompt_question',
38 'enter_interactive_mode',
39 )
40 __required_callable__ = (
41 'create_task',
42 'update_task',
43 'display_progress_message',
44 'change_of_processor_status',
45 'prompt_question',
46 'enter_interactive_mode',
47 )
49 def __instancecheck__(cls, instance: Any) -> bool:
50 return cls.__subclasscheck__(type(instance))
52 def __subclasscheck__(cls, subclass: Any) -> bool:
53 members = all([hasattr(subclass, m) for m in cls.__required_members__])
54 if not members:
55 return False
56 callables = all([callable(getattr(subclass, c)) for c in cls.__required_callable__])
57 return callables
60class UserInterfaceBase(metaclass=UserInterfaceMeta):
61 """
62 An abstract base user interface class that defines the interface for communicating with different user interfaces.
64 This class provides a standardized way for MAFw to interact with various user interfaces including command-line,
65 textual, and graphical interfaces. It defines the required methods that any concrete implementation must provide.
67 The interface allows for task management, progress reporting, status updates, and user interaction capabilities.
69 .. versionadded:: v1.0
70 """
72 always_display_progress_message = 10
73 """
74 Threshold for displaying progress messages.
76 If the total number of events is below this value, then the progress message is always displayed, otherwise
77 follow the standard update frequency.
78 """
79 name = 'base'
80 """The name of the interface"""
82 def create_task(
83 self,
84 task_name: str,
85 task_description: str = '',
86 completed: int = 0,
87 increment: int | None = None,
88 total: int | None = None,
89 **kwargs: Any,
90 ) -> None:
91 """
92 Create a new task.
94 :param task_name: A unique identifier for the task. You cannot have more than 1 task with the same name in
95 the whole execution. If you want to use the processor name, it is recommended to use the
96 :attr:`~mafw.processor.Processor.unique_name`.
97 :type task_name: str
98 :param task_description: A short description for the task. Defaults to None.
99 :type task_description: str, Optional
100 :param completed: The amount of task already completed. Defaults to None.
101 :type completed: int, Optional
102 :param increment: How much of the task has been done since last update. Defaults to None.
103 :type increment: int, Optional
104 :param total: The total amount of task. Defaults to None.
105 :type total: int, Optional
106 """
107 pass
109 def update_task(
110 self, task_name: str, completed: int = 0, increment: int | None = None, total: int | None = None, **kwargs: Any
111 ) -> None:
112 """
113 Update an existing task.
115 :param task_name: A unique identifier for the task. You cannot have more than 1 task with the same name in
116 the whole execution. If you want to use the processor name, it is recommended to use the
117 :attr:`~mafw.processor.Processor.replica_name`.
118 :type task_name: str
119 :param completed: The amount of task already completed. Defaults to None.
120 :type completed: int, Optional
121 :param increment: How much of the task has been done since last update. Defaults to None.
122 :type increment: int, Optional
123 :param total: The total amount of task. Defaults to None.
124 :type total: int, Optional
125 """
126 pass
128 def __enter__(self) -> Self:
129 """Context enter dunder."""
130 return self
132 def __exit__(
133 self, type_: type[BaseException] | None, value: BaseException | None, traceback: TracebackType | None
134 ) -> None:
135 """
136 Context exit dunder.
138 :param type_: Exception type.
139 :param value: Exception value.
140 :param traceback: Exception trace back.
141 """
142 pass
144 def display_progress_message(self, message: str, i_item: int, n_item: int | None, frequency: float) -> None:
145 """
146 Display a message during the process execution.
148 :param message: The message to be displayed.
149 :type message: str
150 :param i_item: The current item enumerator.
151 :type i_item: int
152 :param n_item: The total number of items or None for an indeterminate progress (while loop).
153 :type n_item: int | None
154 :param frequency: How often (in percentage of n_item) to display the message.
155 :type frequency: float
156 """
157 pass
159 def _is_time_to_display_lopping_message(self, i_item: int, n_item: int | None, frequency: float) -> bool:
160 if n_item is None:
161 # let's print the message everytime i_item is a multiple of 10.
162 return i_item % 10 == 0
164 if n_item > self.always_display_progress_message:
165 always = False
166 else:
167 always = True
168 if always:
169 do_display = True
170 else:
171 mod = max([round(frequency * n_item), 1])
172 do_display = i_item == 0 or i_item % mod == 0 or i_item == n_item - 1
173 return do_display
175 def change_of_processor_status(
176 self, processor_name: str, old_status: ProcessorStatus, new_status: ProcessorStatus
177 ) -> None:
178 pass
180 @contextmanager
181 def enter_interactive_mode(self) -> Generator[None, Any, None]:
182 """
183 A context manager for entering interactive mode.
185 This method provides a way to temporarily switch to interactive mode, allowing for direct user interaction
186 during processing. It should be used as a context manager with a ``with`` statement.
188 .. versionadded:: v2.0.0
190 :returns: A context manager that yields control to the interactive section
191 :rtype: Generator
192 """
193 try:
194 yield
195 finally:
196 pass
198 def prompt_question(self, question: str, **kwargs: Any) -> Any:
199 """
200 Prompt the user with a question and return their response.
202 This method should display a question to the user and wait for their input. The implementation may vary
203 depending on the specific user interface being used.
205 .. versionadded:: v2.0.0
207 :param question: The question to be asked to the user.
208 :type question: str
209 :param kwargs: Additional keyword arguments that might be used by specific implementations.
210 :returns: The user's response to the question.
211 :rtype: Any
212 """
213 pass # pragma: no cover