Improve initial recorder device select

Author: yorevs

src/demo/components/vision_demo.py

@@ -1,11 +1,13 @@
-from askai.core.engine.ai_vision import AIVision
-from askai.core.support.shared_instances import shared
-from utils import init_context
+import os
+
+from askai.core.features.router.tools.vision import offline_captioner
 
 if __name__ == "__main__":
-    init_context("vision-demo")
-    vision: AIVision = shared.engine.vision()
+    # init_context("vision-demo")
+    # vision: AIVision = shared.engine.vision()
     load_dir: str = "${HOME}/.config/hhs/askai/cache/pictures/photos"
     image_file: str = "eu-edvaldo-suecia.jpg"
-    result = vision.caption(image_file, load_dir)
-    print(result)
+    # result = vision.caption(image_file, load_dir)
+    # print(result)
+    result2 = offline_captioner(os.path.join(load_dir, image_file))
+    print(result2)

src/main/askai/__classpath__.py

@@ -12,25 +12,31 @@
 
    Copyright (c) 2024, HomeSetup
 """
-
-from askai.core.model.api_keys import ApiKeys
-from hspylib.core.metaclass.classpath import Classpath
-from hspylib.core.tools.commons import parent_path, root_dir
-
 import logging as log
 import os
-import pydantic
 import sys
 import warnings
 
-warnings.simplefilter(action="ignore", category=FutureWarning)
+import pydantic
+from clitt.core.term.commons import is_a_tty
+from hspylib.core.metaclass.classpath import Classpath
+from hspylib.core.tools.commons import parent_path, root_dir, is_debugging
+
+from askai.core.model.api_keys import ApiKeys
+
+if not is_debugging():
+    warnings.simplefilter("ignore", category=FutureWarning)
+    warnings.simplefilter("ignore", category=UserWarning)
+    warnings.simplefilter("ignore", category=DeprecationWarning)
+
+if not is_a_tty():
+    log.getLogger().setLevel(log.ERROR)
 
 if not os.environ.get("USER_AGENT"):
     # The AskAI User Agent, required by the langchain framework
     ASKAI_USER_AGENT: str = "AskAI-User-Agent"
     os.environ["USER_AGENT"] = ASKAI_USER_AGENT
 
-
 try:
     API_KEYS: ApiKeys = ApiKeys()
 except pydantic.v1.error_wrappers.ValidationError as err:
@@ -40,7 +46,7 @@
 
 
 class _Classpath(Classpath):
-    """TODO"""
+    """A class for managing classpath-related operations. Uses the Classpath metaclass."""
 
     def __init__(self):
         super().__init__(parent_path(__file__), parent_path(root_dir()), (parent_path(__file__) / "resources"))

src/main/askai/__main__.py

@@ -12,13 +12,12 @@
 
    Copyright (c) 2024, HomeSetup
 """
-from askai.__classpath__ import classpath
-from askai.core.askai import AskAi
-from askai.core.askai_cli import AskAiCli
-from askai.core.askai_configs import configs
-from askai.core.support.shared_instances import shared
-from askai.tui.askai_app import AskAiApp
-from clitt.core.term.commons import is_a_tty
+import logging as log
+import os
+import sys
+from textwrap import dedent
+from typing import Any, Optional
+
 from clitt.core.tui.tui_application import TUIApplication
 from hspylib.core.enums.charset import Charset
 from hspylib.core.tools.commons import to_bool
@@ -27,15 +26,13 @@
 from hspylib.modules.application.argparse.parser_action import ParserAction
 from hspylib.modules.application.exit_status import ExitStatus
 from hspylib.modules.application.version import Version
-from textwrap import dedent
-from typing import Any, Optional
-
-import logging as log
-import os
-import sys
 
-if not is_a_tty():
-    log.getLogger().setLevel(log.ERROR)
+from askai.__classpath__ import classpath
+from askai.core.askai import AskAi
+from askai.core.askai_cli import AskAiCli
+from askai.core.askai_configs import configs
+from askai.core.support.shared_instances import shared
+from askai.tui.askai_app import AskAiApp
 
 
 class Main(TUIApplication):

src/main/askai/core/askai_cli.py

