Coverage for src / mafw / ui / console_user_interface.py: 100%
30 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-09 09:08 +0000
1"""
2The console user interface.
4The module provides a simple, still efficient user interface ideal for code execution of a headless system where it is
5not possible to observe the output in real-time.
6Nevertheless, important messages are logged via the logging library and thus it is also possible to save them to a file,
7if a proper logging handler is set up.
8"""
10# Copyright 2025 European Union
11# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
12# SPDX-License-Identifier: EUPL-1.2
13from __future__ import annotations
15import logging
16from types import TracebackType
17from typing import Any, Self
19from mafw.enumerators import ProcessorStatus
20from mafw.ui.abstract_user_interface import UserInterfaceBase
22log = logging.getLogger(__name__)
25class ConsoleInterface(UserInterfaceBase):
26 """
27 A console user interface.
29 Ideal for execution in a headless environment.
31 Messages are sent via the logging system, so they can also be saved to a file is a logging handler is properly set
32 up in the execution framework.
33 """
35 name = 'console'
37 def create_task(
38 self,
39 task_name: str,
40 task_description: str = '',
41 completed: int = 0,
42 increment: int | None = None,
43 total: int | None = None,
44 **kwargs: Any,
45 ) -> None:
46 """
47 Create a new task.
49 :param task_name: A unique identifier for the task. You cannot have more than 1 task with the same name in
50 the whole execution. If you want to use the processor name, it is recommended to use the
51 :attr:`~mafw.processor.Processor.unique_name`.
52 :type task_name: str
53 :param task_description: A short description for the task. Defaults to None.
54 :type task_description: str, Optional
55 :param completed: The amount of task already completed. Defaults to None.
56 :type completed: int, Optional
57 :param increment: How much of the task has been done since last update. Defaults to None.
58 :type increment: int, Optional
59 :param total: The total amount of task. Defaults to None.
60 :type total: int, Optional
61 """
62 pass
64 def update_task(
65 self,
66 task_name: str,
67 completed: int | None = None,
68 increment: int | None = None,
69 total: int | None = None,
70 **kwargs: Any,
71 ) -> None:
72 """
73 Update an existing task.
75 :param task_name: A unique identifier for the task. You cannot have more than one task with the same name in
76 the whole execution. If you want to use the processor name, it is recommended to use the
77 :attr:`~mafw.processor.Processor.replica_name`.
78 :type task_name: str
79 :param completed: The amount of task already completed. Defaults to None.
80 :type completed: int, Optional
81 :param increment: How much of the task has been done since last update. Defaults to None.
82 :type increment: int, Optional
83 :param total: The total amount of task. Defaults to None.
84 :type total: int, Optional
85 """
86 pass
88 def __enter__(self) -> Self:
89 """Context enter dunder."""
90 return self
92 def __exit__(
93 self, type_: type[BaseException] | None, value: BaseException | None, traceback: TracebackType | None
94 ) -> None:
95 """
96 Context exit dunder.
98 :param type_: Exception type.
99 :param value: Exception value.
100 :param traceback: Exception trace back.
101 """
102 pass
104 def display_progress_message(self, message: str, i_item: int, n_item: int | None, frequency: float) -> None:
105 """
106 Display a progress message with item counter.
108 This method displays progress information at specified intervals based on the frequency parameter.
109 It formats the message with a counter showing the current item and total items.
111 :param message: The progress message to display.
112 :type message: str
113 :param i_item: The current item number (0-based index).
114 :type i_item: int
115 :param n_item: The total number of items, or None if unknown.
116 :type n_item: int | None
117 :param frequency: The frequency at which messages should be displayed (in seconds).
118 :type frequency: float
119 """
120 if self._is_time_to_display_lopping_message(i_item, n_item, frequency):
121 if n_item is None:
122 n_item = max(1000, i_item)
123 width = len(str(n_item))
124 counter = f'[{i_item + 1:>{width}}/{n_item}] '
125 msg = counter + message
126 log.info(msg)
128 def change_of_processor_status(
129 self, processor_name: str, old_status: ProcessorStatus, new_status: ProcessorStatus
130 ) -> None:
131 """
132 Log a processor status change.
134 This method logs when a processor changes its status, using debug level logging.
136 :param processor_name: The name of the processor whose status changed.
137 :type processor_name: str
138 :param old_status: The previous status of the processor.
139 :type old_status: ProcessorStatus
140 :param new_status: The new status of the processor.
141 :type new_status: ProcessorStatus
142 """
143 msg = f'{processor_name} is {new_status}'
144 log.debug(msg)
146 def prompt_question(self, question: str, **kwargs: Any) -> Any:
147 """
148 Prompt the user with a question and return their input.
150 This method displays a question to the user and waits for their input.
151 Additional keyword arguments can be passed through to the underlying input function.
153 .. versionadded:: v2.0.0
155 :param question: The question to display to the user.
156 :type question: str
157 :param kwargs: Additional keyword arguments to pass to the input function.
158 :return: The user's input.
159 :rtype: Any
160 """
161 return input(question)