navigation_ui.py

import base64
import json
import logging
import os
import subprocess
import sys
import threading
import time
import uuid
from datetime import datetime, timedelta
from typing import Dict, List, Optional

import gradio as gr
import numpy as np
import open3d as o3d
import plotly.graph_objects as go
import requests
from fastapi import APIRouter, FastAPI, HTTPException, status, BackgroundTasks, Response
from pydantic import BaseModel

import asyncio
import uvicorn
from collections import defaultdict

BACKEND_URL = os.getenv("BACKEND_URL", "http://localhost:8001") # fastapi server
API_ENDPOINTS = {
    "submit_task": f"{BACKEND_URL}/predict/video",
    "query_status": f"{BACKEND_URL}/predict/task",
    "get_result": f"{BACKEND_URL}/predict"
}


SCENE_CONFIGS = {
    "scene_1": {
        "description": "Modern Apartment",
        "name": "17DRP5sb8fy",
        "glb_path": "scene_assets/scene1_no_ceiling.glb"  # PLY file path
    },
    "scene_2": {
        "description": "Office Building",
        "name": "r1Q1Z4BcV1o",
        "glb_path": "scene_assets/scene2_no_ceiling.glb"
    },
    "scene_3": {
        "description": "University Campus",
        "name": "dhjEzFoUFzH",
        "glb_path": "scene_assets/scene3_no_ceiling.glb"
    },
}

EPISODE_CONFIGS = {
    "episode_1": {
        "description": "1",
    },
    "episode_2": {
        "description": "2",
    },
    "episode_3": {
        "description": "3",
    },
    "episode_4": {
        "description": "4",
    }
}

MODEL_CHOICES = [] 


###############################################################################

SESSION_TASKS = {}
IP_REQUEST_RECORDS = defaultdict(list)
IP_LIMIT = 5  

def is_request_allowed(ip: str) -> bool:
    now = datetime.now()
    IP_REQUEST_RECORDS[ip] = [t for t in IP_REQUEST_RECORDS[ip] if now - t < timedelta(minutes=1)]
    if len(IP_REQUEST_RECORDS[ip]) < IP_LIMIT:
        IP_REQUEST_RECORDS[ip].append(now)
        return True
    return False

###############################################################################


# Log directory path
LOG_DIR = "~/logs"
os.makedirs(LOG_DIR, exist_ok=True)
ACCESS_LOG = os.path.join(LOG_DIR, "access.log")
SUBMISSION_LOG = os.path.join(LOG_DIR, "submissions.log")

def log_access(user_ip: str = None, user_agent: str = None):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    log_entry = {
        "timestamp": timestamp,
        "type": "access",
        "user_ip": user_ip or "unknown",
        "user_agent": user_agent or "unknown"
    }
    
    with open(ACCESS_LOG, "a") as f:
        f.write(json.dumps(log_entry) + "\n")

def log_submission(scene: str, prompt: str, model: str, user: str = "anonymous", res: str = "unknown"):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    log_entry = {
        "timestamp": timestamp,
        "type": "submission",
        "user": user,
        "scene": scene,
        "prompt": prompt,
        "model": model,
        #"max_step": str(max_step),
        "res": res
    }
    
    with open(SUBMISSION_LOG, "a") as f:
        f.write(json.dumps(log_entry) + "\n")

def read_logs(log_type: str = "all", max_entries: int = 50) -> list:
    logs = []
    
    if log_type in ["all", "access"]:
        try:
            with open(ACCESS_LOG, "r") as f:
                for line in f:
                    logs.append(json.loads(line.strip()))
        except FileNotFoundError:
            pass
    
    if log_type in ["all", "submission"]:
        try:
            with open(SUBMISSION_LOG, "r") as f:
                for line in f:
                    logs.append(json.loads(line.strip()))
        except FileNotFoundError:
            pass
    
    # Sorted by timestemp
    logs.sort(key=lambda x: x["timestamp"], reverse=True)
    return logs[:max_entries]

