"""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, Tuple 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 # P5-M6: physics-domain grouping (single source of truth = afmcore.metrics) try: import sys as _sys _afm_src = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname( os.path.dirname(os.path.abspath(__file__)))))), "src", ) if _afm_src not in _sys.path: _sys.path.insert(0, _afm_src) from afmcore.metrics import METRIC_DEFINITIONS as _MD _METRIC_DOMAIN: Dict[str, str] = { m["key"]: m.get("domain", "electromagnetic") for m in _MD } _METRIC_LABEL: Dict[str, str] = {m["key"]: m["label"] for m in _MD} _METRIC_UNIT: Dict[str, str] = {m["key"]: m.get("unit", "") for m in _MD} except Exception: # noqa: BLE001 _METRIC_DOMAIN = {} _METRIC_LABEL = {} _METRIC_UNIT = {} _DOMAIN_ORDER = ["electromagnetic", "thermal", "structural"] _DOMAIN_LABELS = { "electromagnetic": "Electromagnetic Metrics", "thermal": "Thermal Metrics", "structural": "Structural / Mechanical Metrics", } def _group_metrics_by_domain(metrics: Dict[str, Any]) -> Dict[str, Dict[str, Any]]: """Group a flat metrics dict by physics domain. Returns {domain: {key: value}}. Unknown keys default to 'electromagnetic'. Empty domains are omitted. """ grouped: Dict[str, Dict[str, Any]] = {d: {} for d in _DOMAIN_ORDER} for key, value in metrics.items(): domain = _METRIC_DOMAIN.get(key, "electromagnetic") if domain not in grouped: grouped[domain] = {} grouped[domain][key] = value return {d: v for d, v in grouped.items() if v} def _metric_display(key: str, value: Any) -> Tuple[str, str]: """Return (display_label, display_value) for a metric key.""" label = _METRIC_LABEL.get(key, key) unit = _METRIC_UNIT.get(key, "") if unit: label = "%s [%s]" % (label, unit) if "[" not in label else label if isinstance(value, float): disp = "%.4g" % value else: disp = str(value) return label, disp 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 (P5-M6: grouped by physics domain) doc.add_heading("Results Summary", level=1) results = task_data.get("result_metrics", {}) if results: grouped = _group_metrics_by_domain(results) for domain in _DOMAIN_ORDER: if domain not in grouped: continue doc.add_heading(_DOMAIN_LABELS.get(domain, domain), level=2) 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 grouped[domain].items(): label, disp = _metric_display(key, value) row = table.add_row().cells row[0].text = label row[1].text = disp # 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.""" _raw_metrics = task_data.get("result_metrics", {}) report = { "title": report_title or f"Simulation Report - {task_data.get('task_name', 'Task')}", "generated_at": datetime.now().isoformat(), "task_data": task_data, "metrics_by_domain": _group_metrics_by_domain(_raw_metrics), "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