@@ -12,6 +12,21 @@
 
    Copyright (c) 2024, HomeSetup
 """
+import logging as log
+import os
+from functools import partial
+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 *
@@ -24,20 +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 functools import partial
-from hspylib.modules.eventbus.event import Event
-from pathlib import Path
-from rich.progress import Progress
-from threading import Thread
-from typing import List, TypeAlias
-
-import logging as log
-import nltk
-import os
-import pause
 
 QueryString: TypeAlias = str | List[str] | None
 
@@ -161,7 +162,7 @@ def _startup(self) -> None:
         # List of tasks for progress tracking
         tasks = [
             "Downloading nltk data",
-            "Preloading input history",
+            "Loading input history",
             "Starting scheduler",
             "Setting up recorder",
             "Starting player delay",
@@ -180,7 +181,7 @@ def _startup(self) -> None:
                 self._progress.update(task, advance=1, description="[green]Downloading nltk data")
                 nltk.download("averaged_perceptron_tagger", quiet=True, download_dir=CACHE_DIR)
                 cache.cache_enable = configs.is_cache
-                self._progress.update(task, advance=1, description="[green]Preloading input history")
+                self._progress.update(task, advance=1, description="[green]Loading input history")
                 KeyboardInput.preload_history(cache.load_input_history(commands()))
                 self._progress.update(task, advance=1, description="[green]Starting scheduler")
                 scheduler.start()

src/main/askai/core/commander/commands/tts_stt_cmd.py

@@ -12,17 +12,20 @@
 
    Copyright (c) 2024, HomeSetup
 """
+import os
 from abc import ABC
+from pathlib import Path
+
+import pause
+from clitt.core.tui.mselect.mselect import mselect
+
 from askai.core.askai_configs import configs
 from askai.core.askai_settings import settings
 from askai.core.component.audio_player import player
-from askai.core.component.recorder import recorder
+from askai.core.component.recorder import recorder, InputDevice
 from askai.core.support.shared_instances import shared
 from askai.core.support.text_formatter import text_formatter
 from askai.core.support.utilities import copy_file
-from pathlib import Path
-
-import os
 
 
 class TtsSttCmd(ABC):
@@ -101,15 +104,25 @@ def device_set(name_or_index: str | int | None = None) -> None:
         :param name_or_index: The name or index of the audio input device to set. If None, the default device will be
                               used.
         """
+        device: InputDevice | None = None
         all_devices = recorder.devices
-        if name_or_index.isdecimal() and 0 <= int(name_or_index) <= len(all_devices):
+
+        def _set_device(_device) -> bool:
+            if recorder.set_device(_device):
+                text_formatter.cmd_print(f"`Text-To-Speech` device changed to %GREEN%{_device}%NC%")
+                return True
+            text_formatter.cmd_print(f"%HOM%%ED2%Error: '{_device}' is not an Audio Input device!%NC%")
+            all_devices.remove(_device)
+            pause.seconds(2)
+            return False
+
+        if not name_or_index:
+            device: InputDevice = mselect(
+                all_devices, f"{'-=' * 40}%EOL%AskAI::Select the Audio Input device%EOL%{'=-' * 40}%EOL%")
+        elif name_or_index.isdecimal() and 0 <= int(name_or_index) <= len(all_devices):
             name_or_index = all_devices[int(name_or_index)][1]
-        if device := next((dev for dev in all_devices if dev[1] == name_or_index), None):
-            if recorder.set_device(device):
-                text_formatter.cmd_print(f"`Text-To-Speech` device changed to %GREEN%{device[1]}%NC%")
-            else:
-                text_formatter.cmd_print(f"%RED%Device: '{name_or_index}' failed to initialize!%NC%")
-        else:
+            device = next((dev for dev in all_devices if dev[1] == name_or_index), None)
+        if not (device and _set_device(device)):
             text_formatter.cmd_print(f"%RED%Invalid audio input device: '{name_or_index}'%NC%")
 
     @staticmethod

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

@@ -12,29 +12,28 @@
 
    Copyright (c) 2024, HomeSetup
 """
-from askai.core.askai_configs import configs
-from askai.core.askai_events import events
-from askai.core.askai_messages import msg
-from askai.core.component.cache_service import REC_DIR
-from askai.core.component.scheduler import Scheduler
-from askai.core.support.utilities import display_text, seconds
-from askai.exception.exceptions import InvalidInputDevice, InvalidRecognitionApiError
-from askai.language.language import Language
-from clitt.core.tui.mselect.mselect import mselect
+import logging as log
+import operator
+import sys
+from pathlib import Path
+from typing import Callable, Optional, TypeAlias
+
 from hspylib.core.enums.enumeration import Enumeration
 from hspylib.core.metaclass.classpath import AnyPath
 from hspylib.core.metaclass.singleton import Singleton
 from hspylib.core.preconditions import check_argument, check_state
 from hspylib.core.zoned_datetime import now_ms
 from hspylib.modules.application.exit_status import ExitStatus
-from pathlib import Path
 from speech_recognition import AudioData, Microphone, Recognizer, RequestError, UnknownValueError, WaitTimeoutError
-from typing import Callable, Optional, TypeAlias
 
