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