Change events.reply(_error) with message: str to reply: AIReply - Fixes

Author: yorevs

src/main/askai/core/askai.py

@@ -12,6 +12,22 @@
 
    Copyright (c) 2024, HomeSetup
 """
+import logging as log
+import os
+import re
+import sys
+from enum import Enum
+from pathlib import Path
+from typing import List, Optional, TypeAlias
+
+from click import UsageError
+from hspylib.core.enums.charset import Charset
+from hspylib.core.tools.commons import file_is_not_empty, is_debugging
+from hspylib.core.zoned_datetime import DATE_FORMAT, now, TIME_FORMAT
+from hspylib.modules.application.exit_status import ExitStatus
+from hspylib.modules.eventbus.event import Event
+from openai import RateLimitError
+
 from askai.__classpath__ import classpath
 from askai.core.askai_configs import configs
 from askai.core.askai_events import events
@@ -29,21 +45,6 @@
 from askai.exception.exceptions import (ImpossibleQuery, InaccurateResponse, IntelligibleAudioError,
                                         MaxInteractionsReached, TerminatingQuery)
 from askai.tui.app_icons import AppIcons
-from click import UsageError
-from enum import Enum
-from hspylib.core.enums.charset import Charset
-from hspylib.core.tools.commons import file_is_not_empty, is_debugging
-from hspylib.core.zoned_datetime import DATE_FORMAT, now, TIME_FORMAT
-from hspylib.modules.application.exit_status import ExitStatus
-from hspylib.modules.eventbus.event import Event
-from openai import RateLimitError
-from pathlib import Path
-from typing import AnyStr, List, Optional, TypeAlias
-
-import logging as log
-import os
-import re
-import sys
 
 QueryString: TypeAlias = str | List[str] | None
 
@@ -196,15 +197,15 @@ def _create_console_file(self, overwrite: bool = True) -> None:
             )
             f_console.flush()
 
-    def _reply(self, message: AnyStr) -> None:
+    def _reply(self, reply: AIReply) -> None:
         """Reply to the user with the AI-generated response.
-        :param message: The message to send as a reply to the user.
+        :param reply: The reply message to send as a reply to the user.
         """
         ...
 
-    def _reply_error(self, message: AnyStr) -> None:
+    def _reply_error(self, reply: AIReply) -> None:
         """Reply to the user with an AI-generated error message or system error.
-        :param message: The error message to be displayed to the user.
+        :param reply: The error reply message to be displayed to the user.
         """
         ...
 

src/main/askai/core/askai_cli.py

@@ -12,6 +12,20 @@
 
    Copyright (c) 2024, HomeSetup
 """
+import logging as log
+import os
+from pathlib import Path
+from threading import Thread
+from typing import List, TypeAlias
+
+import nltk
+import pause
+from clitt.core.term.cursor import cursor
+from clitt.core.term.screen import screen
+from clitt.core.tui.line_input.keyboard_input import KeyboardInput
+from hspylib.modules.eventbus.event import Event
+from rich.progress import Progress
+
 from askai.core.askai import AskAi
 from askai.core.askai_configs import configs
 from askai.core.askai_events import *
@@ -25,19 +39,6 @@
 from askai.core.support.shared_instances import shared
 from askai.core.support.text_formatter import text_formatter
 from askai.core.support.utilities import display_text
-from clitt.core.term.cursor import cursor
-from clitt.core.term.screen import screen
-from clitt.core.tui.line_input.keyboard_input import KeyboardInput
-from hspylib.modules.eventbus.event import Event
-from pathlib import Path
-from rich.progress import Progress
-from threading import Thread
-from typing import AnyStr, List, TypeAlias
-
-import logging as log
-import nltk
-import os
-import pause
 
 QueryString: TypeAlias = str | List[str] | None
 
@@ -87,26 +88,26 @@ def run(self) -> None:
             self._reply(msg.goodbye())
         display_text("", markdown=False)
 
-    def _reply(self, message: AnyStr) -> None:
+    def _reply(self, reply: AIReply) -> None:
         """Reply to the user with the AI-generated response.
-        :param message: The message to send as a reply to the user.
+        :param reply: The reply message to send as a reply to the user.
         """
-        if message and (text := msg.translate(message)):
-            log.debug(message)
-            if configs.is_speak:
+        if reply and (text := msg.translate(reply.message)):
+            log.debug(reply.message)
+            if configs.is_speak and reply.is_speakable:
                 self.engine.text_to_speech(text, f"{shared.nickname}")
             elif not configs.is_interactive:
                 display_text(text_formatter.strip_format(text), f"{shared.nickname}", markdown=False)
             else:
                 display_text(text, f"{shared.nickname}")
 
-    def _reply_error(self, message: AnyStr) -> None:
+    def _reply_error(self, reply: AIReply) -> None:
         """Reply to the user with an AI-generated error message or system error.
-        :param message: The error message to be displayed to the user.
+        :param reply: The error reply message to be displayed to the user.
         """
