feat(akande): 🎨 v0.0.4

Major refactoring and optimisations

akande/__init__.py

@@ -14,4 +14,4 @@
 # limitations under the License.
 #
 """The Python Akande module."""
-__version__ = "0.0.3"
+__version__ = "0.0.4"

akande/__main__.py

@@ -43,7 +43,13 @@ async def main():
 
     openai_service = OpenAIImpl()
     akande = Akande(openai_service=openai_service)
-    await akande.run_interaction()
+    try:
+        await akande.run_interaction()
+    except KeyboardInterrupt:
+        logging.info("Keyboard interrupt detected. Exiting...")
+        # Perform any necessary cleanup tasks here
+        # For example, stop the CherryPy server if it's running
+        await akande.stop_server()
 
 
 if __name__ == "__main__":

akande/akande.py

@@ -13,9 +13,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+import cherrypy
 from .cache import SQLiteCache
 from .config import OPENAI_DEFAULT_MODEL
 from .services import OpenAIService
+
 from .utils import generate_pdf, generate_csv
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
@@ -25,7 +27,20 @@
 import hashlib
 import logging
 import pyttsx4
+import threading
 import speech_recognition as sr
+import os
+
+
+# Define ANSI escape codes for colors
+class Colors:
+    RESET = "\033[0m"
+    HEADER = "\033[95m"
+    RED_BACKGROUND = "\033[48;2;179;0;15m"
+    CYAN_BACKGROUND = "\033[48;2;65;175;220m"
+    GREEN_BACKGROUND = "\033[48;2;0;103;0m"
+    BLUE_BACKGROUND = "\033[48;2;0;78;203m"
+    ORANGE_BACKGROUND = "\033[48;2;150;61;0m"
 
 
 class Akande:
@@ -45,6 +60,11 @@ def __init__(self, openai_service: OpenAIService):
             openai_service (OpenAIService): The OpenAI service for
             generating responses.
         """
+        # Initialize the CherryPy server attribute
+        self.server = None
+        self.server_thread = None
+        self.server_running = False
+
         # Create a directory path with the current date
         date_str = datetime.now().strftime("%Y-%m-%d")
         directory_path = Path(date_str)
@@ -161,42 +181,87 @@ async def listen(self) -> str:
     async def run_interaction(self) -> None:
         """Main interaction loop of the voice assistant."""
         while True:
-            welcome_msg = (
-                "\nWelcome to Àkàndé, your AI voice assistant.\n"
-            )
-            instructions = (
-                "\nPress Enter to use voice or type "
-                "your question and press Enter:\n"
-            )
-            choice = (
-                input(welcome_msg + instructions).strip().lower()
-            )  # Normalize input to lower case immediately
+            os.system(
+                "clear"
+            )  # Clear the console for better visualization
+            banner_text = "Àkàndé Voice Assistant"
+            banner_width = len(banner_text) + 4
+            print(f"{Colors.RESET}{' ' * banner_width}")
+            print("  " + banner_text + "  ")
+            print(" " * banner_width + Colors.RESET)
+
+            options = [
+                ("1. Use voice", Colors.BLUE_BACKGROUND),
+                ("2. Ask a question", Colors.GREEN_BACKGROUND),
+                ("3. Start server", Colors.ORANGE_BACKGROUND),
+                ("4. Stop", Colors.RED_BACKGROUND),
+            ]
+
+            for option_text, color in options:
+                print(f"{color}{' ' * banner_width}{Colors.RESET}")
+                print(
+                    f"{color}{option_text:<{banner_width}}{Colors.RESET}"
+                )
+                print(f"{color}{' ' * banner_width}{Colors.RESET}")
+
+            choice = input("\nPlease select an option: ").strip()
 
-            if choice == "stop":
+            if choice == "4":
                 print("\nGoodbye!")
+                await self.stop_server()
                 break
-
-            if choice:
-                prompt = choice
-            else:
+            elif choice == "3":
+                await self.run_server()
+            elif choice == "2":
+                question = input("Please enter your question: ").strip()
+                if question:
+                    print("Processing question...")
+                    response = await self.generate_response(question)
+                    await self.speak(response)
+                    await generate_pdf(question, response)
+                    await generate_csv(question, response)
+                else:
+                    print("No question provided.")
+            elif choice == "1":
                 print("Listening...")
                 prompt = (await self.listen()).lower()
                 if prompt == "stop":
                     print("\nGoodbye!")
+                    await self.stop_server()
                     break
+                elif prompt:
+                    print("Processing voice command...")
+                    response = await self.generate_response(prompt)
+                    await self.speak(response)
+                    await generate_pdf(prompt, response)
+                    await generate_csv(prompt, response)
+                else:
+                    print("No voice command detected.")
+            else:
+                print("Invalid choice. Please select a valid option.")
 
-            if prompt and prompt not in [
-                "stop voice",
-                "stop text",
-                "thank you for your help",
-            ]:
-                response = await self.generate_response(prompt)
-                await self.speak(response)
-                await generate_pdf(prompt, response)
-                await generate_csv(prompt, response)
-            elif prompt == "thank you for your help":
-                await self.speak("You're welcome. Goodbye!")
-                break
+    async def run_server(self) -> None:
+        """Run the CherryPy server in a separate thread."""
+
+        # Define a function to start the server
+        def start_server():
+            from .server.server import AkandeServer
+
+            cherrypy.quickstart(AkandeServer())
+
+            # Set server_running flag to True
+            self.server_running = True
+
+        # Start the server in a separate thread
+        server_thread = threading.Thread(target=start_server)
+        server_thread.start()
+        logging.info("CherryPy server started.")
+
+    async def stop_server(self) -> None:
+        """Stop the CherryPy server."""
+        self.server_running = False
+        cherrypy.engine.exit()
+        logging.info("CherryPy server stopped.")
 
     async def generate_response(self, prompt: str) -> str:
         """

