"""DaisyChain dashboard (CLI: `daisychain-dashboard`). Tailwind-styled page with
a readiness banner, P2P connectivity scan, pooled resource + capacity plan, and
live training status. Config via DAISY_NODES_FILE. Serves on :8080."""
import json
import os
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
try:
from daisychain.dashboard.scanner import scan_cluster
except Exception:
from scanner import scan_cluster # running from the folder directly
NODES_FILE = os.environ.get("DAISY_NODES_FILE", "config/nodes.example.json")
PORT = int(os.environ.get("DAISY_DASH_PORT", "8080"))
EXPECTED_WORLD = os.environ.get("DAISY_EXPECTED_WORLD")
BASE_BATCH = int(os.environ.get("DAISY_BASE_BATCH", "32"))
def load_nodes():
with open(NODES_FILE) as f:
return json.load(f)
def _chip(ok, yes="OK", no="DOWN"):
cls = ("bg-emerald-500/20 text-emerald-600 dark:text-emerald-400" if ok
else "bg-rose-500/20 text-rose-600 dark:text-rose-400")
return f'{yes if ok else no}'
def render(d):
ready = d["ready"]
banner = ("bg-emerald-500", "✓ CLUSTER READY — all nodes connected") if ready \
else ("bg-rose-500", f"✗ NOT READY — {d['reachable']}/{d['total']} nodes reachable")
rows = ""
for n in d["nodes"]:
lat = f'{n["latency_ms"]} ms' if n["latency_ms"] is not None else "—"
res = n.get("resources") or {}
dev = res.get("device", "—")
gpu = (res.get("gpu") or {}).get("name", "")
devlabel = f"{dev}" + (f" ({gpu})" if gpu else "")
rows += f"""
{n['name']}
{n['host']}
{_chip(n['reachable'],'reachable','unreachable')}
{devlabel}
{lat}
"""
plan = d["plan"]
pr = ""
for p in plan["per_node"]:
dv = p["device"] + (f" ({p['gpu']})" if p.get("gpu") else "")
ram = f'{p["ram_gb"]} GB' if p.get("ram_gb") is not None else "—"
pr += f"""
{p['name']}
{dv}
{ram}
{p['capacity']}
{p['weight']}
{p['batch']}
"""
tot_ram = f'{plan["total_ram_gb"]} GB' if plan["total_ram_gb"] is not None else "—"
t = d["training"]
if t:
step, total = t.get("step", 0), t.get("total_steps", 1)
pct = int(100 * (step + 1) / max(1, total)); loss = t.get("cluster_avg_loss")
badge = _chip(True, "DONE") if t.get("done") else 'training…'
train = f"""
step {step} / {total}{badge}
{loss:.5f}
cluster-avg loss
"""
else:
train = '
No active run detected (waiting for rank 0 status)…
'
return f"""
DaisyChain — Cluster
\U0001f33c DaisyChain
Old Hardware Training Pipeline · scanned {d['scanned_at']}
auto-refresh 3s
{banner[1]}
Nodes
{d['reachable']}/{d['total']}
Total cores
{plan['total_cores']}
Total RAM
{tot_ram}
Connectivity scan
Node
Host
Agent
Device
Latency
{rows}
Capacity plan · global batch {plan['global_batch']}
Node
Device
RAM
Capacity
Weight
Batch
{pr}
Live training
{train}
"""
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
data = scan_cluster(load_nodes(),
int(EXPECTED_WORLD) if EXPECTED_WORLD else None, BASE_BATCH)
if self.path.startswith("/api"):
body = json.dumps(data).encode(); ctype = "application/json"
else:
body = render(data).encode(); ctype = "text/html; charset=utf-8"
self.send_response(200); self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body))); self.end_headers()
self.wfile.write(body)
def log_message(self, *a):
pass
def main():
print(f"[dashboard] :{PORT} nodes={NODES_FILE}", flush=True)
ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()
if __name__ == "__main__":
main()