-import logging as log
-import operator
-import pause
-import sys
+from askai.core.askai_configs import configs
+from askai.core.askai_events import events
+from askai.core.askai_messages import msg
+from askai.core.component.cache_service import REC_DIR
+from askai.core.component.scheduler import Scheduler
+from askai.core.support.utilities import seconds, display_text
+from askai.exception.exceptions import InvalidInputDevice, InvalidRecognitionApiError
+from askai.language.language import Language
 
 InputDevice: TypeAlias = tuple[int, str]
 
@@ -240,23 +239,19 @@ def _select_device(self) -> None:
         available: list[str] = list(filter(lambda d: d, map(str.strip, configs.recorder_devices)))
         device: InputDevice | None = None
         devices: list[InputDevice] = list(reversed(self.devices))
-        while not device:
+        while devices and not device:
             if available:
                 for dev in devices:
                     if dev[1] in available and self.set_device(dev):
                         device = dev
                         break
             if not device:
-                device: InputDevice = mselect(
-                    devices, f"{'-=' * 40}%EOL%AskAI::Select the Audio Input device%EOL%{'=-' * 40}%EOL%"
-                )
-                if not device:
+                if not (device := next(devices.__iter__(), None)):
+                    display_text(f"%HOM%%ED2%Error: Unable to setup an Audio Input device!%NC%")
                     sys.exit(ExitStatus.FAILED.val)
-                elif not self.set_device(device):
-                    display_text(f"%HOM%%ED2%Error: '{device[1]}' is not an Audio Input device!%NC%")
-                    devices.remove(device)
-                    device = None
-                    pause.seconds(2)
+            if device and not self.set_device(device):
+                devices.remove(device)
+                device = None
 
 
 assert (recorder := Recorder().INSTANCE) is not None

src/main/askai/core/features/router/tools/vision.py

@@ -1,39 +1,41 @@
+import os
+from textwrap import indent
+
+import torch
+from PIL import Image
+from hspylib.core.config.path_object import PathObject
+from hspylib.core.enums.enumeration import Enumeration
+from hspylib.core.metaclass.classpath import AnyPath
+from transformers import BlipForConditionalGeneration, BlipProcessor
+
 from askai.core.askai_events import events
 from askai.core.askai_messages import msg
 from askai.core.component.cache_service import PICTURE_DIR
 from askai.core.engine.ai_vision import AIVision
 from askai.core.features.validation.accuracy import resolve_x_refs
 from askai.core.model.image_result import ImageResult
 from askai.core.support.shared_instances import shared
-from hspylib.core.config.path_object import PathObject
-from hspylib.core.enums.enumeration import Enumeration
-from hspylib.core.metaclass.classpath import AnyPath
-from PIL import Image
-from textwrap import indent
-from transformers import BlipForConditionalGeneration, BlipProcessor
 
-import os
-import torch
+
+class HFModel(Enumeration):
+    """Available Hugging Face models"""
+
+    # fmt: off
+    SF_BLIP_BASE    = "Salesforce/blip-image-captioning-base"
+    SF_BLIP_LARGE   = "Salesforce/blip-image-captioning-large"
+    # fmt: on
+
+    @staticmethod
+    def default() -> "HFModel":
+        """Return the default HF model."""
+        return HFModel.SF_BLIP_BASE
 
 
 def offline_captioner(path_name: AnyPath) -> str:
     """This tool is used to describe an image.
     :param path_name: The path of the image to describe.
     """
 
-    class HFModel(Enumeration):
-        """Available Hugging Face models"""
-
-        # fmt: off
-        SF_BLIP_BASE            = "Salesforce/blip-image-captioning-base"
-        SF_BLIP_LARGE           = "Salesforce/blip-image-captioning-large"
-        # fmt: on
-
-        @staticmethod
-        def default() -> "HFModel":
-            """Return the default HF model."""
-            return HFModel.SF_BLIP_LARGE
-
     caption: str = "Not available"
 
     posix_path: PathObject = PathObject.of(path_name)
@@ -46,12 +48,16 @@ def default() -> "HFModel":
 
     if posix_path.exists:
         events.reply.emit(message=msg.describe_image(str(posix_path)))
-        hf_model: HFModel = HFModel.default()
         # Use GPU if it's available
         device = "cuda" if torch.cuda.is_available() else "cpu"
         image = Image.open(str(posix_path)).convert("RGB")
-        model = BlipForConditionalGeneration.from_pretrained(hf_model.value).to(device)
-        processor = BlipProcessor.from_pretrained(hf_model.value)
+        model_id: str = HFModel.default().value
+        match model_id.casefold():
+            case model if "blip-" in model:
+                model = BlipForConditionalGeneration.from_pretrained(model_id).to(device)
+                processor = BlipProcessor.from_pretrained(model_id)
+            case _:
+                raise ValueError(f"Unsupported model: '{model_id}'")
         inputs = processor(images=image, return_tensors="pt").to(device)
         outputs = model.generate(**inputs)
         caption = processor.decode(outputs[0], skip_special_tokens=True)