Upload app.py with huggingface_hub

app.py

@@ -1,6 +1,6 @@
 """
 Hugging Face Spaces Cluster - Controller
-Basierend auf simple_cluster.py
+Alles auf Port 7860 (Gradio + API)
 """
 
 import os
@@ -8,22 +8,22 @@ import time
 import uuid
 import threading
 import json
-from http.server import HTTPServer, BaseHTTPRequestHandler
-from urllib.parse import urlparse, parse_qs
+import numpy as np
 import gradio as gr
 
 CONTROLLER_ID = os.getenv("CONTROLLER_ID", "controller")
 
-class SimpleController:
+class Controller:
     def __init__(self):
         self.workers = {}
         self.tasks = {}
         self.lock = threading.Lock()
 
-    def add_worker(self, worker_id, address=""):
+    def register_worker(self, worker_id):
         with self.lock:
-            self.workers[worker_id] = {"address": address, "status": "ready", "tasks": 0}
-        print(f"✅ Worker registriert: {worker_id}")
+            self.workers[worker_id] = {"status": "ready", "tasks": 0}
+        print(f"✅ Worker: {worker_id}")
+        return {"status": "ok"}
 
     def get_worker(self):
         with self.lock:
@@ -33,130 +33,109 @@ class SimpleController:
                     return wid
         return None
 
-    def submit_task(self, task_type, data):
-        task_id = str(uuid.uuid4())
+    def submit_task(self, typ, data):
+        tid = str(uuid.uuid4())
         with self.lock:
-            self.tasks[task_id] = {"type": task_type, "data": data, "status": "pending", "worker_id": None, "result": None}
-        self._distribute(task_id)
-        return task_id
+            self.tasks[tid] = {"type": typ, "data": data, "status": "pending", "result": None}
+        self._distribute(tid)
+        return tid
 
-    def _distribute(self, task_id):
-        worker_id = self.get_worker()
-        if worker_id:
+    def _distribute(self, tid):
+        wid = self.get_worker()
+        if wid:
             with self.lock:
-                self.tasks[task_id]["worker_id"] = worker_id
-                self.tasks[task_id]["status"] = "assigned"
-            print(f"📤 Task {task_id[:8]} → {worker_id}")
+                self.tasks[tid]["worker_id"] = wid
+                self.tasks[tid]["status"] = "assigned"
+            print(f"📤 {tid[:8]} → {wid}")
 
-    def submit_result(self, worker_id, task_id, result):
+    def submit_result(self, wid, tid, result):
         with self.lock:
-            if task_id in self.tasks:
-                self.tasks[task_id]["result"] = result
-                self.tasks[task_id]["status"] = "completed"
-            if worker_id in self.workers:
-                self.workers[worker_id]["status"] = "ready"
-                self.workers[worker_id]["tasks"] += 1
-        print(f"✅ Task {task_id[:8]} abgeschlossen")
-
-    def get_status(self):
+            if tid in self.tasks:
+                self.tasks[tid]["result"] = result
+                self.tasks[tid]["status"] = "completed"
+            if wid in self.workers:
+                self.workers[wid]["status"] = "ready"
+                self.workers[wid]["tasks"] += 1
+        print(f"✅ {tid[:8]} done")
+        return {"status": "ok"}
+
+    def get_task(self, wid):
+        with self.lock:
+            for tid, t in self.tasks.items():
+                if t["status"] == "pending":
+                    t["status"] = "assigned"
+                    t["worker_id"] = wid
+                    return {"id": tid, "type": t["type"], "data": t["data"]}
+        return {}
+
+    def status(self):
         with self.lock:
             return {
-                "workers": {"total": len(self.workers), "ready": sum(1 for w in self.workers.values() if w["status"] == "ready")},
-                "tasks": {"total": len(self.tasks), "completed": sum(1 for t in self.tasks.values() if t["status"] == "completed")},
-                "worker_list": [{"id": wid, "status": info["status"], "tasks": info["tasks"]} for wid, info in self.workers.items()]
+                "workers": len(self.workers),
+                "ready": sum(1 for w in self.workers.values() if w["status"] == "ready"),
+                "tasks": len(self.tasks),
+                "done": sum(1 for t in self.tasks.values() if t["status"] == "completed"),
+                "worker_list": [{"id": k, "status": v["status"], "tasks": v["tasks"]} for k,v in self.workers.items()]
             }
 
