""" Nav3D Agent Demo — Web Dashboard FastAPI app serving a monitoring dashboard for the voxel navigation agent. Supports both ngrok and bore tunneling. """ import os import json import subprocess import asyncio from datetime import datetime from pathlib import Path from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles app = FastAPI(title="Nav3D Agent Demo", version="2.0.0") # ── State ────────────────────────────────────────────────────────────────── _bot_status = { "online": False, "username": "Zelin_", "position": {"x": 0, "y": 0, "z": 0}, "health": 20, "food": 20, "activity": "starting", "last_think": "", "emotion": "neutral", "chat_log": [], "world_time": 0, "dimension": "overworld", "connected_since": "", "tunnel_url": "", "tunnel_method": "", "server_online": False, "players_online": 0, } # ── Load tunnel URL (supports both ngrok and bore) ───────────────────────── def _load_tunnel_url(): # Check bore tunnel first try: tunnel_file = Path("/app/logs/tunnel_url.txt") if tunnel_file.exists(): url = tunnel_file.read_text().strip() if url: return url except: pass # Fallback: check ngrok try: url_file = Path("/app/logs/ngrok_url.txt") if url_file.exists(): url = url_file.read_text().strip() if url: return url except: pass # Try ngrok API try: import urllib.request with urllib.request.urlopen("http://localhost:4040/api/tunnels", timeout=2) as resp: data = json.loads(resp.read()) for t in data.get("tunnels", []): url = t.get("public_url", "") if "tcp" in url: return url except: pass return "" def _load_tunnel_method(): try: method_file = Path("/app/logs/tunnel_method.txt") if method_file.exists(): return method_file.read_text().strip() except: pass # Check if ngrok URL exists try: url_file = Path("/app/logs/ngrok_url.txt") if url_file.exists() and url_file.read_text().strip(): return "ngrok" except: pass return "" def _update_bot_status(): """Read bot status from the shared JSON file""" global _bot_status try: status_file = Path("/app/logs/bot_status.json") if status_file.exists(): data = json.loads(status_file.read_text()) _bot_status.update(data) except: pass _bot_status["tunnel_url"] = _load_tunnel_url() _bot_status["tunnel_method"] = _load_tunnel_method() # ── Routes ───────────────────────────────────────────────────────────────── @app.get("/health") async def health(): _update_bot_status() return { "status": "running", "bot_online": _bot_status.get("online", False), "server_online": _bot_status.get("server_online", False), "timestamp": datetime.now().isoformat(), } @app.get("/api/status") async def api_status(): _update_bot_status() return _bot_status @app.get("/api/tunnel") async def api_tunnel(): url = _load_tunnel_url() method = _load_tunnel_method() return {"url": url, "method": method, "active": bool(url)} @app.get("/api/debug") async def api_debug(): """Debug endpoint with server logs and process info""" info = { "tunnel_url": _load_tunnel_url(), "tunnel_method": _load_tunnel_method(), "bot_status_exists": Path("/app/logs/bot_status.json").exists(), "paper_jar_exists": Path("/app/server/paper.jar").exists(), "plugins": [], } # Check plugins plugins_dir = Path("/app/server/plugins") if plugins_dir.exists(): info["plugins"] = [f.name for f in plugins_dir.glob("*.jar")] # Get last 50 lines of MC server log try: log_file = Path("/app/server/logs/latest.log") if log_file.exists(): lines = log_file.read_text().splitlines()[-100:] info["mc_server_log"] = lines except: pass # Get bot stdout log try: bot_log = Path("/app/logs/bot_stdout.log") if bot_log.exists(): lines = bot_log.read_text().splitlines()[-50:] info["bot_stdout"] = lines except: pass # Get tunnel log try: method = _load_tunnel_method() if method == "bore": tunnel_log = Path("/app/logs/bore.log") else: tunnel_log = Path("/app/logs/ngrok.log") if tunnel_log.exists(): info["tunnel_log"] = tunnel_log.read_text().splitlines()[-20:] except: pass return info @app.get("/", response_class=HTMLResponse) async def dashboard(): _update_bot_status() return HTML_TEMPLATE @app.get("/api/chat-log") async def chat_log(): _update_bot_status() return {"messages": _bot_status.get("chat_log", [])[-50:]} @app.post("/api/bot/restart") async def restart_bot(): """Restart the mineflayer bot process""" try: subprocess.run(["pkill", "-f", "node bot.js"], capture_output=True) await asyncio.sleep(2) subprocess.Popen( ["node", "/app/bot.js"], cwd="/app", stdout=open("/app/logs/bot_stdout.log", "a"), stderr=open("/app/logs/bot_stderr.log", "a"), ) return {"status": "restarting"} except Exception as e: return {"status": "error", "message": str(e)} # ── WebSocket for live updates ───────────────────────────────────────────── _ws_clients = [] @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() _ws_clients.append(websocket) try: while True: _update_bot_status() await websocket.send_json(_bot_status) await asyncio.sleep(2) except WebSocketDisconnect: _ws_clients.remove(websocket) except: if websocket in _ws_clients: _ws_clients.remove(websocket) # ── HTML Dashboard ───────────────────────────────────────────────────────── HTML_TEMPLATE = """ Nav3D Agent Demo

Nav3D Agent Demo

OFFLINE

Agent Status

Position
0, 0, 0
Activity
--
Emotion
--
Dimension
--

Vitals

Health 0/20
Hunger 0/20

Current Thought

Waiting for agent to connect...

Connect to Server

Add this server in Minecraft (any version 1.8-1.21+):
Loading...

Server Info

Players Online
0
World Time
--
Connected Since
--

Chat Log

--:-- Waiting for messages...

Debug Logs

Click Refresh to load debug info...
"""