ready

Author: sudo-junaiddev

.gitignore

@@ -1,2 +1,3 @@
 __pycache__/
-venv/
+venv/
+test.py

README.md

@@ -1,3 +1,19 @@
-# UNDER DEVELOPMENT
+# Simple AI UI
+Simple AI UI is a simple , minimalist and lightweight web GUI. It offers an server also to host your own AI in your local network or on Internet.
+
+### Features
+- Robust API infrastructure.
+- Simple , minimalist and lightweight UI
+- Privacy (Your data is stored in your computer)
+
+### Supported Platforms for Hosting
+- Windows
+- Mac
+- Linux
+**NOTE :** Ollama must be installed on the system.
+
+### Flavors
+- Python
+- Rust-lang (coming soon)
+
 
-Made By Junaid (www.abujuni.dev)

app.py

@@ -1,22 +1,23 @@
 from flask import Flask, render_template, jsonify, request, send_from_directory
 import requests
-from ollamaClient import OllamaClient
 import threading
 import queue
+from ollamaClient import OllamaClient
 
 # Initialize Flask app and Ollama client
 app = Flask(__name__)
 ai = OllamaClient()
 
-# Queue for handling requests
-request_queue = queue.Queue()
+# Queue for handling asynchronous chat requests
+request_queue: queue.Queue = queue.Queue()
 
 
-def process_requests():
+def process_requests() -> None:
+    """Worker thread function to process queued chat requests."""
     while True:
         task = request_queue.get()
         if task is None:
-            break  # Exit thread when None is received
+            break  # Exit when a termination signal is received
 
         request_data, response_queue = task
         prompt = request_data.get("prompt")
@@ -25,32 +26,27 @@ def process_requests():
         if not prompt or not model:
             response_queue.put({"error": "Invalid request data"})
         else:
-            response = ai.chat(prompt=prompt, model=model)
+            response = ai.chat(model=model, prompt=prompt)
             response_queue.put({"response": response})
 
 
-# Start the background worker thread
-worker_thread = threading.Thread(target=process_requests, daemon=True)
-worker_thread.start()
+# Start background worker thread
+threading.Thread(target=process_requests, daemon=True).start()
 
 
+# Routes
 @app.route("/")
 def index():
     return render_template("index.html"), 200
 
 
-@app.route("/static/<name>")
-def file_manager_static(name):
-    return send_from_directory(directory="./static", path=name), 200
-
-
-@app.route("/fonts/<name>")
-def file_manager_fonts(name):
-    return send_from_directory(directory="./fonts", path=name), 200
+@app.route("/static/<path:filename>")
+def serve_static(filename: str):
+    return send_from_directory("static", filename), 200
 
 
 @app.errorhandler(Exception)
-def error_page(error):
+def handle_error(error: Exception):
     if hasattr(error, "code") and error.code == 404:
         return render_template("404.html"), 404
     return render_template("error.html", error=str(error)), 500
@@ -62,7 +58,7 @@ def api_connection():
         res = requests.get("http://localhost:11434", timeout=5)
         status_code = res.status_code
     except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
-        print("[LOG] ERROR: UNABLE TO CONNECT TO SERVER")
+        app.logger.error("Unable to connect to server")
         status_code = 404
 
     return (
@@ -79,25 +75,24 @@ def api_connection():
 
 @app.route("/api/connection/stats")
 def api_connection_stats():
-    list_of_models = ai.list_models()
-    list_of_active_models = ai.list_active_models()
+    models = ai.list_models()
+    active_models = ai.list_active_models()
 
-    if not list_of_models and not list_of_active_models:
+    if not models and not active_models:
         return jsonify({"error": "No models found"}), 404
 
-    return jsonify({"available": list_of_models, "active": list_of_active_models}), 200
+    return jsonify({"available": models, "active": active_models}), 200
 
 
 @app.route("/api/chat", methods=["POST"])
 def api_chat():
     data = request.json
-    response_queue = queue.Queue()
+    response_queue: queue.Queue = queue.Queue()
     request_queue.put((data, response_queue))
-
-    # Wait for the response from the queue
-    response = response_queue.get()
-    return jsonify(response), 200 if "response" in response else 400
+    response = response_queue.get()  # Blocking wait for response
+    status = 200 if "response" in response else 400
+    return jsonify(response), status
 
 
 if __name__ == "__main__":
-    app.run(debug=True)
+    app.run()

ollamaClient.py

@@ -10,96 +10,77 @@
 }
 
 
