Upload app.py with huggingface_hub

app.py

@@ -2,18 +2,16 @@
 Hugging Face Spaces Cluster - Worker App
 =========================================
 Läuft auf jedem Worker Space und führt Tasks aus.
-
-Deployment:
-1. Diese Datei als app.py auf Hugging Face Space hochladen
-2. requirements.txt hochladen
-3. Space startet automatisch
 """
 
 import os
 import time
 import requests
 import numpy as np
-from huggingface_hub import HfApi
+import threading
+import json
+
+import gradio as gr
 
 # Hugging Face Konfiguration
 HF_TOKEN = os.getenv("HF_TOKEN", "")
@@ -21,8 +19,13 @@ WORKER_ID = os.getenv("WORKER_ID", "worker-1")
 CONTROLLER_URL = os.getenv("CONTROLLER_URL", "")
 SPACE_NAME = os.getenv("SPACE_NAME", "")
 
-# API für Task-Empfang
-import gradio as gr
+print(f"\n{'='*60}")
+print(f"🔧 Worker startet...")
+print(f"   WORKER_ID: {WORKER_ID}")
+print(f"   CONTROLLER_URL: {CONTROLLER_URL}")
+print(f"   SPACE_NAME: {SPACE_NAME}")
+print(f"   HF_TOKEN: {HF_TOKEN[:10]}...")
+print(f"{'='*60}\n")
 
 # ============================================
 # Task-Verarbeitung
@@ -32,44 +35,135 @@ def process_task(task_type, data):
     """Verarbeitet einen Task vom Controller"""
     print(f"[{WORKER_ID}] Verarbeite Task: {task_type}")
     
+    if isinstance(data, list):
+        data = np.array(data)
+    
     if task_type == "sum":
-        result = np.sum(data)
+        result = float(np.sum(data))
     elif task_type == "mean":
-        result = np.mean(data)
+        result = float(np.mean(data))
     elif task_type == "matrix_multiply":
         a, b = data
-        result = np.dot(a, b)
+        result = np.dot(np.array(a), np.array(b)).tolist()
     elif task_type == "inference":
-        # Beispiel: ML-Inferenz
-        result = np.mean(data)  # Platzhalter
+        result = float(np.mean(data))
     else:
         result = None
     
-    print(f"[{WORKER_ID}] Task abgeschlossen: {result}")
+    print(f"[{WORKER_ID}] Ergebnis: {result}")
     return result
 
-def health_check():
-    """Sendet Health-Check an Controller"""
+# ============================================
+# Beim Controller registrieren
+# ============================================
+
+def register_with_controller():
+    """Registriert sich beim Controller"""
+    if not CONTROLLER_URL:
+        print("❌ CONTROLLER_URL nicht gesetzt!")
+        return False
+    
+    try:
+        # Versuche HTTP und HTTPS
+        urls_to_try = [
+            CONTROLLER_URL,
+            CONTROLLER_URL.replace("https://", "http://"),
+        ]
+        
+        for url in urls_to_try:
+            try:
+                response = requests.post(
+                    f"{url}/api/register",
+                    json={"worker_id": WORKER_ID, "status": "ready"},
+                    timeout=10
+                )
+                if response.status_code == 200:
+                    print(f"✅ Beim Controller registriert: {url}")
+                    return True
+            except Exception as e:
+                continue
+        
+        print(f"❌ Registrierung fehlgeschlagen: {CONTROLLER_URL}")
+        return False
+        
+    except Exception as e:
+        print(f"❌ Fehler bei Registrierung: {e}")
+        return False
+
+def poll_controller():
+    """Pollt Controller auf neue Tasks"""
     if not CONTROLLER_URL:
         return
     
     try:
-        requests.post(
-            f"{CONTROLLER_URL}/api/register",
-            json={"worker_id": WORKER_ID, "status": "ready"},
+        response = requests.get(
+            f"{CONTROLLER_URL}/api/get_task",
+            params={"worker_id": WORKER_ID},
             timeout=5
         )
+        
+        if response.status_code == 200 and response.text.strip():
+            try:
+                task = response.json()
+                if task and task.get("type"):
+                    print(f"📥 Task erhalten: {task.get('id', 'unknown')[:8]} ({task['type']})")
+                    result = process_task(task["type"], task["data"])
+                    
+                    # Ergebnis zurück senden
+                    requests.post(
+                        f"{CONTROLLER_URL}/api/submit_result",
+                        json={
+                            "worker_id": WORKER_ID,
+                            "task_id": task.get("id"),
+                            "result": result
+                        },
+                        timeout=5
+                    )
+                    print(f"✅ Ergebnis gesendet")
+            except json.JSONDecodeError:
+                pass  # Leere Antwort, kein Task verfügbar
+        else:
+            pass  # Kein Task verfügbar
+                
     except Exception as e:
-        print(f"Health-Check fehlgeschlagen: {e}")
+        pass  # Still schweigen beim Polling
+
+def start_polling():
+    """Startet Background-Polling"""
+    # Erst registrieren
+    register_with_controller()
+    
+    # Dann pollen
+    while True:
+        poll_controller()
+        time.sleep(3)
+
+# Background-Thread starten
+threading.Thread(target=start_polling, daemon=True).start()
 
 # ============================================
-# Gradio Interface (für Space)
+# Gradio Interface
 # ============================================
 
+def get_worker_status():
+    """Zeigt Worker-Status"""
+    registered = "🟢" if CONTROLLER_URL else "🔴"
+    return f"""
+    ## Worker Status
+    
+    **Worker ID:** {WORKER_ID}
+    **Space:** {SPACE_NAME}
+    **Controller:** {CONTROLLER_URL or 'Nicht konfiguriert'}
+    **Status:** {registered} {'Verbunden' if CONTROLLER_URL else 'Keine URL'}
+    
+    ### Verfügbare Tasks:
+    - sum: Array-Summe
+    - mean: Array-Durchschnitt
+    - matrix_multiply: Matrix-Multiplikation
+    """
+
 def worker_interface(task_json):
     """Gradio Interface für manuelle Tasks"""
-    import json
-    
     try:
         task = json.loads(task_json)
         task_type = task.get("type", "sum")
@@ -85,32 +179,36 @@ def worker_interface(task_json):
     except Exception as e:
         return json.dumps({"status": "error", "message": str(e)})
 
-def get_worker_status():
-    """Zeigt Worker-Status"""
-    return f"""
-    ## Worker Status
-    
-    **Worker ID:** {WORKER_ID}
-    **Space:** {SPACE_NAME}
-    **Controller:** {CONTROLLER_URL or 'Nicht konfiguriert'}
-    **Status:** 🟢 Ready
+def test_connection():
+    """Testet Verbindung zum Controller"""
+    if not CONTROLLER_URL:
+        return "❌ CONTROLLER_URL nicht gesetzt!"
     
