nUR MCP Server

Instructions

生成机器人运行报告

参数:
- robot_id: 机器人ID
- start_time: 开始时间戳
- end_time: 结束时间戳
- report_path: 报告保存路径

返回:
- 报告生成结果

Implementation Reference

• src/nonead_universal_robots_mcp/server.py:1150-1182 (handler)

  The MCP tool handler function decorated with @mcp.tool(), implementing generate_robot_report by calling advanced_data_analyzer.generate_report and handling errors.

  @mcp.tool()
  def generate_robot_report(robot_id: str, start_time: float = None, end_time: float = None, report_path: str = None):
      """
      生成机器人运行报告
      
      参数:
      - robot_id: 机器人ID
      - start_time: 开始时间戳
      - end_time: 结束时间戳
      - report_path: 报告保存路径
      
      返回:
      - 报告生成结果
      """
      try:
          if advanced_data_analyzer is None:
              return return_msg("高级数据分析器未初始化")
          
          # 生成报告
          report = advanced_data_analyzer.generate_report(
              robot_id=robot_id,
              start_time=start_time,
              end_time=end_time,
              report_path=report_path
          )
          
          if 'error' in report:
              return return_msg({"error": report['error']})
          
          return return_msg({"success": True, "report": report})
      except Exception as e:
          logger.error(f"生成机器人报告失败: {str(e)}")
          return return_msg(f"生成机器人报告失败: {str(e)}")

• URBasic/advanced_data_analyzer.py:1767-1857 (helper)

  The core helper method in AdvancedDataAnalyzer class that generates the comprehensive robot report by loading data, performing multiple analyses (statistical, performance, energy, anomaly, correlation), compiling results into a JSON report, and optionally saving to file.

  def generate_report(self, 
                     start_time: Optional[float] = None,
                     end_time: Optional[float] = None,
                     robot_id: Optional[str] = None,
                     report_path: Optional[str] = None) -> Dict[str, Any]:
      """
      生成综合分析报告
      
      Args:
          start_time: 开始时间
          end_time: 结束时间
          robot_id: 机器人ID
          report_path: 报告保存路径
          
      Returns:
          Dict[str, Any]: 报告内容
      """
      # 加载数据
      df = self.load_data(
          start_time=start_time,
          end_time=end_time,
          robot_id=robot_id
      )
      
      if df.empty:
          return {'error': '没有找到数据'}
      
      report = {
          'metadata': {
              'generated_at': time.time(),
              'generated_at_str': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
              'robot_id': robot_id,
              'time_range': {
                  'start': start_time,
                  'end': end_time
              },
              'data_points': len(df)
          },
          'analyses': {}
      }
      
      # 执行各种分析
      
      # 1. 统计分析
      numeric_cols = df.select_dtypes(include=['float64', 'int64']).columns.tolist()
      if numeric_cols:
          report['analyses']['statistical'] = self.analyze(
              df,
              AnalysisType.STATISTICAL,
              {'columns': numeric_cols[:5]}  # 分析前5个数值列
          )
      
      # 2. 性能分析（如果有相关列）
      if 'operation_type' in df.columns and 'execution_time' in df.columns:
          report['analyses']['performance'] = self.analyze(
              df,
              AnalysisType.PERFORMANCE,
              {'operation_column': 'operation_type', 'duration_column': 'execution_time'}
          )
      
      # 3. 能耗分析（如果有相关列）
      if 'voltage' in df.columns and 'current' in df.columns:
          report['analyses']['energy'] = self.analyze(
              df,
              AnalysisType.ENERGY,
              {'voltage_column': 'voltage', 'current_column': 'current'}
          )
      
      # 4. 异常检测（对温度等关键参数）
      temperature_cols = [col for col in df.columns if 'temp' in col.lower() or 'temperature' in col.lower()]
      if temperature_cols:
          report['analyses']['anomalies'] = self.analyze(
              df,
              AnalysisType.ANOMALY,
              {'columns': temperature_cols}
          )
      
      # 5. 相关性分析
      report['analyses']['correlation'] = self.analyze(
          df,
          AnalysisType.CORRELATION
      )
      
      # 保存报告
      if report_path:
          os.makedirs(os.path.dirname(report_path), exist_ok=True)
          with open(report_path, 'w', encoding='utf-8') as f:
              json.dump(report, f, ensure_ascii=False, indent=2)
          logger.info(f"分析报告已保存到: {report_path}")
      
      return report