-# Helper functions
-def fetch_data(url):
+def fetch_data(url: str) -> dict:
     """Fetch data from a given URL using a GET request."""
     try:
         response = requests.get(url, timeout=5)
-        response.raise_for_status()  # Raise an exception for HTTP errors
+        response.raise_for_status()
         return response.json()
     except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
         print(f"[LOG] ERROR: Unable to connect to server - {e}")
-        return {}
     except requests.exceptions.RequestException as e:
         print(f"[LOG] ERROR: Request failed - {e}")
-        return {}
+    return {}
 
 
-def send_data(url, data):
+def send_data(url: str, data: dict) -> str:
     """Send data to a given URL using a POST request."""
     headers = {"Content-Type": "application/json"}
     try:
         response = requests.post(url, json=data, headers=headers, stream=True)
-        response.raise_for_status()  # Raise an exception for HTTP errors
-        return response.text  # Return text instead of raw content
+        response.raise_for_status()
+        return response.text
     except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
         print(f"[LOG] ERROR: Unable to connect to server - {e}")
-        return ""
     except requests.exceptions.RequestException as e:
         print(f"[LOG] ERROR: Request failed - {e}")
-        return ""
+    return ""
+
+
+def parse_models(data: dict) -> list:
+    """Parse and return a list of models from response data."""
+    models = data.get("models", [])
+    return [
+        {"name": item.get("name", ""), "model": item.get("model", "")}
+        for item in models
+    ]
 
 
-# Ollama Client Class
 class OllamaClient:
     @staticmethod
-    def list_models():
-        """Fetch and return a list of available models."""
+    def list_models() -> list:
         data = fetch_data(OLLAMA_SERVER + API_ENDPOINTS["list-models"])
-        if not data:
-            return []
-
-        models = data.get("models", [])
-        return [
-            {"name": item.get("name", ""), "model": item.get("model", "")}
-            for item in models
-        ]
+        return parse_models(data)
 
     @staticmethod
-    def list_active_models():
-        """Fetch and return a list of active models."""
+    def list_active_models() -> list:
         data = fetch_data(OLLAMA_SERVER + API_ENDPOINTS["list-active-models"])
-        if not data:
-            return []
-
-        models = data.get("models", [])
-        return [
-            {"name": item.get("name", ""), "model": item.get("model", "")}
-            for item in models
-        ]
+        return parse_models(data)
 
     @staticmethod
-    def chat(model, prompt):
-        """Send a chat prompt to a specified model and return the response."""
+    def chat(model: str, prompt: str) -> str:
         data = {
             "model": model,
             "messages": [{"role": "user", "content": prompt}],
         }
-
         response_text = send_data(OLLAMA_SERVER + API_ENDPOINTS["chat"], data)
         if not response_text:
             return ""
 
-        # Parse the response (assuming it's a stream of JSON objects)
-        parsed_lines = []
-        for line in response_text.split("\n"):
-            if line.strip():
-                try:
-                    parsed_lines.append(json.loads(line))
-                except json.JSONDecodeError as e:
-                    print(f"[LOG] ERROR: Failed to parse JSON - {e}")
-
-        # Extract the message contents
-        response_messages = [
-            item.get("message", {}).get("content", "")
-            for item in parsed_lines
-            if "message" in item
-        ]
+        response_lines = [line for line in response_text.splitlines() if line.strip()]
+        response_messages = []
+        for line in response_lines:
+            try:
+                json_line = json.loads(line)
+                if "message" in json_line:
+                    response_messages.append(
+                        json_line.get("message", {}).get("content", "")
+                    )
+            except json.JSONDecodeError as e:
+                print(f"[LOG] ERROR: Failed to parse JSON - {e}")
 
         response_str = "".join(response_messages)
-
         print(
-            f"[LOG] INCOMING DATA:\nmodel: {model}\nprompt: {prompt}\nOUTGOING:\nResponse: {response_str}"
+            f"[LOG] INCOMING DATA:\nModel: {model}\nPrompt: {prompt}\nResponse: {response_str}"
         )
-
         return response_str