-    ### Verfügbare Tasks:
-    - sum: Array-Summe
-    - mean: Array-Durchschnitt
-    - matrix_multiply: Matrix-Multiplikation
-    - inference: ML-Inferenz
-    """
+    try:
+        response = requests.get(f"{CONTROLLER_URL}/api/cluster_status", timeout=5)
+        if response.status_code == 200:
+            return f"✅ Controller erreichbar: {CONTROLLER_URL}"
+        else:
+            return f"❌ Controller antwortet nicht: {response.status_code}"
+    except Exception as e:
+        return f"❌ Verbindung fehlgeschlagen: {e}"
 
 # Gradio UI
 with gr.Blocks(title=f"Cluster Worker - {WORKER_ID}") as demo:
-    gr.Markdown(f"# 🤗 Cluster Worker: {WORKER_ID}")
+    gr.Markdown(f"# 🔧 Cluster Worker: {WORKER_ID}")
     
     with gr.Row():
         with gr.Column():
             status_md = gr.Markdown(get_worker_status())
+            refresh_btn = gr.Button("🔄 Status aktualisieren")
+            refresh_btn.click(fn=get_worker_status, outputs=status_md)
         
         with gr.Column():
+            gr.Markdown("### Verbindung testen")
+            test_btn = gr.Button("🔗 Teste Controller")
+            test_output = gr.Textbox(label="Ergebnis")
+            test_btn.click(fn=test_connection, outputs=test_output)
+            
             gr.Markdown("### Task manuell ausführen")
             task_input = gr.Textbox(
                 label="Task JSON",
@@ -120,70 +218,12 @@ with gr.Blocks(title=f"Cluster Worker - {WORKER_ID}") as demo:
             run_btn = gr.Button("Task ausführen")
             result_output = gr.Textbox(label="Ergebnis")
     
-    run_btn.click(
-        fn=worker_interface,
-        inputs=task_input,
-        outputs=result_output
-    )
-
-# ============================================
-# Background Polling (für Controller-Tasks)
-# ============================================
-
-def poll_controller():
-    """Pollt Controller auf neue Tasks"""
-    if not CONTROLLER_URL:
-        return
-    
-    try:
-        response = requests.get(
-            f"{CONTROLLER_URL}/api/get_task",
-            params={"worker_id": WORKER_ID},
-            timeout=5
-        )
-        
-        if response.status_code == 200:
-            task = response.json()
-            if task.get("type"):
-                result = process_task(task["type"], task["data"])
-                
-                # Ergebnis zurück senden
-                requests.post(
-                    f"{CONTROLLER_URL}/api/submit_result",
-                    json={
-                        "worker_id": WORKER_ID,
-                        "task_id": task.get("id"),
-                        "result": result
-                    },
-                    timeout=5
-                )
-    except Exception as e:
-        print(f"Polling fehlgeschlagen: {e}")
-
-# Background-Task starten
-import threading
-
-def start_polling():
-    while True:
-        poll_controller()
-        time.sleep(2)  # Alle 2 Sekunden pollern
-
-threading.Thread(target=start_polling, daemon=True).start()
+    run_btn.click(fn=worker_interface, inputs=task_input, outputs=result_output)
 
 # ============================================
 # Main
 # ============================================
 
 if __name__ == "__main__":
-    print(f"🚀 Starte Worker: {WORKER_ID}")
-    print(f"   Space: {SPACE_NAME}")
-    print(f"   Controller: {CONTROLLER_URL}")
-    
-    # Health-Check senden
-    health_check()
-    
-    # Gradio starten
+    print(f"🚀 Starte Gradio UI...")
     demo.launch(server_name="0.0.0.0", server_port=7860)
-
-# Für Hugging Face Spaces - die App muss 'app' heißen
-app = demo