Create app.py

app.py

@@ -0,0 +1,524 @@
+"""
+SharePUTER Worker Node — Serverless (HF Space Docker / any container)
+Executes task fragments assigned by the head node.
+Supports node types: RAM, CPU, GPU
+"""
+
+from fastapi import FastAPI, Request
+from fastapi.responses import HTMLResponse, JSONResponse
+import httpx
+import asyncio
+import uuid
+import os
+import sys
+import io
+import json
+import time
+import traceback
+import multiprocessing
+import psutil
+from datetime import datetime
+from contextlib import redirect_stdout, redirect_stderr
+from typing import Optional
+
+app = FastAPI(title="SharePUTER Worker Node")
+
+# ─── Configuration ──────────────────────────────────────────────────────────
+HEAD_NODE_URL = os.environ.get("SACCP_HEAD_URL", "http://localhost:7860")
+NODE_TYPE = os.environ.get("SACCP_NODE_TYPE", "CPU").upper()  # RAM, CPU, GPU
+NODE_OWNER = os.environ.get("SACCP_NODE_OWNER", "anonymous")
+NODE_ID: Optional[str] = None
+REGISTERED = False
+
+# Auto-detect specs
+def detect_specs():
+    specs = {
+        "cpu_count": multiprocessing.cpu_count(),
+        "ram_total_gb": round(psutil.virtual_memory().total / (1024**3), 2),
+        "ram_available_gb": round(psutil.virtual_memory().available / (1024**3), 2),
+        "platform": sys.platform,
+        "python_version": sys.version.split()[0],
+    }
+
+    # Check for GPU
+    try:
+        import torch
+        if torch.cuda.is_available():
+            specs["gpu_available"] = True
+            specs["gpu_name"] = torch.cuda.get_device_name(0)
+            specs["gpu_memory_gb"] = round(torch.cuda.get_device_properties(0).total_mem / (1024**3), 2)
+        else:
+            specs["gpu_available"] = False
+    except ImportError:
+        specs["gpu_available"] = False
+
+    return specs
+
+
+SPECS = detect_specs()
+
+
+def safe_execute(code: str, timeout: int = 300) -> dict:
+    """Safely execute Python code in an isolated namespace with timeout."""
+    result = {
+        "status": "completed",
+        "result": None,
+        "stdout": "",
+        "stderr": "",
+        "error": None,
+        "execution_time": 0
+    }
+
+    stdout_capture = io.StringIO()
+    stderr_capture = io.StringIO()
+    namespace = {"__builtins__": __builtins__}
+
+    start_time = time.time()
+    try:
+        with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture):
+            exec(code, namespace)
+
+        result["execution_time"] = round(time.time() - start_time, 3)
+        result["stdout"] = stdout_capture.getvalue()
+        result["stderr"] = stderr_capture.getvalue()
+
+        # Try to capture meaningful result
+        # Look for common result patterns
+        for var_name in ["result", "output", "answer", "ret", "__fragment_results__"]:
+            if var_name in namespace:
+                val = namespace[var_name]
+                try:
+                    json.dumps(val)  # Check JSON serializable
+                    result["result"] = val
+                except (TypeError, ValueError):
+                    result["result"] = str(val)
+                break
+
+        # If no named result, capture last expression result via stdout
+        if result["result"] is None and result["stdout"]:
+            result["result"] = result["stdout"].strip()
+
+    except Exception as e:
+        result["status"] = "failed"
+        result["error"] = f"{type(e).__name__}: {str(e)}\n{traceback.format_exc()}"
+        result["execution_time"] = round(time.time() - start_time, 3)
+        result["stdout"] = stdout_capture.getvalue()
+        result["stderr"] = stderr_capture.getvalue()
+
+    return result
+
+
+# ─── Homepage ───────────────────────────────────────────────────────────────
+@app.get("/", response_class=HTMLResponse)
+async def homepage():
+    gpu_info = ""
+    if SPECS.get("gpu_available"):
+        gpu_info = f"""
+        <div class="spec-item">
+            <span class="spec-label">GPU</span>
+            <span class="spec-value">{SPECS.get('gpu_name', 'Unknown')}</span>
+        </div>
+        <div class="spec-item">
+            <span class="spec-label">GPU Memory</span>
+            <span class="spec-value">{SPECS.get('gpu_memory_gb', 0)} GB</span>
+        </div>"""
+
+    type_colors = {"RAM": "#ff6b6b", "CPU": "#00aaff", "GPU": "#aa66ff"}
+    type_color = type_colors.get(NODE_TYPE, "#00ff88")
+
+    current_ram = round(psutil.virtual_memory().available / (1024**3), 2)
+    cpu_percent = psutil.cpu_percent(interval=0.5)
+
+    html = f"""<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>SharePUTER™ Worker Node</title>
+    <style>
+        @import url('https://fonts.googleapis.com/css2?family=JetBrains+Mono:wght@300;400;700&family=Inter:wght@300;400;600;700;900&display=swap');
+        * {{ margin:0; padding:0; box-sizing:border-box; }}
+        body {{
+            font-family:'Inter',sans-serif;
+            background:#0a0a0f;
+            color:#e0e0e0;
+            min-height:100vh;
+            display:flex;
+            align-items:center;
+            justify-content:center;
+        }}
+        .bg {{
+            position:fixed; top:0; left:0; right:0; bottom:0;
+            background-image:
+                radial-gradient(circle at 20% 30%, {type_color}11 0%, transparent 50%),
+                radial-gradient(circle at 80% 70%, {type_color}08 0%, transparent 50%);
+        }}
+        .card {{
+            position:relative; z-index:1;
+            background:rgba(255,255,255,0.02);
+            border:1px solid rgba(255,255,255,0.06);
+            border-radius:24px;
+            padding:48px;
+            max-width:560px;
+            width:90%;
+            backdrop-filter:blur(20px);
+        }}
+        .node-type {{
+            display:inline-block;
+            padding:6px 20px;
+            border-radius:30px;
+            font-family:'JetBrains Mono',monospace;
+            font-size:0.85rem;
+            font-weight:700;
+            letter-spacing:2px;
+            background:{type_color}22;
+            color:{type_color};
+            border:1px solid {type_color}44;
+            margin-bottom:20px;
+        }}
+        h1 {{
+            font-size:2rem;
+            font-weight:900;
+            margin-bottom:4px;
+            background:linear-gradient(135deg,#fff,{type_color});
+            -webkit-background-clip:text;
+            -webkit-text-fill-color:transparent;
+        }}
+        .subtitle {{ color:#888; font-size:0.9rem; margin-bottom:32px; }}
+        .status {{
+            display:flex; align-items:center; gap:10px;
+            margin-bottom:32px;
+            font-family:'JetBrains Mono',monospace;
+            font-size:0.85rem;
+        }}
+        .pulse {{
+            width:10px; height:10px; border-radius:50%;
+            background:#00ff88;
+            animation:pulse 2s ease-in-out infinite;
+        }}
+        @keyframes pulse {{
+            0%,100% {{ box-shadow:0 0 0 0 rgba(0,255,136,0.4); }}
+            50% {{ box-shadow:0 0 0 8px rgba(0,255,136,0); }}
+        }}
+        .specs {{
+            background:rgba(255,255,255,0.02);
+            border:1px solid rgba(255,255,255,0.04);
+            border-radius:12px;
+            padding:20px;
+            margin-bottom:24px;
+        }}
+        .specs h3 {{
+            font-size:0.75rem;
+            text-transform:uppercase;
+            letter-spacing:2px;
+            color:#888;
+            margin-bottom:16px;
+        }}
+        .spec-item {{
+            display:flex; justify-content:space-between;
+            padding:8px 0;
+            border-bottom:1px solid rgba(255,255,255,0.03);
+        }}
+        .spec-item:last-child {{ border-bottom:none; }}
+        .spec-label {{ color:#888; font-size:0.9rem; }}
+        .spec-value {{
+            color:{type_color};
+            font-family:'JetBrains Mono',monospace;
+            font-weight:600;
+            font-size:0.9rem;
+        }}
+        .config {{
+            background:rgba(255,255,255,0.02);
+            border:1px solid rgba(255,255,255,0.04);
+            border-radius:12px;
+            padding:20px;
+        }}
+        .config h3 {{
+            font-size:0.75rem;
+            text-transform:uppercase;
+            letter-spacing:2px;
+            color:#888;
+            margin-bottom:16px;
+        }}
+        .config label {{ display:block; color:#aaa; font-size:0.85rem; margin-bottom:4px; }}
+        .config input, .config select {{
+            width:100%;
+            padding:10px 14px;
+            border-radius:8px;
+            border:1px solid rgba(255,255,255,0.1);
+            background:rgba(255,255,255,0.04);
+            color:#fff;
+            font-family:'JetBrains Mono',monospace;
+            font-size:0.85rem;
+            margin-bottom:14px;
+            outline:none;
+        }}
+        .config input:focus, .config select:focus {{
+            border-color:{type_color}88;
+        }}
+        .btn {{
+            width:100%;
+            padding:12px;
+            border:none;
+            border-radius:10px;
+            background:linear-gradient(135deg,{type_color},{type_color}aa);
+            color:#000;
+            font-weight:700;
+            font-size:0.95rem;
+            cursor:pointer;
+            transition:all 0.2s;
+            font-family:'Inter',sans-serif;
+        }}
+        .btn:hover {{ transform:translateY(-2px); box-shadow:0 4px 20px {type_color}44; }}
+        .footer {{ text-align:center; margin-top:24px; color:#555; font-size:0.75rem; }}
+        #configResult {{
+            margin-top:12px; padding:10px; border-radius:8px;
+            font-family:'JetBrains Mono',monospace; font-size:0.8rem;
+            display:none;
+        }}
+    </style>
+</head>
+<body>
+    <div class="bg"></div>
+    <div class="card">
+        <div class="node-type">{NODE_TYPE} NODE</div>
+        <h1>SharePUTER™ Worker</h1>
+        <p class="subtitle">SACCP Network Compute Node</p>
+
+        <div class="status">
+            <div class="pulse"></div>
+            <span style="color:#00ff88;">{'REGISTERED & WORKING' if REGISTERED else 'ONLINE — NOT REGISTERED'}</span>
+        </div>
+
+        <div class="specs">
+            <h3>System Specifications</h3>
+            <div class="spec-item">
+                <span class="spec-label">Node ID</span>
+                <span class="spec-value">{NODE_ID or 'Not Registered'}</span>
+            </div>
+            <div class="spec-item">
+                <span class="spec-label">CPU Cores</span>
+                <span class="spec-value">{SPECS['cpu_count']}</span>
+            </div>
+            <div class="spec-item">
+                <span class="spec-label">CPU Usage</span>
+                <span class="spec-value">{cpu_percent}%</span>
+            </div>
+            <div class="spec-item">
+                <span class="spec-label">Total RAM</span>
+                <span class="spec-value">{SPECS['ram_total_gb']} GB</span>
+            </div>
+            <div class="spec-item">
+                <span class="spec-label">Available RAM</span>
+                <span class="spec-value">{current_ram} GB</span>
+            </div>
+            <div class="spec-item">
+                <span class="spec-label">Python</span>
+                <span class="spec-value">{SPECS['python_version']}</span>
+            </div>
+            {gpu_info}
+        </div>
+
+        <div class="config">
+            <h3>Node Configuration</h3>
+            <label>Head Node URL</label>
+            <input type="text" id="headUrl" value="{HEAD_NODE_URL}" />
+            <label>Node Type</label>
+            <select id="nodeType">
+                <option value="RAM" {"selected" if NODE_TYPE=="RAM" else ""}>RAM — Provide memory</option>
+                <option value="CPU" {"selected" if NODE_TYPE=="CPU" else ""}>CPU — Provide compute</option>
+                <option value="GPU" {"selected" if NODE_TYPE=="GPU" else ""}>GPU — Provide GPU</option>
+            </select>
+            <label>Owner Username</label>
+            <input type="text" id="owner" value="{NODE_OWNER}" />
+            <button class="btn" onclick="configure()">
+                {'Reconfigure' if REGISTERED else 'Register & Start Working'}
+            </button>
+            <div id="configResult"></div>
+        </div>
+
+        <div class="footer">SharePUTER™ SACCP Network — Worker Node v1.0</div>
+    </div>
+
+    <script>
+    async function configure() {{
+        const headUrl = document.getElementById('headUrl').value;
+        const nodeType = document.getElementById('nodeType').value;
+        const owner = document.getElementById('owner').value;
+        const resultDiv = document.getElementById('configResult');
+
+        resultDiv.style.display = 'block';
+        resultDiv.style.background = 'rgba(0,136,255,0.1)';
+        resultDiv.style.color = '#00aaff';
+        resultDiv.innerText = 'Registering with head node...';
+
+        try {{
+            const resp = await fetch('/configure', {{
+                method: 'POST',
+                headers: {{'Content-Type': 'application/json'}},
+                body: JSON.stringify({{head_url: headUrl, node_type: nodeType, owner: owner}})
+            }});
+            const data = await resp.json();
+            if (resp.ok) {{
+                resultDiv.style.background = 'rgba(0,255,136,0.1)';
+                resultDiv.style.color = '#00ff88';
+                resultDiv.innerText = '✓ ' + (data.message || 'Registered!') + '\\nNode ID: ' + (data.node_id || 'N/A');
+                setTimeout(() => location.reload(), 2000);
+            }} else {{
+                resultDiv.style.background = 'rgba(255,68,68,0.1)';
+                resultDiv.style.color = '#ff4444';
+                resultDiv.innerText = '✗ ' + (data.detail || data.error || 'Failed');
+            }}
+        }} catch (e) {{
+            resultDiv.style.background = 'rgba(255,68,68,0.1)';
+            resultDiv.style.color = '#ff4444';
+            resultDiv.innerText = '✗ Network error: ' + e.message;
+        }}
+    }}
+    </script>
+</body>
+</html>"""
+    return HTMLResponse(content=html)
+
+
+@app.post("/configure")
+async def configure_node(data: dict):
+    global HEAD_NODE_URL, NODE_TYPE, NODE_OWNER, NODE_ID, REGISTERED
+
+    HEAD_NODE_URL = data.get("head_url", HEAD_NODE_URL)
+    NODE_TYPE = data.get("node_type", NODE_TYPE).upper()
+    NODE_OWNER = data.get("owner", NODE_OWNER)
+
+    # Register with head node
+    try:
+        async with httpx.AsyncClient(timeout=15.0) as client:
+            resp = await client.post(f"{HEAD_NODE_URL}/api/register_node", json={
+                "node_type": NODE_TYPE,
+                "node_url": os.environ.get("SPACE_HOST", f"http://localhost:7861"),
+                "owner": NODE_OWNER,
+                "specs": SPECS
+            })
+            if resp.status_code == 200:
+                result = resp.json()
+                NODE_ID = result.get("node_id")
+                REGISTERED = True
+                return {"message": "Registered successfully", "node_id": NODE_ID, "pay_rate": result.get("pay_rate")}
+            else:
+                return JSONResponse(content={"error": resp.text}, status_code=500)
+    except Exception as e:
+        return JSONResponse(content={"error": str(e)}, status_code=500)
+
+
+@app.get("/health")
+async def health():
+    return {
+        "status": "online",
+        "node_id": NODE_ID,
+        "node_type": NODE_TYPE,
+        "registered": REGISTERED,
+        "specs": SPECS
+    }
+
+
+# ─── Worker Loop (runs in background) ──────────────────────────────────────
+async def worker_loop():
+    """Continuously poll head node for work and execute fragments."""
+    global NODE_ID, REGISTERED
+
+    # Wait for registration
+    while not REGISTERED:
+        await asyncio.sleep(5)
+        # Auto-register if env vars are set
+        if os.environ.get("SACCP_AUTO_REGISTER", "false").lower() == "true":
+            try:
+                async with httpx.AsyncClient(timeout=15.0) as client:
+                    resp = await client.post(f"{HEAD_NODE_URL}/api/register_node", json={
+                        "node_type": NODE_TYPE,
+                        "node_url": os.environ.get("SPACE_HOST", "http://localhost:7861"),
+                        "owner": NODE_OWNER,
+                        "specs": SPECS
+                    })
+                    if resp.status_code == 200:
+                        result = resp.json()
+                        NODE_ID = result.get("node_id")
+                        REGISTERED = True
+                        print(f"[SACCP] Auto-registered as {NODE_ID}")
+            except Exception as e:
+                print(f"[SACCP] Auto-register failed: {e}")
+
+    print(f"[SACCP] Worker loop started — Node: {NODE_ID} Type: {NODE_TYPE}")
+
+    heartbeat_interval = 30
+    last_heartbeat = 0
+
+    while True:
+        try:
+            # Heartbeat
+            if time.time() - last_heartbeat > heartbeat_interval:
+                try:
+                    async with httpx.AsyncClient(timeout=10.0) as client:
+                        await client.post(f"{HEAD_NODE_URL}/api/node_heartbeat", json={
+                            "node_id": NODE_ID,
+                            "status": "online",
+                            "current_load": {
+                                "cpu_percent": psutil.cpu_percent(),
+                                "ram_available_gb": round(psutil.virtual_memory().available / (1024**3), 2)
+                            }
+                        })
+                    last_heartbeat = time.time()
+                except Exception:
+                    pass
+
+            # Get work
+            async with httpx.AsyncClient(timeout=15.0) as client:
+                resp = await client.get(f"{HEAD_NODE_URL}/api/get_work?node_id={NODE_ID}")
+                if resp.status_code != 200:
+                    await asyncio.sleep(5)
+                    continue
+
+                data = resp.json()
+                work = data.get("work")
+
+                if not work:
+                    await asyncio.sleep(3)
+                    continue
+
+                fragment_id = work["fragment_id"]
+                code = work.get("code", "")
+                frag_type = work.get("fragment_type", "compute")
+
+                print(f"[SACCP] Executing fragment {fragment_id} (type: {frag_type})")
+
+                if frag_type == "setup":
+                    # Setup fragments just validate imports
+                    exec_result = safe_execute(code, timeout=60)
+                elif frag_type == "aggregate":
+                    exec_result = {"status": "completed", "result": "aggregation_placeholder", "stdout": "", "error": None}
+                else:
+                    exec_result = safe_execute(code, timeout=300)
+
+                # Submit result
+                await client.post(f"{HEAD_NODE_URL}/api/submit_result", json={
+                    "fragment_id": fragment_id,
+                    "node_id": NODE_ID,
+                    "status": exec_result["status"],
+                    "result": exec_result.get("result"),
+                    "error": exec_result.get("error"),
+                    "stdout": exec_result.get("stdout", "")
+                })
+
+                print(f"[SACCP] Fragment {fragment_id} -> {exec_result['status']}")
+
+        except Exception as e:
+            print(f"[SACCP] Worker error: {e}")
+            await asyncio.sleep(5)
+
+
+@app.on_event("startup")
+async def startup():
+    asyncio.create_task(worker_loop())
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7861)