def format_logs_for_display(logs: list) -> str:
    if not logs:
        return "No log record"
    
    markdown = "### System Access Log\n\n"
    markdown += "| Time | Type | User/IP | Details |\n"
    markdown += "|------|------|---------|----------|\n"
    
    for log in logs:
        timestamp = log.get("timestamp", "unknown")
        log_type = "Access" if log.get("type") == "access" else "Submission"
        
        if log_type == "Access":
            user = log.get("user_ip", "unknown")
            details = f"User-Agent: {log.get('user_agent', 'unknown')}"
        else:
            user = log.get("user", "anonymous")
            result = log.get('res', 'unknown')
            if result != "success": 
                if len(result) > 40:  # Adjust this threshold as needed
                    result = f"{result[:20]}...{result[-20:]}"
            details = f"Scene: {log.get('scene', 'unknown')}, Prompt: {log.get('prompt', '')}, Model: {log.get('model', 'unknown')}, result: {result}"
        
        markdown += f"| {timestamp} | {log_type} | {user} | {details} |\n"
    
    return markdown


def submit_to_backend(
    scene: str,
    prompt: str,
    episode: str,
    user: str = "Gradio-user",
) -> dict:
    job_id = str(uuid.uuid4())

    scene_index = scene.split("_")[-1]
    episode_index = episode.split("_")[-1]

    data = {
        "task_type": "vln_eval",  # Identify task type
        "instruction": prompt,
        "scene_index": scene_index,
        "episode_index": episode_index,
    }
    
    payload = {
        "user": user,
        "task": "robot_navigation",
        "job_id": job_id,
        "data": json.dumps(data)
    }
    
    try:
        headers = {"Content-Type": "application/json"}
        response = requests.post(
            API_ENDPOINTS["submit_task"],
            json=payload,
            headers=headers,
            timeout=600
        )
        return response.json()
    except Exception as e:
        return {"status": "error", "message": str(e)}

def get_task_status(task_id: str) -> dict:
    try:
        response = requests.get(f"{API_ENDPOINTS['query_status']}/{task_id}", timeout=600)
        try:
            return response.json()  
        except json.JSONDecodeError:
            return {"status": "error", "message": response.text}
    except Exception as e:
        return {"status": "error", "message": str(e)}


def get_task_result(task_id: str) -> Optional[dict]:
    try:
        response = requests.get(
            f"{API_ENDPOINTS['get_result']}/{task_id}",
            timeout=5
        )
        return response.json()
    except Exception as e:
        print(f"Error fetching result: {e}")
        return None

def run_simulation(
    scene: str,
    prompt: str,
    episode: str,
    history: list,
    request: gr.Request
) -> dict:
    model = "InternNav-VLA"

    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    scene_desc = SCENE_CONFIGS.get(scene, {}).get("description", scene)

    user_ip = request.client.host if request else "unknown"
    session_id = request.session_hash

    if not is_request_allowed(user_ip):
        log_submission(scene, prompt, model, user_ip, "IP blocked temporarily")
        raise gr.Error("Too many requests from this IP. Please wait and try again one minute later.")
    
    submission_result = submit_to_backend(scene, prompt, episode)
    print("submission_result: ", submission_result)

    if submission_result.get("status") != "pending":
        log_submission(scene, prompt, model, user_ip, "Submission failed")
        raise gr.Error(f"Submission failed: {submission_result.get('message', 'unknown issue')}")
    
    try:
        task_id = submission_result["task_id"]
        SESSION_TASKS[session_id] = task_id

        gr.Info(f"Simulation started, task_id: {task_id}")
        time.sleep(5)
        # Get Task Status
        status = get_task_status(task_id)
        print("first status: ", status)
        result_folder = status.get("result", "")
    except Exception as e:
        log_submission(scene, prompt, model, user_ip, str(e))
        raise gr.Error(f"error occurred when parsing submission result from backend: {str(e)}")

    while True:
        status = get_task_status(task_id)
        if status.get("status") == "completed":
            break
        elif status.get("status") == "failed":
            break
        time.sleep(1)
    if status.get("status") == "completed":
        import base64
        video_bytes = base64.b64decode(status.get("video"))
        receive_time = time.time()
        with open(f"received_video_{receive_time}.mp4", "wb") as f:
            f.write(video_bytes)
        video_path = f"received_video_{receive_time}.mp4"
        new_entry = {
            "timestamp": timestamp,
            "scene": scene,
            "model": model,
            "prompt": prompt,
            "video_path": video_path
        }
        
        updated_history = history + [new_entry]
        
        if len(updated_history) > 10:
            updated_history = updated_history[:10]
        
        print("updated_history:", updated_history)
        log_submission(scene, prompt, model, user_ip, "success")
        gr.Info("Simulation completed successfully!")
        yield video_path, updated_history

    elif status.get("status") == "failed":
        log_submission(scene, prompt, model, user_ip, status.get('result', 'backend error'))
        raise gr.Error(f"task execution fails: {status.get('result', 'backend error')}")
        yield None, history

    elif status.get("status") == "terminated":
        log_submission(scene, prompt, model, user_ip, "terminated")
        video_path = os.path.join(result_folder, "output.mp4")
        if os.path.exists(video_path):
            return f" task {task_id} terminated with some results", video_path, history
        else:
            return f" task {task_id} terminated without any results", None, history

    else:
        log_submission(scene, prompt, model, user_ip, "missing task's status from backend")
        yield None, history

