pcb-afm-simulation-system/web/backend/app/services/report_generator.py

"""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