execute_s_line_site_task_workflow
Automates S-line robot task workflows by processing site information to select maps, retrieve subareas, build tasks, and submit them for execution.
Instructions
Executes complete S-line robot task workflow with site information.
Automated process: Site info → Map selection → Subarea retrieval → Task building → Task submission
Args:
robot_id: The ID of the target robot.
task_parameters: Task parameters including map criteria and task settings.
Returns:
A dictionary containing the workflow execution result.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| robot_id | Yes | ||
| task_parameters | Yes |
Implementation Reference
- src/gs_openapi/main.py:259-278 (handler)Primary MCP tool handler and registration for 'execute_s_line_site_task_workflow'. Delegates to GausiumMCP instance.
@mcp.tool() async def execute_s_line_site_task_workflow( robot_id: str, task_parameters: dict ): """Executes complete S-line robot task workflow with site information. Automated process: Site info → Map selection → Subarea retrieval → Task building → Task submission Args: robot_id: The ID of the target robot. task_parameters: Task parameters including map criteria and task settings. Returns: A dictionary containing the workflow execution result. """ return await mcp.execute_s_line_site_task_workflow( robot_id=robot_id, task_parameters=task_parameters ) - GausiumMCP class method implementing the workflow by delegating to TaskExecutionEngine.
async def execute_s_line_site_task_workflow( self, robot_id: str, task_parameters: Dict[str, Any] ) -> Dict[str, Any]: """ 执行S线有站点任务完整工作流。 自动化流程：站点信息 → 地图选择 → 分区获取 → 任务构建 → 任务下发 Args: robot_id: 机器人ID task_parameters: 任务参数 Returns: 工作流执行结果 """ return await self.task_engine.execute_s_line_site_task( robot_id=robot_id, task_parameters=task_parameters ) - Core implementation of the S-line site task workflow in TaskExecutionEngine, handling site info retrieval, map selection, subarea fetching, task building, and submission.
async def execute_s_line_site_task( self, robot_id: str, task_parameters: Dict[str, Any] ) -> Dict[str, Any]: """ 执行S线有站点任务。 工作流： 1. 获取站点信息 2. 解析可用地图 3. 获取目标地图分区 4. 构建并下发有站点临时任务 Args: robot_id: 机器人ID task_parameters: 任务参数 Returns: 任务执行结果 """ logger.info(f"Starting S-line site task execution for robot: {robot_id}") async with GausiumAPIClient() as client: try: # 1. 获取站点信息 site_info = await client.call_endpoint( 'get_site_info', path_params={'robot_id': robot_id} ) # 2. 解析可用地图 available_maps = self._extract_maps_from_site(site_info) if not available_maps: raise ValueError("No maps found in site information") # 3. 选择目标地图 target_map_id = self._select_map( available_maps, task_parameters.get('map_criteria', {}) ) # 4. 获取地图分区 subareas = await client.call_endpoint( 'get_map_subareas', path_params={'map_id': target_map_id} ) # 5. 构建任务数据 task_data = self._build_site_task_data( target_map_id, subareas, task_parameters ) # 6. 下发有站点临时任务 task_result = await client.call_endpoint( 'submit_temp_site_task', json_data=task_data ) logger.info(f"S-line site task executed successfully: {task_result}") return task_result except Exception as e: logger.error(f"S-line site task execution failed: {str(e)}") raise