###################################################################################################################
def update_history_display(history: list) -> list:
    print("update_history_display")
    updates = []
    
    for i in range(10):
        if i < len(history):
            entry = history[i]
            updates.extend([
                gr.update(visible=True),
                gr.update(visible=True, label=f"Simulation {i+1}  scene: {entry['scene']}, prompt: {entry['prompt']}", open=False),
                gr.update(value=entry['video_path'], visible=True),
                gr.update(value=f"{entry['timestamp']}")
            ])
            print(f'update video')
            print(entry['video_path'])
        else:
            updates.extend([
                gr.update(visible=False),
                gr.update(visible=False),
                gr.update(value=None, visible=False),
                gr.update(value="")
            ])
    print("update_history_display end!!")
    return updates

def update_scene_display(scene: str):
    print(f"update_scene_display {scene}")
    config = SCENE_CONFIGS.get(scene, {})
    glb_path = config.get("glb_path", "")

    # Validate if file path exists
    if not os.path.exists(glb_path):
        return None, None

    return None, glb_path

def update_episode_display(scene: str, episode: str):
    print(f"update_episode_display {scene} {episode}")
    config = SCENE_CONFIGS.get(scene, {})
    scene_name = config.get("name", "")
    episode_id = int(episode[-1])
    image_path = os.path.join("scene_assets", f"{scene_name}_{episode_id-1}.jpg")
    print(f"image_path {image_path}")
    # vaild if file path exists
    if not os.path.exists(image_path):
        return None

    return image_path
def update_log_display():
    logs = read_logs()
    return format_logs_for_display(logs)
##############################################################################


def cleanup_session(request: gr.Request):
    session_id = request.session_hash
    task_id = SESSION_TASKS.pop(session_id, None)
    if task_id:
        try:
            requests.post(f"{BACKEND_URL}/predict/terminate/{task_id}", timeout=3)
            print(f"Task Terminated: {task_id}")
        except Exception as e:
            print(f"Task Termination Failed: {task_id}: {e}")



###############################################################################

custom_css = """
#simulation-panel {
    border-radius: 8px;
    padding: 20px;
    background: #f9f9f9;
    box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}
#result-panel {
    border-radius: 8px;
    padding: 20px;
    background: #f0f8ff;
}
.dark #simulation-panel { background: #2a2a2a; }
.dark #result-panel { background: #1a2a3a; }

.history-container {
    max-height: 600px;
    overflow-y: auto;
    margin-top: 20px;
}

.history-accordion {
    margin-bottom: 10px;
}

.scene-preview {
    height: 400px;
    border: 1px solid #ddd;
    border-radius: 8px;
    overflow: hidden;
}
"""