-        if message and (text := msg.translate(message)):
-            log.error(message)
-            if configs.is_speak:
+        if reply and (text := msg.translate(reply.message)):
+            log.error(reply.message)
+            if configs.is_speak and reply.is_speakable:
                 self.engine.text_to_speech(f"Error: {text}", f"{shared.nickname}")
             else:
                 display_text(f"%RED%Error: {text}%NC%", f"{shared.nickname}")
@@ -124,12 +125,12 @@ def _cb_reply_event(self, ev: Event) -> None:
         reply: AIReply
         if reply := ev.args.reply:
             if reply.is_error:
-                self._reply_error(str(reply))
+                self._reply_error(reply)
             else:
-                if ev.args.reply.verbosity <= 1 or configs.is_debug:
+                if ev.args.reply.verbosity.match(configs.verbosity) or configs.is_debug:
                     if ev.args.erase_last:
                         cursor.erase_line()
-                    self._reply(str(reply))
+                    self._reply(reply)
 
     def _cb_mic_listening_event(self, ev: Event) -> None:
         """Callback to handle microphone listening events.

src/main/askai/core/askai_configs.py

@@ -12,15 +12,17 @@
 
    Copyright (c) 2024, HomeSetup
 """
-from askai.__classpath__ import classpath
-from askai.core.askai_settings import settings
-from askai.language.language import Language
+import locale
+import os
+from shutil import which
+
 from hspylib.core.enums.charset import Charset
 from hspylib.core.metaclass.singleton import Singleton
-from shutil import which
 
-import locale
-import os
+from askai.__classpath__ import classpath
+from askai.core.askai_settings import settings
+from askai.core.enums.verbosity import Verbosity
+from askai.language.language import Language
 
 
 class AskAiConfigs(metaclass=Singleton):
@@ -98,6 +100,10 @@ def ttl(self) -> int:
     def ttl(self, value: int) -> None:
         settings.put("askai.cache.ttl.minutes", value)
 
+    @property
+    def verbosity(self) -> Verbosity:
+        return Verbosity.of_value(settings.get_int("askai.verbosity.level"))
+
     @property
     def chunk_size(self) -> int:
         return settings.get_int("askai.text.splitter.chunk.size")

src/main/askai/core/askai_settings.py

@@ -12,21 +12,22 @@
 
    Copyright (c) 2024, HomeSetup
 """
-from askai.__classpath__ import classpath
+import logging as log
+import os
+import re
 from contextlib import redirect_stdout
-from hspylib.core.metaclass.singleton import Singleton
-from hspylib.core.tools.commons import console_out, to_bool
 from io import StringIO
 from pathlib import Path
+from typing import Any, Optional
+
+from hspylib.core.metaclass.singleton import Singleton
+from hspylib.core.tools.commons import console_out, to_bool
 from rich.table import Table
 from setman.settings.settings import Settings
 from setman.settings.settings_config import SettingsConfig
 from setman.settings.settings_entry import SettingsEntry
-from typing import Any, Optional
 
-import logging as log
-import os
-import re
+from askai.__classpath__ import classpath
 
 # AskAI config directory.
 ASKAI_DIR: Path = Path(f'{os.getenv("ASKAI_DIR", os.getenv("HHS_DIR", str(Path.home())))}/askai')
@@ -43,7 +44,7 @@ class AskAiSettings(metaclass=Singleton):
     INSTANCE: "AskAiSettings"
 
     # Current settings version. Updating this value will trigger a database recreation using the defaults.
-    _ACTUAL_VERSION: str = "0.1.91"
+    _ACTUAL_VERSION: str = "0.1.92"
 
     RESOURCE_DIR = str(classpath.resource_path())
 
@@ -99,6 +100,7 @@ def defaults(self) -> None:
         self._settings.put("askai.preferred.language", "askai", "")
         self._settings.put("askai.default.engine", "askai", "openai")
         self._settings.put("askai.default.engine.model", "askai", "gpt-4o-mini")
+        self._settings.put("askai.verbosity.level", "askai", 3)
         self._settings.put("askai.text.to.speech.tempo", "askai", 1)
         self._settings.put("askai.text.splitter.chunk.size", "askai", 1000)
         self._settings.put("askai.text.splitter.chunk.overlap", "askai", 100)

src/main/askai/core/component/audio_player.py

@@ -12,21 +12,21 @@
 
    Copyright (c) 2024, HomeSetup
 """
-from askai.__classpath__ import classpath
-from askai.core.askai_configs import configs
-from clitt.core.term.terminal import Terminal
+import logging as log
+import time
 from functools import lru_cache
+from pathlib import Path
+from shutil import which
+from typing import Literal
+
+from clitt.core.term.terminal import Terminal
 from hspylib.core.metaclass.singleton import Singleton
 from hspylib.core.preconditions import check_argument
 from hspylib.core.tools.commons import file_is_not_empty
 from hspylib.core.tools.text_tools import ensure_endswith
 from hspylib.modules.application.exit_status import ExitStatus
