planner_step.py

from model import DialogueAgent
import json
import re
import threading
import time
import keyboard
from retrieval_data import get_retrival_data
#system prompt
tmp1="""
你是战斗机辅助系统，当前任务是{task}，子任务是{subtask}，请根据以下步骤指导驾驶员完成任务：
接下来需要你根据上述信息来一步一步指导驾驶员了，请你根据之前的信息（示例动作序列，相关数据名，当前数据），来生成需要提供给驾驶员的指导信息，
注意，你需要根据当前信息来判断是否需要适当修改当前step，请输出你认为合适的step。
除此之外，在你生成具体操作时，还需要生成一个判断函数用于判断当前步骤是否完成，用于接下来的步骤规划，
这个函数应该python代码的形式，函数内涉及到的数据应该从之前检索出的实际数据中选择，注意不要任意生成可能不存在的数据。
格式参考如下：
example:

## normal case
input:
<step info>
3:Set the MSL PREP switch ON. This commands the WCS to start missile preparation for the AIM-7 and AIM-54. When the individual missiles are tuned and ready the corresponding missile status windows turns white to indicate a ready missile. This should take approx. 2 minutes.
</step info>

<common data>
'ALT_MSL_FT': 1597, 'LATITUDE_DEGREES': "0°41'0", 'LONGITUDE_DEGREES': "40°2'54", 'HEADING_DEGREES': 219.0157, 'IAS_US_INT': 320
</common data>

<specific data>
name: PLT_MISSLE_PREP
description: Initiates missile preparation sequence for AIM-7 and AIM-54 weapon systems.
api_variant: momentary_last_position
value: OFF

name: PLT_MSL_PREP_ON
description: Indicates missile preparation status for pilot awareness.
api_variant: None
value: OFF
</specific data>

<text>
Set the MSL PREP switch ON. This commands the WCS to start missile preparation for the AIM-7 and AIM-54. When the individual missiles are tuned and ready the corresponding missile status windows turns white to indicate a ready missile. This should take approx. 2 minutes
</text>
<check func>
result = dcs_states['PLT_MSL_PREP_ON'] == "ON"
</check func>

## finished case
If you determine that the current step has already been completed, there is no need to remind the human to execute it again. For example, referring to the above step, if:
<data>
name: PLT_MISSLE_PREP
description: Initiates missile preparation for AIM-7 and AIM-54 weapon systems.
value: ON

name: PLT_MSL_PREP_ON
description: Indicates missile preparation status for pilot awareness.
value: ON
</data>

At this point, if the action is considered complete, you should directly output:
<text>
You should set the MSL PREP switch ON. but it has been done. Let's move on to the next step.
</text>
<check func>
result = True
</check func>

## multi-options case
for some multi-options step, you can use more complex func to detec, for example:

## keep-silent case
When encountering descriptive statements in some step, you do not need to provide a specific command or check function.
for example:


## screen type case
When processing api_variant data of screen type, follow these guidelines:
1. For screen-type information (visual interface data), generate string-based queries that will be co-processed with the screen capture by a Vision-Language Model (VLM). 

2. Format requirements:
   - Use natural language questions in quotation marks
   - Ensure questions target specific visual elements
   - Maintain aircraft system terminology consistency

for example:
输入指令为
"Line up target with ADL (within 20 deg) in HUD"
并且检索到的相关数据并没有能够表现ADL的数据，此时必须去解析HUD画面，因此你的检测输出是：

result = video_states['F_14_HUD'] == "Does the target have already been line up with ADL(within 20 deg)?"

Note: Please output the step instructions and check functions strictly in the following format, and do not output anything else:
- Step instructions should be contained in <text>...</text> tags.
- Check functions should be contained in <check func>...</check func> tags.
"""

#user prompt
tmp2 = """
<step info>
{step_num}:{step_ins}
</step info>

<common data>
{common_data}
</common data>

<specific data>
{specific_data}
</specific data>
Based on the above information, please provide the user with a language prompt and a check function.

Please remember that some unnecessary data may be provided to you, such as redundant information, irrelevant information, etc.

Please combine the context and actual scenario and use only the required data in the check function.

If the api_variant of a data is not None, please pay special attention to it.

For example, momentary_last_position means that this key is a sequential key, which is suitable for detecting whether the key is pressed and should not be relied on to determine the system status.
"""