with gr.Blocks(title="Robot Navigation Inference", css=custom_css) as demo:
    gr.Markdown("""
    # 🧭 Habitat Robot Navigation Demo
    ### Simulation Test Based on Habitat Framework
    """)
    
    history_state = gr.State([])

    with gr.Row():
        with gr.Column(elem_id="simulation-panel"):
            gr.Markdown("### Simulation Task Configuration")
            with gr.Row():
                scene_dropdown = gr.Dropdown(
                    label="Select Scene",
                    choices=list(SCENE_CONFIGS.keys()),
                    value="scene_1",
                    interactive=True,
            )
                episode_dropdown = gr.Dropdown(
                    label="Select Start Position",
                    choices=list(EPISODE_CONFIGS.keys()),
                    value="episode_1",
                    interactive=True,
                )

            with gr.Row():
                scene_preview = gr.Model3D(elem_classes=["scene-preview"],
                camera_position=(90.0, 120, 20000.0),
                #display_mode="solid"
                )
                fps_preview = gr.Image(label="FPS Preview")
            
            scene_description = gr.Markdown("### Scene preview")
            
            prompt_input = gr.Textbox(
                label="Navigation Instruction",
                value="Exit the bedroom and turn left. Walk straight passing the gray couch and stop near the rug.",
                placeholder="e.g.: 'Exit the bedroom and turn left. Walk straight passing the gray couch and stop near the rug.'",
                lines=2,
                max_lines=4
            )
                        
            scene_dropdown.change(
                update_scene_display,
                inputs=scene_dropdown,
                outputs=[scene_description, scene_preview]
            ).then(
                update_episode_display,
                inputs=[scene_dropdown, episode_dropdown],
                outputs=[fps_preview]
            )
            
            episode_dropdown.change(
                update_episode_display,
                inputs=[scene_dropdown, episode_dropdown],
                outputs=[fps_preview]
            )
            
            submit_btn = gr.Button("Start Navigation Simulation", variant="primary")

       
        with gr.Column(elem_id="result-panel"):
            gr.Markdown("### Latest Simulation Result")

            # Video Output
            video_output = gr.Video(
                label="Live",
                interactive=False,
                format="mp4",
                autoplay=True,
                # streaming=True
            )
            
            with gr.Column() as history_container:
                gr.Markdown("### History")
                gr.Markdown("#### History will be reset after refresh")
                
                history_slots = []
                for i in range(10):
                    with gr.Column(visible=False) as slot:
                        with gr.Accordion(visible=False, open=False) as accordion:
                            video = gr.Video(interactive=False)  
                            detail_md = gr.Markdown() 
                    history_slots.append((slot, accordion, video, detail_md))  
    
    with gr.Accordion("View System Log (DEV ONLY)", open=False):
        logs_display = gr.Markdown()
        refresh_logs_btn = gr.Button("Refresh Log", variant="secondary")
        
        refresh_logs_btn.click(
            update_log_display,
            outputs=logs_display
        )

    gr.Examples(
        examples=[
            ["scene_1", "Exit the bedroom and turn left. Walk straight passing the gray couch and stop near the rug.", "episode_0"],
            ["scene_2", "Go from reception to conference room passing the water cooler.", "episode_1"],
            ["scene_3", "From the classroom, go to the library via the main hall.", "episode_2"],
            ["scene_4", "From emergency room to pharmacy passing nurse station.", "episode_3"]
        ],
        inputs=[scene_dropdown, prompt_input, episode_dropdown],
        label="Navigation Task Example"
    )
    
    submit_btn.click(
        fn=run_simulation,
        inputs=[scene_dropdown, prompt_input, episode_dropdown, history_state],
        outputs=[video_output, history_state],
        queue=True,
        api_name="run_simulation"
    ).then(
        fn=update_history_display,
        inputs=history_state,
        outputs=[comp for slot in history_slots for comp in slot],
        queue=True
    ).then(
        fn=update_log_display,
        outputs=logs_display,
    )


    demo.load(
        fn=lambda: update_scene_display("scene_1"),
        outputs=[scene_description, scene_preview]
    ).then(
        fn=update_log_display,
        outputs=logs_display
    )
    demo.load(
        fn=lambda: update_episode_display("scene_1", "episode_1"),
        outputs=[fps_preview]
    )

    def record_access(request: gr.Request):
        user_ip = request.client.host if request else "unknown"
        user_agent = request.headers.get("user-agent", "unknown")
        log_access(user_ip, user_agent)
        return update_log_display()   

    demo.load(
        fn=record_access,
        inputs=None,
        outputs=logs_display,
        queue=False
    )

    demo.queue(default_concurrency_limit=8)

    demo.unload(fn=cleanup_session)

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=5750, debug=True, share = True, allowed_paths=["/mnt"])