File size: 3,242 Bytes
1a6b15f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | """AgentGlancer Dashboard Server — serves connectome snapshots + static files"""
import http.server
import json, os, glob, time
from pathlib import Path
PORT = 9011
SNAPSHOT_DIR = Path("/tmp/agent-connectome-snapshots")
STATIC_DIR = Path("/home/y7/hayula/web/glancer")
class GlancerServer(http.server.SimpleHTTPRequestHandler):
def __init__(self, *args, **kwargs):
super().__init__(*args, directory=str(STATIC_DIR), **kwargs)
def do_GET(self):
if self.path == '/api/connectome':
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Cache-Control', 'no-cache')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
# Read latest snapshot
data = {"agent_list": [], "hubs": [], "bottlenecks": [], "critical_paths": []}
if SNAPSHOT_DIR.exists():
snapshots = sorted(SNAPSHOT_DIR.glob("connectome-*.json"), reverse=True)
if snapshots:
try:
with open(snapshots[0]) as f:
data = json.load(f)
except:
pass
# If no snapshot data, return demo data
if not data.get("agent_list"):
data = generate_demo_data()
self.wfile.write(json.dumps(data).encode())
else:
super().do_GET()
def log_message(self, format, *args):
pass # Quiet
def generate_demo_data():
"""Generate demo data when FFAM snapshots not available."""
return {
"agent_list": ["rushd", "wafa", "awf", "dragon", "hermes", "musa", "zeus", "haytham",
"saif", "averroes", "orphanim", "uta", "bait", "0xZeus", "exploiter"],
"hubs": [
{"agent": "dragon", "degree": 13, "type": "router"},
{"agent": "rushd", "degree": 12, "type": "router"},
{"agent": "haytham", "degree": 11, "type": "aggregator"},
{"agent": "awf", "degree": 10, "type": "worker"},
{"agent": "hermes", "degree": 9, "type": "router"},
],
"bottlenecks": [
{"agent": "rushd", "betweenness": 0.45, "severity": "moderate", "recommendation": "Consider load balancing"},
{"agent": "awf", "betweenness": 0.32, "severity": "moderate", "recommendation": "Monitor"}
],
"critical_paths": [
{"path": ["rushd", "dragon", "wafa"], "frequency": 6},
{"path": ["haytham", "musa"], "frequency": 8},
{"path": ["hermes", "awf", "dragon"], "frequency": 5},
{"path": ["rushd", "awf", "wafa"], "frequency": 4},
],
"stats": {"agents": 15, "skills": 5, "events": 200}
}
if __name__ == "__main__":
import socketserver
print(f"[AgentGlancer] Serving on http://localhost:{PORT}")
print(f"[AgentGlancer] Reading snapshots from: {SNAPSHOT_DIR}")
with socketserver.ThreadingTCPServer(("0.0.0.0", PORT), GlancerServer) as httpd:
try:
httpd.serve_forever()
except KeyboardInterrupt:
httpd.shutdown()
|