""" Permanent Magnet Discovery Pipeline v2 CrystaLLM → NequIP relaxation → MP property prediction Key changes from v1: - Removed ALIGNN endpoints (systematic ~1.6 eV/atom overestimation confirmed; MnBi false negative) - Added NequIP-OAM-XL structural relaxation as required preprocessing step - Formation energy: Materials Project route (not JARVIS) - Curie temperature and magnetic saturation: dedicated routes (hermes) Route interfaces (file-object pattern): Relax, formation energy, Curie temp, magnetic saturation all take: {"file": {"url": "...", "filename": "...", "type": "cif", ...}} NOT raw CIF bytes. Async note: relaxation is async — returns action_id; poll for completion. """ from ouro_client import OuroClient # ── Route IDs ────────────────────────────────────────────────────────────────── CRYSTALLM_ROUTE = "a3088687-0fce-45fd-8dbb-952be5fc4d1c" RELAX_ROUTE = "d040d3b6-faad-40cf-9d7c-999a5c769ed8" # NequIP MLIP (async, webhook) EFORM_ROUTE = "1ce64a21-efb5-4c01-8073-366fdc895fe7" # MP formation energy TC_ROUTE = "daf42af4-a3e4-4f9e-af65-6ecaafc26334" # Curie temperature MSAT_ROUTE = "d1fdf6d1-2b35-47af-956f-1b83c2fca036" # Magnetic saturation # ── Screening thresholds ────────────────────────────────────────────────────── THRESHOLDS = { "e_form_max": 0.050, # eV/atom (generous; MP ground truth calibration target) "tc_min": 400.0, # K — useful PM needs measurable T_C "msat_min": 50.0, # emu/g — soft magnetic saturation floor } RARE_EARTHS = {"La", "Ce", "Pr", "Nd", "Sm", "Eu", "Gd", "Tb", "Dy", "Ho", "Er", "Tm", "Yb", "Lu"} # ── Helpers ──────────────────────────────────────────────────────────────────── def _has_rare_earth(composition: str) -> bool: import re elements = set(re.sub(r"[0-9+\s()]", "", composition)) return bool(elements & RARE_EARTHS) def _build_file_obj(cif_content: str, client: OuroClient, comp: str) -> dict: """Upload CIF text and return a route-ready file object with URL.""" import re import hashlib safe = re.sub(r"[^a-zA-Z0-9]", "_", comp) fname = f"{safe}_{hashlib.md5(cif_content.encode()).hexdigest()[:6]}.cif" asset = client.files.upload( file_content_text=cif_content, file_name=fname, name=fname, org_id=client.org_id, team_id=client.team_id, visibility="private", description=f"Auto-uploaded by PM pipeline for {comp}", ) return { "url": asset["url"], "filename": fname, "type": "cif", "org_id": client.org_id, "team_id": client.team_id, "visibility": "private", } def _poll_action(client: OuroClient, action_id: str, timeout: int = 120, poll_every: float = 3.0) -> dict: """Poll an async action until completion.""" import time start = time.time() while time.time() - start < timeout: action = client.actions.get(action_id) if action["status"] in ("completed", "failed"): return action time.sleep(poll_every) raise TimeoutError(f"Action {action_id} did not complete within {timeout}s") def _extract_eform(resp: dict) -> float: for key in ("formation_energy_per_atom", "formation_energy", "eform", "energy"): if key in resp: return float(resp[key]) vals = [v for v in resp.values() if isinstance(v, (int, float))] return float(vals[0]) if vals else 0.0 def _extract_tc(resp: dict) -> float: for key in ("curie_temperature", "tc", "t_c", "temperature"): if isinstance(resp, dict) and key in resp: return float(resp[key]) vals = [v for v in (resp.values() if isinstance(resp, dict) else []) if isinstance(v, (int, float))] return float(vals[0]) if vals else 0.0 def _extract_msat(resp: dict) -> float: for key in ("magnetic_saturation", "msat", "m_sat", "saturation"): if isinstance(resp, dict) and key in resp: return float(resp[key]) vals = [v for v in (resp.values() if isinstance(resp, dict) else []) if isinstance(v, (int, float))] return float(vals[0]) if vals else 0.0 # ── Main screening function ─────────────────────────────────────────────────── def screen_candidates(compositions: list[str], client=None) -> dict: """ CrystaLLM → NequIP relaxation → MP property prediction pipeline. Args: compositions: list of chemical formulas, e.g. ["MnBi", "FeCo"] client: OuroClient instance (creates one if None) Returns: dict with keys: compositions_screened, passing_count, passing, all_results """ client = client or OuroClient() results = [] for comp in compositions: if _has_rare_earth(comp): results.append({"composition": comp, "status": "skipped_rare_earth"}) continue try: # Step 1: CrystaLLM generate gen_resp = client.routes.execute( route_id=CRYSTALLM_ROUTE, parameters={"composition": comp}, ) cif = ( gen_resp.get("cif") or gen_resp.get("structure") or gen_resp.get("result", "") ) if not cif or len(cif) < 50: results.append({"composition": comp, "status": "error_crystallm", "error": "empty CIF"}) continue # Step 2: Upload CIF for downstream routes file_obj = _build_file_obj(cif, client, comp) # Step 3: Structural relaxation (NequIP, async) relax_resp = client.routes.execute( route_id=RELAX_ROUTE, parameters={ "file": file_obj, "fmax": 0.03, "max_steps": 400, "optimize_cell": True, }, ) action_id = relax_resp.get("action_id") if not action_id: # Synchronous fallback relaxed_cif = cif relax_status = "sync_fallback" else: action = _poll_action(client, action_id) if action["status"] == "failed": results.append({ "composition": comp, "status": "error_relax", "error": action.get("error", "relaxation failed"), }) continue output_files = action.get("output_files", []) if not output_files: results.append({"composition": comp, "status": "error_relax", "error": "no output file"}) continue relaxed_asset = client.files.get(output_files[0]["id"]) relaxed_cif = relaxed_asset.get("content", relaxed_asset.get("cif", "")) relax_status = "relaxed" if not relaxed_cif: results.append({"composition": comp, "status": "error_relax", "error": "empty relaxed CIF"}) continue relaxed_file = _build_file_obj(relaxed_cif, client, f"{comp}_relaxed") # Step 4: MP Formation energy eform_resp = client.routes.execute( route_id=EFORM_ROUTE, parameters={"file": relaxed_file}, ) e_form = _extract_eform(eform_resp) # Step 5: Curie temperature tc_resp = client.routes.execute( route_id=TC_ROUTE, parameters={"file": relaxed_file}, ) tc = _extract_tc(tc_resp) # Step 6: Magnetic saturation msat_resp = client.routes.execute( route_id=MSAT_ROUTE, parameters={"file": relaxed_file}, ) msat = _extract_msat(msat_resp) # Screen passes = ( float(e_form) <= THRESHOLDS["e_form_max"] and float(tc) >= THRESHOLDS["tc_min"] and float(msat) >= THRESHOLDS["msat_min"] ) results.append({ "composition": comp, "e_form": round(float(e_form), 4), "tc": round(float(tc), 1), "msat": round(float(msat), 1), "relax_status": relax_status, "passes": passes, "status": "pass" if passes else "below_threshold", }) except TimeoutError as exc: results.append({"composition": comp, "status": "error_timeout", "error": str(exc)}) except Exception as exc: results.append({"composition": comp, "status": "error", "error": str(exc)}) passing = [r for r in results if r.get("status") == "pass"] return { "compositions_screened": len(compositions), "passing_count": len(passing), "passing": passing, "all_results": results, } if __name__ == "__main__": compositions = [ "MnBi", "MnAl", "MnGa", "FeCo", "FeNi", "CoPt", "Fe3N", "Fe5C2", ] print("Permanent Magnet Discovery Pipeline v2") print(f"Compositions: {compositions}") print(f"Thresholds: {THRESHOLDS}") print("-" * 60) outcome = screen_candidates(compositions) print(f"Screened: {outcome['compositions_screened']}") print(f"Passing: {outcome['passing_count']}") for r in outcome["all_results"]: status = r.get("status", "?") if status == "pass": print(f" [PASS] {r['composition']}: E_form={r['e_form']} eV/atom, " f"Tc={r['tc']} K, Msat={r['msat']} emu/g") elif status in ("skipped_rare_earth", "error_crystallm", "error_relax", "error_timeout"): print(f" [--] {r['composition']}: {status}") else: print(f" [FAIL] {r['composition']}: E_form={r.get('e_form','?')}, " f"Tc={r.get('tc','?')}, Msat={r.get('msat','?')}")