Class Completion_Estimation_Experts

DiscussNav.py:234–290

        return observe_results

class Completion_Estimation_Experts:
    def summarize_observation(self, curr_observe, landmarks):
        direction_id = int(curr_observe.split("Navigable Viewpoint")[0].replace("Direction","").replace("(lower position indicates down stairs)","").replace("(higher position indicates up stairs)","").strip())
        directions = ["Front, range(right 0 to right 30)", "Font Right, range(right 30 to right 60)", "Right, range(right 60 to right 90)", "Right, range(right 90 to right 120)", "Rear Right, range(right 120 to right 150)", "Rear Right, range(right 150 to right 180)",
                    "Rear Left, range(left 180 to left 150)", "Rear Left, range(left 150 to left 120)", "Left, range(left 120 to left 90)", "Left, range(left 90 to left 60)", "Front Left, range(left 60 to left 30)", "Front Left, range(left 30 to left 0)"]
        direction = directions[direction_id]
        curr_observe = "Scene Description"+curr_observe.split("Scene Description")[1]
        prompt = [
            {"role": "system", "content": "You are a trajectory summary expert. Your task is to simplify environment description as short and clear as possible."},
            {"role": "user", "content": f"Given Environment Description \"{curr_observe}\", Summarization:"}
        ]
        
        return f"Direction {direction} " + gpt_response(prompt, "gpt-4", 1)[0]["message"]["content"]

    def summarize_thought(self, thought):
        prompt = [
            {"role": "system", "content": "You are a trajectory summary expert. Your task is to simplify navigation thought process as short and clear as possible."},
            {"role": "user", "content": f"Given Thought Process \"{thought}\", Summarization:"}
        ]
        
        return gpt_response(prompt, "gpt-4", 1)[0]["message"]["content"]

    def save_history(self, next_vp, thought, curr_observe, nav_history, landmarks): 
        curr_observe = self.summarize_observation(curr_observe, landmarks)
        thought = self.summarize_thought(thought)
        nav_history.append({
            "viewpoint": next_vp,
            "observation": curr_observe,
            "thought": thought
        })
        
        return nav_history
    
    def review_history(self, nav_history):
        nav_history_str = " -> ".join(["Step "+str(idx+1)+" Observation: "+item["observation"]+" Thought: "+item["thought"] for idx, item in enumerate(nav_history)])
        logger.info("History: "+nav_history_str)
        
        return nav_history_str

    def estimate_completion(self, actions, landmarks, history_traj):
        prompt = [
            {"role": "system", "content": "You are a completion estimation expert. Your task is to estimate what actions in the instruction have been executed based on navigation history and landmarks. \
                All actions in the instruction are given following the temporal order. Your answer includes two parts: \"Thought\" and \"Executed Actions\". \
                In the \"Thought\", you must follow procedures to analyze as detailed as possible what actions have been executed: \
                (1) What given landmarks of actions have appeared in the navigation history? \
                (2) Analyze the direction change at each step in the navigation history. \
                (3) Estimate each action in the instruction based on each step in the navigation history to check their completion. \
                In the \"Executed Actions\", you must only write down actions that have been executed without other words. \
                You must strictly refer original actions in the given instruction to estimate."},
            {"role": "user", "content": f"Given Navigation History \"{history_traj}\" and Landmarks \"{landmarks}\", estimate what actions in instruction \"{actions}\" have been executed."}   
        ]
        response = gpt_response(prompt, "gpt-4", 1)[0]["message"]["content"]
        if "Executed Actions" in response:
            logger.info("Executed Actions "+response)
            return response.split("Executed Actions")[1].strip()
        else:
            return response