app.py

'''
author:        caishaofei <caishaofei@stu.pku.edu.cn>
date:          2024-09-20 20:10:44
Copyright © Team CraftJarvis All rights reserved
'''
import re
import os
import cv2
import time
from pathlib import Path
import argparse
import requests
import gradio as gr
import torch
import numpy as np
from io import BytesIO
from PIL import Image, ImageDraw
from rocket.arm.sessions import Session, Pointer

COLORS = [
    (255, 0, 0), (0, 255, 0), (0, 0, 255), 
    (255, 255, 0), (255, 0, 255), (0, 255, 255),
    (255, 255, 255), (0, 0, 0), (128, 128, 128),
    (128, 0, 0), (128, 128, 0), (0, 128, 0),
    (128, 0, 128), (0, 128, 128), (0, 0, 128),
]

SEGMENT_MAPPING = {
    "Hunt": 0, "Use": 3, "Mine": 2, "Interact": 3, "Craft": 4, "Switch": 5, "Approach": 6
}

NOOP_ACTION = {
    "back": 0,
    "drop": 0,
    "forward": 0,
    "hotbar.1": 0,
    "hotbar.2": 0,
    "hotbar.3": 0,
    "hotbar.4": 0,
    "hotbar.5": 0,
    "hotbar.6": 0,
    "hotbar.7": 0,
    "hotbar.8": 0,
    "hotbar.9": 0,
    "inventory": 0,
    "jump": 0,
    "left": 0,
    "right": 0,
    "sneak": 0,
    "sprint": 0,
    "camera": np.array([0, 0]),
    "attack": 0,
    "use": 0,
}

def reset_fn(env_name, session):
    image = session.reset(env_name)
    return image, session

def step_fn(act_key, session):
    action = NOOP_ACTION.copy()
    if act_key != "null":
        action[act_key] = 1
    image = session.step(action)
    return image, session

def loop_step_fn(steps, session):
    for i in range(steps):
        image = session.step()
        status = f"Running Agent `Rocket` steps: {i+1}/{steps}. "
        yield image, session.num_steps, status, session

def clear_memory_fn(session):
    image = session.current_image
    session.clear_agent_memory()
    return image, "0", session

def get_points_with_draw(image, label, session, evt: gr.SelectData):
    points = session.points
    point_label = session.points_label
    x, y = evt.index[0], evt.index[1]
    point_radius, point_color = 5, (0, 255, 0) if label == 'Add Points' else (255, 0, 0)
    points.append([x, y])
    point_label.append(1 if label == 'Add Points' else 0)
    cv2.circle(image, (x, y), point_radius, point_color, -1)
    return image, session

def clear_points_fn(session):
    session.clear_points()
    return session.current_image, session

def segment_fn(session):
    if len(session.points) == 0:
        return session.current_image, session
    session.segment()
    image = session.apply_mask()
    return image, session

def clear_segment_fn(session):
    session.clear_obj_mask()
    session.tracking_flag = False
    return session.current_image, False, session

def set_tracking_mode(tracking_flag, session):
    session.tracking_flag = tracking_flag
    return session

def set_segment_type(segment_type, session):
    session.segment_type = segment_type
    return session

def play_fn(session):
    image = session.step()
    return image, session

memory_length = gr.Textbox(value="0", interactive=False, show_label=False)

def make_video_fn(session, make_video, save_video, progress=gr.Progress()):
    images = session.image_history
    if len(images) == 0:
        return session, make_video, save_video
    filepath = "rocket.mp4"
    h, w = images[0].shape[:2]
    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    video = cv2.VideoWriter(filepath, fourcc, 20.0, (w, h))
    for image in progress.tqdm(images):
        image = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
        video.write(image)
    video.release()
    session.image_history = []
    return session, gr.Button("Make Video", visible=False), gr.DownloadButton("Download!", value=filepath, visible=True)