akande/server/server.py

@@ -0,0 +1,135 @@
+import cherrypy
+import json
+import logging
+import os
+import io
+import speech_recognition as sr
+from pydub import AudioSegment
+from pydub.exceptions import CouldntDecodeError
+from akande.config import OPENAI_DEFAULT_MODEL
+from akande.services import OpenAIImpl
+
+
+class AkandeServer:
+    def __init__(self):
+        self.openai_service = OpenAIImpl()
+        self.logger = logging.getLogger(__name__)
+
+    @cherrypy.expose
+    def index(self):
+        return open("./public/index.html")
+
+    @cherrypy.expose
+    def static(self, path):
+        return open(f"./public/{path}")
+
+    @cherrypy.expose
+    def process_question(self):
+        try:
+            request_data = json.loads(cherrypy.request.body.read())
+            question = request_data.get("question")
+            self.logger.info(f"Received question: {question}")
+
+            response_object = (
+                self.openai_service.generate_response_sync(
+                    question, OPENAI_DEFAULT_MODEL, None
+                )
+            )
+            message_content = response_object.choices[0].message.content
+            return json.dumps({"response": message_content})
+
+        except Exception as e:
+            self.logger.error(f"Failed to process question: {e}")
+            return json.dumps({"response": "An error occurred"})
+
+    @cherrypy.expose
+    @cherrypy.tools.allow(methods=["POST"])
+    def process_audio_question(self):
+        try:
+            audio_data = cherrypy.request.body.read()
+            wav_file_path = self.convert_to_wav(audio_data)
+            processed_result = self.process_audio(wav_file_path)
+
+            question_data = {
+                "response": "Audio data processed successfully",
+                "result": processed_result,
+            }
+
+            question = question_data.get("result").get("text")
+            response_object = (
+                self.openai_service.generate_response_sync(
+                    question, OPENAI_DEFAULT_MODEL, None
+                )
+            )
+
+            if os.path.exists(wav_file_path):
+                os.remove(wav_file_path)
+                self.logger.info(f"WAV file removed: {wav_file_path}")
+
+            message_content = response_object.choices[0].message.content
+            return json.dumps({"response": message_content})
+
+        except Exception as e:
+            self.logger.error("Failed to process audio:", exc_info=True)
+            cherrypy.response.status = 500
+            return json.dumps(
+                {"error": "Failed to process audio", "details": str(e)}
+            ).encode("utf-8")
+
+    @staticmethod
+    def convert_to_wav(audio_data):
+        try:
+            for input_format in ["webm", "mp3", "mp4", "ogg", "flac"]:
+                try:
+                    audio_segment = AudioSegment.from_file(
+                        io.BytesIO(audio_data), format=input_format
+                    )
+                    break
+                except CouldntDecodeError:
+                    pass
+
+            else:
+                raise ValueError("Unsupported audio format")
+
+            audio_segment = audio_segment.set_channels(
+                1
+            ).set_frame_rate(16000)
+            directory_path = "./"
+            filename = "audio.wav"
+            file_path = os.path.join(directory_path, filename)
+            audio_segment.export(file_path, format="wav")
+            return file_path
+
+        except Exception as e:
+            raise RuntimeError(f"Error converting audio: {e}")
+
+    @staticmethod
+    def process_audio(file_path):
+        try:
+            recognizer = sr.Recognizer()
+            with sr.AudioFile(file_path) as source:
+                audio_data = recognizer.record(source)
+
+            text = recognizer.recognize_google(audio_data)
+            return {"text": text, "success": True}
+
+        except sr.UnknownValueError:
+            return {
+                "error": "Audio could not be understood",
+                "success": False,
+            }
+        except sr.RequestError as e:
+            return {
+                "error": f"Speech recognition service error {e}",
+                "success": False,
+            }
+
+
+def main():
+    logging.basicConfig(level=logging.INFO)
+    cherrypy.config.update({"server.socket_port": 8080})
+    cherrypy.quickstart(AkandeServer())
+
+
+if __name__ == "__main__":
+    main()

akande/services.py

@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-from abc import ABC, abstractmethod
+from abc import ABC
 import asyncio
 import logging
 from typing import Any, Dict
@@ -24,7 +24,6 @@
 class OpenAIService(ABC):
     """Base class for OpenAI services."""
 
-    @abstractmethod
     async def generate_response(
         self, prompt: str, model: str, params: Dict[str, Any]
     ) -> Dict[str, Any]:
@@ -94,3 +93,12 @@ async def generate_response(
         except Exception as exc:
             logging.error("OpenAI API error: %s", exc)
             return {"error": str(exc)}
+
+    def generate_response_sync(
+        self, user_prompt, model=OPENAI_DEFAULT_MODEL, params=None
+    ):
+        response = asyncio.run(
+            self.generate_response(user_prompt, model, params)
+        )
+        logging.info(f"Generated response: {response}")
+        return response