import subprocess import sys import threading import json import socket import os import logging # ========================================== # ZONE 1: AUTO-INSTALLER & LOG SILENCER # ========================================== def bootstrap(): """Checks for required libraries and installs them if missing.""" required = ["flask", "requests", "markdown"] installed_any = False for lib in required: try: __import__(lib) except ImportError: installed_any = True print(f"--- [SETUP] {lib} not found. Installing... ---") subprocess.check_call([sys.executable, "-m", "pip", "install", lib]) # Only clear the screen if we actually did a messy install if installed_any: os.system('cls' if os.name == 'nt' else 'clear') bootstrap() # Now import the libraries we just verified from flask import Flask, render_template_string, request, redirect, url_for, jsonify import requests import markdown # Silence the Flask "Development Server" warnings and request logs log = logging.getLogger('werkzeug') log.setLevel(logging.ERROR) #os.environ['WERKZEUG_RUN_MAIN'] = 'true' # ========================================== # ZONE 2: APP LOGIC & MODEL MANAGEMENT # ========================================== app = Flask(__name__) chat_history = [] stream_state = {"text": "", "done": True} current_model = "llama3.1" def get_local_ip(): """Detects the computer's local IP for the Kobo to connect to.""" s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: s.connect(('10.255.255.255', 1)) IP = s.getsockname()[0] except Exception: IP = '127.0.0.1' finally: s.close() return IP def print_dashboard(ollama_status): """Refreshes the terminal UI with current server info.""" my_ip = get_local_ip() my_port = 5001 os.system('cls' if os.name == 'nt' else 'clear') print("\n" + "╔" + "═"*48 + "╗") print(f"║ {'INKWELL AI SERVER':^46} ║") print("╠" + "═"*48 + "╣") print(f"║ IP: http://{my_ip}:{my_port:<21} ║") print(f"║ AI: {ollama_status:<38} ║") print(f"║ MODEL: {current_model:<37} ║") print("╚" + "═"*48 + "╝") print(" [Press CTRL+C to stop the server]\n") def get_available_models(): """Fetches the list of downloaded models from Ollama.""" try: r = requests.get("http://localhost:11434/api/tags", timeout=2) if r.status_code == 200: return [m['name'] for m in r.json().get('models', [])] except: return [] return [] def generate_ai_response(prompt, user_ip): """Streams response from Ollama in a separate thread.""" global stream_state, chat_history, current_model stream_state["text"] = "" stream_state["done"] = False try: r = requests.post("http://localhost:11434/api/generate", json={"model": current_model, "prompt": prompt, "stream": True}, stream=True) r.raise_for_status() for line in r.iter_lines(): if line: chunk = json.loads(line) stream_state["text"] += chunk.get("response", "") except Exception as e: stream_state["text"] = f"Connection Error: {e}" stream_state["done"] = True chat_history.append({"role": f"AI ({current_model})", "content": markdown.markdown(stream_state["text"])}) HTML_PAGE = """ InkWell AI
Model: {{ active_model }}
Clear Undo Refresh
{% for msg in history %}
{{ msg.role }}: {{ msg.content | safe }}
{% endfor %}

/// THINKING... ///

Available Models:
{% if ollama_online %} {% if all_models %} {% for m in all_models %} {{ m }} {% endfor %} {% else %}

Connection Successful!
Your model library is empty. Please pull a model on your laptop to begin.
Example: ollama pull llama3.1

{% endif %} {% else %}

Ollama is NOT running!
Please start Ollama on your computer.

{% endif %}
""" @app.route("/") def home(): models = get_available_models() try: requests.get("http://localhost:11434/api/tags", timeout=1) ollama_status = "ONLINE" service_online = True except: ollama_status = "OFFLINE (Start Ollama!)" service_online = False # Redraw the terminal box so it updates when the Kobo refreshes print_dashboard(ollama_status) return render_template_string(HTML_PAGE, history=chat_history, active_model=current_model, all_models=models, ollama_online=service_online) @app.route("/select/") def select_model(name): global current_model current_model = name return redirect(url_for('home')) @app.route("/ask", methods=["POST"]) def ask(): user_input = request.form.get("prompt") user_ip = request.remote_addr if user_input: chat_history.append({"role": f"User ({user_ip})", "content": user_input}) thread = threading.Thread(target=generate_ai_response, args=(user_input, user_ip)) thread.start() return jsonify({"status": "started"}) @app.route("/status") def status(): md_text = markdown.markdown(stream_state["text"]) return jsonify({"text": md_text, "done": stream_state["done"]}) @app.route("/clear") def clear(): chat_history.clear() return redirect(url_for('home')) @app.route("/undo") def undo(): if len(chat_history) >= 2: chat_history.pop(); chat_history.pop() return redirect(url_for('home')) # ========================================== # ZONE 3: THE LAUNCHER # ========================================== if __name__ == "__main__": # Initial UI draw print_dashboard("CHECKING...") app.run(host='0.0.0.0', port=5001, debug=False, use_reloader=False)