app.py

"""
Minecraft AI Agent Research — Monitoring Dashboard
Gradio-based monitoring interface for the MC server.
This also serves the API endpoints needed for bot discovery.
"""

import gradio as gr
import subprocess
import json
import os
import time
import threading
import requests
import psutil

# ── State ─────────────────────────────────────────────────────────────────────
_state = {
    "mc_online": False,
    "ngrok_url": "",
    "ngrok_tcp": "",
    "players": [],
    "uptime": 0,
    "start_time": time.time(),
    "mc_logs": [],
    "tps": "N/A",
}

def get_ngrok_url():
    """Fetch the ngrok tunnel URL from local API"""
    try:
        resp = requests.get("http://localhost:4040/api/tunnels", timeout=3)
        data = resp.json()
        for t in data.get("tunnels", []):
            url = t.get("public_url", "")
            if url.startswith("tcp://"):
                _state["ngrok_url"] = url
                _state["ngrok_tcp"] = url
                return url
    except:
        pass
    return _state.get("ngrok_url", "Not available")

def get_mc_status():
    """Check if MC server is running by reading latest log"""
    try:
        log_path = "/app/mc-server/logs/latest.log"
        if os.path.exists(log_path):
            with open(log_path, "r") as f:
                lines = f.readlines()
                _state["mc_logs"] = lines[-50:] if len(lines) > 50 else lines
                for line in lines:
                    if "Done (" in line:
                        _state["mc_online"] = True
                    if "lost connection" in line or "Stopping server" in line:
                        pass  # don't mark offline for individual disconnects
    except:
        pass
    return _state["mc_online"]

def parse_mc_address(tcp_url):
    """Parse tcp://host:port into host, port"""
    if not tcp_url or not tcp_url.startswith("tcp://"):
        return "N/A", "N/A"
    addr = tcp_url.replace("tcp://", "")
    parts = addr.rsplit(":", 1)
    if len(parts) == 2:
        return parts[0], parts[1]
    return addr, "25565"

def check_server_status():
    """Background status checker"""
    while True:
        get_mc_status()
        get_ngrok_url()
        _state["uptime"] = int(time.time() - _state["start_time"])
        time.sleep(15)

# Start background checker
_status_thread = threading.Thread(target=check_server_status, daemon=True)
_status_thread.start()

# ── Gradio UI ─────────────────────────────────────────────────────────────────

def refresh_status():
    """Get current status for display"""
    get_mc_status()
    get_ngrok_url()

    mc_status = "🟢 Online" if _state["mc_online"] else "🔴 Offline"
    uptime_h = _state["uptime"] // 3600
    uptime_m = (_state["uptime"] % 3600) // 60

    ngrok_url = _state.get("ngrok_url", "Not available")
    host, port = parse_mc_address(ngrok_url)

    # Format connect info
    if ngrok_url != "Not available":
        connect_info = f"**Server Address:** `{host}:{port}`\n\nConnect with Minecraft 1.21.4 using this address to watch the AI agent play."
    else:
        connect_info = "Tunnel not available yet. Please wait..."

    # Resource usage
    try:
        mem = psutil.virtual_memory()
        cpu = psutil.cpu_percent(interval=0.5)
        resources = f"CPU: {cpu}% | RAM: {mem.percent}% ({mem.used // 1024 // 1024}MB / {mem.total // 1024 // 1024}MB)"
    except:
        resources = "N/A"

    return mc_status, f"{uptime_h}h {uptime_m}m", ngrok_url, connect_info, resources

def get_logs():
    """Get recent MC server logs"""
    get_mc_status()
    logs = _state.get("mc_logs", [])
    return "".join(logs[-30:])

def get_connect_info():
    """Get connection info formatted for users"""
    get_ngrok_url()
    ngrok_url = _state.get("ngrok_url", "")
    if not ngrok_url:
        return "Server tunnel not ready yet. Check back in a minute."

    host, port = parse_mc_address(ngrok_url)
    return f"""## Connect to Watch

**Minecraft Java Edition 1.21.4**

Server: `{host}`
Port: `{port}`

Add this as a server in your Minecraft multiplayer menu to watch the AI agent play in real time.

The agent (Zelin_) connects autonomously and plays survival mode — exploring, mining, building, and fighting mobs."""

# ── Build Gradio Interface ────────────────────────────────────────────────────

