execute_s_line_site_task_workflow
Executes complete S-line robot task workflow by processing site information, selecting maps, retrieving subareas, and submitting tasks.
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
- Actual handler: Core implementation of the S-line site task workflow. Performs: 1) get site info, 2) extract maps from site, 3) select target map, 4) get subareas, 5) build task data, 6) submit temp site task.
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 - src/gs_openapi/main.py:259-278 (registration)MCP tool registration: Decorated with @mcp.tool() to register 'execute_s_line_site_task_workflow' as an MCP tool, then delegates to mcp.execute_s_line_site_task_workflow.
@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 ) - Method on GausiumMCP class that delegates to task_engine.execute_s_line_site_task.
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 ) - Helper: Extracts a list of maps from site info by iterating building -> floors -> maps.
def _extract_maps_from_site(self, site_info: Dict[str, Any]) -> List[Dict[str, Any]]: """ 从站点信息中提取地图列表。 Args: site_info: 站点信息 Returns: 地图列表 """ maps = [] buildings = site_info.get('buildings', []) for building in buildings: floors = building.get('floors', []) for floor in floors: floor_maps = floor.get('maps', []) if isinstance(floor_maps, list): maps.extend(floor_maps) return maps - Helper: Builds the task data dict for submitting a site task, including mapId, subareas, taskType, and cleaningMode.
def _build_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