main (#137)

Reviewed-on: https://interstellardevelopment.org/code/code/React-Group/interstellar_ai/pulls/137
Reviewed-by: Patrick <patrick_pluto@noreply.localhost>
Co-authored-by: sageTheDM <info@photofuel.tech>
Co-committed-by: sageTheDM <info@photofuel.tech>

py/ai.py

@@ -4,10 +4,11 @@ import google.generativeai as genai
 import anthropic
 import ollama
 
-
 class AI:
     @staticmethod
     def process_local(model, messages, return_class, access_token):
+        """Process chat messages using the Ollama model locally."""
+        # Stream the chat response from the Ollama model
         stream = ollama.chat(
             model=model,
             messages=messages,
@@ -15,50 +16,61 @@ class AI:
             options={"temperature": 0.5},
         )
 
+        # Initialize the AI response for the given access token
         with return_class.ai_response_lock:
             return_class.ai_response[access_token] = ""
 
+        # Collect the response chunks and append to the response for the given access token
         for chunk in stream:
             with return_class.ai_response_lock:
                 return_class.ai_response[access_token] += chunk["message"]["content"]
 
     @staticmethod
     def process_mistralai(model, messages, return_class, access_token, api_key):
+        """Process chat messages using the Mistral AI model."""
         client = Mistral(api_key=api_key)
 
+        # Stream the chat response from the Mistral model
         stream_response = client.chat.stream(model=model, messages=messages)
 
+        # Initialize the AI response for the given access token
         with return_class.ai_response_lock:
             return_class.ai_response[access_token] = ""
 
+        # Collect the response chunks and append to the response for the given access token
         for chunk in stream_response:
             with return_class.ai_response_lock:
-                return_class.ai_response[access_token] += chunk.data.choices[
-                    0
-                ].delta.content
+                return_class.ai_response[access_token] += chunk.data.choices[0].delta.content
 
     @staticmethod
     def process_openai(model, messages, return_class, access_token, api_key):
+        """Process chat messages using the OpenAI model."""
         client = OpenAI(api_key=api_key)
 
+        # Stream the chat response from the OpenAI model
         stream_response = client.chat.completions.create(
             model=model, messages=messages, stream=True
         )
 
+        # Initialize the AI response for the given access token
         with return_class.ai_response_lock:
             return_class.ai_response[access_token] = ""
 
+        # Collect the response chunks and append to the response for the given access token
         for chunk in stream_response:
             with return_class.ai_response_lock:
                 return_class.ai_response[access_token] += chunk.choices[0].delta.content
 
     @staticmethod
     def process_anthropic(model, messages, return_class, access_token, api_key):
+        """Process chat messages using the Anthropic model."""
         client = anthropic.Anthropic(api_key=api_key)
 
+        # Initialize the AI response for the given access token
         with return_class.ai_response_lock:
             return_class.ai_response[access_token] = ""
 
+        # Stream the chat response from the Anthropic model
         with client.messages.stream(
             max_tokens=1024,
             model=model,
@@ -70,24 +82,27 @@ class AI:
 
     @staticmethod
     def process_google(model, messages, return_class, access_token, api_key):
-        message = messages[-1]["content"]
-        messages.pop()
+        """Process chat messages using the Google Generative AI model."""
+        message = messages[-1]["content"]  # Get the latest message content
+        messages.pop()  # Remove the latest message from the list
 
+        # Prepare messages for the Google Generative AI format
         for msg in messages:
             msg["parts"] = msg.pop()["content"]
 
+        # Change 'assistant' role to 'model' for compatibility
         for msg in messages:
             if msg["role"] == "assistant":
                 msg["role"] = "model"
 
+        # Configure the Google Generative AI client
         genai.configure(api_key=api_key)
 
+        # Start a chat session with the specified model and message history
         model = genai.GenerativeModel(model)
+        chat = model.start_chat(history=messages)
 
-        chat = model.start_chat(
-            history=messages,
-        )
-
+        # Send the message and stream the response
         response = chat.send_message(message, stream=True)
         for chunk in response:
             return_class.ai_response[access_token] += chunk.text

py/api.py

@@ -1,5 +1,4 @@
 from time import sleep
-
 from flask import Flask, request, jsonify
 from flask_cors import CORS
 import secrets
@@ -13,53 +12,58 @@ from tts import TTS
 
 class API:
     def __init__(self):
-        self.crypt_size = 64
-        self.app = Flask(__name__)
-        self.ai_response = {}
-        self.ai = AI()
-        self.db = DB()
-        self.weather = Weather()
-        self.voice = VoiceRecognition()
-        self.tts = TTS()
-        self.db.load_database()
-        self.ai_response_lock = threading.Lock()
-        CORS(self.app)
+        # Initialize the API class with necessary components and configurations
+        self.crypt_size = 64  # Size for generating secure tokens
+        self.app = Flask(__name__)  # Create a Flask app instance
+        self.ai_response = {}  # Dictionary to store AI responses keyed by access token
+        self.ai = AI()  # AI processing instance
+        self.db = DB()  # Database instance
+        self.weather = Weather()  # Weather data retrieval instance
+        self.voice = VoiceRecognition()  # Voice recognition instance
+        self.tts = TTS()  # Text-to-Speech instance
+        self.db.load_database()  # Load the database on startup
+        self.ai_response_lock = threading.Lock()  # Lock for thread-safe access to AI responses
+        CORS(self.app)  # Enable Cross-Origin Resource Sharing
 
     def run(self):
+        # Route to create a new AI session
         @self.app.route("/interstellar_ai/api/ai_create", methods=["GET"])
         def create_ai():
-            access_token = secrets.token_urlsafe(self.crypt_size)
+            access_token = secrets.token_urlsafe(self.crypt_size)  # Generate a secure access token
 
+            # Ensure the access token is unique
             while access_token in self.ai_response:
                 access_token = secrets.token_urlsafe(self.crypt_size)
 
-            self.ai_response[access_token] = ""
+            self.ai_response[access_token] = ""  # Initialize the response for the new session
             return jsonify({"status": 200, "access_token": access_token})
 
+        # Route to send messages to the AI
         @self.app.route("/interstellar_ai/api/ai_send", methods=["POST"])
         def send_ai():
-            data = request.get_json()
-            messages = data.get("messages")
-            model_type = data.get("model_type")
-            ai_model = data.get("ai_model")
-            access_token = data.get("access_token")
+            data = request.get_json()  # Get JSON data from the request
+            messages = data.get("messages")  # Extract messages
+            model_type = data.get("model_type")  # Extract model type
+            ai_model = data.get("ai_model")  # Extract AI model name
+            access_token = data.get("access_token")  # Extract access token
 
-            print(model_type)
+            print(model_type)  # Debugging output
             if access_token not in self.ai_response:
-                return jsonify({"status": 401, "error": "Invalid access token"})
+                return jsonify({"status": 401, "error": "Invalid access token"})  # Token validation
 
+            # Start a new thread to process AI requests based on the model type
             if model_type == "local":
                 thread = threading.Thread(
                     target=self.ai.process_local,
                     args=(ai_model, messages, self, access_token),
                 )
                 thread.start()
-                thread.join()
-                sleep(0.5)
+                thread.join()  # Wait for the thread to complete
+                sleep(0.5)  # Sleep for a short duration
                 return jsonify({"status": 200})
             elif model_type == "mistral":
-                print(model_type)
-                api_key = data.get("api_key")
+                print(model_type)  # Debugging output
+                api_key = data.get("api_key")  # Get API key
                 thread = threading.Thread(
                     target=self.ai.process_mistralai,
                     args=(ai_model, messages, self, access_token, api_key),
@@ -69,7 +73,7 @@ class API:
                 sleep(0.5)
                 return jsonify({"status": 200})
             elif model_type == "openai":
-                api_key = data.get("api_key")
+                api_key = data.get("api_key")  # Get API key
                 thread = threading.Thread(
                     target=self.ai.process_openai,
                     args=(ai_model, messages, self, access_token, api_key),
@@ -79,7 +83,7 @@ class API:
                 sleep(0.5)
                 return jsonify({"status": 200})
             elif model_type == "anthropic":
-                api_key = data.get("api_key")
+                api_key = data.get("api_key")  # Get API key
                 thread = threading.Thread(
                     target=self.ai.process_anthropic,
                     args=(ai_model, messages, self, access_token, api_key),
@@ -89,7 +93,7 @@ class API:
                 sleep(0.5)
                 return jsonify({"status": 200})
             elif model_type == "google":
-                api_key = data.get("api_key")
+                api_key = data.get("api_key")  # Get API key
                 thread = threading.Thread(
                     target=self.ai.process_google,
                     args=(ai_model, messages, self, access_token, api_key),
@@ -99,79 +103,85 @@ class API:
                 sleep(0.5)
                 return jsonify({"status": 200})
 
-            return jsonify({"status": 401, "error": "Invalid AI model type"})
+            return jsonify({"status": 401, "error": "Invalid AI model type"})  # Model type validation
 
+        # Route to retrieve AI response based on access token
         @self.app.route("/interstellar_ai/api/ai_get", methods=["GET"])
         def get_ai():
-            data = request.args.get("access_token")
+            data = request.args.get("access_token")  # Get access token from query parameters
             if data not in self.ai_response:
-                return jsonify({"status": 401, "error": "Invalid access token"})
-            return jsonify({"status": 200, "response": self.ai_response[data]})
+                return jsonify({"status": 401, "error": "Invalid access token"})  # Token validation
+            return jsonify({"status": 200, "response": self.ai_response[data]})  # Return AI response
 
+        # Route for database operations
         @self.app.route("/interstellar_ai/db", methods=["POST"])
         def db_manipulate():
-            sent_data = request.get_json()
-            action = sent_data.get("action")
+            sent_data = request.get_json()  # Get JSON data from the request
+            action = sent_data.get("action")  # Extract action type
             if action == "create_account":
-                return jsonify({"status": 200, "response": self.db.add_user(sent_data)})
+                return jsonify({"status": 200, "response": self.db.add_user(sent_data)})  # Add user
             elif action == "change_password":
                 return jsonify(
-                    {"status": 200, "response": self.db.update_password(sent_data)}
+                    {"status": 200, "response": self.db.update_password(sent_data)}  # Update password
                 )
             elif action == "get_settings":
                 return jsonify(
-                    {"status": 200, "response": self.db.get_settings(sent_data)}
+                    {"status": 200, "response": self.db.get_settings(sent_data)}  # Get user settings
                 )
             elif action == "change_settings":
                 return jsonify(
-                    {"status": 200, "response": self.db.change_settings(sent_data)}
+                    {"status": 200, "response": self.db.change_settings(sent_data)}  # Change user settings
                 )
             elif action == "get_history":
                 return jsonify(
-                    {"status": 200, "response": self.db.get_history(sent_data)}
+                    {"status": 200, "response": self.db.get_history(sent_data)}  # Get user history
                 )
             elif action == "change_history":
                 return jsonify(
-                    {"status": 200, "response": self.db.change_history(sent_data)}
+                    {"status": 200, "response": self.db.change_history(sent_data)}  # Change user history
                 )
             elif action == "check_credentials":
                 return jsonify(
-                    {"status": 200, "response": self.db.check_credentials(sent_data)}
+                    {"status": 200, "response": self.db.check_credentials(sent_data)}  # Check user credentials
                 )
             elif action == "delete_account":
                 return jsonify(
-                    {"status": 200, "response": self.db.delete_user(sent_data)}
+                    {"status": 200, "response": self.db.delete_user(sent_data)}  # Delete user account
                 )
             elif action == "get_email":
                 return jsonify(
-                    {"status": 200, "response": self.db.get_email(sent_data)}
+                    {"status": 200, "response": self.db.get_email(sent_data)}  # Get user email
                 )
             elif action == "get_name":
-                return jsonify({"status": 200, "response": self.db.get_name(sent_data)})
+                return jsonify({"status": 200, "response": self.db.get_name(sent_data)})  # Get user name
 
-            return jsonify({"status": 401, "response": "Invalid action"})
+            return jsonify({"status": 401, "response": "Invalid action"})  # Action validation
 
+        # Route for voice recognition
         @self.app.route("/interstellar_ai/api/voice_recognition", methods=["POST"])
         def voice_recognition():
-            audio = request.files.get("audio")
-            text = self.voice.recognition(audio)
-            return jsonify({"status": 200, "response": text})
+            audio = request.files.get("audio")  # Get audio file from request
+            text = self.voice.recognition(audio)  # Perform voice recognition
+            return jsonify({"status": 200, "response": text})  # Return recognized text
 
+        # Route for weather information retrieval
         @self.app.route("/interstellar_ai/api/weather", methods=["POST"])
         def get_weather():
-            sent_data = request.get_json()
-            unit_type = sent_data.get("unit_type")
-            city = sent_data.get("city")
-            weather_data = self.weather.getweather(unit_type, city)
-            return jsonify({"status": 200, "response": weather_data})
+            sent_data = request.get_json()  # Get JSON data from the request
+            unit_type = sent_data.get("unit_type")  # Extract unit type (metric, imperial)
+            city = sent_data.get("city")  # Extract city name
+            weather_data = self.weather.getweather(unit_type, city)  # Get weather data
+            return jsonify({"status": 200, "response": weather_data})  # Return weather data
 
-        self.app.run(debug=True, host="0.0.0.0", port=5000)
+        self.app.run(debug=True, host="0.0.0.0", port=5000)  # Start the Flask app
 
+        # Route for Text-to-Speech conversion
         @self.app.route("/interstellar_ai/api/tts", methods=["POST"])
         def tts():
-            text = request.args.get("text")
-            return jsonify({"status": 200, "response": self.tts.gen_tts(text)})
+            text = request.args.get("text")  # Get text from query parameters
+            return jsonify({"status": 200, "response": self.tts.gen_tts(text)})  # Generate TTS and return response
 
 
+# Initialize the API class and run the application
 api = API()
 api.run()

py/db.py

@@ -3,12 +3,13 @@ import json
 import os
 import pycouchdb
 
-
 class DB:
     def __init__(self):
+        # Initialize the database dictionary to store user data
         self.database = {}
 
     def ensure_username(self, data):
+        # Ensure a username can be retrieved either from username or email
         if "username" in data:
             return data.get("username")
         elif "email" in data:
@@ -18,15 +19,17 @@ class DB:
 
     @staticmethod
     def hash_password(password):
-        salt = "your_secret_salt"
+        # Hash the password with a salt for secure storage
+        salt = "your_secret_salt"  # Consider using a secure random salt
         hashed_password = hashlib.sha256((password + salt).encode()).hexdigest()
         return hashed_password
 
     def add_user(self, data):
+        # Add a new user to the database if username is unique
         username = data.get("username")
         password = data.get("password")
         email = data.get("email")
-        hashed_password = self.hash_password(password)
+        hashed_password = self.hash_password(password)  # Hash the password
         user_data = {
             "hashed_password": hashed_password,
             "email": email,
@@ -35,112 +38,123 @@ class DB:
         }
         if username not in self.database:
             self.database[username] = user_data
-            self.save_database()
+            self.save_database()  # Save changes to the database
             return True
-        return False
+        return False  # User already exists
 
     def delete_user(self, data):
+        # Delete a user from the database if credentials are valid
         username = self.ensure_username(data)
         if not self.check_credentials(data):
-            return False
+            return False  # Invalid credentials
 
-        del self.database[username]
-        self.save_database()
+        del self.database[username]  # Remove user from database
+        self.save_database()  # Save changes
         return True
 
     def update_password(self, data):
+        # Update the user's password if credentials are valid
         username = self.ensure_username(data)
         new_password = data.get("new_password")
         if not self.check_credentials(data):
-            return False
+            return False  # Invalid credentials
 
-        hashed_new_password = self.hash_password(new_password)
+        hashed_new_password = self.hash_password(new_password)  # Hash the new password
         self.database[username].update({"hashed_password": hashed_new_password})
-        self.save_database()
+        self.save_database()  # Save changes
         return True
 
     def check_credentials(self, data):
+        # Verify if provided credentials match stored data
         username = self.ensure_username(data)
         password = data.get("password")
         if username not in self.database:
-            return False
+            return False  # User not found
 
         stored_hashed_password = self.database[username]["hashed_password"]
         entered_hashed_password = self.hash_password(password)
-        return stored_hashed_password == entered_hashed_password
+        return stored_hashed_password == entered_hashed_password  # Check hashed password
 
     def change_settings(self, data):
+        # Change user settings if credentials are valid
         username = self.ensure_username(data)
         if not self.check_credentials(data):
-            return False
+            return False  # Invalid credentials
 
-        self.database[username]["settings"] = data.get("data")
-        self.save_database()
+        self.database[username]["settings"] = data.get("data")  # Update settings
+        self.save_database()  # Save changes
         return True
 
     def get_settings(self, data):
+        # Retrieve user settings if credentials are valid
         username = self.ensure_username(data)
         if not self.check_credentials(data):
-            return None
+            return None  # Invalid credentials
 
-        send_back = self.database[username].get("settings")
+        send_back = self.database[username].get("settings")  # Get settings
         return send_back
 
     def change_history(self, data):
+        # Change user history if credentials are valid
         username = self.ensure_username(data)
         if not self.check_credentials(data):
-            return False
+            return False  # Invalid credentials
 
-        self.database[username]["history"] = data.get("data")
-        self.save_database()
+        self.database[username]["history"] = data.get("data")  # Update history
+        self.save_database()  # Save changes
         return True
 
     def get_history(self, data):
+        # Retrieve user history if credentials are valid
         username = self.ensure_username(data)
         if not self.check_credentials(data):
-            return None
+            return None  # Invalid credentials
 
-        send_back = self.database[username].get("history")
+        send_back = self.database[username].get("history")  # Get history
         return send_back
 
     def get_email(self, data):
+        # Retrieve user email if credentials are valid
         username = self.ensure_username(data)
         if not self.check_credentials(data):
-            return None
+            return None  # Invalid credentials
 
-        send_back = self.database[username].get("email")
+        send_back = self.database[username].get("email")  # Get email
         return send_back
 
     def get_name(self, data):
+        # Retrieve username if credentials are valid
         if not self.check_credentials(data):
-            return None
+            return None  # Invalid credentials
 
-        send_back = self.ensure_username(data)
+        send_back = self.ensure_username(data)  # Get username
         return send_back
 
     def save_database(self):
+        # Save the database to the specified storage (CouchDB or JSON file)
         if os.environ.get("PRODUCTION") == "YES":
             server = pycouchdb.Server("http://admin:admin@localhost:5984/")
             db = server.database("interstellar_ai")
-            db.save(self.database)
+            db.save(self.database)  # Save to CouchDB
 
         else:
             with open("database.json", "w") as file:
-                json.dump(self.database, file)
+                json.dump(self.database, file)  # Save to JSON file
 
     def load_database(self):
+        # Load the database from the specified storage (CouchDB or JSON file)
         if os.environ.get("PRODUCTION") == "YES":
             server = pycouchdb.Server("http://admin:admin@localhost:5984/")
             db = server.database("interstellar_ai")
             if db:
-                self.database = db
+                self.database = db  # Load from CouchDB
             else:
-                server.create("interstellar_ai")
+                server.create("interstellar_ai")  # Create database if it doesn't exist
                 db = server.database("interstellar_ai")
-                db.save(self.database)
+                db.save(self.database)  # Save initial empty database
         else:
             try:
                 with open("database.json", "r") as file:
-                    self.database = json.load(file)
+                    self.database = json.load(file)  # Load from JSON file
             except FileNotFoundError:
-                pass
+                pass  # File not found, do nothing

py/tts.py

@@ -1,6 +1,5 @@
 import pyttsx3
 
-
 class TTS:
     @staticmethod
     def gen_tts(text):

py/voice.py

@@ -6,20 +6,28 @@ from pydub import AudioSegment
 class VoiceRecognition:
     @staticmethod
     def recognition(audio):
+        # Read the audio file into a BytesIO buffer
         audio_buffer = io.BytesIO(audio.read())
 
+        # Load the audio file using pydub
         audio_segment = AudioSegment.from_file(audio_buffer, format="ogg")
 
+        # Export the audio to a WAV format in a BytesIO buffer
         wav_io = io.BytesIO()
         audio_segment.export(wav_io, format="wav")
-        wav_io.seek(0)
+        wav_io.seek(0)  # Reset the buffer pointer to the start
 
-        model_size = "base"
+        # Load the Whisper model
+        model_size = "base"  # Specify the model size
         model = WhisperModel(model_size, device="cpu", compute_type="int8")
 
+        # Transcribe the audio
         segments, _ = model.transcribe(wav_io)
         transcription = ""
+
+        # Combine the transcribed segments into a single string
         for segment in segments:
             transcription += segment.text + " "
-        result = transcription.strip()
+        
+        result = transcription.strip()  # Strip any leading/trailing whitespace
         return result

py/weather.py

@@ -15,6 +15,7 @@ class Weather:
             async with python_weather.Client(unit=unit_type) as client:
                 weather = await client.get(city)
 
+                # Collect weather data
                 data = {
                     "temperature": weather.temperature,
                     "humidity": weather.humidity,