updated prompts

import time

import math

from ai2thor.platform import CloudRendering

from typing import Dict, List, Tuple, Optional, Any

from vagen.env.utils.env_utils import NoLoggerWarnings, set_seed

from vagen.env.utils.context_utils import parse_llm_raw_response, convert_numpy_to_PIL

from vagen.env.utils.context_utils import convert_numpy_to_PIL

from vagen.env.utils.parse_utils import parse_function_map

from .env_config import NavigationEnvConfig

from .prompt import system_prompt, system_prompt_vision, init_observation_template, action_template, format_prompt

from .prompt import system_prompt,init_observation_template, action_template, format_prompt

class NavigationEnv(BaseEnv):

    """Navigation environment from embodied bench. """   

    SUCCESS_THRESHOLD = 1

    SUCCESS_THRESHOLD = 2

    ValidEvalSets = [

        'base', 'common_sense', 'complex_instruction', 'visual_appearance', 'long_horizon'

    # Action descriptions

    DISCRETE_SKILLSET = [

        "Move forward by 0.25",

        "Move backward by 0.25",

        "Move rightward by 0.25",

        "Move leftward by 0.25",

        "Move forward by 0.5 meter",

        "Move backward by 0.5 meter",

        "Move rightward by 0.5 meter",

        "Move leftward by 0.5 meter",

        "Rotate to the right by 90 degrees.",

        "Rotate to the left by 90 degrees.",

        "Tilt the camera upward by 30 degrees.",

        self.number_of_episodes = len(self.dataset)

        self._current_episode_num = 0

        self._current_step = 0

        self._max_episode_steps = 20

        self._max_episode_steps = 30

        self._episode_start_time = 0

        self.is_holding = False

        self.episode_log = []

        info['episode_elapsed_seconds'] = time.time() - self._episode_start_time

        info['task_success'] = success

        info['last_action_success'] = self.env.last_event.metadata['lastActionSuccess']

        info["env_feedback"] ="Last action is executed successfully." if info['last_action_success'] else "Last action is not executed successfully."

        self.info = info

        # Update total reward

        self.total_reward += self.reward

            action_index: Index of the action to execute

        """

        if action_index == 1:  # Move forward by 0.25 meter

            self._last_event = self.env.step(action="MoveAhead", moveMagnitude=0.25)

            self._last_event = self.env.step(action="MoveAhead", moveMagnitude=0.5)

        elif action_index == 2:  # Move backward by 0.25 meter

            self._last_event = self.env.step(action="MoveBack", moveMagnitude=0.25)

            self._last_event = self.env.step(action="MoveBack", moveMagnitude=0.5)

        elif action_index == 3:  # Move right by 0.25 meter

            self._last_event = self.env.step(action="MoveRight", moveMagnitude=0.25)

            self._last_event = self.env.step(action="MoveRight", moveMagnitude=0.5)

        elif action_index == 4:  # Move left by 0.25 meter

            self._last_event = self.env.step(action="MoveLeft", moveMagnitude=0.25)

            self._last_event = self.env.step(action="MoveLeft", moveMagnitude=0.5)

        elif action_index == 5:  # Rotate clockwise by 90 degrees

            self._last_event = self.env.step(action="RotateRight", degrees=90)

        elif action_index == 6:  # Rotate counterclockwise by 90 degrees

                reward=self.reward,

                done=self.measure_success()[0],

                instruction=self.episode_language_instruction,

                env_feedback=self.info["env_feedback"]

            ) + "\n" + format_prompt_text

        return {

            add_example=True  # Always true for system prompt

        )

        if self.config.render_mode == "vision":

            return system_prompt_vision() + '\n' + format_prompt_text

        else:

        return system_prompt() + '\n' + format_prompt_text

    def compute_reward(self):

    env = NavigationEnv(config)

    print(env.system_prompt())

    obs, info = env.reset(seed=0)

    obs, info = env.reset(seed=3)

    print(obs["obs_str"])

    i = 0

    os.makedirs("./test_navigation", exist_ok=True)

    fov: int = 100

    multiview: bool = False

    render_mode: str= 'vision'

    max_actions_per_step: int = 10

    max_actions_per_step: int = 5

    max_action_penalty: float = -0.1

    format_reward: float = 0.5

    gpu_device: int = 0

def system_prompt():

    return """You are a home robot and perform navigation tasks according to instructions.

Navigation Guide

Goal: Achieve the human instruction

Actions you can take: moveahead, moveback, moveright, moveleft, rotateright, rotateleft, lookup, lookdown. 

moveahead: Move forward by 0.25 meter

moveback: Move backward by 0.25 meter

moveright: Move rightward by 0.25 meter

moveleft: Move leftward by 0.25 meter

rotateright: Rotate to the right by 90 degrees

rotateleft: Rotate to the left by 90 degrees

lookup: Tilt the camera upward by 30 degrees

lookdown: Tilt the camera downward by 30 degrees

Rewards:

Format correct: +0.5

Achieve the human instruction: +10.0

"""

def system_prompt_vision():

    return """You are a home robot and perform navigation tasks according to instructions.

Navigation Guide

You should follow the human instruction and navigate to the target location.

Actions you can take: moveahead, moveback, moveright, moveleft, rotateright, rotateleft, lookup, lookdown. 

moveahead: Move forward by 0.25 meter

moveback: Move backward by 0.25 meter

moveright: Move rightward by 0.25 meter

moveleft: Move leftward by 0.25 meter

moveahead: Move forward by 0.4 meter

moveback: Move backward by 0.4 meter

moveright: Move rightward by 0.4 meter

moveleft: Move leftward by 0.4 meter

rotateright: Rotate to the right by 90 degrees

rotateleft: Rotate to the left by 90 degrees

lookup: Tilt the camera upward by 30 degrees

Achieve the human instruction: +10.0

The instruction will be provided with each observation. Look at the image carefully and navigate to complete the instruction.

"""

Hints:

1. You can take multiple actions at a time, in most cases, if you find the target object is far away from you, you can call moveahead, moveleft and move right multiple times.

2. If you find yourself seems to be stuck, you can lookdown to see if there's any object above or below you, you can also rotate to see if there's any object behind you.

Example:

Round 1:

<image>

Reasoning: I can see the garbage can in the upper left corner of the image, next to the kitchen sink. To move there, we can go forward-left, but since there's a kitchen counter directly ahead, we should go left first. Following the strategy, I can go by first moving leftward.

Actions: moveleft, moveleft

Round 2:

Env_feedback: Last action is executed successfully.

<image>

Reasoning: From the secene, I see that by moving leftward, we are getting closer to the garbage can. Now, the garbage can is in front of me, slightly to the left. And there's a large area ahead of us. Following the strategy, I can go by first moving forward then moving leftward.

Actions: moveahead, moveahead,moveahead,moveleft

Round 3:

Env_feedback: Last action is executed successfully.

<image>

Reasoning: From the image we can see the garbage can is very close to us, still to our front-left. Moving leftward might be blocked but i can see that there is still space in front of me to get closer to the garbage can. Following the strategy, we can take about two steps forward then one step left to reach the garbage can.

Actions: moveahead, moveahead,moveleft

Round 4:

Env_feedback: Success"""

def init_observation_template(observation, instruction):

    return f"""[Initial Observation]:

{observation}

Human Instruction: {instruction}

Decide your next action(s).

"""

Decide your next action(s)."""

def action_template(valid_action, observation, reward, done, instruction):

def action_template(valid_action, observation, reward, done, instruction,env_feedback):

    return f"""After your answer, the extracted valid action is {valid_action}.

After that, the observation is:

{observation}

The environment feedback is: {env_feedback}

reward: {reward}

done: {done}

After that, the observation is:

{observation}

Human Instruction: {instruction}

Decide your next action(s).

"""

Decide your next action(s)."""

def free_think_format_prompt(max_actions_per_step, action_sep, add_example=True):

    base_prompt = f"""You can take up to {max_actions_per_step} action(s) at a time, separated by {action_sep}.

<think>...</think><answer>...</answer>"""

    if add_example:

        example = f"""e.g. <think>I can see from the sight the target object is right in the top left of me, I will move forward, then move left to access it.</think><answer>moveahead{action_sep}moveahead{action_sep}moveahead{action_sep}moveahead{action_sep}moveahead{action_sep}moveleft{action_sep}moveleft</answer>"""

        example = f"""e.g. <think>I can see from the sight the target object is right in the top left of me, I will move forward, then move left to access it.</think><answer>moveahead{action_sep}moveahead{action_sep}moveahead{action_sep}moveleft{action_sep}moveleft</answer>"""

        return base_prompt + '\n' + example

    return base_prompt

<answer>...</answer>"""

    if add_example:

        example = f"""e.g. <answer>moveahead{action_sep}moveahead{action_sep}moveahead{action_sep}moveahead{action_sep}moveahead{action_sep}moveleft{action_sep}moveleft</answer>"""

        example = f"""e.g. <answer>moveahead{action_sep}moveahead{action_sep}moveahead{action_sep}moveleft{action_sep}moveleft</answer>"""

        return base_prompt + '\n' + example

    return base_prompt

def grounding_format_prompt(max_actions_per_step, action_sep, add_example=True):

    base_prompt = f"""You can take up to {max_actions_per_step} action(s) at a time, separated by {action_sep}.

You should first give the current state, then your thought process, and finally your answer.

You should first give the description of current state, then your thought process, and finally your answer.

The state should be described in detail about what you see in the environment.

Your response should be in the format of:

<current_state>...</current_state><think>...</think><answer>...</answer>"""