execute_s_line_no_site_task_workflow
Automates S-line robot task workflows by mapping, selecting subareas, building tasks, and submitting them using robot serial numbers and 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
- src/gs_openapi/main.py:280-299 (registration)MCP tool registration using @mcp.tool() decorator. This is the entry point for the tool invocation, delegating to GausiumMCP instance.
@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 ) - GausiumMCP class method that wraps and delegates the workflow execution to TaskExecutionEngine.
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 ) - Core handler implementing the S-line no-site task workflow: fetches robot maps, selects map, retrieves 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 - Helper function to construct the task data dictionary for submitting no-site temporary tasks.
def _build_no_site_task_data( self, map_id: str, subareas: Dict[str, Any], task_parameters: Dict[str, Any] ) -> Dict[str, Any]: """ 构建无站点任务数据。 Args: map_id: 地图ID subareas: 地图分区信息 task_parameters: 任务参数 Returns: 任务数据 """ # TODO: 根据实际API要求构建任务数据 task_data = { "mapId": map_id, "subareas": subareas, "taskType": task_parameters.get('task_type', 'cleaning'), "cleaningMode": task_parameters.get('cleaning_mode', '__middle_cleaning'), **task_parameters } return task_data - Helper function to select the target map from the robot's available maps based on criteria.
def _select_map_from_list( self, available_maps: List[Dict[str, Any]], criteria: Dict[str, Any] ) -> str: """ 从地图列表中选择目标地图。 Args: available_maps: 可用地图列表 criteria: 选择条件 Returns: 选中的地图ID """ if not criteria: return available_maps[0]['mapId'] # TODO: 实现地图选择逻辑 for map_info in available_maps: if self._matches_criteria(map_info, criteria): return map_info['mapId'] return available_maps[0]['mapId']