pcb-afm-simulation-system/scripts/test_metrics_extension.py

"""P5-M6: unit tests for metrics extension (thermal + structural) and
domain-grouped report generation.

Covers:
- New thermal/structural metrics present in METRIC_DEFINITIONS with domain
- parse_export + extract_all_metrics picks up new metrics automatically
- check_required_metrics unaffected (new metrics required=False)
- report_generator._group_metrics_by_domain correct grouping
- report_generator JSON fallback includes metrics_by_domain
- Boundary: empty metrics, unknown keys, mixed domains
- robust_motorcad enable_thermal parameter exists (signature check)

All source is ASCII only. Run: python scripts/test_metrics_extension.py
exit 0 = PASS.
"""
import os
import sys
import tempfile
import unittest

_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_SRC = os.path.join(_ROOT, "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from afmcore.metrics import (  # noqa: E402
    METRIC_DEFINITIONS,
    METRIC_KEYS,
    REQUIRED_METRICS,
    check_required_metrics,
    extract_all_metrics,
    parse_export,
)

# report_generator lives under web/backend; add its dir to path
_REPORT_DIR = os.path.join(_ROOT, "web", "backend", "app", "services")
if _REPORT_DIR not in sys.path:
    sys.path.insert(0, _REPORT_DIR)
from report_generator import (  # noqa: E402
    ReportGenerator,
    _DOMAIN_ORDER,
    _group_metrics_by_domain,
    _metric_display,
)


THERMAL_KEYS = [
    "winding_hotspot_temp_c", "magnet_temp_c", "stator_temp_c",
    "bearing_temp_c", "temp_rise_c", "thermal_resistance_k_w",
]
STRUCTURAL_KEYS = [
    "axial_force_n", "radial_force_n", "max_stress_mpa", "deformation_mm",
]


class TestMetricDefinitions(unittest.TestCase):
    def test_thermal_metrics_present(self):
        keys = {m["key"] for m in METRIC_DEFINITIONS}
        for k in THERMAL_KEYS:
            self.assertIn(k, keys, "missing thermal metric: %s" % k)

    def test_structural_metrics_present(self):
        keys = {m["key"] for m in METRIC_DEFINITIONS}
        for k in STRUCTURAL_KEYS:
            self.assertIn(k, keys, "missing structural metric: %s" % k)

    def test_thermal_metrics_have_domain(self):
        for m in METRIC_DEFINITIONS:
            if m["key"] in THERMAL_KEYS:
                self.assertEqual(m.get("domain"), "thermal",
                                 "%s should have domain=thermal" % m["key"])

    def test_structural_metrics_have_domain(self):
        for m in METRIC_DEFINITIONS:
            if m["key"] in STRUCTURAL_KEYS:
                self.assertEqual(m.get("domain"), "structural",
                                 "%s should have domain=structural" % m["key"])

    def test_new_metrics_not_required(self):
        for k in THERMAL_KEYS + STRUCTURAL_KEYS:
            self.assertNotIn(k, REQUIRED_METRICS,
                             "%s should be required=False" % k)

    def test_total_metric_count(self):
        # original 25 + 6 thermal + 4 structural = 35
        self.assertEqual(len(METRIC_DEFINITIONS), 35)


class TestMetricExtraction(unittest.TestCase):
    """Construct a mock Motor-CAD export CSV with thermal/structural fields
    and verify extract_all_metrics picks them up automatically."""

    def _make_export(self, fields):
        """Write a mock semicolon-CSV export and return its path."""
        lines = ["E-Magnetics"]
        for field, value in fields:
            lines.append("%s;%s" % (field, value))
        fd, path = tempfile.mkstemp(suffix=".csv")
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write("\n".join(lines))
        self.addCleanup(os.unlink, path)
        return path

    def test_extract_thermal_metrics(self):
        path = self._make_export([
            ("Average torque (virtual work)", "1.5"),
            ("Magnet Temperature", "62.5"),
            ("Winding Hotspot Temperature", "88.3"),
            ("Stator Temperature", "70.1"),
            ("Bearing Temperature", "55.0"),
            ("Temperature Rise", "48.2"),
            ("Thermal Resistance", "0.85"),
        ])
        parsed = parse_export(path)
        metrics = extract_all_metrics(parsed)
        self.assertAlmostEqual(metrics["magnet_temp_c"], 62.5, places=2)
        self.assertAlmostEqual(metrics["winding_hotspot_temp_c"], 88.3, places=2)
        self.assertAlmostEqual(metrics["stator_temp_c"], 70.1, places=2)
        self.assertAlmostEqual(metrics["bearing_temp_c"], 55.0, places=2)
        self.assertAlmostEqual(metrics["temp_rise_c"], 48.2, places=2)
        self.assertAlmostEqual(metrics["thermal_resistance_k_w"], 0.85, places=2)

    def test_extract_structural_metrics(self):
        path = self._make_export([
            ("Axial Force", "125.5"),
            ("Radial Force", "45.2"),
            ("Maximum Stress", "180.3"),
            ("Max Deformation", "0.12"),
        ])
        parsed = parse_export(path)
        metrics = extract_all_metrics(parsed)
        self.assertAlmostEqual(metrics["axial_force_n"], 125.5, places=2)
        self.assertAlmostEqual(metrics["radial_force_n"], 45.2, places=2)
        self.assertAlmostEqual(metrics["max_stress_mpa"], 180.3, places=2)
        self.assertAlmostEqual(metrics["deformation_mm"], 0.12, places=2)

    def test_chinese_alias_thermal(self):
        path = self._make_export([
            ("\u6c38\u78c1\u4f53\u6e29\u5ea6", "70.0"),  # magnet temp
            ("\u8f74\u5411\u529b", "200.0"),  # axial force
        ])
        parsed = parse_export(path)
        metrics = extract_all_metrics(parsed)
        self.assertAlmostEqual(metrics["magnet_temp_c"], 70.0, places=2)
        self.assertAlmostEqual(metrics["axial_force_n"], 200.0, places=2)

    def test_required_check_unaffected(self):
        # Only tavg/ripple/efficiency/total_losses are required;
        # missing thermal/structural metrics must not fail the check.
        ok, missing = check_required_metrics({"tavg_nm": 1.0, "ripple_pct": 2.0,
                                                "efficiency_pct": 90.0, "total_losses_w": 10.0})
        self.assertTrue(ok)
        self.assertEqual(missing, [])

    def test_required_check_fails_on_missing_core(self):
        ok, missing = check_required_metrics({"tavg_nm": 1.0})
        self.assertFalse(ok)
        self.assertIn("ripple_pct", missing)


