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

"""Platform P2 regression tests: topology registry + adapter-driven executor.

Covers (docs/PLATFORM_DESIGN_V2.md P2):
1. Topology registry: registration, lookup, parameter system, validation.
2. plan_schema topology validation integration.
3. Adapter registry + get_adapter("motorcad").
4. MotorCADTaskExecutor wired through get_adapter (metrics flattened to
   top level, FAILED points raise, _compute_metrics works) - via a fake
   adapter, so NO real Motor-CAD is launched.

Run:  python scripts/test_platform_registry.py
Exit code 0 = all PASS. Pure ASCII source.
"""
import os
import sys

ROOT = os.path.dirname(os.path.abspath(__file__))
PROJ = os.path.dirname(ROOT)
sys.path.insert(0, ROOT)
sys.path.insert(0, PROJ)
sys.path.insert(0, os.path.join(PROJ, "src"))
sys.path.insert(0, os.path.join(PROJ, "web", "backend"))

PASS = 0
FAIL = 0


def check(name, cond, detail=""):
    global PASS, FAIL
    if cond:
        PASS += 1
        print("  [PASS] %s%s" % (name, (" - " + detail) if detail else ""))
    else:
        FAIL += 1
        print("  [FAIL] %s%s" % (name, (" - " + detail) if detail else ""))


def section(title):
    print("\n== %s ==" % title)


# ---------------------------------------------------------------------------
section("1. Topology registry")
from afmcore.topology import (  # noqa: E402
    TOPOLOGY_REGISTRY,
    get_topology,
    get_topology_or_none,
    list_topologies,
    list_topology_codes,
    is_supported,
    is_active,
    param_names_for,
    validate_params,
    to_dict,
    register_topology,
    TopologyDefinition,
)

check("registry has 3 topologies", len(TOPOLOGY_REGISTRY) == 3)
check("codes sorted", list_topology_codes() == ["DRSS", "SDSR", "SSSR"])
sssr = get_topology("SSSR")
check("SSSR label_zh", sssr.label_zh == "\u5355\u5b9a\u5b50\u5355\u8f6c\u5b50", sssr.label_zh)
check("SSSR active", is_active("SSSR"))
check("SSSR 37 params", len(sssr.all_params()) == 37, str(len(sssr.all_params())))
check("SSSR 8 groups", len(sssr.param_system) == 8)
check("SSSR sizing", "Airgap" in sssr.sizing_params and "Magnet_Thickness" in sssr.sizing_params)
check("SSSR scan vars", set(sssr.default_scan_vars) >= {"Airgap", "Magnet_Thickness", "RMSCurrent"})
check("SSSR single airgap", sssr.airgap_count == 1 and sssr.stator_count == 1 and sssr.rotor_count == 1)

drss = get_topology("drss")  # case-insensitive lookup
check("DRSS planned + double airgap",
      drss.status == "planned" and drss.airgap_count == 2 and drss.rotor_count == 2)
check("DRSS not active", not is_active("DRSS"))
check("SDSR planned", get_topology("SDSR").status == "planned")

check("is_supported('XXX') False", not is_supported("XXX"))
check("get_topology_or_none unknown None", get_topology_or_none("XXX") is None)

r = validate_params("SSSR", {"Airgap": 1.0, "RMSCurrent": 15.0, "Foo": 3})
check("validate known", set(r["known"]) == {"Airgap", "RMSCurrent"})
check("validate unknown", r["unknown"] == ["Foo"])
check("validate supported", r["unsupported"] is False)

r2 = validate_params("XXX", {"Airgap": 1.0})
check("unknown topology -> unsupported", r2["unsupported"] is True)

snap = to_dict()
check("to_dict has 3", len(snap) == 3)
check("to_dict SSSR params", len(snap["SSSR"]["params"]) == 37)

# custom registration (idempotent / override)
register_topology(TopologyDefinition(code="SSSR", label_zh="x", label_en="y", status="active"))
check("re-register same code overrides", get_topology("SSSR").label_en == "y")
register_topology(TopologyDefinition(code="SSSR", label_zh="\u5355\u5b9a\u5b50\u5355\u8f6c\u5b50",
                                     label_en="Single Stator Single Rotor", status="active",
                                     param_system=sssr.param_system,
                                     sizing_params=sssr.sizing_params,
                                     default_scan_vars=sssr.default_scan_vars))
