"""Report generation service (P4-M4). Generates Word simulation reports from task results. Includes cover, parameters, results summary, metrics, AI analysis. """ import json import os from datetime import datetime from typing import Any, Dict, List, Optional try: from docx import Document from docx.shared import Inches, Pt, RGBColor from docx.enum.text import WD_ALIGN_PARAGRAPH HAS_DOCX = True except ImportError: HAS_DOCX = False class ReportGenerator: """Generate simulation reports from task results.""" def __init__(self, output_dir: Optional[str] = None): self.output_dir = output_dir or os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "output", "reports" ) os.makedirs(self.output_dir, exist_ok=True) def generate_report(self, task_data: Dict[str, Any], ai_analysis: Optional[Dict[str, Any]] = None, report_title: Optional[str] = None) -> str: """Generate a Word report from task data. Args: task_data: Task data including parameters, results, metrics ai_analysis: Optional AI analysis results report_title: Custom report title Returns: Path to generated report file """ if not HAS_DOCX: return self._generate_json_report(task_data, ai_analysis, report_title) doc = Document() # Title title = report_title or f"Simulation Report - {task_data.get('task_name', 'Task')}" heading = doc.add_heading(title, level=0) heading.alignment = WD_ALIGN_PARAGRAPH.CENTER # Metadata doc.add_paragraph(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") doc.add_paragraph(f"Task ID: {task_data.get('task_id', 'N/A')}") doc.add_paragraph(f"Status: {task_data.get('status', 'N/A')}") # Parameters section doc.add_heading("Simulation Parameters", level=1) params = task_data.get("plan_data", {}) if params: table = doc.add_table(rows=1, cols=2) table.style = "Table Grid" hdr = table.rows[0].cells hdr[0].text = "Parameter" hdr[1].text = "Value" for key, value in params.items(): row = table.add_row().cells row[0].text = str(key) row[1].text = str(value) # Results summary doc.add_heading("Results Summary", level=1) results = task_data.get("result_metrics", {}) if results: table = doc.add_table(rows=1, cols=2) table.style = "Table Grid" hdr = table.rows[0].cells hdr[0].text = "Metric" hdr[1].text = "Value" for key, value in results.items(): row = table.add_row().cells row[0].text = str(key) row[1].text = str(value) # Per-point results doc.add_heading("Per-Point Results", level=1) points = task_data.get("points", []) if points: table = doc.add_table(rows=1, cols=4) table.style = "Table Grid" hdr = table.rows[0].cells hdr[0].text = "Point" hdr[1].text = "Parameters" hdr[2].text = "Status" hdr[3].text = "Duration (s)" for i, point in enumerate(points): row = table.add_row().cells row[0].text = str(i + 1) row[1].text = json.dumps(point.get("params", {}), ensure_ascii=False) row[2].text = point.get("status", "N/A") row[3].text = str(point.get("duration_s", "N/A")) # AI Analysis section if ai_analysis: doc.add_heading("AI Analysis", level=1) if "summary" in ai_analysis: doc.add_paragraph(ai_analysis["summary"]) if "convergence" in ai_analysis: doc.add_heading("Convergence Assessment", level=2) conv = ai_analysis["convergence"] doc.add_paragraph(f"Converged: {conv.get('converged', 'N/A')}") doc.add_paragraph(f"Confidence: {conv.get('confidence', 'N/A')}") if "recommendations" in ai_analysis: doc.add_heading("Recommendations", level=2) for rec in ai_analysis["recommendations"]: doc.add_paragraph(rec, style="List Bullet") # Save filename = f"report_{task_data.get('task_id', 'task')}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.docx" filepath = os.path.join(self.output_dir, filename) doc.save(filepath) return filepath def _generate_json_report(self, task_data: Dict[str, Any], ai_analysis: Optional[Dict[str, Any]], report_title: Optional[str]) -> str: """Fallback: generate JSON report when python-docx is unavailable.""" report = { "title": report_title or f"Simulation Report - {task_data.get('task_name', 'Task')}", "generated_at": datetime.now().isoformat(), "task_data": task_data, "ai_analysis": ai_analysis, } filename = f"report_{task_data.get('task_id', 'task')}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" filepath = os.path.join(self.output_dir, filename) with open(filepath, "w", encoding="utf-8") as f: json.dump(report, f, ensure_ascii=False, indent=2) return filepath def list_reports(self) -> List[Dict[str, Any]]: """List all generated reports.""" reports = [] if os.path.exists(self.output_dir): for f in os.listdir(self.output_dir): if f.endswith((".docx", ".json")): filepath = os.path.join(self.output_dir, f) reports.append({ "filename": f, "path": filepath, "size": os.path.getsize(filepath), "created_at": datetime.fromtimestamp(os.path.getctime(filepath)).isoformat(), }) return sorted(reports, key=lambda x: x["created_at"], reverse=True) _generator: Optional[ReportGenerator] = None def get_report_generator() -> ReportGenerator: global _generator if _generator is None: _generator = ReportGenerator() return _generator