File size: 3,755 Bytes
c8d12fe | 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 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | #!/usr/bin/env python3
"""
Register an edge node with the central SmartClass server.
This script registers a new edge device (e.g., Raspberry Pi 5) with the central
server, establishing its identity and section assignment.
Usage:
python scripts/register_edge_node.py --node-id rpi5-room301 --section AIML-3-A --ip 192.168.1.101
"""
import argparse
import json
import socket
import sys
import urllib.request
import urllib.error
from datetime import datetime
def get_system_info() -> dict:
"""Gather system information for registration."""
import platform
return {
"hostname": socket.gethostname(),
"platform": platform.platform(),
"architecture": platform.machine(),
"python_version": platform.python_version(),
}
def register_node(
api_url: str,
node_id: str,
section_key: str,
ip_address: str,
api_key: str = "",
) -> bool:
"""Register edge node with central API server."""
system_info = get_system_info()
payload = {
"node_id": node_id,
"section_key": section_key,
"ip_address": ip_address,
"hostname": system_info["hostname"],
"platform": system_info["platform"],
"architecture": system_info["architecture"],
"registered_at": datetime.now().isoformat(),
}
print(f"📡 Registering edge node: {node_id}")
print(f" Section: {section_key}")
print(f" IP: {ip_address}")
print(f" Hostname: {system_info['hostname']}")
headers = {
"Content-Type": "application/json",
}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
try:
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
f"{api_url}/api/v1/edge-nodes/register",
data=data,
headers=headers,
method="POST",
)
with urllib.request.urlopen(req, timeout=15) as resp:
result = json.loads(resp.read().decode())
print(f"\n✅ Registration successful!")
print(f" Node ID: {result.get('node_id', node_id)}")
print(f" Status: {result.get('status', 'registered')}")
return True
except urllib.error.HTTPError as e:
body = e.read().decode() if e.fp else ""
print(f"\n❌ Registration failed: HTTP {e.code}")
print(f" Response: {body[:200]}")
return False
except urllib.error.URLError as e:
print(f"\n❌ Connection failed: {e.reason}")
print(f" Is the API server running at {api_url}?")
return False
def main():
parser = argparse.ArgumentParser(description="Register SmartClass edge node")
parser.add_argument("--node-id", required=True, help="Unique node identifier")
parser.add_argument("--section", required=True, help="Section key (e.g., AIML-3-A)")
parser.add_argument("--ip", default="", help="Node IP address (auto-detected if empty)")
parser.add_argument("--api-url", default="http://localhost:8000", help="Central API URL")
parser.add_argument("--api-key", default="", help="API authentication key")
args = parser.parse_args()
# Auto-detect IP if not provided
ip_address = args.ip
if not ip_address:
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
ip_address = s.getsockname()[0]
s.close()
except Exception:
ip_address = "127.0.0.1"
success = register_node(
api_url=args.api_url,
node_id=args.node_id,
section_key=args.section,
ip_address=ip_address,
api_key=args.api_key,
)
sys.exit(0 if success else 1)
if __name__ == "__main__":
main()
|