class CopilotAgent():
    def __init__(self,task_name,subtask_name,init_state,dcs_state,model="gpt-4o",logger=None,history=None):
        self.task_name = task_name
        self.subtask_name = subtask_name
        self.logger = logger
        self.history = history
        self.init_state = init_state
        self.dcs_state = dcs_state
        self.step_specific_data = {}
        self.step_specific_data_names = []
        # self.step_info = get_retrival_data(subtask_name,slient=False)
        self.step_info = [{'instruction': 'Master Arm switch - ON (UP)', 'data': [{'key': 'PLT_MASTER_ARM_SW', 'description': ' Activates or deactivates weapon systems for arming and firing.'}, {'key': 'PLT_WEAPON_SELECTOR', 'description': ' Selects and activates weapon systems during flight operations.'}, {'key': 'PLT_MASTER_ARM_COVER', 'description': " Protects and enables the pilot's weapon arming and firing controls."}]}, {'instruction': 'HUD Display Mode - A/A', 'data': [{'key': 'PLT_HUD_MODE_A2A', 'description': ' Switches HUD to air-to-air mode for target tracking and weapon engagement.'}, {'key': 'PLT_HUD_MODE', 'description': ' Selects display mode for pilot鈥檚 heads-up visual interface.'}, {'key': 'PLT_HUD_MODE_A2G', 'description': ' Activates HUD air-to-ground mode for targeting and weapon impact point display during ground engagements.'}]}, {'instruction': 'Set the MSL PREP switch ON. This commands the WCS to start missile preparation for the AIM-7 and AIM-54. When the individual missiles are tuned and ready the corresponding missile status windows turns white to indicate a ready missile. This should take approx. 2 minutes.', 'data': [{'key': 'PLT_MISSLE_PREP', 'description': ' Initiates missile preparation for AIM-7 and AIM-54 weapon systems.'}, {'key': 'PLT_MSL_PREP_OFF', 'description': ' Indicates missile preparation is deactivated during weapon operations.'}, {'key': 'PLT_MSL_PREP_ON', 'description': ' Indicates missile preparation status for pilot awareness.'}, {'key': 'PLT_MSL_PREP_ON', 'description': ' Indicates missile preparation status for pilot awareness.'}, {'key': 'PLT_WEAPON_STORE_5_READY_L', 'description': ' Indicates weapon store 5 readiness status for the pilot.'}, {'key': 'PLT_WEAPON_STORE_3_READY_L', 'description': ' Indicates weapon store 3 readiness status for the pilot.'}]}, {'instruction': 'Press the WEAPON SELECTOR button in, hold it in and cycle the Weapon Selector UP to SP/PH.', 'data': [{'key': 'PLT_WEAPON_SELECTOR', 'description': ' Selects and activates weapon systems during flight operations.'}, {'key': 'PLT_Trigger', 'description': ' Activates weapons systems for target engagement.'}, {'key': 'PLT_HOT_TRIGGER', 'description': ' Indicates weapon system activation readiness to the pilot.'}]}, {'instruction': 'Select NORM (used to engage a target tracked in STT; WCS is capable of using both CW and pulse doppler for guidance) or BRSIT (Boresight, tracks the strongest target return within the boresighted flood area) Mode. We will chose NORM mode.', 'data': [{'key': 'RIO_RADAR_PDSTT', 'description': ' Enables precise tracking of a single target using Pulse Doppler radar.'}, {'key': 'RIO_RADAR_PSTT', 'description': ' Tracks a single target using pulse radar mode for precise engagement.'}, {'key': 'RIO_DDD_LIGHT_PSTT_2', 'description': ' Indicates active Pulse Doppler STT mode for target tracking.'}]}, {'instruction': 'You need a radar STT (Single Target Track) lock to the target. This can be done through JESTER, through the RIO cockpit or through one of the pilot ACM modes with the Target Designate Switch (Forward engages the PAL (Pilot Automatic Lockon Mode, Up/Down engages VSL HI/LO). The WCS (Weapon Control System) just needs a radar lock to compute ranging and lead information.', 'data': [{'key': 'PLT_GUN_LEAD_DISP', 'description': ' Displays gun lead targeting information for accurate shooting during combat.'}, {'key': 'PLT_WEAPON_SELECTOR', 'description': ' Selects and activates weapon systems during flight operations.'}, {'key': 'RIO_LANTIRN_3_TCS', 'description': ' Provides infrared targeting and navigation for precision ground attacks.'}, {'key': 'PLT_WEAPON_SELECTOR', 'description': ' Selects and activates weapon systems during flight operations.'}, {'key': 'PLT_BIT_SWITCH', 'description': ' Rotates to select and test aircraft system diagnostics.'}, {'key': 'PLT_Trigger', 'description': ' Activates weapons systems for target engagement.'}, {'key': 'RIO_RADAR_PSTT', 'description': ' Tracks a single target using pulse radar mode for precise engagement.'}, {'key': 'RIO_RADAR_PDSTT', 'description': ' Enables precise tracking of a single target using Pulse Doppler radar.'}, {'key': 'RIO_TID_TRACKHOLD', 'description': ' Maintains radar track on selected target for tactical information display.'}]}, {'instruction': 'Line up target with ADL (within 20 deg)', 'data': [{'key': 'RIO_WEAPON_MSL_SPD', 'description': ' Adjusts missile acquisition gate for precise target selection.'}, {'key': 'PLT_MSL_MODE_BORE', 'description': ' Indicates missile boresight mode activation for targeting.'}, {'key': 'RIO_RADAR_VSL', 'description': ' Adjusts radar vertical scan range for target acquisition at specific altitudes.'}]}, {'instruction': 'Steer the aircraft to line up the T-shaped steering cue with the center of the black circle (ASE, or Allowable Steering Error circle) on the HSD to maximize probability of kill.', 'data': [{'key': 'PLT_HSD_DIS_MODE', 'description': ' Displays navigation or tactical information for pilot situational awareness.'}, {'key': 'PLT_HSD_TEST', 'description': " Tests pilot's Horizontal Situation Display for navigational accuracy and situational awareness."}, {'key': 'HSD_MAN_CRS_S', 'description': ' Displays manual course selection for navigation and situational awareness.'}, {'key': 'PLT_AHRS_HEMISPHERE', 'description': ' Determines compass hemisphere for accurate aircraft heading and orientation.'}, {'key': 'PLT_AOA_OPT', 'description': ' Indicates optimal angle of attack for pilot during flight maneuvers.'}, {'key': 'PLT_AFCS_ROLL', 'description': ' Stabilizes aircraft roll axis during normal flight conditions.'}]}, {'instruction': 'Press the TRIGGER SECOND STAGE (SPACE) on the stick to fire missile when ready.', 'data': [{'key': 'PLT_Trigger', 'description': ' Activates weapons systems for target engagement.'}, {'key': 'PLT_WEAPON_SELECTOR', 'description': ' Selects and activates weapon systems during flight operations.'}, {'key': 'PLT_HOT_TRIGGER', 'description': ' Indicates weapon system activation readiness to the pilot.'}]}, {'instruction': 'Keep tracking the target to maintain radar lock until missile impact.', 'data': [{'key': 'PLT_HUD_LIGHT_SAM', 'description': ' Alerts pilot to SAM radar lock or missile launch threats.'}, {'key': 'RIO_RADAR_PSTT', 'description': ' Tracks a single target using pulse radar mode for precise engagement.'}, {'key': 'RIO_TID_TRACKHOLD', 'description': ' Maintains radar track on selected target for tactical information display.'}]}]
        self.agent = DialogueAgent(model,system_msg = tmp1.format(task = task_name,subtask=subtask_name))
        self.step_max_count = len(self.step_info)
        self.step_count = 0

    def generate_step_info(self):
        for step in self.step_info:
            print(f"####Step {self.step_count+1}/{self.step_max_count}####")

            specific_data_str = ""
            for data in step["data"]:
                self.step_specific_data_names.append(data['key'])
                # 从mapping中获取可能的值
                value_scale = data.get("value_scale",{})
                api_variant = data.get("api_variant",None)
                specific_data_str += f"***\nname: {data['key']}\ndescription: {data['description']}\nvalue: {self.dcs_state.get_chosen_data([data['key']])}\n value_scale: {value_scale}\napi_variant:{api_variant}\n***\n"
    
            llm_output = tmp2.format(step_num=self.step_count+1,
                                    step_ins=step["instruction"],
                                    common_data=self.dcs_state.get_common_data(),
                                    specific_data=specific_data_str)
            
            output,_ = self.agent.generate_response(llm_output)

            ## 解析输出
            text_pattern = r'<text>(.*?)</text>'
            check_pattern = r'<check func>(.*?)</check func>'
            text_match = re.search(text_pattern, output, re.DOTALL)
            check_match = re.search(check_pattern, output, re.DOTALL)
            if text_match:
                text = text_match.group(1).strip()
                print(text)
            else:
                print("No text found in the output")

            if check_match:
                check_func_code = check_match.group(1).strip()
                print(check_func_code)
                # 创建一个事件对象
                event = threading.Event()
                # 启动一个线程来监控检查函数
                check_thread = threading.Thread(target=self.monitor_check_function, args=(check_func_code, event))
                check_thread.start()
                # 等待事件被设置
                event.wait()
                # check_thread.join()
                self.step_count += 1

    def execute_check_function(self, check_func_code):
        local_vars = {'dcs_states': self.dcs_state.get_chosen_data(self.step_specific_data_names),'video_staes':{}}
        exec(check_func_code, {}, local_vars)
        return local_vars.get('result', False)

    def monitor_check_function(self, check_func_code, event):
        while True:
            if self.execute_check_function(check_func_code):
                print("Condition met! Notifying main program...")
                # 设置事件，通知主程序
                event.set()
                break
            if keyboard.is_pressed("esc"):
                event.set()
                break
            time.sleep(1)  # 每秒检查一次

func_sysy_prompt = """
ins:{ins}

relevant data:
{data}

common data:
{common_data}


"""


#调用后返回查询相关的文档信息辅助回答
def search_relevant_docs(query):
    pass

#重点函数，用于生成指导信息和检查函数
def generate_instruction_and_check_func(ins,data,common_data):
    pass

#基于prompt调用外部VLM分析指定显示器的信息，将分析结果返回
def analysis_screen(screen_name,prompt,bool=False):
    #param: screen_name:屏幕名称
    #       prompt:分析提示
    #       bool:输出是否为bool型数据，如果是则返回True/False，否则返回具体的分析结果
    pass

#基于当前指令，检索当前指令最相关的数据
def get_relervant_data(ins,feedback=None):
    #param: ins:当前指令
    #       feedback:反馈信息，如果存在，则需要基于反馈信息进一步细化检索
    pass

#调用后读取下一条指导操作，避免使用外部程序化的方式实现状态跳转
def next_step():
    pass