-from pathlib import Path
-from shutil import which
-from typing import Literal
 
-import logging as log
-import time
+from askai.__classpath__ import classpath
 
 
 class AudioPlayer(metaclass=Singleton):
@@ -46,9 +46,7 @@ def play_audio_file(path_to_audio_file: str | Path, tempo: int = 1) -> bool:
         :param tempo: The playback speed (default is 1).
         :return: True if the audio file is played successfully, otherwise False.
         """
-        if not configs.is_speak:
-            return True
-        elif file_is_not_empty(str(path_to_audio_file)):
+        if file_is_not_empty(str(path_to_audio_file)):
             try:
                 out, code = Terminal.shell_exec(
                     f'ffplay -af "atempo={tempo}" -v 0 -nodisp -autoexit {path_to_audio_file}'

src/main/askai/core/enums/verbosity.py

@@ -0,0 +1,35 @@
+from hspylib.core.enums.enumeration import Enumeration
+
+
+class Verbosity(Enumeration):
+    """Represents different verbosity levels for logging or output.
+    Attributes:
+        MINIMUM (int): The minimum verbosity level.
+        LOW (int): A low verbosity level.
+        NORMAL (int): The normal verbosity level.
+        DETAILED (int): A detailed verbosity level.
+        FULL (int): The full verbosity level.
+    """
+
+    # fmt: off
+    MINIMUM     = 1
+    LOW         = 2
+    NORMAL      = 3
+    DETAILED    = 4
+    FULL        = 5
+    # fmt: on
+
+    @property
+    def val(self) -> int:
+        """Gets the integer value of the verbosity level.
+        :return: The integer representation of the verbosity level.
+        """
+        return int(self.value)
+
+    def match(self, level: 'Verbosity') -> bool:
+        """Checks if the current verbosity level is less than or equal to the given level.
+        :param level: The verbosity level to compare against.
+        :return: True if the current level is less than or equal to the given level, otherwise False.
+        """
+        return self.val <= level.val
+

src/main/askai/core/model/ai_reply.py

@@ -14,44 +14,68 @@
 """
 
 from dataclasses import dataclass
-from typing import AnyStr, Literal, TypeAlias
+from typing import AnyStr
 
-Verbosity: TypeAlias = Literal[1, 2, 3, 4, 5]
+from askai.core.enums.verbosity import Verbosity
 
 
 @dataclass(frozen=True)
 class AIReply:
-    """Data class that represents AI replies."""
+    """Data class that represents AI replies.
+    :param message: The reply message.
+    :param is_success: Indicates whether the reply is successful.
+    :param is_debug: Indicates whether the reply is for debugging.
+    :param verbosity: The verbosity level of the reply.
+    :param is_speakable: Indicates whether the reply is speakable.
+    """
 
     message: str = ""
     is_success: bool = True
     is_debug: bool = False
-    verbosity: Verbosity = 1
+    verbosity: Verbosity = Verbosity.MINIMUM
     is_speakable: bool = True
 
     def __str__(self) -> str:
         return self.message
 
     @staticmethod
-    def info(message: AnyStr, verbosity: Verbosity = 1, speakable: bool = True) -> "AIReply":
-        """TODO"""
+    def info(message: AnyStr, verbosity: Verbosity = Verbosity.NORMAL, speakable: bool = True) -> "AIReply":
+        """Creates an info reply.
+        :param message: The reply message.
+        :param verbosity: The verbosity level of the reply.
+        :param speakable: Indicates whether the reply is speakable.
+        :return: An AIReply instance with info settings.
+        """
         return AIReply(str(message), True, False, verbosity, speakable)
 
     @staticmethod
     def error(message: AnyStr) -> "AIReply":
-        """TODO"""
-        return AIReply(str(message), True, False, 1, False)
+        """Creates an error reply.
+        :param message: The reply message.
+        :return: An AIReply instance with error settings.
+        """
+        return AIReply(str(message), False, False, Verbosity.NORMAL, False)
 
     @staticmethod
     def debug(message: AnyStr) -> "AIReply":
-        """TODO"""
-        return AIReply(str(message), True, True, 1, False)
+        """Creates a debug reply.
+        :param message: The reply message.
+        :return: An AIReply instance with debug settings.
+        """
+        return AIReply(str(message), True, True, Verbosity.NORMAL, False)
 
     @staticmethod
-    def mute(message: AnyStr, verbosity: Verbosity = 1) -> "AIReply":
-        """TODO"""
+    def mute(message: AnyStr, verbosity: Verbosity = Verbosity.NORMAL) -> "AIReply":
+        """Creates a mute reply.
+        :param message: The reply message.
+        :param verbosity: The verbosity level of the reply.
+        :return: An AIReply instance with mute settings.
+        """
         return AIReply(str(message), True, False, verbosity, False)
 
     @property
     def is_error(self) -> bool:
+        """Checks if the reply indicates an error.
+        :return: True if the reply is not successful, otherwise False.
+        """
         return not self.is_success