def save_video_fn(session, make_video, save_video):
    return session, gr.Button("Make Video", visible=True), gr.DownloadButton("Download!", visible=False)

def choose_sam_fn(sam_choice, session):
    session.sam_choice = sam_choice
    session.load_sam()
    return session

def molmo_fn(molmo_text, molmo_session, rocket_session, display_image):
    image = rocket_session.current_image.copy()
    points = molmo_session.gen_point(image=image, prompt=molmo_text)
    molmo_result = molmo_session.molmo_result
    for x, y in points:
        x, y = int(x), int(y)
        point_radius, point_color = 5, (0, 255, 0) 
        rocket_session.points.append([x, y])
        rocket_session.points_label.append(1)
        cv2.circle(display_image, (x, y), point_radius, point_color, -1)
    return molmo_result, display_image

def extract_points(data):
    # 匹配 x 和 y 坐标的值，支持 <points> 和 <point> 标签
    pattern = r'x\d?="([-+]?\d*\.\d+|\d+)" y\d?="([-+]?\d*\.\d+|\d+)"'
    points = re.findall(pattern, data)
    # 将提取到的坐标转换为浮点数
    points = [(float(x)/100*640, float(y)/100*360) for x, y in points]
    return points

def draw_gradio_components(args):

    with gr.Blocks() as demo:
        
        gr.Markdown(
            """
            # Welcome to Explore ROCKET-1 in Minecraft!!
            ## Please follow next steps to interact with the agent:
            1. Reset the environment by selecting an environment name.
            2. Select a SAM2 checkpoint to load.
            3. Use your mouse to add or remove points on the image. 
            4. Select the segment type you want to perform.
            5. Enable `tracking` mode if you want to track objects while stepping actions. 
            6. Click `New Segment` to segment the image based on the points you added. 
            7. Call the agent by clicking `Call Rocket` to run the agent for a certain number of steps. 
            ## Hints:
            1. You can use the `Make Video` button to generate a video of the agent's actions.
            2. You can use the `Clear Memory` button to clear the ROCKET-1's memory. 
            3. You can use the `Clear Segment` button to clear SAM's memory. 
            4. You can use the `Manually Step` button to manually step the agent. 
            """
        )
        
        rocket_session = gr.State(Session(
            sam_path=args.sam_path,
        ))
        molmo_session = gr.State(Pointer(
            model_id="molmo-72b-0924", 
            model_url="http://172.17.30.127:8000/v1", 
        ))
        with gr.Row():
            
            with gr.Column(scale=2):
                # start_image = Image.open("start.png").resize((640, 360))
                start_image = np.zeros((360, 640, 3), dtype=np.uint8)
                
                with gr.Group():
                    display_image = gr.Image(
                        value=np.array(start_image), 
                        interactive=False, 
                        show_label=False, 
                        label="Real-time Environment Observation", 
                        streaming=True
                    )
                    display_status = gr.Textbox("Status Bar", interactive=False, show_label=False)
            
            with gr.Column(scale=1):
                
                sam_choice = gr.Radio(
                    choices=["large", "base", "small", "tiny"],
                    value="base",
                    label="Select SAM2 checkpoint",
                )
                sam_choice.select(fn=choose_sam_fn, inputs=[sam_choice, rocket_session], outputs=[rocket_session], show_progress=False)
                
                with gr.Group():
                    add_or_remove = gr.Radio(
                        choices=["Add Points", "Remove Areas"], 
                        value="Add Points", 
                        label="Use you mouse to add or remove points",
                    )
                    clear_points_btn = gr.Button("Clear Points")
                    clear_points_btn.click(clear_points_fn, inputs=[rocket_session], outputs=[display_image, rocket_session], show_progress=True)
                
                with gr.Group():
                    segment_type = gr.Radio(
                        choices=["Approach", "Interact", "Hunt", "Mine", "Craft", "Switch"],
                        value="Approach", 
                        label="What do you want with this segment?",
                    )
                    track_flag = gr.Checkbox(True, label="Enable tracking objects while steping actions")
                    track_flag.select(fn=set_tracking_mode, inputs=[track_flag, rocket_session], outputs=[rocket_session], show_progress=False)
                    with gr.Group(), gr.Row():
                        new_segment_btn = gr.Button("New Segment")
                        clear_segment_btn = gr.Button("Clear Segment")
                        new_segment_btn.click(segment_fn, inputs=[rocket_session], outputs=[display_image, rocket_session], show_progress=True)
                        clear_segment_btn.click(clear_segment_fn, inputs=[rocket_session], outputs=[display_image, track_flag, rocket_session], show_progress=True)

            display_image.select(get_points_with_draw, inputs=[display_image, add_or_remove, rocket_session], outputs=[display_image, rocket_session])
            segment_type.select(set_segment_type, inputs=[segment_type, rocket_session], outputs=[rocket_session], show_progress=False)

        with gr.Row():
            with gr.Group():
                env_list = [f"rocket/{x.stem}" for x in Path("../env_configs/rocket").glob("*.yaml") if 'base' not in x.name != 'base']
                env_name = gr.Dropdown(env_list, multiselect=False, min_width=200, show_label=False, label="Env Name")
                reset_btn = gr.Button("Reset Environment")
                reset_btn.click(fn=reset_fn, inputs=[env_name, rocket_session], outputs=[display_image, rocket_session], show_progress=True)

            with gr.Group():
                action_list = [x for x in NOOP_ACTION.keys()]
                act_key = gr.Dropdown(action_list, multiselect=False, min_width=200, show_label=False, label="Action")
                step_btn = gr.Button("Manually Step")
                step_btn.click(fn=step_fn, inputs=[act_key, rocket_session], outputs=[display_image, rocket_session], show_progress=False)

            with gr.Group():
                steps = gr.Slider(1, 600, 30, 1, label="Steps", show_label=False)
                play_btn = gr.Button("Call Rocket")
                play_btn.click(fn=loop_step_fn, inputs=[steps, rocket_session], outputs=[display_image, memory_length, display_status, rocket_session], show_progress=False)

            with gr.Group():
                memory_length.render()
                clear_states_btn = gr.Button("Clear Memory")
                clear_states_btn.click(fn=clear_memory_fn, inputs=rocket_session, outputs=[display_image, memory_length, rocket_session], show_progress=False)
            
            make_video_btn = gr.Button("Make Video")
            save_video_btn = gr.DownloadButton("Download!!", visible=False)
            make_video_btn.click(make_video_fn, inputs=[rocket_session, make_video_btn, save_video_btn], outputs=[rocket_session, make_video_btn, save_video_btn], show_progress=False)
            save_video_btn.click(save_video_fn, inputs=[rocket_session, make_video_btn, save_video_btn], outputs=[rocket_session, make_video_btn, save_video_btn], show_progress=False)
        with gr.Row():
            with gr.Group():
                molmo_text = gr.Textbox("pinpoint the", label="Molmo Text", show_label=True, min_width=200)
                molmo_btn = gr.Button("Generate")
                output_text = gr.Textbox("", label="Molmo Output", show_label=False, min_width=200)
                molmo_btn.click(molmo_fn, inputs=[molmo_text, molmo_session, rocket_session, display_image],outputs=[output_text, display_image],show_progress=False)

        demo.queue()
        demo.launch(share=False,server_port=args.port)

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument("--port", type=int, default=7860)
    parser.add_argument("--sam-path", type=str, default="/app/ROCKET-1/rocket/realtime_sam/checkpoints")
    parser.add_argument("--molmo-id", type=str, default="molmo-72b-0924")
    parser.add_argument("--molmo-url", type=str, default="http://127.0.0.1:8000/v1")
    args = parser.parse_args()
    draw_gradio_components(args)