import http.server
import socketserver
import json
import os
import subprocess
import argparse
import sys
from pathlib import Path
# Global constants
DEFAULT_PORT = 7330
DEFAULT_HOST = "127.0.0.1"
CONF_DIR = "/etc/llama.cpp.d"
LINK_PATH = "/etc/llama.cpp.conf"
SERVICE_NAME = "llama.cpp"
STATIC_DIR = os.path.join(os.path.dirname(__file__), "static")
class ModelManager:
@staticmethod
def getModels():
"""Returns a list of model config names and the currently active one."""
if not os.path.exists(CONF_DIR):
return [], None
models = [f for f in os.listdir(CONF_DIR) if f.endswith(".conf")]
models.sort()
current_model = None
if os.path.islink(LINK_PATH):
target = os.readlink(LINK_PATH)
current_model = os.path.basename(target)
return models, current_model
@staticmethod
def switchModel(model_name):
"""Updates the symlink and restarts the systemd service."""
target_path = os.path.join(CONF_DIR, model_name)
if not os.path.exists(target_path):
raise FileNotFoundError(f"Config file {model_name} not found")
# Update symlink
# Use 'ln -sf' via subprocess to handle potential permission needs or atomic switch
# But we'll try os.symlink first, though it requires removing existing link
try:
if os.path.exists(LINK_PATH) or os.path.islink(LINK_PATH):
os.remove(LINK_PATH)
os.symlink(target_path, LINK_PATH)
except PermissionError as e:
return False, f"Permission denied: {str(e)}"
except Exception as e:
return False, str(e)
# Restart service
try:
subprocess.run(["systemctl", "restart", SERVICE_NAME], check=True)
return True, "Success"
except subprocess.CalledProcessError as e:
return False, f"Failed to restart service: {str(e)}"
except Exception as e:
return False, str(e)
class RequestHandler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
if self.path == "/api/models":
self.handleGetModels()
elif self.path == "/api/logs":
self.handleGetLogs()
else:
self.serveStatic()
def do_POST(self):
if self.path == "/api/switch":
self.handleSwitchModel()
else:
self.send_error(404)
def serveStatic(self):
"""Serves files from the static directory."""
path = self.path.split('?')[0]
if path == "/":
path = "/index.html"
file_path = os.path.join(STATIC_DIR, path.lstrip("/"))
if os.path.exists(file_path) and os.path.isfile(file_path):
content_type = self.getContentType(file_path)
self.send_response(200)
self.send_header("Content-type", content_type)
self.end_headers()
with open(file_path, "rb") as f:
self.wfile.write(f.read())
else:
self.send_error(404)
def getContentType(self, file_path):
if file_path.endswith(".html"): return "text/html"
if file_path.endswith(".css"): return "text/css"
if file_path.endswith(".js"): return "application/javascript"
return "application/octet-stream"
def handleGetModels(self):
try:
models, current = ModelManager.getModels()
response = {
"models": models,
"current": current
}
self.send_json_response(200, response)
except Exception as e:
self.send_json_response(500, {"error": str(e)})
def handleGetLogs(self):
try:
# -u: service name
# -n 100: last 100 lines
# --no-pager: don't pipe to less/more
# -r: reverse order (newest first)
cmd = ["journalctl", "-u", SERVICE_NAME, "-n", "100", "--no-pager", "-r"]
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
if result.returncode != 0:
# If journalctl fails, it might be due to permissions or service not found
# We return the error message
logs = [f"Error fetching logs: {result.stderr.strip() or 'Unknown error'}"]
else:
logs = result.stdout.splitlines()
self.send_json_response(200, {"logs": logs})
except Exception as e:
self.send_json_response(500, {"error": str(e)})
def handleSwitchModel(self):
content_length = int(self.headers.get('Content-Length', 0))
post_data = self.rfile.read(content_length)
try:
data = json.loads(post_data)
model_name = data.get("model")
if not model_name:
self.send_json_response(400, {"error": "Model name required"})
return
success, message = ModelManager.switchModel(model_name)
if success:
self.send_json_response(200, {"status": "ok"})
else:
self.send_json_response(500, {"error": message})
except Exception as e:
self.send_json_response(500, {"error": str(e)})
def send_json_response(self, status_code, data):
self.send_response(status_code)
self.send_header("Content-type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(data).encode("utf-8"))
def runServer():
parser = argparse.ArgumentParser(description="LLM Model Switcher")
parser.add_argument("--host", default=DEFAULT_HOST, help="Host address to bind to. If a path (contains /), it will be treated as a unix socket. Use '' to bind to all interfaces. (default: '%(default)s')")
parser.add_argument("--port", type=int, default=DEFAULT_PORT, help="Port to bind to (TCP only). (default: %(default)s)")
args = parser.parse_args()
if os.geteuid() != 0:
print("WARNING: Not running as root. This script requires sudo to modify /etc and restart systemd services.")
# Check if host looks like a unix socket path
if '/' in args.host:
socket_path = args.host
if os.path.exists(socket_path):
os.unlink(socket_path)
with socketserver.UnixStreamServer(socket_path, RequestHandler) as httpd:
# Set permissions for the socket so others can access it if needed (e.g. nginx)
# Default to 777 for maximum accessibility given the local tool nature, or 666
os.chmod(socket_path, 0o666)
print(f"LLM Model Switcher started on unix socket: {socket_path}")
try:
httpd.serve_forever()
except KeyboardInterrupt:
pass
finally:
if os.path.exists(socket_path):
os.unlink(socket_path)
else:
# TCP Server
socketserver.TCPServer.allow_reuse_address = True
with socketserver.TCPServer((args.host, args.port), RequestHandler) as httpd:
host_display = args.host if args.host else "0.0.0.0"
print(f"LLM Model Switcher started at http://{host_display}:{args.port}")
try:
httpd.serve_forever()
except KeyboardInterrupt:
pass
if __name__ == "__main__":
runServer()