Batch Mode

def get_close(self) -> bool:
    """
    Check if the plan is closed.
    :return: True if the plan need closed, False otherwise.
    """

    return self.plan.get("close", False)

def get_host_agent_request(self) -> str:
    """
    Get the request for the host agent.
    :return: The request for the host agent.
    """

    object_name = self.get_operation_object()

    request = (
        f"Open and select the application of {object_name}, and output the FINISH status immediately, without assigning any subtask"
        "You must output the selected application with their control text and label even if it is already open."
    )

    return request

def get_host_request(self) -> str:
    """
    Get the request for the host agent.
    :return: The request for the host agent.
    """

    task = self.get_task()
    object_name = self.get_operation_object()
    if object_name in self.support_apps:
        request = task
    else:
        request = (
            f"Your task is '{task}'. And open the application of {object_name}. "
            "You must output the selected application with their control text and label even if it is already open."
        )
    return request

def get_initial_request(self) -> str:
    """
    Get the initial request in the plan.
    :return: The initial request.
    """

    task = self.get_task()
    object_name = self.get_operation_object()

    request = f"{task} in {object_name}"

    return request

def get_operation_object(self) -> str:
    """
    Get the operation object in the step.
    :return: The operation object.
    """

    return self.plan.get("object", None).lower()

def get_root_path(self) -> str:
    """
    Get the root path of the plan.
    :return: The root path of the plan.
    """

    return os.path.dirname(os.path.abspath(self.plan_file))

def get_steps(self) -> List[str]:
    """
    Get the steps in the plan.
    :return: The steps in the plan.
    """

    return self.plan.get("steps", [])

def get_support_apps(self) -> List[str]:
    """
    Get the support apps in the plan.
    :return: The support apps in the plan.
    """

    return self.support_apps

def get_task(self) -> str:
    """
    Get the task name.
    :return: The task name.
    """

    return self.plan.get("task", "")

def next_step(self) -> Optional[str]:
    """
    Get the next step in the plan.
    :return: The next step.
    """

    if self.remaining_steps:
        step = self.remaining_steps.pop(0)
        return step

    return None

def task_finished(self) -> bool:
    """
    Check if the task is finished.
    :return: True if the task is finished, False otherwise.
    """

    return not self.remaining_steps

def create_new_round(self) -> None:
    """
    Create a new round.
    """

    # Get a request for the new round.
    request = self.next_request()

    # Create a new round and return None if the session is finished.
    if self.is_finished():
        return None

    self._host_agent.set_state(ContinueHostAgentState())

    round = BaseRound(
        request=request,
        agent=self._host_agent,
        context=self.context,
        should_evaluate=ufo_config.system.eva_round,
        id=self.total_rounds,
    )

    self.add_round(round.id, round)

    return round

def get_app_com(self, object_name: str) -> str:
    """
    Get the COM object name based on the object name.
    :param object_name: The name of the object.
    :return: The COM object name.
    """
    application_mapping = {
        ".docx": "Word.Application",
        ".xlsx": "Excel.Application",
        ".pptx": "PowerPoint.Application",
    }
    self.app_name = application_mapping.get(object_name)
    return self.app_name

def get_app_name(self, object_name: str) -> str:
    """
    Get the application name based on the object name.
    :param object_name: The name of the object.
    :return: The application name.
    """
    application_mapping = {
        ".docx": "WINWORD.EXE",
        ".xlsx": "EXCEL.EXE",
        ".pptx": "POWERPNT.EXE",
        # "outlook": "olk.exe",
        # "onenote": "ONENOTE.EXE",
    }
    self.app_name = application_mapping.get(object_name)
    return self.app_name

def next_request(self) -> str:
    """
    Get the request for the host agent.
    :return: The request for the host agent.
    """

    if self.total_rounds == 0:
        console.print(self.plan_reader.get_host_request(), style="cyan")
        return self.plan_reader.get_host_request()
    else:
        self._finish = True
        return

def record_task_done(self) -> None:
    """
    Record the task done.
    """
    is_record = ufo_config.system.task_status
    if is_record:
        file_path = ufo_config.system.get(
            "TASK_STATUS_FILE",
            os.path.join(self.plan_file, "../..", "tasks_status.json"),
        )
        task_done = json.load(open(file_path, "r"))
        task_done[self.task_name] = True
        json.dump(
            task_done,
            open(file_path, "w"),
            indent=4,
        )

def request_to_evaluate(self) -> str:
    """
    Get the request to evaluate.
    return: The request(s) to evaluate.
    """
    return self.plan_reader.get_task()

def run(self) -> None:
    """
    Run the session.
    """
    self.setup_application_environment()
    try:
        super().run()
        self.record_task_done()
    except Exception as e:
        import traceback

        traceback.print_exc()
        print(f"An error occurred: {e}")
    # Close the APP if the user ask so.
    self.terminate_application_processes()

def setup_application_environment(self):
    """
    Sets up the application environment by determining the application name and
    command based on the operation object, and then launching the application.

    Raises:
        Exception: If an error occurs during the execution of the command or
                   while interacting with the application via COM.
    """
    self.object_name = self.plan_reader.get_operation_object()
    if self.object_name:
        suffix = os.path.splitext(self.object_name)[1]
        self.app_name = self.get_app_name(suffix)
        print("app_name:", self.app_name)
        if self.app_name not in self.support_apps:
            print(f"The app {self.app_name} is not supported.")
            return  # The app is not supported, so we don't need to setup the environment.
        file = self.plan_reader.get_file_path()
        code_snippet = f"import os\nos.system('start {self.app_name} \"{file}\"')"
        code_snippet = code_snippet.replace("\\", "\\\\")  # escape backslashes
        try:
            exec(code_snippet, globals())
            app_com = self.get_app_com(suffix)
            time.sleep(2)  # wait for the app to boot
            word_app = win32com.client.Dispatch(app_com)
            word_app.WindowState = 1  # wdWindowStateMaximize
        except Exception as e:
            print(f"An error occurred: {e}")

def terminate_application_processes(self):
    """
    Terminates specific application processes based on the provided conditions.
    """
    if self.close:
        if self.object_name:
            for process in psutil.process_iter(["name"]):
                if process.info["name"] == self.app_name:
                    os.system(f"taskkill /f /im {self.app_name}")
                    time.sleep(1)
        else:
            app_names = ["WINWORD.EXE", "EXCEL.EXE", "POWERPNT.EXE"]
            for process in psutil.process_iter(["name"]):
                if process.info["name"] in app_names:
                    os.system(f"taskkill /f /im {process.info['name']}")
                    time.sleep(1)