execute_s_line_no_site_task_workflow
Execute complete S-line robot task workflows by automatically processing maps, selecting areas, building tasks, and submitting them for GS cleaning robots using only robot serial number and task parameters.
Instructions
Executes complete S-line robot task workflow without site information.
Automated process: Map list → Map selection → Subarea retrieval → Task building → Task submission
Args:
robot_sn: The serial number 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_sn | Yes | ||
| task_parameters | Yes |
Implementation Reference
- Core handler implementing the S-line no-site task workflow: fetches robot maps, selects target map, retrieves map subareas, builds task data, and submits temporary no-site task via API.async def execute_s_line_no_site_task( self, robot_sn: str, task_parameters: Dict[str, Any] ) -> Dict[str, Any]: """ 执行S线无站点任务。 工作流： 1. 获取机器人地图列表 2. 获取目标地图分区 3. 构建并下发无站点临时任务 Args: robot_sn: 机器人序列号 task_parameters: 任务参数 Returns: 任务执行结果 """ logger.info(f"Starting S-line no-site task execution for robot: {robot_sn}") async with GausiumAPIClient() as client: try: # 1. 获取机器人地图列表 maps_response = await client.call_endpoint( 'list_maps', json_data={'robotSn': robot_sn} ) # 处理Gausium特殊响应格式 if maps_response.get('code') != 0: raise RuntimeError(f"Failed to get maps: {maps_response.get('msg')}") available_maps = maps_response.get('data', []) if not available_maps: raise ValueError("No maps found for robot") # 2. 选择目标地图 target_map_id = self._select_map_from_list( available_maps, task_parameters.get('map_criteria', {}) ) # 3. 获取地图分区 subareas = await client.call_endpoint( 'get_map_subareas', path_params={'map_id': target_map_id} ) # 4. 构建任务数据 task_data = self._build_no_site_task_data( target_map_id, subareas, task_parameters ) # 5. 下发无站点临时任务 task_result = await client.call_endpoint( 'submit_temp_no_site_task', json_data=task_data ) logger.info(f"S-line no-site task executed successfully: {task_result}") return task_result except Exception as e: logger.error(f"S-line no-site task execution failed: {str(e)}") raise
- src/gs_openapi/main.py:280-299 (registration)MCP tool registration using @mcp.tool() decorator. Defines input schema via type hints and docstring. Thin wrapper delegating to GausiumMCP instance method.@mcp.tool() async def execute_s_line_no_site_task_workflow( robot_sn: str, task_parameters: dict ): """Executes complete S-line robot task workflow without site information. Automated process: Map list → Map selection → Subarea retrieval → Task building → Task submission Args: robot_sn: The serial number 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_no_site_task_workflow( robot_sn=robot_sn, task_parameters=task_parameters )
- Helper method in GausiumMCP class that delegates the workflow execution to the TaskExecutionEngine instance.async def execute_s_line_no_site_task_workflow( self, robot_sn: str, task_parameters: Dict[str, Any] ) -> Dict[str, Any]: """ 执行S线无站点任务完整工作流。 自动化流程：地图列表 → 地图选择 → 分区获取 → 任务构建 → 任务下发 Args: robot_sn: 机器人序列号 task_parameters: 任务参数 Returns: 工作流执行结果 """ return await self.task_engine.execute_s_line_no_site_task( robot_sn=robot_sn, task_parameters=task_parameters )