check("SSSR restored", get_topology("SSSR").label_en == "Single Stator Single Rotor")

# ---------------------------------------------------------------------------
section("2. plan_schema topology validation")
from src.plan_schema import SimulationPlan  # noqa: E402

_, errs = SimulationPlan(topology="SSSR").validate()
check("SSSR no topology error", all("topology not supported" not in e for e in errs), str(errs))
_, errs = SimulationPlan(topology="TORUS").validate()
check("TORUS rejected", any("topology not supported" in e for e in errs))
_, errs = SimulationPlan(topology="DRSS").validate()
check("DRSS accepted (registered)", all("topology not supported" not in e for e in errs))
rt = SimulationPlan.from_dict(SimulationPlan(topology="SSSR").to_dict())
check("serialization round-trip", rt.topology == "SSSR")

# ---------------------------------------------------------------------------
section("3. Adapter registry")
from afmcore.adapters import (  # noqa: E402
    SimulationAdapter,
    register_adapter,
    get_adapter,
    registered_tools,
)
import afmcore.adapters.motorcad  # noqa: F401,E402  (registers "motorcad")

check("motorcad registered", "motorcad" in registered_tools())
mc = get_adapter("motorcad", model_path="models/x.mot")
check("get_adapter returns SimulationAdapter", isinstance(mc, SimulationAdapter))
check("adapter tool meta", mc.tool_name == "motorcad" and mc.tool_label == "Motor-CAD (ANSYS)")
try:
    get_adapter("maxwell")
    check("unknown tool raises", False)
except KeyError:
    check("unknown tool raises", True)

# ---------------------------------------------------------------------------
section("4. MotorCADTaskExecutor through adapter (fake, no Motor-CAD)")
CALLS = []


class FakeAdapter(SimulationAdapter):
    tool_name = "fake"

    def __init__(self, **kw):
        super().__init__(**kw)
        self.points = 0

    def connect(self):
        CALLS.append("connect")

    def disconnect(self):
        CALLS.append("disconnect")

    def load_model(self, model_path):
        CALLS.append("load:" + model_path)

    def set_parameter(self, name, value):
        CALLS.append("set:%s" % name)

    def run_simulation(self, mode="electromagnetic"):
        CALLS.append("run")

    def extract_metrics(self, output_dir, tag=""):
        return {"metrics": {}, "status": "OK", "error": None, "raw_path": ""}

    def run_point(self, model_path, params=None, output_dir="output", tag=""):
        CALLS.append("run_point:" + str(tag))
        self.points += 1
        if tag == "fail":
            return {"metrics": {}, "status": "FAILED", "error": "boom", "raw_path": ""}
        return {"metrics": {"tavg_nm": 12.5 + self.points, "efficiency_pct": 93.1},
                "status": "OK", "error": None, "raw_path": "x", "solve_time_s": 1.5}


register_adapter("fake", FakeAdapter)

from task_executor import MotorCADTaskExecutor  # noqa: E402

ex = MotorCADTaskExecutor(model_path="models/x.mot", tool="fake", web_base_url="http://127.0.0.1:1")
res = ex._run_simulation_point({"Airgap": 1.0}, 0)
check("OK point flattened", res.get("tavg_nm") == 13.5 and res.get("status") == "OK", str(res))
computed = ex._compute_metrics([res, res])
check("compute_metrics mean", computed.get("tavg_nm_mean") == 13.5, str(computed))
check("compute_metrics counts", computed["successful_points"] == 2 and computed["failed_points"] == 0)
try:
    ex._run_simulation_point({"Airgap": 1.0}, "fail")
    check("FAILED point raises", False)
except RuntimeError:
    check("FAILED point raises", True)

# execute_task end-to-end with 3 points (no real backend calls needed; adapter
# drives the simulation; requests is available so progress reports just 404)
ex.execute_task({"task_id": "T-P2", "parameters": [{"Airgap": 1.0}, {"Airgap": 1.5}, {"Airgap": 2.0}]})
check("execute_task created adapter", ex._adapter is not None)
ex.cleanup()
check("cleanup disconnected", "disconnect" in CALLS and ex._adapter is None)

# ---------------------------------------------------------------------------
print("\n================================")
print("PASS: %d   FAIL: %d" % (PASS, FAIL))
sys.exit(0 if FAIL == 0 else 1)