""" 组件详细配置文件 包含各种组件类型及其子组件的配置信息 """ from typing import Dict, List, Any, Optional class ScriptComponentConfig: """脚本组件配置""" # 脚本组件类型 RUN_SCRIPT = "run_script" # 运行脚本 SET_VARIABLES = "set_variables" # 设置task.variables # 脚本组件详细配置 @classmethod def get_components(cls) -> List[Dict[str, Any]]: """获取脚本组件列表""" return [ { "type": "script", "sub_type": cls.RUN_SCRIPT, "name": "运行脚本", "description": "执行JavaScript代码并返回结果", "icon": "code", # 图标名称，前端可用 "params": [ { "name": "function_name", "label": "函数名", "type": "string", "required": False, "description": "定义脚本中的主函数名称", "value_types": [ { "type": "simple", "label": "简单值", "default": True }, { "type": "expression", "label": "表达式", "default": False } ] }, { "name": "params", "label": "函数参数", "type": "array", "required": False, "description": "传递给脚本的参数", "value_types": [ { "type": "simple", "label": "简单值", "default": True }, { "type": "expression", "label": "表达式", "default": False } ] } ] }, { "type": "script", "sub_type": cls.SET_VARIABLES, "name": "设置task.variables", "description": "设置和管理任务变量", "icon": "variable", "params": [ { "name": "function_name", "label": "函数名", "type": "string", "required": False, "description": "定义脚本中的主函数名称", "value_types": [ { "type": "simple", "label": "简单值", "default": True }, { "type": "expression", "label": "表达式", "default": False } ] }, { "name": "params", "label": "函数参数", "type": "array", "required": False, "description": "传递给脚本的参数", "value_types": [ { "type": "simple", "label": "简单值", "default": True }, { "type": "expression", "label": "表达式", "default": False } ] } ] } ] class HttpComponentConfig: """HTTP请求组件配置""" # HTTP请求组件类型 HTTP_REQUEST = "http_request" # HTTP请求 # HTTP请求组件详细配置 @classmethod def get_components(cls) -> List[Dict[str, Any]]: """获取HTTP请求组件列表""" return [ { "type": "http", "sub_type": cls.HTTP_REQUEST, "name": "HTTP请求", "description": "发送HTTP请求并处理响应", "icon": "http", "params": [ { "name": "method", "label": "请求方法", "type": "select", "options": ["GET", "POST", "PUT", "DELETE", "PATCH"], "required": True, "description": "HTTP请求方法" }, { "name": "url", "label": "请求URL", "type": "string", "required": True, "description": "请求的目标URL" }, { "name": "headers", "label": "请求头", "type": "object", "required": False, "description": "HTTP请求头" }, { "name": "body", "label": "请求体", "type": "object", "required": False, "description": "HTTP请求体" }, { "name": "timeout", "label": "超时时间", "type": "number", "required": False, "description": "请求超时时间（毫秒）" } ] } ] class FlowComponentConfig: """流程控制组件配置""" # 流程控制组件类型 IF = "if" # 条件判断 IF_ELSE = "if_else" # 条件分支 FOR_EACH = "for_each" # 循环遍历 WHILE = "while" # 条件循环 # 流程控制组件详细配置 @classmethod def get_components(cls) -> List[Dict[str, Any]]: """获取流程控制组件列表""" return [ { "type": "flow", "sub_type": cls.IF, "name": "条件判断", "description": "根据条件执行不同的操作", "icon": "branch", "params": [ { "name": "condition", "label": "条件表达式", "type": "expression", "required": True, "description": "条件判断表达式" } ] }, { "type": "flow", "sub_type": cls.IF_ELSE, "name": "条件分支", "description": "根据条件执行不同的分支", "icon": "branch-multiple", "params": [ { "name": "condition", "label": "条件表达式", "type": "expression", "required": True, "description": "条件判断表达式" } ] }, { "type": "flow", "sub_type": cls.FOR_EACH, "name": "循环遍历", "description": "遍历数组或对象的每个元素", "icon": "loop", "params": [ { "name": "collection", "label": "集合表达式", "type": "expression", "required": True, "description": "要遍历的数组或对象" }, { "name": "item_name", "label": "元素变量名", "type": "string", "required": True, "description": "当前元素的变量名" }, { "name": "index_name", "label": "索引变量名", "type": "string", "required": False, "description": "当前索引的变量名" } ] }, { "type": "flow", "sub_type": cls.WHILE, "name": "条件循环", "description": "当条件为真时重复执行", "icon": "loop-circular", "params": [ { "name": "condition", "label": "条件表达式", "type": "expression", "required": True, "description": "循环条件表达式" } ] } ] class SubtaskComponentConfig: """子任务组件配置""" # 子任务组件类型 SUBTASK = "subtask" # 子任务 # 子任务组件详细配置 @classmethod def get_components(cls) -> List[Dict[str, Any]]: """获取子任务组件列表""" return [ { "type": "subtask", "sub_type": cls.SUBTASK, "name": "子任务", "description": "执行已定义的任务作为子任务", "icon": "subtask", "params": [ { "name": "task_id", "label": "选择子任务", "type": "select", "required": True, "description": "选择要执行的子任务（从已创建的任务中选择）", "data_source": "available_subtasks", # 指示前端从API返回的available_subtasks字段获取数据 "display_field": "name", # 显示任务名称 "value_field": "task_id" # 使用任务ID作为值 }, { "name": "params", "label": "任务参数", "type": "object", "required": False, "description": "传递给子任务的参数" }, { "name": "wait_complete", "label": "等待完成", "type": "boolean", "required": False, "default": True, "description": "是否等待子任务完成后再继续执行" } ] } ] # 组件配置管理类 class ComponentDetailConfig: """组件详细配置管理""" # 组件类型中文名称映射 @classmethod def get_component_type_names(cls) -> Dict[str, str]: """获取组件类型的中文名称映射""" return { # 基础类型 "script": "脚本", "http": "HTTP请求", "flow": "流程", "robot": "机器人调度", "site": "库位", "device": "设备", "subtask": "子任务", "task": "任务", "basic": "基础", # 脚本组件 "run_script": "运行脚本", "set_task_variables": "设置任务变量", "runscript": "运行脚本", "settaskvariables": "设置任务变量", # HTTP请求组件 "http_get_request": "GET请求", "http_post_request": "POST请求", "httpgetrequest": "GET请求", "httppostrequest": "POST请求", # 流程控制组件 "if": "条件判断", "if_else": "条件分支", "if_else_if": "多条件分支", "for_each": "循环遍历", "while": "条件循环", "break": "跳出循环", "return": "返回", "delay": "延时", "parallel_execute": "并行执行", "serial_execute": "串行执行", "throw_exception": "抛出异常", "foreach": "循环遍历", "ifelse": "条件分支", "ifelseif": "多条件分支", "parallelexecute": "并行执行", "serialexecute": "串行执行", "throwexception": "抛出异常", # 机器人调度组件 "select_robot": "选择机器人", "get_robot_position": "获取机器人位置", "robot_action": "机器人动作", "change_robot_destination": "更改机器人目的地", "get_robot_battery": "获取机器人电量", "get_robot_pgv_code": "获取机器人PGV码", "changerobotdestination": "更改机器人目的地", "getrobotbattery": "获取机器人电量", "getrobotpgvcode": "获取机器人PGV码", "getrobotposition": "获取机器人位置", "robotaction": "机器人动作", "selectrobot": "选择机器人", # 库位组件 "batch_set_site": "批量设置库位", "get_dense_site": "获取密集库位", "query_site": "查询库位", "lock_site": "锁定库位", "unlock_site": "解锁库位", "get_locked_sites_by_task": "获取任务锁定的库位", "get_site_extension_property": "获取库位扩展属性", "set_site_extension_property": "设置库位扩展属性", "set_site_goods": "设置库位货物", "set_site_empty": "设置库位为空", "set_site_occupied": "设置库位为占用", "set_site_tag": "设置库位标签", "batchsetsite": "批量设置库位", "getdensesite": "获取密集库位", "getlockedsitesbytask": "获取任务锁定的库位", "getsiteextensionproperty": "获取库位扩展属性", "locksite": "锁定库位", "querysite": "查询库位", "setsiteempty": "设置库位为空", "setsiteextensionproperty": "设置库位扩展属性", "setsitegoods": "设置库位货物", "setsiteoccupied": "设置库位为占用", "setsitetag": "设置库位标签", "unlocksite": "解锁库位", # 任务组件 "cache_data": "缓存数据", "clear_cache_data": "清除缓存数据", "get_cache_data": "获取缓存数据", "set_task_status": "设置任务状态", "jump_to_block": "跳转到块", "get_task_input_param": "获取任务输入参数", "cachedata": "缓存数据", "clearcachedata": "清除缓存数据", "getcachedata": "获取缓存数据", "gettaskinputparam": "获取任务输入参数", "jumptoblock": "跳转到块", "settaskstatus": "设置任务状态", # 基础组件 "check_task_instance_id_exists": "检查任务实例ID是否存在", "create_unique_id": "创建唯一ID", "current_timestamp": "当前时间戳", "current_time": "当前时间", "execute_sql": "执行SQL", "query_sql": "查询SQL", "string_md5_encrypt": "字符串MD5加密", "string_to_json_array": "字符串转JSON数组", "string_to_json_object": "字符串转JSON对象", "print": "打印", "checktaskinstanceidexists": "检查任务实例ID是否存在", "createuniqueid": "创建唯一ID", "currenttime": "当前时间", "currenttimestamp": "当前时间戳", "executesql": "执行SQL", "querysql": "查询SQL", "stringmd5encrypt": "字符串MD5加密", "stringtojsonarray": "字符串转JSON数组", "stringtojsonobject": "字符串转JSON对象", # 设备组件 "wait_modbus_value": "等待Modbus值", "write_modbus_value": "写入Modbus值", "waitmodbusvalue": "等待Modbus值", "writemodbusvalue": "写入Modbus值" } @classmethod def get_all_components(cls) -> List[Dict[str, Any]]: """获取所有组件详细配置""" all_components = [] # 添加子任务组件（放在第一位） all_components.extend(SubtaskComponentConfig.get_components()) # 添加脚本组件 all_components.extend(ScriptComponentConfig.get_components()) # 添加HTTP请求组件 all_components.extend(HttpComponentConfig.get_components()) # 添加流程控制组件 all_components.extend(FlowComponentConfig.get_components()) # 可以继续添加其他类型的组件... return all_components @classmethod def get_components_by_type(cls, component_type: str) -> List[Dict[str, Any]]: """根据组件类型获取组件列表""" all_components = cls.get_all_components() return [comp for comp in all_components if comp["type"] == component_type]