with gr.Blocks(
    title="Minecraft AI Agent Research",
    theme=gr.themes.Soft(),
    css="""
    .status-card { padding: 20px; border-radius: 10px; margin: 10px 0; }
    .connect-box { background: #1a1a2e; color: #e0e0e0; padding: 20px; border-radius: 10px; font-family: monospace; }
    """
) as demo:

    gr.Markdown("""
    # 🤖 Minecraft AI Agent Research
    Autonomous AI agent navigating and interacting in a Minecraft world.
    """)

    with gr.Row():
        with gr.Column(scale=1):
            gr.Markdown("### Server Status")
            status_text = gr.Textbox(label="MC Server", value="Checking...", interactive=False)
            uptime_text = gr.Textbox(label="Uptime", value="0h 0m", interactive=False)
            tunnel_text = gr.Textbox(label="Tunnel URL", value="Connecting...", interactive=False)
            resources_text = gr.Textbox(label="Resources", value="N/A", interactive=False)
            refresh_btn = gr.Button("🔄 Refresh", variant="primary")

        with gr.Column(scale=1):
            connect_md = gr.Markdown(value="Loading connection info...")
            refresh_btn2 = gr.Button("📋 Get Connection Info")

    with gr.Row():
        with gr.Column():
            gr.Markdown("### Recent Logs")
            logs_text = gr.Textbox(label="Server Logs", value="", lines=15, interactive=False, max_lines=15)
            logs_btn = gr.Button("📜 Refresh Logs")

    # Wire up buttons
    refresh_btn.click(
        fn=refresh_status,
        outputs=[status_text, uptime_text, tunnel_text, connect_md, resources_text]
    )
    refresh_btn2.click(fn=get_connect_info, outputs=[connect_md])
    logs_btn.click(fn=get_logs, outputs=[logs_text])

    # Auto-refresh every 30s
    demo.load(fn=refresh_status, outputs=[status_text, uptime_text, tunnel_text, connect_md, resources_text])

# ── API Endpoints (via Gradio mounting) ───────────────────────────────────────
# We need to add custom API routes for the bot to discover the server

import http.server
import socketserver
from functools import partial

class APIHandler(http.server.BaseHTTPRequestHandler):
    """Simple API server for bot discovery"""

    def do_GET(self):
        if self.path == "/api/ngrok":
            self.send_json({"url": _state.get("ngrok_url", ""), "tcp": _state.get("ngrok_tcp", "")})
        elif self.path == "/api/status":
            self.send_json({
                "mc_server": "online" if _state.get("mc_online") else "offline",
                "ngrok_url": _state.get("ngrok_url", ""),
                "uptime": _state.get("uptime", 0),
            })
        elif self.path == "/health":
            self.send_json({"status": "ok", "mc": _state.get("mc_online", False)})
        else:
            self.send_response(404)
            self.end_headers()

    def send_json(self, data):
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(json.dumps(data).encode())

    def log_message(self, format, *args):
        pass  # Suppress API logs

# Start API server on a different port (7861) — Gradio on 7860
def run_api():
    with socketserver.TCPServer(("", 7861), APIHandler) as httpd:
        httpd.serve_forever()

api_thread = threading.Thread(target=run_api, daemon=True)
api_thread.start()

# Mount API handler on the main port too via middleware
# We'll use a custom Gradio launch with the API routes
# Actually, let's use a simpler approach: patch Gradio's FastAPI app

original_launch = demo.launch

def patched_launch(*args, **kwargs):
    # Launch Gradio first
    result = original_launch(*args, **kwargs)

    # Add API routes to Gradio's FastAPI app
    try:
        from fastapi import FastAPI
        app = demo.app  # Gradio creates a FastAPI app

        @app.get("/api/ngrok")
        async def api_ngrok():
            get_ngrok_url()
            return {"url": _state.get("ngrok_url", ""), "tcp": _state.get("ngrok_tcp", "")}

        @app.get("/api/status")
        async def api_status():
            return {
                "mc_server": "online" if _state.get("mc_online") else "offline",
                "ngrok_url": _state.get("ngrok_url", ""),
                "uptime": _state.get("uptime", 0),
            }

        @app.get("/health")
        async def health():
            return {"status": "ok", "mc": _state.get("mc_online", False)}

        print("[API] Routes mounted on Gradio FastAPI app")
    except Exception as e:
        print(f"[API] Could not mount routes: {e}")

    return result

demo.launch = patched_launch

# Launch
if __name__ == "__main__":
    demo.launch(
        server_name="0.0.0.0",
        server_port=7860,
        show_error=True,
        quiet=True,
    )