-controller = SimpleController()
-
-# HTTP Server für Worker API
-class ControllerHandler(BaseHTTPRequestHandler):
-    def do_POST(self):
-        length = int(self.headers.get('Content-Length', 0))
-        body = json.loads(self.rfile.read(length).decode()) if length else {}
-        
-        if self.path == "/api/register":
-            controller.add_worker(body.get("worker_id", "unknown"))
-            self.send_json({"status": "ok"})
-        elif self.path == "/api/submit_result":
-            controller.submit_result(body["worker_id"], body["task_id"], body["result"])
-            self.send_json({"status": "ok"})
-        else:
-            self.send_error(404)
-    
-    def do_GET(self):
-        params = parse_qs(urlparse(self.path).query)
-        
-        if self.path.startswith("/api/get_task"):
-            worker_id = params.get("worker_id", [""])[0]
-            with controller.lock:
-                for tid, t in controller.tasks.items():
-                    if t["status"] == "pending":
-                        t["status"] = "assigned"
-                        t["worker_id"] = worker_id
-                        self.send_json({"id": tid, "type": t["type"], "data": t["data"]})
-                        return
-            self.send_json({})
-        elif self.path == "/api/cluster_status":
-            self.send_json(controller.get_status())
-        else:
-            self.send_response(200)
-            self.send_header("Content-Type", "text/html")
-            self.end_headers()
-            self.wfile.write(b"<h1>Cluster Controller</h1>")
-    
-    def send_json(self, data):
-        self.send_response(200)
-        self.send_header("Content-Type", "application/json")
-        self.end_headers()
-        self.wfile.write(json.dumps(data).encode())
-    
-    def log_message(self, *args): pass
-
-def run_server(port=7861):
-    server = HTTPServer(("0.0.0.0", port), ControllerHandler)
-    print(f"🌐 API Server läuft auf Port {port}")
-    server.serve_forever()
+ctrl = Controller()
 
-# UI Functions
+# UI
 def get_status():
-    s = controller.get_status()
-    workers = "<br>".join([f"• {w['id']}: {'🟢' if w['status']=='ready' else '🔴'} ({w['tasks']} Tasks)" for w in s["worker_list"]]) or "Keine Worker"
-    return f"""## Cluster Status
-**Worker:** {s['workers']['total']} ({s['workers']['ready']} ready)
-**Tasks:** {s['tasks']['total']} ({s['tasks']['completed']} done)
+    s = ctrl.status()
+    wlist = "<br>".join([f"• {w['id']}: {'🟢' if w['status']=='ready' else '🔴'} ({w['tasks']})" for w in s["worker_list"]]) or "Keine"
+    return f"""## Status
+Worker: {s['workers']} ({s['ready']} ready)
+Tasks: {s['tasks']} ({s['done']} done)
 
-**Worker:**
-{workers}"""
+{wlist}"""
 
-def submit_task(typ, data):
+def send_task(typ, data):
     try:
-        d = json.loads(data)
-        tid = controller.submit_task(typ, d)
-        return f"✅ Task: `{tid[:8]}`"
+        d = json.loads(data) if isinstance(data, str) else data
+        tid = ctrl.submit_task(typ, d)
+        return f"✅ {tid[:8]}"
     except Exception as e:
         return f"❌ {e}"
 
-# Gradio UI
-with gr.Blocks(title="Cluster Controller") as demo:
+with gr.Blocks(title="Controller") as demo:
     gr.Markdown("# 🤗 Cluster Controller")
     with gr.Tabs():
         with gr.Tab("Status"):
-            btn = gr.Button("🔄 Refresh")
+            btn = gr.Button("🔄")
             out = gr.Markdown()
             btn.click(fn=get_status, outputs=out)
             demo.load(fn=get_status, outputs=out)
         with gr.Tab("Task"):
             t = gr.Dropdown(choices=["sum", "mean"], value="sum", label="Typ")
-            d = gr.Textbox(label="Daten (JSON)", value="[1,2,3,4,5]")
-            b = gr.Button("Senden")
-            r = gr.Textbox(label="Ergebnis")
-            b.click(fn=submit_task, inputs=[t,d], outputs=r)
-        with gr.Tab("Info"):
-            gr.Markdown(f"**API:** http://0.0.0.0:7861\n- POST /api/register\n- GET /api/get_task?worker_id=x\n- POST /api/submit_result")
+            d = gr.Textbox(label="Daten", value="[1,2,3,4,5]")
+            b = gr.Button("Send")
+            r = gr.Textbox(label="Result")
+            b.click(fn=send_task, inputs=[t,d], outputs=r)
+
+# API Routes - direkt an Gradio App hängen
+from fastapi import Request
+
+def setup_api():
+    @demo.app.post("/api/register")
+    async def register(req: Request):
+        data = await req.json()
+        return ctrl.register_worker(data.get("worker_id", "unknown"))
+    
+    @demo.app.get("/api/get_task")
+    async def get_task(worker_id: str):
+        return ctrl.get_task(worker_id)
+    
+    @demo.app.post("/api/submit_result")
+    async def submit_result(req: Request):
+        data = await req.json()
+        return ctrl.submit_result(data["worker_id"], data["task_id"], data["result"])
+    
+    @demo.app.get("/api/cluster_status")
+    async def cluster_status():
+        return ctrl.status()
 
 if __name__ == "__main__":
-    # Start HTTP Server im Hintergrund
-    threading.Thread(target=run_server, daemon=True).start()
-    time.sleep(1)
-    print("🚀 Starte Controller UI...")
+    setup_api()
+    print("🚀 Start...")
     demo.launch(server_name="0.0.0.0", server_port=7860)