class TestDomainGrouping(unittest.TestCase):
    def test_group_by_domain(self):
        metrics = {
            "tavg_nm": 1.5, "ripple_pct": 5.0,  # electromagnetic (default)
            "magnet_temp_c": 60.0, "temp_rise_c": 40.0,  # thermal
            "axial_force_n": 100.0, "max_stress_mpa": 150.0,  # structural
        }
        grouped = _group_metrics_by_domain(metrics)
        self.assertIn("electromagnetic", grouped)
        self.assertIn("thermal", grouped)
        self.assertIn("structural", grouped)
        self.assertEqual(set(grouped["thermal"].keys()), {"magnet_temp_c", "temp_rise_c"})
        self.assertEqual(set(grouped["structural"].keys()), {"axial_force_n", "max_stress_mpa"})

    def test_empty_metrics(self):
        self.assertEqual(_group_metrics_by_domain({}), {})

    def test_unknown_key_defaults_electromagnetic(self):
        grouped = _group_metrics_by_domain({"unknown_metric_xyz": 42.0})
        self.assertIn("electromagnetic", grouped)
        self.assertEqual(grouped["electromagnetic"]["unknown_metric_xyz"], 42.0)

    def test_domain_order(self):
        self.assertEqual(_DOMAIN_ORDER, ["electromagnetic", "thermal", "structural"])

    def test_metric_display(self):
        label, value = _metric_display("tavg_nm", 1.5)
        self.assertIn("Average Torque", label)
        self.assertEqual(value, "1.5")

    def test_metric_display_float_format(self):
        _, value = _metric_display("efficiency_pct", 92.3456789)
        self.assertEqual(value, "92.35")  # %.4g


class TestReportJsonFallback(unittest.TestCase):
    def test_json_report_includes_metrics_by_domain(self):
        rg = ReportGenerator(output_dir=tempfile.mkdtemp())
        task_data = {
            "task_id": "test-001",
            "task_name": "Test Task",
            "status": "completed",
            "result_metrics": {
                "tavg_nm": 1.5,
                "magnet_temp_c": 60.0,
                "axial_force_n": 100.0,
            },
        }
        path = rg._generate_json_report(task_data, None, None)
        self.addCleanup(os.unlink, path)
        import json
        with open(path, "r", encoding="utf-8") as f:
            report = json.load(f)
        self.assertIn("metrics_by_domain", report)
        self.assertIn("thermal", report["metrics_by_domain"])
        self.assertIn("structural", report["metrics_by_domain"])
        self.assertEqual(report["metrics_by_domain"]["thermal"]["magnet_temp_c"], 60.0)


class TestRobustMotorcadThermalParam(unittest.TestCase):
    """Verify enable_thermal parameter exists in RobustMotorCADSolver."""

    def test_init_has_enable_thermal(self):
        sys.path.insert(0, _ROOT)
        from scripts.robust_motorcad import RobustMotorCADSolver
        import inspect
        sig = inspect.signature(RobustMotorCADSolver.__init__)
        self.assertIn("enable_thermal", sig.parameters)
        self.assertFalse(sig.parameters["enable_thermal"].default)

    def test_run_single_point_has_enable_thermal(self):
        sys.path.insert(0, _ROOT)
        from scripts.robust_motorcad import RobustMotorCADSolver
        import inspect
        sig = inspect.signature(RobustMotorCADSolver.run_single_point)
        self.assertIn("enable_thermal", sig.parameters)
        self.assertIsNone(sig.parameters["enable_thermal"].default)


if __name__ == "__main__":
    unittest.main(verbosity=2)