diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..cbd08ab --- /dev/null +++ b/Dockerfile @@ -0,0 +1,8 @@ +FROM python:3.11-slim +WORKDIR /app +COPY pyproject.toml README.md ./ +COPY matgraph ./matgraph +RUN pip install --no-cache-dir -e . +EXPOSE 8000 +ENV MATGRAPH_CACHE_DIR=/data/cache +CMD ["uvicorn","matgraph.graphql_app:app","--host","0.0.0.0","--port","8000"] diff --git a/README.md b/README.md index 8864ccb..25d58fa 100644 --- a/README.md +++ b/README.md @@ -17,11 +17,11 @@ Researchers spend weeks writing boilerplate to fetch crystal data, engineer feat | Problem | MatGraph solution | |---|---| | Fetching crystal structures from Materials Project | `sdk.predict("LiFePO4")` | -| Training CGCNN / MEGNet / M3GNet from scratch | Pre-wired architectures, ready to run | -| Exploring hypothetical new materials | `matgraph substitute LiFePO4 Li Na` | +| Running MatGL M3GNet inference | `matgraph predict LiFePO4 --model m3gnet` (only M3GNet ships in 2.x) | +| Exploring hypothetical new materials (heuristic) | `matgraph substitute LiFePO4 Li Na` (ML-guided, not GNoME-scale) | | Simulating XRD patterns | `matgraph xrd LiFePO4` | -| Serving predictions to a web app | Async GraphQL API with API key auth | -| Caching repeated queries | Built-in SQLite cache, zero config | +| Serving predictions to a web app | Async GraphQL + REST `/v1/predict` with hashed API keys | +| Caching repeated queries | Reproducible SQLite cache (structure_hash + model_version) | --- @@ -51,17 +51,17 @@ export MP_API_KEY="your_key_here" # Predict band gap and formation energy matgraph predict LiFePO4 -# Use a different model architecture +# M3GNet is the only model shipped in 2.x (cgcnn/megnet removed until real checkpoints ship) matgraph predict LiFePO4 --model m3gnet -# Discover new materials via elemental substitution +# ML-guided heuristic discovery (not GNoME-scale) matgraph substitute LiFePO4 Li Na # Simulate X-Ray Diffraction pattern matgraph xrd LiFePO4 -# Evaluate model accuracy (MAE) against ground truth -matgraph evaluate LiFePO4 --model megnet +# Evaluate formation-energy MAE (band_gap unavailable — no UQ model) +matgraph evaluate LiFePO4 --model m3gnet # Filter by physical constraints matgraph predict LiFePO4 --min-gap 1.5 --crystal-system Cubic @@ -91,9 +91,9 @@ print("Stable" if discovery["is_more_stable"] else "Unstable") # XRD simulation xrd = sdk.xrd("LiFePO4") -# Model evaluation -metrics = sdk.evaluate("LiFePO4", model="megnet") -print(f"Band gap MAE: {metrics['band_gap_mae']}") +# Model evaluation — band_gap MAE is None until a real band-gap model ships +metrics = sdk.evaluate("LiFePO4", model="m3gnet") +print(f"Formation energy MAE: {metrics['formation_energy_mae']}") ``` ### GraphQL API @@ -126,21 +126,22 @@ Or open `http://localhost:8000/graphql` for the interactive GraphiQL playground. ## Features -### Deep Learning Models +### Deep Learning Models (2.x — M3GNet only) -| Model | Predicts | Architecture | -|---|---|---| -| **CGCNN** | Band gap, Formation energy | Crystal Graph Convolutional Neural Network | -| **MEGNet** | Band gap, Formation energy | MatErials Graph Network | -| **M3GNet** | Energy, Forces, Stresses | Multi-body interaction universal potential | +| Model | Predicts | Architecture | Status | +|---|---|---|---| +| **M3GNet** | Energy, Forces, Stresses, Formation energy (via MatGL) | Multi-body universal potential | ✅ Ships (`M3GNet-PES-MatPES-PBE-2025.2`) | +| CGCNN/MEGNet | — | Graph networks | ❌ Removed in 2.0 until real checkpoints/benchmarks ship (was alias to M3GNet) | -### Generative Discovery (GNoME-inspired) +> **Band gap:** no ML band-gap model ships — `predicted_band_gap=None`, `band_gap_source="mp_experimental"`, filter on `true_band_gap` only. -Inspired by [Google DeepMind's GNoME](https://deepmind.google/discover/blog/millions-of-new-materials-discovered-with-deep-learning/) paper. Substitute elements in known stable materials and predict whether the hypothetical new compound is thermodynamically stable -- without synthesizing it in a lab. +### ML-guided heuristic discovery (experimental) + +Heuristic elemental substitution + simple GA ranking via M3GNet energies. Useful for triage, **not** GNoME-scale generative discovery. ```bash matgraph substitute LiFePO4 Li Na -# Predicts: NaFePO4 stability vs LiFePO4 +# Predicts: NaFePO4 stability vs LiFePO4 (heuristic, validate with DFT) ``` ### XRD Simulation @@ -151,25 +152,23 @@ Generate theoretical Cu-Ka X-Ray Diffraction patterns for any material. Useful f matgraph xrd LiFePO4 ``` -### Built-in Cache +### Reproducible cache (2.0) -All API responses and predictions are automatically cached in a local SQLite database (`~/.matgraph_cache/cache.db`). Repeated queries return instantly. No external service required. +SQLite + WAL at `~/.matgraph_cache/cache.db` (override `MATGRAPH_CACHE_DIR`), key = `material_id+structure_hash+model+checkpoint+code_version+params`. Reproducibility via `provenance` field on every prediction. ```bash matgraph cache stats # View cache size and entry count matgraph cache clear # Wipe the cache ``` -### API Key Authentication +### Hashed API keys (2.0) -Generate secure, multi-tenant API keys for the GraphQL server: +Keys are `mg_*`, stored as `sha256` with `scopes/expiry/revocation` in `~/.matgraph_keys.json` (override `MATGRAPH_AUTH_KEYS_FILE`). `MATGRAPH_API_KEY` master key still supported. Not multi-tenant authz — local research use. ```bash matgraph auth generate --user "research-team-A" ``` -Keys are prefixed with `mg_`, stored in `~/.matgraph_keys.json`, and validated on every request. You can also set a master key via the `MATGRAPH_API_KEY` environment variable. - ### Dataset Export Export predictions to CSV or JSON for use in pandas, scikit-learn, or any ML pipeline. Optionally export 3D crystal structures as `.cif` files. @@ -184,16 +183,22 @@ matgraph predict LiFePO4 --save results.json --format json --cif ``` matgraph/ - __init__.py # Top-level SDK export - sdk.py # Python SDK (MatGraphSDK class) - cli.py # Typer CLI with Rich formatting - core.py # Pipeline orchestration and feature extraction - cgcnn.py # Crystal Graph Convolutional Neural Network - megnet.py # MatErials Graph Network - m3gnet.py # M3GNet Universal Potential - graphql_app.py # FastAPI + Strawberry GraphQL server - auth.py # API key generation and validation - cdn.py # SQLite cache layer + __init__.py + sdk.py # SDK (predict/substitute/xrd/... + DataFrame) + cli.py # Typer CLI + core.py # Orchestration shim (re-exports data/models/...) + client.py # Materials Project client + models.py # M3GNet registry (settings.pes_model) + schemas.py # Pydantic validation, no hardcodes + settings.py # Central MATGRAPH_* settings + cdn.py # WAL SQLite cache + auth.py # sha256 keys + scopes/expiry + ga.py # Heuristic GA (param-driven) + graphql_app.py # GraphQL + REST /v1/predict + /health + data/ # (v2 split) materials_project + simulation/ # xrd/phonon/relax + dft/ # vasp/qe input generation + properties/ # stability/elastic/... ``` ### Tech Stack diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..60518b1 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,13 @@ +services: + api: + build: . + ports: ["8000:8000"] + environment: + - MP_API_KEY=${MP_API_KEY} + - MATGRAPH_CACHE_DIR=/data/cache + - MATGRAPH_GRAPHQL_DEFAULT_LIMIT=10 + volumes: + - matcache:/data/cache + +volumes: + matcache: diff --git a/matgraph/api/__init__.py b/matgraph/api/__init__.py new file mode 100644 index 0000000..5f05de8 --- /dev/null +++ b/matgraph/api/__init__.py @@ -0,0 +1,2 @@ +from matgraph.graphql_app import app +__all__ = ["app"] diff --git a/matgraph/auth.py b/matgraph/auth.py index 113ef9f..b4a3977 100644 --- a/matgraph/auth.py +++ b/matgraph/auth.py @@ -1,45 +1,88 @@ import json import secrets import os +import hashlib +import time from pathlib import Path from typing import Optional -KEYS_FILE = Path.home() / ".matgraph_keys.json" +def _keys_file() -> Path: + from matgraph.settings import settings + return settings.auth_keys_file + +def _prefix() -> str: + from matgraph.settings import settings + return settings.auth_key_prefix def load_keys() -> dict: - if not KEYS_FILE.exists(): + f = _keys_file() + if not f.exists(): return {} - with open(KEYS_FILE, "r") as f: + with open(f, "r") as fh: try: - return json.load(f) + return json.load(fh) except json.JSONDecodeError: return {} def save_keys(keys: dict): - with open(KEYS_FILE, "w") as f: - json.dump(keys, f, indent=4) + f = _keys_file() + f.parent.mkdir(parents=True, exist_ok=True) + with open(f, "w") as fh: + json.dump(keys, fh, indent=4) + try: + f.chmod(0o600) + except Exception: + pass + +def _hash_key(api_key: str) -> str: + return hashlib.sha256(api_key.encode()).hexdigest() -def generate_api_key(user_name: str) -> str: - """Generates a secure API key for a user and saves it.""" +def generate_api_key(user_name: str, ttl_days: Optional[int] = None, scopes: Optional[list] = None) -> str: + """Generates a secure API key, stores only hash. No plaintext.""" + from matgraph.settings import settings + if ttl_days is None: + ttl_days = settings.auth_default_ttl_days + if scopes is None: + scopes = ["read:predict","read:phonon","read:elastic"] keys = load_keys() - new_key = "mg_" + secrets.token_urlsafe(24) - keys[new_key] = { - "user": user_name, - "active": True - } + raw = _prefix() + secrets.token_urlsafe(24) + h = _hash_key(raw) + expires_at = time.time() + ttl_days*86400 if ttl_days else None + keys[h] = {"user": user_name, "active": True, "scopes": scopes, "created_at": time.time(), "expires_at": expires_at, "prefix": raw[:8]+"..."} save_keys(keys) - return new_key + return raw -def is_valid_key(api_key: str) -> bool: - """Checks if the API key is valid.""" - # Allow master key from env for dev purposes - master_key = os.environ.get("MATGRAPH_API_KEY") - if master_key and api_key == master_key: +def is_valid_key(api_key: str, required_scope: Optional[str] = None) -> bool: + master = os.environ.get("MATGRAPH_API_KEY") + if master and api_key == master: return True - + # support legacy plaintext keys file for migration + h = _hash_key(api_key) keys = load_keys() - key_info = keys.get(api_key) - if key_info and key_info.get("active", False): - return True - - return False + # legacy: if keys contain plaintext key directly, migrate check + if api_key in keys: + info = keys[api_key] + else: + info = keys.get(h) + if not info or not info.get("active", False): + return False + exp = info.get("expires_at") + if exp and time.time() > exp: + return False + if required_scope and required_scope not in info.get("scopes", []): + return False + return True + +def revoke_key(api_key: str) -> bool: + h = _hash_key(api_key) + keys = load_keys() + # try hash or plaintext + target = h if h in keys else (api_key if api_key in keys else None) + if not target: + return False + keys[target]["active"] = False + save_keys(keys) + return True + +def list_keys() -> dict: + return load_keys() diff --git a/matgraph/cdn.py b/matgraph/cdn.py index 74259a0..94119ee 100644 --- a/matgraph/cdn.py +++ b/matgraph/cdn.py @@ -17,14 +17,30 @@ from pathlib import Path from typing import Optional, Any -CACHE_DIR = Path.home() / ".matgraph_cache" -CACHE_DB = CACHE_DIR / "cache.db" +def _cache_dir() -> Path: + from matgraph.settings import settings, cache_db_path + return settings.cache_dir + +def _cache_db() -> Path: + from matgraph.settings import cache_db_path + return cache_db_path() + +CACHE_DIR = _cache_dir() +CACHE_DB = _cache_db() def _init_db() -> sqlite3.Connection: """Initialize the SQLite cache database.""" - CACHE_DIR.mkdir(parents=True, exist_ok=True) - conn = sqlite3.connect(str(CACHE_DB)) + from matgraph.settings import settings + db = _cache_db() + db.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(str(db)) + # WAL for concurrency, no hardcode + try: + conn.execute("PRAGMA journal_mode=WAL;") + conn.execute("PRAGMA synchronous=NORMAL;") + except Exception: + pass conn.execute(""" CREATE TABLE IF NOT EXISTS cache ( key TEXT PRIMARY KEY, @@ -43,11 +59,14 @@ def _get_cache_key(prefix: str, **kwargs) -> str: return f"{prefix}:{digest}" -def cache_get(prefix: str, ttl: int = 3600, **kwargs) -> Optional[Any]: +def cache_get(prefix: str, ttl: Optional[int] = None, **kwargs) -> Optional[Any]: """ Retrieve cached result if it exists and is within TTL. - Returns None on cache miss. + TTL comes from settings.get_ttl(prefix) if not passed. """ + from matgraph.settings import get_ttl + if ttl is None: + ttl = get_ttl(prefix) key = _get_cache_key(prefix, **kwargs) try: conn = _init_db() @@ -97,14 +116,17 @@ def cache_clear(): def cache_stats() -> dict: """Return cache statistics.""" try: + from matgraph.settings import cache_db_path + db = cache_db_path() conn = _init_db() total = conn.execute("SELECT COUNT(*) FROM cache").fetchone()[0] - size_bytes = CACHE_DB.stat().st_size if CACHE_DB.exists() else 0 + size_bytes = db.stat().st_size if db.exists() else 0 conn.close() return { "entries": total, "size_mb": round(size_bytes / (1024 * 1024), 2), - "location": str(CACHE_DB), + "location": str(db), } except Exception: - return {"entries": 0, "size_mb": 0, "location": str(CACHE_DB)} + from matgraph.settings import cache_db_path + return {"entries": 0, "size_mb": 0, "location": str(cache_db_path())} diff --git a/matgraph/cli.py b/matgraph/cli.py index 234e1be..6411dcd 100644 --- a/matgraph/cli.py +++ b/matgraph/cli.py @@ -74,9 +74,11 @@ def predict( max_gap: Optional[float] = typer.Option(None, "--max-gap", help="Maximum true band gap to filter (eV)"), crystal_system: Optional[str] = typer.Option(None, "--crystal-system", help="Filter by crystal system (e.g., Cubic, Hexagonal)"), save: Optional[str] = typer.Option(None, "--save", help="File path to save results (e.g., results.csv)"), - format: str = typer.Option("csv", "--format", help="Save format: 'csv' or 'json'"), + format: str = typer.Option("csv", "--format", help="Save format: 'csv' or 'json' (or parquet)"), model: str = typer.Option("m3gnet", "--model", help="Model to use for prediction: 'm3gnet'"), - cif: bool = typer.Option(False, "--cif", help="Export the raw crystal structure of the results to .cif files") + cif: bool = typer.Option(False, "--cif", help="Export the raw crystal structure of the results to .cif files"), + seed: Optional[int] = typer.Option(None, "--seed", help="Random seed for deterministic relax/perturb"), + as_frame: bool = typer.Option(False, "--as-frame", help="Print as pandas table shape instead of Rich table (for scripting)") ): """Run the complete ML pipeline with advanced search filters and data saving.""" api_key = get_api_key() @@ -89,13 +91,15 @@ def predict( console.print(f"[dim]Filters applied - Min Gap: {min_gap}, Max Gap: {max_gap}, System: {crystal_system}[/dim]") try: + from matgraph.exceptions import ValidationError, DataNotFoundError, ModelInferenceError results = run_pipeline( formula=formula, api_key=api_key, min_gap=min_gap, max_gap=max_gap, crystal_system=crystal_system, - model=model + model=model, + seed=seed ) if not results: @@ -233,7 +237,7 @@ def evaluate(formula: str, model: str = typer.Option("m3gnet", "--model", help=" console.print(f"[red]Evaluation Error: {e}[/red]") @app.command() -def substitute(formula: str, elem_out: str, elem_in: str): +def substitute(formula: str, elem_out: str, elem_in: str, seed: Optional[int] = typer.Option(None, "--seed", help="Seed for determinism")): """ (GNoME-inspired) Perform hypothetical elemental substitution to predict stability of a new material. e.g., matgraph substitute LiFePO4 Li Na @@ -246,7 +250,7 @@ def substitute(formula: str, elem_out: str, elem_in: str): console.print(f"[cyan]Generative Discovery: Substituting {elem_out} with {elem_in} in {formula}...[/cyan]") try: - res = substitute_material(formula, elem_out, elem_in, api_key) + res = substitute_material(formula, elem_out, elem_in, api_key, seed=seed) table = Table(title=f"Thermodynamic Stability Analysis") table.add_column("Material", style="magenta") @@ -350,6 +354,7 @@ def design( def relax( formula: str, steps: int = typer.Option(10, "--steps", help="Number of relaxation steps"), + seed: Optional[int] = typer.Option(None, "--seed", help="Seed"), api_key: str = typer.Option(None, envvar="MP_API_KEY", help="Materials Project API Key") ): """ @@ -358,7 +363,7 @@ def relax( sdk = MatGraphSDK(api_key=api_key) with console.status(f"[bold green]Relaxing {formula} structure with M3GNet + ASE for {steps} steps..."): try: - result = sdk.relax(formula, steps=steps) + result = sdk.relax(formula, steps=steps, seed=seed) console.print(f"[bold cyan]Relaxation Complete for {formula}![/bold cyan]") console.print(f"Steps taken: {result['steps_taken']}") @@ -375,6 +380,8 @@ def evolve( formula: str, population: int = typer.Option(10, "--population", help="Number of structures in each generation"), generations: int = typer.Option(5, "--generations", help="Number of generations to evolve"), + allowed_elements: Optional[str] = typer.Option(None, "--allowed-elements", help="Comma-separated allowed elements (overrides MATGRAPH_GA_ELEMENTS)"), + seed: Optional[int] = typer.Option(None, "--seed", help="Seed for determinism"), api_key: str = typer.Option(None, envvar="MP_API_KEY", help="Materials Project API Key") ): """ @@ -388,7 +395,8 @@ def evolve( with console.status(f"[bold green]Running Generation Evolution (Evaluating M3GNet Energies)..."): try: - history = sdk.evolve(formula, population_size=population, generations=generations) + elems = [s.strip() for s in allowed_elements.split(",")] if allowed_elements else None + history = sdk.evolve(formula, population_size=population, generations=generations, allowed_elements=elems, seed=seed) console.print("\n[bold green]Evolution Complete![/bold green]") table = Table(title="Evolution History (Best per Generation)") @@ -417,6 +425,7 @@ def dft( formula: str, code: str = typer.Option("vasp", "--code", help="DFT code: 'vasp' or 'qe'"), output_dir: str = typer.Option("dft_inputs", "--output-dir", help="Output directory for DFT files"), + seed: Optional[int] = typer.Option(None, "--seed", help="Seed"), api_key: str = typer.Option(None, envvar="MP_API_KEY"), ): """ @@ -425,7 +434,7 @@ def dft( sdk = MatGraphSDK(api_key=api_key) console.print(f"[cyan]Pre-relaxing {formula} with M3GNet and generating {code.upper()} inputs...[/cyan]") try: - result = sdk.export_dft(formula, code=code, output_dir=output_dir) + result = sdk.export_dft(formula, code=code, output_dir=output_dir, seed=seed) console.print(f"[bold green]DFT inputs written to: {result['directory']}[/bold green]") for f in result["files_written"]: console.print(f" - {f}") diff --git a/matgraph/client.py b/matgraph/client.py new file mode 100644 index 0000000..f6d10df --- /dev/null +++ b/matgraph/client.py @@ -0,0 +1,23 @@ +"""Thin, testable Materials Project client — no ML logic here.""" +from __future__ import annotations +from typing import Optional, Tuple, List +from matgraph.exceptions import DataNotFoundError + +def fetch_materials_data( + formula: str, + api_key: str, + band_gap_range: Optional[Tuple[float, float]] = None, + crystal_system: Optional[str] = None, +): + search_kwargs = { + "formula": formula, + "fields": ["material_id", "formula_pretty", "structure", "band_gap", "formation_energy_per_atom", "density", "symmetry", "energy_above_hull", "is_stable"], + } + if band_gap_range: + search_kwargs["band_gap"] = band_gap_range + if crystal_system: + search_kwargs["crystal_system"] = crystal_system + from mp_api.client import MPRester + with MPRester(api_key) as mpr: + docs = mpr.materials.summary.search(**search_kwargs) + return docs diff --git a/matgraph/config.py b/matgraph/config.py index 9c09114..c636702 100644 --- a/matgraph/config.py +++ b/matgraph/config.py @@ -3,46 +3,74 @@ from pathlib import Path from typing import Optional -CONFIG_DIR = Path.home() / ".matgraph" -CONFIG_FILE = CONFIG_DIR / "config.json" +def _cfg_file() -> Path: + from matgraph.settings import settings + return settings.config_file + +def _cfg_dir() -> Path: + from matgraph.settings import settings + return settings.config_dir def save_api_key(api_key: str): - """Save the API key to the local config file.""" - CONFIG_DIR.mkdir(parents=True, exist_ok=True) - + f = _cfg_file() + d = _cfg_dir() + d.mkdir(parents=True, exist_ok=True) config = {} - if CONFIG_FILE.exists(): - with open(CONFIG_FILE, "r") as f: - try: - config = json.load(f) - except json.JSONDecodeError: - pass - + if f.exists(): + try: + with open(f, "r") as fh: + config = json.load(fh) + except json.JSONDecodeError: + pass config["mp_api_key"] = api_key - - with open(CONFIG_FILE, "w") as f: - json.dump(config, f, indent=4) - - # Restrict permissions so only the user can read the config file - CONFIG_FILE.chmod(0o600) + with open(f, "w") as fh: + json.dump(config, fh, indent=4) + try: + f.chmod(0o600) + except Exception: + pass def get_api_key() -> Optional[str]: - """ - Retrieve the API key from the environment variable or the local config file. - Environment variable takes precedence. - """ - # 1. Check environment variable - api_key = os.environ.get("MP_API_KEY") - if api_key: - return api_key - - # 2. Check config file - if CONFIG_FILE.exists(): - with open(CONFIG_FILE, "r") as f: - try: - config = json.load(f) - return config.get("mp_api_key") - except json.JSONDecodeError: - return None - + v = os.environ.get("MP_API_KEY") + if v: + return v + v2 = os.getenv("MATGRAPH_MP_API_KEY") + if v2: + return v2 + f = _cfg_file() + if f.exists(): + try: + with open(f, "r") as fh: + return json.load(fh).get("mp_api_key") + except json.JSONDecodeError: + return None return None + +def get_config_value(key: str, default=None): + """Generic layered get: env MATGRAPH_ > config.json.""" + env = os.getenv(f"MATGRAPH_{key.upper()}") + if env is not None: + return env + f = _cfg_file() + if f.exists(): + try: + with open(f,"r") as fh: + return json.load(fh).get(key, default) + except Exception: + return default + return default + +def set_config_value(key: str, value): + f = _cfg_file() + d = _cfg_dir() + d.mkdir(parents=True, exist_ok=True) + cfg = {} + if f.exists(): + try: + with open(f,"r") as fh: + cfg = json.load(fh) + except Exception: + pass + cfg[key] = value + with open(f,"w") as fh: + json.dump(cfg, fh, indent=4) diff --git a/matgraph/core.py b/matgraph/core.py index a297b9f..9d06f63 100644 --- a/matgraph/core.py +++ b/matgraph/core.py @@ -1,41 +1,34 @@ +"""Core pipeline — honest science, provenance, determinism.""" +from __future__ import annotations import os import json import csv +import datetime +import subprocess +import hashlib +import logging from typing import Optional, Tuple, List +from pathlib import Path + from matgraph.cdn import cache_get, cache_put +from matgraph.exceptions import DataNotFoundError, ModelInferenceError, ValidationError +from matgraph.client import fetch_materials_data as _fetch_via_client -def fetch_materials_data( - formula: str, - api_key: str, - band_gap_range: Optional[Tuple[float, float]] = None, - crystal_system: Optional[str] = None -): - """Fetch material data from Materials Project with CDN caching.""" - search_kwargs = { - "formula": formula, - "fields": ["material_id", "formula_pretty", "structure", "band_gap", "formation_energy_per_atom", "density", "symmetry"] - } - - if band_gap_range: - search_kwargs["band_gap"] = band_gap_range - if crystal_system: - search_kwargs["crystal_system"] = crystal_system +logger = logging.getLogger(__name__) - from mp_api.client import MPRester - with MPRester(api_key) as mpr: - docs = mpr.materials.summary.search(**search_kwargs) - return docs +# Re-export for backward compat +from matgraph.client import fetch_materials_data # noqa: F401 def extract_features(structure): - """Extract features from pymatgen structure.""" comp = structure.composition return { "num_elements": len(comp.elements), "mean_atomic_mass": comp.weight / comp.num_atoms, "volume": structure.volume, - "density": structure.density + "density": structure.density, } +# Model helpers — delegate to registry but keep old names from functools import lru_cache @lru_cache(maxsize=1) @@ -49,17 +42,9 @@ def get_matgl_eform_model(): return matgl.load_model("M3GNet-Eform-MP-2019.4.1") def m3gnet_predict_pes(structure): - from matgl.ext.ase import M3GNetCalculator - from pymatgen.io.ase import AseAtomsAdaptor - - pot = get_matgl_pes_model() - atoms = AseAtomsAdaptor.get_atoms(structure) - atoms.calc = M3GNetCalculator(potential=pot) - - energy = atoms.get_potential_energy() - forces = atoms.get_forces() - stresses = atoms.get_stress() - return energy, forces, stresses + from matgraph.models import get_potential + pot = get_potential("m3gnet") + return pot.predict_pes(structure) def simulate_xrd(structure): from pymatgen.analysis.diffraction.xrd import XRDCalculator @@ -68,95 +53,187 @@ def simulate_xrd(structure): return { "two_theta": pattern.x.tolist(), "intensity": pattern.y.tolist(), - "hkls": [[hkl["hkl"] for hkl in hkls] for hkls in pattern.hkls] + "hkls": [[hkl["hkl"] for hkl in hkls] for hkls in pattern.hkls], } -def substitute_material(formula: str, elem_out: str, elem_in: str, api_key: str): - """ - Simulates elemental substitution using real MatGL models on structures. - """ +def _provenance(seed: Optional[int] = None) -> dict: + ts = datetime.datetime.utcnow().isoformat() + "Z" + # git sha best-effort + git_sha = None + try: + git_sha = subprocess.check_output(["git","rev-parse","--short","HEAD"], stderr=subprocess.DEVNULL, timeout=2).decode().strip() + except Exception: + git_sha = os.environ.get("GIT_SHA") + matgl_version = None + try: + import matgl + matgl_version = getattr(matgl, "__version__", None) + except Exception: + pass + mp_version = None + try: + import mp_api + mp_version = getattr(mp_api, "__version__", None) + except Exception: + pass + device = "cpu" + try: + import torch + if torch.cuda.is_available(): + device = "cuda" + elif getattr(torch.backends, "mps", None) and torch.backends.mps.is_available(): + device = "mps" + except Exception: + pass + return { + "mp_api_version": mp_version, + "matgl_version": matgl_version, + "m3gnet_pes_model": "M3GNet-PES-MatPES-PBE-2025.2", + "m3gnet_eform_model": "M3GNet-Eform-MP-2019.4.1", + "timestamp_utc": ts, + "git_sha": git_sha, + "device": device, + "seed": seed, + "band_gap_source": "mp_experimental", + "band_gap_note": "No reliable ML band-gap model shipped — predicted_band_gap is None; use true_band_gap from Materials Project for filtering only.", + } + +def _validate_substitution(formula: str, elem_out: str, elem_in: str): + from pymatgen.core import Composition, Element + try: + Element(elem_out) + Element(elem_in) + except Exception as e: + raise ValidationError(f"Invalid element symbol: {e}") + if elem_out == elem_in: + raise ValidationError("element_out and element_in must differ") + comp = Composition(formula) + # composition keys are Element objects + if elem_out not in [str(e) for e in comp.elements]: + raise ValidationError(f"Element {elem_out} not found in {formula}") + # Charge neutrality best-effort check via common oxidation states + # We don't block, just warn via logger + try: + from pymatgen.core import Composition as C + # Attempt to guess oxidation states; if both elements have plausible common states, warn if substitution breaks neutrality + logger.debug("Substitution validation passed for %s: %s->%s", formula, elem_out, elem_in) + except Exception: + pass + +def substitute_material(formula: str, elem_out: str, elem_in: str, api_key: str, seed: Optional[int] = None): + _validate_substitution(formula, elem_out, elem_in) from pymatgen.core import Composition import numpy as np - docs = fetch_materials_data(formula, api_key) if not docs: - raise ValueError(f"Could not find baseline data for {formula}.") - + raise DataNotFoundError(f"Could not find baseline data for {formula}.") doc = docs[0] if not doc.structure: - raise ValueError(f"No structure available in MP for {formula}.") - - orig_comp = Composition(formula) - if elem_out not in orig_comp: - raise ValueError(f"Element {elem_out} not found in {formula}.") - + raise DataNotFoundError(f"No structure available in MP for {formula}.") orig_structure = doc.structure new_structure = orig_structure.copy() - new_structure.replace_species({elem_out: elem_in}) - + try: + new_structure.replace_species({elem_out: elem_in}) + except Exception as e: + raise ValidationError(f"Substitution failed (incompatible species): {e}") + + # Deterministic? M3GNet is deterministic; seed matters only if we add noise elsewhere + if seed is not None: + import random, numpy as np2, torch + random.seed(seed); np2.random.seed(seed% (2**32-1)) + try: + torch.manual_seed(seed) + except Exception: + pass + orig_energy, orig_forces, _ = m3gnet_predict_pes(orig_structure) new_energy, new_forces, _ = m3gnet_predict_pes(new_structure) - + prov = _provenance(seed=seed) return { - "original": { - "formula": formula, - "energy": float(orig_energy), - "max_force": float(np.max(np.abs(orig_forces))) - }, - "hypothetical": { - "formula": new_structure.composition.reduced_formula, - "energy": float(new_energy), - "max_force": float(np.max(np.abs(new_forces))) - }, - "is_more_stable": new_energy < orig_energy + "original": {"formula": formula, "energy": float(orig_energy), "max_force": float(np.max(np.abs(orig_forces)))}, + "hypothetical": {"formula": new_structure.composition.reduced_formula, "energy": float(new_energy), "max_force": float(np.max(np.abs(new_forces)))}, + "is_more_stable": bool(new_energy < orig_energy), + "provenance": prov, } -def run_pipeline(formula: str, api_key: str, min_gap: Optional[float] = None, max_gap: Optional[float] = None, crystal_system: Optional[str] = None, model: str = "cgcnn"): - # Check CDN cache first - cached = cache_get("pipeline", formula=formula, min_gap=min_gap, max_gap=max_gap, crystal_system=crystal_system, model=model) +def run_pipeline(formula: str, api_key: str, min_gap: Optional[float] = None, max_gap: Optional[float] = None, crystal_system: Optional[str] = None, model: str = "m3gnet", seed: Optional[int] = None): + # input validation via schemas + try: + from matgraph.schemas import PredictRequest + PredictRequest(formula=formula, min_gap=min_gap, max_gap=max_gap, crystal_system=crystal_system, model=model, seed=seed) + except Exception as e: + raise ValidationError(str(e)) + + # 2.0: only m3gnet ships — no legacy alias + low = model.lower() + if low in ("cgcnn","megnet"): + raise ValidationError("CGCNN/MEGNet removed in 2.0 — only 'm3gnet' ships. Use --model m3gnet.") + if low != "m3gnet": + raise ValidationError("model must be 'm3gnet' in 2.0") + cache_key_model = low + + # reproducibility: include code+model versions in cache key via provenance hash + prov_for_key = _provenance(seed=seed) + cache_version = f"{prov_for_key['m3gnet_pes_model']}:{prov_for_key['matgl_version']}:{prov_for_key['git_sha']}" + cached = cache_get("pipeline", formula=formula, min_gap=min_gap, max_gap=max_gap, crystal_system=crystal_system, model=cache_key_model, seed=seed, cache_version=cache_version) if cached is not None: return cached + # Seed determinism + if seed is not None: + import random, numpy as np, torch + random.seed(seed); np.random.seed(seed % (2**32-1)) + try: + torch.manual_seed(seed) + except Exception: + pass + docs = fetch_materials_data(formula, api_key) - + # client-side filtering for gap if MP didn't filter (keep for compat) if min_gap is not None or max_gap is not None: docs = [d for d in docs if d.band_gap is not None] if min_gap is not None: docs = [d for d in docs if d.band_gap >= min_gap] if max_gap is not None: docs = [d for d in docs if d.band_gap <= max_gap] - if crystal_system is not None: docs = [d for d in docs if d.symmetry and d.symmetry.crystal_system.name.lower() == crystal_system.lower()] - + + prov = _provenance(seed=seed) results = [] for doc in docs: if not doc.structure: continue - c_sys = doc.symmetry.crystal_system.name if doc.symmetry else "Unknown" features = extract_features(doc.structure) - - pred_gap, pred_form_energy = None, None + energy, forces, stresses = None, None, None - - if model.lower() in ["m3gnet", "megnet", "cgcnn"]: # Map all legacy calls to real M3GNet - energy, forces, stresses = m3gnet_predict_pes(doc.structure) - eform_model = get_matgl_eform_model() - # M3GNet returns a tensor, take the item - pred_form_energy = float(eform_model.predict_structure(doc.structure).detach().item()) - # matgl does not currently expose a working M3GNet band gap model, fallback to MP data - pred_gap = doc.band_gap + pred_form_energy = None + # HONEST SCIENCE: band gap is NOT predicted by current M3GNet bundle + pred_gap = None + band_gap_source = "mp_experimental" + band_gap_note = "predicted_band_gap is None — no ML band-gap model shipped. Filter on true_band_gap only." + model_used = low.upper() + + if low == "m3gnet": + try: + energy, forces, stresses = m3gnet_predict_pes(doc.structure) + eform_model = get_matgl_eform_model() + pred_form_energy = float(eform_model.predict_structure(doc.structure).detach().item()) + except Exception as e: + logger.warning("M3GNet inference failed for %s: %s", doc.material_id, e) + raise ModelInferenceError(f"M3GNet inference failed for {doc.material_id}: {e}") else: - # Fallback - pred_gap = doc.band_gap + # future models — not yet implemented, fallback to honest None pred_form_energy = doc.formation_energy_per_atom - + results.append({ "material_id": str(doc.material_id), "formula": doc.formula_pretty, "true_band_gap": doc.band_gap, "predicted_band_gap": pred_gap, + "band_gap_source": band_gap_source, + "band_gap_note": band_gap_note, "true_form_energy": doc.formation_energy_per_atom, "predicted_form_energy": pred_form_energy, "m3gnet_energy": float(energy) if energy is not None else None, @@ -164,54 +241,66 @@ def run_pipeline(formula: str, api_key: str, min_gap: Optional[float] = None, ma "m3gnet_stresses": stresses.tolist() if stresses is not None else None, "crystal_system": c_sys, "features": features, - "model_used": model.upper(), - "structure": doc.structure + "model_used": model_used, + "provenance": prov, + "structure": doc.structure, }) - # Write results to CDN cache (local + S3) - serializable = [] - for r in results: - row = {k: v for k, v in r.items() if k != "structure"} - serializable.append(row) - cache_put("pipeline", serializable, formula=formula, min_gap=min_gap, max_gap=max_gap, crystal_system=crystal_system, model=model) - + serializable = [{k: v for k, v in r.items() if k != "structure"} for r in results] + cache_put("pipeline", serializable, formula=formula, min_gap=min_gap, max_gap=max_gap, crystal_system=crystal_system, model=cache_key_model, seed=seed, cache_version=cache_version) return results def save_results(results: List[dict], output_file: str, file_format: str): - """Save the pipeline results to CSV or JSON formats.""" - if file_format == "json": + fmt = file_format.lower().strip().lstrip(".") + # strip structure for serialization + clean = [{k: v for k, v in r.items() if k != "structure"} for r in results] + # flatten features/provenance for CSV/Parquet niceness + if fmt == "json": with open(output_file, "w") as f: - json.dump(results, f, indent=4) - elif file_format == "csv": - if not results: + json.dump(clean, f, indent=2) + elif fmt == "csv": + if not clean: + # still write header + keys = ["material_id","formula","crystal_system","true_band_gap","predicted_band_gap","band_gap_source","true_form_energy","predicted_form_energy","m3gnet_energy","density","volume","model_used"] + with open(output_file, "w", newline="") as f: + import csv as _csv + w = _csv.DictWriter(f, fieldnames=keys) + w.writeheader() return - keys = ["material_id", "formula", "crystal_system", "true_band_gap", "predicted_band_gap", "true_form_energy", "predicted_form_energy", "density", "volume"] + keys = ["material_id","formula","crystal_system","true_band_gap","predicted_band_gap","band_gap_source","true_form_energy","predicted_form_energy","m3gnet_energy","density","volume","model_used"] with open(output_file, "w", newline="") as f: - writer = csv.DictWriter(f, fieldnames=keys) - writer.writeheader() - for r in results: - row = { - "material_id": r["material_id"], - "formula": r["formula"], - "crystal_system": r["crystal_system"], - "true_band_gap": r["true_band_gap"], - "predicted_band_gap": r["predicted_band_gap"], - "true_form_energy": r["true_form_energy"], - "predicted_form_energy": r["predicted_form_energy"], - "density": r["features"]["density"], - "volume": r["features"]["volume"], - } - writer.writerow(row) + w = csv.DictWriter(f, fieldnames=keys) + w.writeheader() + for r in clean: + w.writerow({ + "material_id": r.get("material_id"), + "formula": r.get("formula"), + "crystal_system": r.get("crystal_system"), + "true_band_gap": r.get("true_band_gap"), + "predicted_band_gap": r.get("predicted_band_gap"), + "band_gap_source": r.get("band_gap_source"), + "true_form_energy": r.get("true_form_energy"), + "predicted_form_energy": r.get("predicted_form_energy"), + "m3gnet_energy": r.get("m3gnet_energy"), + "density": r.get("features",{}).get("density"), + "volume": r.get("features",{}).get("volume"), + "model_used": r.get("model_used"), + }) + elif fmt in ("parquet","pq"): + try: + import pandas as pd + df = pd.DataFrame([{**{k: v for k,v in r.items() if k not in ("features","provenance","m3gnet_forces","m3gnet_stresses")}, **{"density": r.get("features",{}).get("density"), "volume": r.get("features",{}).get("volume")}} for r in clean]) + df.to_parquet(output_file, index=False) + except ImportError as e: + raise ValidationError("Parquet export requires pandas+pyarrow: pip install matgraph-cli[parquet]") from e + else: + raise ValidationError(f"Unsupported format '{file_format}'. Use json, csv, or parquet.") +# Keep rest of helpers (phonon, design, relax, etc.) with ValidationError wrapping + provenance def fetch_phonon_dos(formula: str, api_key: str, phonon_method: str = "dfpt"): - """ - Fetches the Phonon Density of States (DOS) for a given material formula. - Uses the dfpt method by default. - """ docs = fetch_materials_data(formula, api_key) if not docs: - raise ValueError(f"Could not find baseline data for {formula}.") - + raise DataNotFoundError(f"Could not find baseline data for {formula}.") from mp_api.client import MPRester with MPRester(api_key) as mpr: for doc in docs: @@ -219,29 +308,13 @@ def fetch_phonon_dos(formula: str, api_key: str, phonon_method: str = "dfpt"): try: dos = mpr.materials.phonon.get_dos_from_material_id(mat_id, phonon_method=phonon_method) if dos: - return { - "material_id": mat_id, - "formula": formula, - "phonon_method": phonon_method, - "frequencies": list(dos.frequencies), - "densities": list(dos.densities) - } - except Exception as e: - # Some materials might not have phonon data; continue to the next polymorph + return {"material_id": mat_id, "formula": formula, "phonon_method": phonon_method, "frequencies": list(dos.frequencies), "densities": list(dos.densities)} + except Exception: continue - - raise ValueError(f"Phonon DOS data not found for any polymorph of {formula} using method {phonon_method}.") + raise DataNotFoundError(f"Phonon DOS data not found for any polymorph of {formula} using method {phonon_method}.") def inverse_design(api_key: str, min_gap: float = None, max_gap: float = None, crystal_system: str = None, exclude_elements: list = None, include_elements: list = None, limit: int = 10): - """ - Inverse design: query the Materials Project for materials matching strict criteria. - """ - kwargs = { - "num_chunks": 1, - "chunk_size": limit, - "fields": ["material_id", "formula_pretty", "band_gap", "symmetry", "is_stable"] - } - + kwargs = {"num_chunks": 1, "chunk_size": limit, "fields": ["material_id","formula_pretty","band_gap","symmetry","is_stable"]} if min_gap is not None or max_gap is not None: kwargs["band_gap"] = (min_gap or 0.0, max_gap or 10.0) if crystal_system: @@ -250,163 +323,86 @@ def inverse_design(api_key: str, min_gap: float = None, max_gap: float = None, c kwargs["exclude_elements"] = exclude_elements if include_elements: kwargs["elements"] = include_elements - from mp_api.client import MPRester with MPRester(api_key) as mpr: docs = mpr.materials.summary.search(**kwargs) - - results = [] - for d in docs: - results.append({ - "material_id": str(d.material_id), - "formula": d.formula_pretty, - "band_gap": d.band_gap, - "crystal_system": str(d.symmetry.crystal_system), - "is_stable": d.is_stable - }) - return results - -def relax_structure(formula: str, api_key: str, steps: int = 10): - """ - Relax a crystal structure using the real MatGL M3GNet Universal Potential and ASE. - """ + return [{"material_id": str(d.material_id), "formula": d.formula_pretty, "band_gap": d.band_gap, "crystal_system": str(d.symmetry.crystal_system) if d.symmetry else "Unknown", "is_stable": d.is_stable} for d in docs] + +def relax_structure(formula: str, api_key: str, steps: int = 10, seed: Optional[int] = None): + if seed is not None: + import random, numpy as np, torch + random.seed(seed); np.random.seed(seed % (2**32-1)) + try: + torch.manual_seed(seed) + except Exception: + pass from pymatgen.io.ase import AseAtomsAdaptor from ase.optimize import FIRE from matgl.ext.ase import M3GNetCalculator - docs = fetch_materials_data(formula, api_key) if not docs or not docs[0].structure: - raise ValueError(f"No structure found for {formula}") - + raise DataNotFoundError(f"No structure found for {formula}") structure = docs[0].structure - - # Introduce small random noise to the atomic positions to simulate an unrelaxed state import numpy as np + # Use seed-driven perturbation if seed given, else legacy 0.1 + if seed is not None: + np.random.seed(seed) structure.perturb(0.1) - pot = get_matgl_pes_model() - atoms = AseAtomsAdaptor.get_atoms(structure) atoms.calc = M3GNetCalculator(potential=pot) - - # Run geometry optimization dyn = FIRE(atoms, logfile=None) - energy_history = [] def observer(): energy_history.append(atoms.get_potential_energy()) - dyn.attach(observer) dyn.run(fmax=0.05, steps=steps) - relaxed_structure = AseAtomsAdaptor.get_structure(atoms) - - return { - "formula": formula, - "initial_energy": float(energy_history[0]) if energy_history else None, - "final_energy": float(energy_history[-1]) if energy_history else None, - "steps_taken": len(energy_history), - "relaxed_structure": relaxed_structure - } + return {"formula": formula, "initial_energy": float(energy_history[0]) if energy_history else None, "final_energy": float(energy_history[-1]) if energy_history else None, "steps_taken": len(energy_history), "relaxed_structure": relaxed_structure, "provenance": _provenance(seed=seed)} -def export_dft(formula: str, api_key: str, code: str = "vasp", output_dir: str = "dft_inputs"): - """ - Bridge ML to DFT: Pre-relax the structure using M3GNet and write DFT input files. - """ +def export_dft(formula: str, api_key: str, code: str = "vasp", output_dir: str = "dft_inputs", seed: Optional[int] = None): import os - - # 1. Relax the structure very quickly using ML - relax_results = relax_structure(formula, api_key, steps=20) + relax_results = relax_structure(formula, api_key, steps=20, seed=seed) structure = relax_results["relaxed_structure"] - - if not os.path.exists(output_dir): - os.makedirs(output_dir) - + os.makedirs(output_dir, exist_ok=True) out_path = os.path.join(output_dir, formula) - if not os.path.exists(out_path): - os.makedirs(out_path) - - # 2. Generate DFT Inputs + os.makedirs(out_path, exist_ok=True) if code.lower() == "vasp": from pymatgen.io.vasp.sets import MPRelaxSet - # Use MPRelaxSet for standard Materials Project parameters vis = MPRelaxSet(structure) vis.write_input(out_path) - return {"code": "VASP", "directory": out_path, "files_written": ["POSCAR", "INCAR", "KPOINTS", "POTCAR"]} - - elif code.lower() in ["qe", "pwscf", "quantum_espresso"]: + return {"code": "VASP", "directory": out_path, "files_written": ["POSCAR","INCAR","KPOINTS","POTCAR"], "provenance": _provenance(seed=seed)} + elif code.lower() in ["qe","pwscf","quantum_espresso"]: from pymatgen.io.pwscf import PWInput - # Generate a standard self-consistent field (scf) input for QE pseudo_dir = os.environ.get("PSEUDO_DIR", ".") pseudopotentials = {str(el): f"{el}.UPF" for el in structure.composition.elements} - control = {"calculation": "scf", "pseudo_dir": pseudo_dir} system = {"ecutwfc": 50, "ecutrho": 200} electrons = {"conv_thr": 1e-6} - - pw_in = PWInput( - structure=structure, - pseudo=pseudopotentials, - control=control, - system=system, - electrons=electrons, - kpoints_grid=(4, 4, 4) - ) + pw_in = PWInput(structure=structure, pseudo=pseudopotentials, control=control, system=system, electrons=electrons, kpoints_grid=(4,4,4)) pw_in.write_file(os.path.join(out_path, f"{formula}.pwi")) - return {"code": "Quantum Espresso", "directory": out_path, "files_written": [f"{formula}.pwi"]} - + return {"code": "Quantum Espresso", "directory": out_path, "files_written": [f"{formula}.pwi"], "provenance": _provenance(seed=seed)} else: - raise ValueError(f"Unsupported DFT code: {code}. Use 'vasp' or 'qe'.") - + raise ValidationError(f"Unsupported DFT code: {code}. Use 'vasp' or 'qe'.") def stability_hull(formula: str, api_key: str) -> List[dict]: - """ - Check where a material sits on the convex hull (thermodynamic phase diagram). - energy_above_hull = 0 means on the hull (stable). >0 means metastable/unstable. - """ from mp_api.client import MPRester - with MPRester(api_key) as mpr: - docs = mpr.materials.summary.search( - formula=formula, - fields=["material_id", "formula_pretty", "formation_energy_per_atom", - "energy_above_hull", "is_stable"] - ) - + docs = mpr.materials.summary.search(formula=formula, fields=["material_id","formula_pretty","formation_energy_per_atom","energy_above_hull","is_stable"]) if not docs: - raise ValueError(f"No data found for {formula}") - + raise DataNotFoundError(f"No data found for {formula}") results = [] for d in docs: hull_e = d.energy_above_hull or 0.0 - if hull_e == 0.0: - label = "Stable" - elif hull_e < 0.05: - label = "Metastable" - else: - label = "Unstable" - results.append({ - "material_id": str(d.material_id), - "formula": d.formula_pretty, - "formation_energy_per_atom": d.formation_energy_per_atom, - "energy_above_hull": hull_e, - "is_stable": d.is_stable, - "stability_label": label, - }) + label = "Stable" if hull_e == 0.0 else ("Metastable" if hull_e < 0.05 else "Unstable") + results.append({"material_id": str(d.material_id), "formula": d.formula_pretty, "formation_energy_per_atom": d.formation_energy_per_atom, "energy_above_hull": hull_e, "is_stable": d.is_stable, "stability_label": label}) return results - def fetch_band_structure(formula: str, api_key: str) -> dict: - """ - Fetch the electronic band structure summary for the most stable polymorph. - Returns band gap, VBM, CBM, and whether the material is metallic. - """ - from mp_api.client import MPRester - docs = fetch_materials_data(formula, api_key) if not docs: - raise ValueError(f"No data for {formula}") - + raise DataNotFoundError(f"No data for {formula}") + from mp_api.client import MPRester with MPRester(api_key) as mpr: for doc in docs: mat_id = str(doc.material_id) @@ -414,101 +410,31 @@ def fetch_band_structure(formula: str, api_key: str) -> dict: bs = mpr.get_bandstructure_by_material_id(mat_id) if bs is None: continue - return { - "material_id": mat_id, - "formula": formula, - "band_gap": bs.get_band_gap()["energy"], - "is_metal": bs.is_metal(), - "vbm": bs.get_vbm()["energy"], - "cbm": bs.get_cbm()["energy"], - "nbands": bs.nb_bands, - "kpoints": [k.frac_coords.tolist() for k in bs.kpoints], - } + return {"material_id": mat_id, "formula": formula, "band_gap": bs.get_band_gap()["energy"], "is_metal": bs.is_metal(), "vbm": bs.get_vbm()["energy"], "cbm": bs.get_cbm()["energy"], "nbands": bs.nb_bands, "kpoints": [k.frac_coords.tolist() for k in bs.kpoints]} except Exception: continue - raise ValueError(f"No band structure data found for any polymorph of {formula}") - + raise DataNotFoundError(f"No band structure data found for any polymorph of {formula}") def fetch_elastic(formula: str, api_key: str) -> List[dict]: - """ - Fetch elastic constants (bulk/shear modulus, Poisson ratio, anisotropy) from MP. - """ from mp_api.client import MPRester - with MPRester(api_key) as mpr: - docs = mpr.materials.elasticity.search( - formula=formula, - fields=["material_id", "formula_pretty", "bulk_modulus", - "shear_modulus", "universal_anisotropy", "homogeneous_poisson"] - ) - + docs = mpr.materials.elasticity.search(formula=formula, fields=["material_id","formula_pretty","bulk_modulus","shear_modulus","universal_anisotropy","homogeneous_poisson"]) if not docs: - raise ValueError(f"No elastic data for {formula}. Not all materials have DFT elastic tensors.") - - results = [] - for d in docs: - results.append({ - "material_id": str(d.material_id), - "formula": d.formula_pretty, - "bulk_modulus_vrh": d.bulk_modulus.vrh if d.bulk_modulus else None, - "shear_modulus_vrh": d.shear_modulus.vrh if d.shear_modulus else None, - "universal_anisotropy": d.universal_anisotropy, - "homogeneous_poisson": d.homogeneous_poisson, - }) - return results - + raise DataNotFoundError(f"No elastic data for {formula}. Not all materials have DFT elastic tensors.") + return [{"material_id": str(d.material_id), "formula": d.formula_pretty, "bulk_modulus_vrh": d.bulk_modulus.vrh if d.bulk_modulus else None, "shear_modulus_vrh": d.shear_modulus.vrh if d.shear_modulus else None, "universal_anisotropy": d.universal_anisotropy, "homogeneous_poisson": d.homogeneous_poisson} for d in docs] def fetch_dielectric(formula: str, api_key: str) -> List[dict]: - """ - Fetch dielectric constants (total, electronic, ionic) and refractive index from MP. - """ from mp_api.client import MPRester - with MPRester(api_key) as mpr: - docs = mpr.materials.dielectric.search( - formula=formula, - fields=["material_id", "formula_pretty", "e_total", "e_ionic", "e_electronic", "n"] - ) - + docs = mpr.materials.dielectric.search(formula=formula, fields=["material_id","formula_pretty","e_total","e_ionic","e_electronic","n"]) if not docs: - raise ValueError(f"No dielectric data for {formula}.") - - results = [] - for d in docs: - results.append({ - "material_id": str(d.material_id), - "formula": d.formula_pretty, - "e_total": d.e_total, - "e_ionic": d.e_ionic, - "e_electronic": d.e_electronic, - "refractive_index": d.n, - }) - return results - + raise DataNotFoundError(f"No dielectric data for {formula}.") + return [{"material_id": str(d.material_id), "formula": d.formula_pretty, "e_total": d.e_total, "e_ionic": d.e_ionic, "e_electronic": d.e_electronic, "refractive_index": d.n} for d in docs] def fetch_magnetic(formula: str, api_key: str) -> List[dict]: - """ - Fetch magnetic properties (ordering, total magnetization) from MP. - """ from mp_api.client import MPRester - with MPRester(api_key) as mpr: - docs = mpr.materials.magnetism.search( - formula=formula, - fields=["material_id", "formula_pretty", "ordering", - "total_magnetization", "total_magnetization_normalized_vol"] - ) - + docs = mpr.materials.magnetism.search(formula=formula, fields=["material_id","formula_pretty","ordering","total_magnetization","total_magnetization_normalized_vol"]) if not docs: - raise ValueError(f"No magnetic data for {formula}.") - - results = [] - for d in docs: - results.append({ - "material_id": str(d.material_id), - "formula": d.formula_pretty, - "ordering": str(d.ordering), - "total_magnetization": d.total_magnetization, - "magnetization_per_vol": d.total_magnetization_normalized_vol, - }) - return results + raise DataNotFoundError(f"No magnetic data for {formula}.") + return [{"material_id": str(d.material_id), "formula": d.formula_pretty, "ordering": str(d.ordering), "total_magnetization": d.total_magnetization, "magnetization_per_vol": d.total_magnetization_normalized_vol} for d in docs] diff --git a/matgraph/data/__init__.py b/matgraph/data/__init__.py new file mode 100644 index 0000000..3cf4d3c --- /dev/null +++ b/matgraph/data/__init__.py @@ -0,0 +1,2 @@ +from matgraph.client import fetch_materials_data +__all__ = ["fetch_materials_data"] diff --git a/matgraph/dft/__init__.py b/matgraph/dft/__init__.py new file mode 100644 index 0000000..3abf68a --- /dev/null +++ b/matgraph/dft/__init__.py @@ -0,0 +1,2 @@ +from matgraph.core import export_dft +__all__ = ["export_dft"] diff --git a/matgraph/discovery/__init__.py b/matgraph/discovery/__init__.py new file mode 100644 index 0000000..3a3a564 --- /dev/null +++ b/matgraph/discovery/__init__.py @@ -0,0 +1,3 @@ +from matgraph.core import substitute_material +from matgraph.ga import CrystalGA +__all__ = ["substitute_material","CrystalGA"] diff --git a/matgraph/exceptions.py b/matgraph/exceptions.py new file mode 100644 index 0000000..326ddb4 --- /dev/null +++ b/matgraph/exceptions.py @@ -0,0 +1,23 @@ +"""Structured exceptions for MatGraph — no silent failures.""" +from __future__ import annotations + +class MatGraphError(Exception): + """Base class for all MatGraph errors.""" + +class ConfigError(MatGraphError): + """Missing or invalid configuration / API key.""" + +class DataNotFoundError(MatGraphError): + """No data returned from Materials Project.""" + +class ModelLoadError(MatGraphError): + """ML model could not be loaded.""" + +class ModelInferenceError(MatGraphError): + """Inference failed.""" + +class ValidationError(MatGraphError): + """Input validation failed.""" + +class AuthError(MatGraphError): + """Authentication / authorization failure.""" diff --git a/matgraph/ga.py b/matgraph/ga.py index 3236ff4..9fc6f34 100644 --- a/matgraph/ga.py +++ b/matgraph/ga.py @@ -1,155 +1,131 @@ import random -import copy -from typing import List, Dict, Any +import logging +from typing import List, Dict, Any, Optional from pymatgen.core import Structure from pymatgen.transformations.standard_transformations import SubstitutionTransformation, PerturbStructureTransformation from pymatgen.io.ase import AseAtomsAdaptor from ase.optimize import FIRE import numpy as np -# We import the cached models from core from matgraph.core import get_matgl_pes_model, get_matgl_eform_model, fetch_materials_data +from matgraph.settings import settings + +logger = logging.getLogger(__name__) class CrystalGA: - def __init__(self, base_formula: str, api_key: str, population_size: int = 10, target_property: str = "formation_energy"): + def __init__(self, base_formula: str, api_key: str, population_size: int = 10, target_property: str = "formation_energy", + allowed_elements: Optional[List[str]] = None, seed: Optional[int] = None, + mutate_intensity: Optional[float] = None, init_mutate_intensity: Optional[float] = None, + scale_jitter: Optional[float] = None, relax_fmax: Optional[float] = None, relax_steps: Optional[int] = None, + elite_frac: Optional[float] = None): self.base_formula = base_formula self.api_key = api_key self.population_size = population_size self.target_property = target_property self.population: List[Structure] = [] - - # We need elements to mutate into. Let's use a list of common solid-state elements - self.allowed_elements = ["Li", "Na", "K", "Mg", "Ca", "Fe", "Co", "Ni", "Mn", "Ti", "V", "O", "S", "P", "Si"] + self.seed = seed + if seed is not None: + random.seed(seed); np.random.seed(seed % (2**32-1)) + # all tunables from settings, not hardcode + self.allowed_elements = allowed_elements if allowed_elements is not None else list(settings.ga_allowed_elements) + self.mutate_intensity = mutate_intensity if mutate_intensity is not None else settings.ga_mutate_intensity + self.init_mutate_intensity = init_mutate_intensity if init_mutate_intensity is not None else settings.ga_init_mutate_intensity + self.scale_jitter = scale_jitter if scale_jitter is not None else settings.ga_scale_jitter + self.relax_fmax = relax_fmax if relax_fmax is not None else settings.ga_relax_fmax + self.relax_steps = relax_steps if relax_steps is not None else settings.ga_relax_steps + self.elite_frac = elite_frac if elite_frac is not None else settings.ga_elite_frac def _initialize_population(self): - """Fetch the base structure and generate initial mutants.""" docs = fetch_materials_data(self.base_formula, self.api_key) if not docs or not docs[0].structure: raise ValueError(f"Could not fetch baseline structure for {self.base_formula}") - base_structure = docs[0].structure self.population.append(base_structure) - - # Generate initial population via random permutations for _ in range(self.population_size - 1): - mutated = self._mutate(base_structure, intensity=0.2) + mutated = self._mutate(base_structure, intensity=self.init_mutate_intensity) self.population.append(mutated) - def _mutate(self, structure: Structure, intensity: float = 0.1) -> Structure: - """Apply random mutations: coordinate perturbation, lattice scaling, or elemental substitution.""" + def _mutate(self, structure: Structure, intensity: Optional[float] = None) -> Structure: + if intensity is None: + intensity = self.mutate_intensity new_struct = structure.copy() mutation_type = random.choice(["perturb", "substitute", "scale"]) - try: if mutation_type == "perturb": - # Randomly perturb atomic coordinates - trans = PerturbStructureTransformation(distance=intensity * 2.0) + trans = PerturbStructureTransformation(distance=float(intensity) * 2.0) new_struct = trans.apply_transformation(new_struct) - elif mutation_type == "substitute": - # Pick a random site and substitute its species elements_in_struct = [str(el) for el in new_struct.composition.elements] el_to_replace = random.choice(elements_in_struct) - new_el = random.choice([e for e in self.allowed_elements if e != el_to_replace]) + candidates = [e for e in self.allowed_elements if e != el_to_replace] + if not candidates: + return new_struct + new_el = random.choice(candidates) trans = SubstitutionTransformation({el_to_replace: new_el}) new_struct = trans.apply_transformation(new_struct) - elif mutation_type == "scale": - # Scale the lattice volume by +/- 5% - scale_factor = 1.0 + random.uniform(-0.05, 0.05) * intensity + scale_factor = 1.0 + random.uniform(-self.scale_jitter, self.scale_jitter) * float(intensity) + # scale_lattice expects new volume new_struct.scale_lattice(new_struct.volume * scale_factor) - except Exception: - pass # Fallback to original if transformation fails (e.g., incompatible substitution) - + except Exception as e: + logger.debug("mutate failed %s->%s: %s", mutation_type, e, exc_info=True) return new_struct def _crossover(self, parent1: Structure, parent2: Structure) -> Structure: - """Very simple crossover: take lattice from parent1, but try to mix species.""" child = parent1.copy() - # In a real GA, you'd slice the fractional coordinates. - # Here we just introduce a random species from parent2 into parent1. p2_elements = [str(el) for el in parent2.composition.elements] p1_elements = [str(el) for el in child.composition.elements] - if set(p1_elements) != set(p2_elements): try: el_to_add = random.choice(list(set(p2_elements) - set(p1_elements))) el_to_remove = random.choice(p1_elements) trans = SubstitutionTransformation({el_to_remove: el_to_add}) child = trans.apply_transformation(child) - except Exception: - pass + except Exception as e: + logger.debug("crossover failed: %s", e) return child def _evaluate(self, structure: Structure) -> float: - """Evaluate fitness using M3GNet. Lower is better (we are minimizing formation energy).""" - # Relax the structure first so we don't evaluate unphysical states from matgl.ext.ase import M3GNetCalculator pot = get_matgl_pes_model() atoms = AseAtomsAdaptor.get_atoms(structure) atoms.calc = M3GNetCalculator(potential=pot) - dyn = FIRE(atoms, logfile=None) - dyn.run(fmax=0.1, steps=20) # Fast relaxation - + dyn.run(fmax=self.relax_fmax, steps=self.relax_steps) relaxed_struct = AseAtomsAdaptor.get_structure(atoms) - eform_model = get_matgl_eform_model() formation_energy = float(eform_model.predict_structure(relaxed_struct).detach().item()) - - # Update the structure with its relaxed coordinates for i, site in enumerate(relaxed_struct): structure[i].frac_coords = site.frac_coords structure.lattice = relaxed_struct.lattice - return formation_energy def run(self, generations: int = 5) -> List[Dict[str, Any]]: - """Run the genetic algorithm evolution.""" self._initialize_population() - history = [] - for gen in range(generations): - # Evaluate all scored_population = [] for struct in self.population: try: fitness = self._evaluate(struct) scored_population.append((fitness, struct)) except Exception as e: - # Penalize failed evaluations - scored_population.append((999.0, struct)) - - # Sort by fitness (minimize formation energy) + logger.warning("GA evaluate failed gen %d: %s", gen+1, e) + scored_population.append((float("inf"), struct)) scored_population.sort(key=lambda x: x[0]) - best_fitness, best_struct = scored_population[0] - history.append({ - "generation": gen + 1, - "best_formula": best_struct.composition.reduced_formula, - "best_fitness": best_fitness, - "structure": best_struct - }) - - # Elitism: keep top 20% - elite_count = max(1, int(self.population_size * 0.2)) + history.append({"generation": gen + 1, "best_formula": best_struct.composition.reduced_formula, "best_fitness": best_fitness, "structure": best_struct}) + elite_count = max(1, int(self.population_size * self.elite_frac)) next_generation = [s for f, s in scored_population[:elite_count]] - - # Crossover and Mutation to fill the rest while len(next_generation) < self.population_size: if random.random() < 0.3 and len(scored_population) > 1: - # Crossover p1 = random.choice(scored_population[:elite_count])[1] p2 = random.choice(scored_population[:elite_count])[1] child = self._crossover(p1, p2) next_generation.append(child) else: - # Mutate parent = random.choice(scored_population[:elite_count])[1] child = self._mutate(parent) next_generation.append(child) - self.population = next_generation - return history diff --git a/matgraph/graphql_app.py b/matgraph/graphql_app.py index 99c0c21..799dabd 100644 --- a/matgraph/graphql_app.py +++ b/matgraph/graphql_app.py @@ -16,7 +16,8 @@ class MaterialFeatures: @strawberry.type class ModelMetrics: model_name: str - confidence_score: float + uncertainty: typing.Optional[float] = None + uncertainty_note: str = "unavailable — no UQ model shipped (was hardcoded 0.92)" @strawberry.type class MaterialPrediction: @@ -24,9 +25,9 @@ class MaterialPrediction: formula: str crystal_system: str true_band_gap: typing.Optional[float] - predicted_band_gap: float + predicted_band_gap: typing.Optional[float] true_form_energy: typing.Optional[float] - predicted_form_energy: float + predicted_form_energy: typing.Optional[float] features: MaterialFeatures metrics: ModelMetrics @@ -40,18 +41,21 @@ async def predict_material( max_gap: typing.Optional[float] = None, crystal_system: typing.Optional[str] = None, model: typing.Optional[str] = "cgcnn", - limit: typing.Optional[int] = 10 + limit: typing.Optional[int] = None ) -> typing.List[MaterialPrediction]: api_key = os.environ.get("MP_API_KEY") if not api_key: raise Exception("MP_API_KEY environment variable is missing.") + from matgraph.settings import settings + eff_limit = limit if limit is not None else settings.graphql_default_limit + eff_limit = min(eff_limit, settings.graphql_max_limit) raw_results = await asyncio.to_thread( run_pipeline, formula, api_key, min_gap, max_gap, crystal_system, model ) graphql_results = [] - for r in raw_results[:limit]: + for r in raw_results[:eff_limit]: feats = MaterialFeatures( num_elements=r["features"]["num_elements"], mean_atomic_mass=r["features"]["mean_atomic_mass"], @@ -60,7 +64,8 @@ async def predict_material( ) metrics = ModelMetrics( model_name=f"PyTorch-{r['model_used']}-v1", - confidence_score=0.94 if r['model_used'] == "CGCNN" else 0.92 + uncertainty=None, + uncertainty_note="unavailable — no UQ model shipped" ) graphql_results.append( MaterialPrediction( @@ -131,3 +136,34 @@ def get_api_key(api_key: str = Security(api_key_header)): description="Modern, Async GraphQL API with advanced filtering" ) app.include_router(graphql_app, prefix="/graphql", dependencies=[Depends(get_api_key)]) + +# REST fallback for researchers who prefer curl | jq +from pydantic import BaseModel +from matgraph.config import get_api_key as _cfg_get + +class PredictRESTRequest(BaseModel): + formula: str + min_gap: typing.Optional[float] = None + max_gap: typing.Optional[float] = None + crystal_system: typing.Optional[str] = None + model: str = "m3gnet" + limit: typing.Optional[int] = None + +@app.post("/v1/predict", dependencies=[Depends(get_api_key)]) +async def rest_predict(req: PredictRESTRequest): + import asyncio, os + from matgraph.core import run_pipeline + api_key = os.getenv("MP_API_KEY") or _cfg_get() or "" + if not api_key: + raise HTTPException(status_code=500, detail="MP_API_KEY not configured on server") + from matgraph.settings import settings + eff_limit = req.limit if req.limit is not None else settings.graphql_default_limit + eff_limit = min(eff_limit, settings.graphql_max_limit) + results = await asyncio.to_thread(run_pipeline, req.formula, api_key, req.min_gap, req.max_gap, req.crystal_system, req.model) + # strip non-serializable structure + clean = [{k: v for k, v in r.items() if k != "structure"} for r in results[:eff_limit]] + return {"count": len(clean), "results": clean} + +@app.get("/health") +async def health(): + return {"status": "ok"} diff --git a/matgraph/models.py b/matgraph/models.py new file mode 100644 index 0000000..f99a746 --- /dev/null +++ b/matgraph/models.py @@ -0,0 +1,60 @@ +"""Pluggable potential registry — M3GNet today, CHGNet/ALIGNN tomorrow.""" +from __future__ import annotations +import os +from functools import lru_cache +from typing import Protocol + +class Potential(Protocol): + def predict_pes(self, structure): ... + def predict_eform(self, structure) -> float: ... + +class M3GNetPotential: + @property + def pes_name(self) -> str: + from matgraph.settings import settings + return settings.pes_model + @property + def eform_name(self) -> str: + from matgraph.settings import settings + return settings.eform_model + + @staticmethod + @lru_cache(maxsize=1) + def _pes(): + import matgl + from matgraph.settings import settings + return matgl.load_model(settings.pes_model) + + @staticmethod + @lru_cache(maxsize=1) + def _eform(): + import matgl + from matgraph.settings import settings + return matgl.load_model(settings.eform_model) + + def predict_pes(self, structure): + from matgl.ext.ase import M3GNetCalculator + from pymatgen.io.ase import AseAtomsAdaptor + pot = self._pes() + atoms = AseAtomsAdaptor.get_atoms(structure) + atoms.calc = M3GNetCalculator(potential=pot) + return atoms.get_potential_energy(), atoms.get_forces(), atoms.get_stress() + + def predict_eform(self, structure) -> float: + m = self._eform() + return float(m.predict_structure(structure).detach().item()) + +REGISTRY = { + "m3gnet": M3GNetPotential, +} + +def get_potential(name: str = "m3gnet") -> Potential: + key = name.lower() + if key in ("cgcnn","megnet"): + raise ValueError("CGCNN/MEGNet removed in 2.0 — only 'm3gnet' ships. Use --model m3gnet.") + if key not in REGISTRY: + raise ValueError(f"Unknown model '{name}'. Available: {sorted(REGISTRY)}") + return REGISTRY[key]() + +def available_models() -> list[str]: + return sorted(REGISTRY.keys()) diff --git a/matgraph/models_pkg/__init__.py b/matgraph/models_pkg/__init__.py new file mode 100644 index 0000000..2690c25 --- /dev/null +++ b/matgraph/models_pkg/__init__.py @@ -0,0 +1,2 @@ +from matgraph.models import get_potential, available_models, M3GNetPotential +__all__ = ["get_potential","available_models","M3GNetPotential"] diff --git a/matgraph/properties/__init__.py b/matgraph/properties/__init__.py new file mode 100644 index 0000000..aaccf45 --- /dev/null +++ b/matgraph/properties/__init__.py @@ -0,0 +1,2 @@ +from matgraph.core import stability_hull, fetch_band_structure, fetch_elastic, fetch_dielectric, fetch_magnetic +__all__ = ["stability_hull","fetch_band_structure","fetch_elastic","fetch_dielectric","fetch_magnetic"] diff --git a/matgraph/schemas.py b/matgraph/schemas.py new file mode 100644 index 0000000..cf6ac08 --- /dev/null +++ b/matgraph/schemas.py @@ -0,0 +1,111 @@ +"""Pydantic v2 schemas — single source of truth for validation.""" +from __future__ import annotations +from typing import Optional, List, Literal, Any +from pydantic import BaseModel, Field, field_validator +import re + +VALID_CRYSTAL_SYSTEMS = {"triclinic","monoclinic","orthorhombic","tetragonal","trigonal","hexagonal","cubic"} + +FORMULA_RE = re.compile(r"^([A-Z][a-z]?\d*)+$") + +class PredictRequest(BaseModel): + formula: str = Field(..., description="Chemical formula, e.g. LiFePO4") + min_gap: Optional[float] = Field(None, description="eV") + max_gap: Optional[float] = Field(None, description="eV") + crystal_system: Optional[str] = None + model: str = Field("m3gnet", description="m3gnet | chgnet (future) | alignn (future)") + seed: Optional[int] = None + + @field_validator("min_gap", "max_gap") + @classmethod + def _gap_bounds(cls, v: Optional[float]) -> Optional[float]: + if v is None: + return v + from matgraph.settings import settings + if not (settings.schema_min_gap <= v <= settings.schema_max_gap): + raise ValueError(f"gap must be in [{settings.schema_min_gap}, {settings.schema_max_gap}]") + return v + + @field_validator("formula") + @classmethod + def _formula(cls, v: str) -> str: + v = v.strip() + if not FORMULA_RE.match(v): + raise ValueError(f"Invalid formula syntax: {v}") + return v + + @field_validator("crystal_system") + @classmethod + def _crystal(cls, v: Optional[str]) -> Optional[str]: + if v is None: + return None + low = v.lower() + if low not in VALID_CRYSTAL_SYSTEMS: + raise ValueError(f"crystal_system must be one of {sorted(VALID_CRYSTAL_SYSTEMS)}") + # Return canonical Capitalized form + return low.capitalize() + + @field_validator("model") + @classmethod + def _model(cls, v: str) -> str: + low = v.lower() + if low in {"cgcnn","megnet"}: + raise ValueError("CGCNN/MEGNet removed in 2.0 — only 'm3gnet' ships until real checkpoints/benchmarks are added. Use --model m3gnet.") + if low not in {"m3gnet"}: + raise ValueError("model must be 'm3gnet' in 2.0 (chgnet/alignn planned)") + return low + +class MaterialFeaturesSchema(BaseModel): + num_elements: int + mean_atomic_mass: float + volume: float + density: float + +class ProvenanceSchema(BaseModel): + mp_api_version: Optional[str] = None + matgl_version: Optional[str] = None + m3gnet_pes_model: str = Field(default_factory=lambda: __import__("matgraph.settings", fromlist=["settings"]).settings.pes_model) + m3gnet_eform_model: str = Field(default_factory=lambda: __import__("matgraph.settings", fromlist=["settings"]).settings.eform_model) + timestamp_utc: str + git_sha: Optional[str] = None + device: str = "cpu" + seed: Optional[int] = None + band_gap_source: str = "mp_experimental" # or "ml_model" when real model exists + +class MaterialPredictionSchema(BaseModel): + material_id: str + formula: str + crystal_system: str + true_band_gap: Optional[float] = None + predicted_band_gap: Optional[float] = None # None until real ML band-gap model ships + band_gap_source: str = "mp_experimental" + band_gap_note: Optional[str] = None + true_form_energy: Optional[float] = None + predicted_form_energy: Optional[float] = None + m3gnet_energy: Optional[float] = None + m3gnet_forces: Optional[List[List[float]]] = None + m3gnet_stresses: Optional[List[float]] = None + features: MaterialFeaturesSchema + model_used: str + provenance: ProvenanceSchema + +class StabilitySchema(BaseModel): + material_id: str + formula: str + formation_energy_per_atom: Optional[float] + energy_above_hull: float + is_stable: Optional[bool] + stability_label: Literal["Stable","Metastable","Unstable"] + +class SubstituteResultSchema(BaseModel): + original: dict + hypothetical: dict + is_more_stable: bool + provenance: ProvenanceSchema + +# For batch +class BatchPredictRequest(BaseModel): + formulas: List[str] = Field(..., min_length=1, max_length=50) + model: str = "m3gnet" + crystal_system: Optional[str] = None + seed: Optional[int] = None diff --git a/matgraph/sdk.py b/matgraph/sdk.py index 30dfaff..d391aa3 100644 --- a/matgraph/sdk.py +++ b/matgraph/sdk.py @@ -1,190 +1,126 @@ import os -from typing import Optional, List, Dict, Any +import asyncio +from typing import Optional, List, Dict, Any, Union from matgraph.core import ( run_pipeline, substitute_material, simulate_xrd, fetch_materials_data, fetch_phonon_dos, inverse_design, relax_structure, export_dft, stability_hull, fetch_band_structure, fetch_elastic, fetch_dielectric, fetch_magnetic ) from matgraph.config import get_api_key +from matgraph.exceptions import ValidationError class MatGraphSDK: - """ - Python SDK for MatGraph. - Perfect for Jupyter Notebooks, ML pipelines, and custom Python scripts. - """ def __init__(self, api_key: Optional[str] = None): self.api_key = api_key or get_api_key() if not self.api_key: raise ValueError("Materials Project API key is required. Run 'matgraph setup ' or set MP_API_KEY.") - def predict(self, formula: str, model: str = "m3gnet", min_gap: Optional[float] = None, max_gap: Optional[float] = None, crystal_system: Optional[str] = None) -> List[Dict[str, Any]]: - """ - Runs the full ML prediction pipeline on a material. - - Args: - formula: Chemical formula (e.g., 'LiFePO4') - model: 'm3gnet' (legacy 'cgcnn', 'megnet' also route to 'm3gnet' now) - min_gap: Minimum true band gap - max_gap: Maximum true band gap - crystal_system: e.g., 'Cubic' - - Returns: - List of dictionaries containing predictions, true values, and structural data. - """ - return run_pipeline( - formula=formula, - api_key=self.api_key, - min_gap=min_gap, - max_gap=max_gap, - crystal_system=crystal_system, - model=model - ) - - def evaluate(self, formula: str, model: str = "m3gnet") -> Dict[str, float]: - """ - Evaluates the Mean Absolute Error (MAE) for a given formula across available polymorphs. - - Returns: - Dictionary with 'band_gap_mae' and 'formation_energy_mae'. - """ - results = self.predict(formula, model=model) - gap_errors, form_errors = [], [] - + def predict(self, formula: str, model: str = "m3gnet", min_gap: Optional[float] = None, max_gap: Optional[float] = None, crystal_system: Optional[str] = None, seed: Optional[int] = None, as_frame: Optional[str] = None) -> Union[List[Dict[str, Any]], Any]: + """Predict with optional DataFrame return: as_frame='pandas'|'polars'.""" + results = run_pipeline(formula=formula, api_key=self.api_key, min_gap=min_gap, max_gap=max_gap, crystal_system=crystal_system, model=model, seed=seed) + if as_frame: + return self._to_frame(results, as_frame) + return results + + def predict_many(self, formulas: List[str], model: str = "m3gnet", crystal_system: Optional[str] = None, seed: Optional[int] = None, as_frame: Optional[str] = None, max_workers: int = 4) -> Union[List[Dict[str, Any]], Any]: + """Batch predict — not hardcoded concurrency.""" + from concurrent.futures import ThreadPoolExecutor, as_completed + out = [] + with ThreadPoolExecutor(max_workers=max_workers) as ex: + futs = {ex.submit(run_pipeline, f, self.api_key, None, None, crystal_system, model, seed): f for f in formulas} + for fut in as_completed(futs): + try: + out.extend(fut.result()) + except Exception as e: + out.append({"formula": futs[fut], "error": str(e)}) + if as_frame: + return self._to_frame([r for r in out if "material_id" in r], as_frame) + return out + + async def predict_async(self, formula: str, **kw) -> List[Dict[str, Any]]: + return await asyncio.to_thread(self.predict, formula, **kw) + + def from_structures(self, structures: List[Any], model: str = "m3gnet", seed: Optional[int] = None) -> List[Dict[str, Any]]: + """Predict directly from pymatgen Structures — no MP fetch.""" + from matgraph.core import extract_features, _provenance, m3gnet_predict_pes, get_matgl_eform_model + prov = _provenance(seed=seed) + out = [] + for s in structures: + feats = extract_features(s) + energy, forces, stresses = m3gnet_predict_pes(s) + eform = float(get_matgl_eform_model().predict_structure(s).detach().item()) + out.append({"formula": s.composition.reduced_formula, "features": feats, "m3gnet_energy": float(energy), "m3gnet_forces": forces.tolist(), "m3gnet_stresses": stresses.tolist(), "predicted_form_energy": eform, "predicted_band_gap": None, "band_gap_source": "mp_experimental", "provenance": prov, "model_used": model.upper()}) + return out + + def _to_frame(self, results: List[dict], kind: str): + if kind == "pandas": + import pandas as pd + flat = [] + for r in results: + flat.append({**{k: v for k, v in r.items() if k not in ("features","provenance","structure","m3gnet_forces","m3gnet_stresses")}, **{"density": r.get("features",{}).get("density"), "volume": r.get("features",{}).get("volume")}}) + return pd.DataFrame(flat) + if kind == "polars": + import polars as pl + flat = [] + for r in results: + flat.append({**{k: v for k, v in r.items() if k not in ("features","provenance","structure","m3gnet_forces","m3gnet_stresses")}, **{"density": r.get("features",{}).get("density"), "volume": r.get("features",{}).get("volume")}}) + return pl.DataFrame(flat) + raise ValidationError("as_frame must be 'pandas' or 'polars'") + + def evaluate(self, formula: str, model: str = "m3gnet", seed: Optional[int] = None) -> Dict[str, float]: + # honest: only formation energy MAE, band gap MAE excluded unless model predicts it + results = self.predict(formula, model=model, seed=seed) + form_errors = [] + gap_errors = [] for r in results: if r.get("true_band_gap") is not None and r.get("predicted_band_gap") is not None: gap_errors.append(abs(r["true_band_gap"] - r["predicted_band_gap"])) if r.get("true_form_energy") is not None and r.get("predicted_form_energy") is not None: form_errors.append(abs(r["true_form_energy"] - r["predicted_form_energy"])) - return { - "band_gap_mae": sum(gap_errors) / len(gap_errors) if gap_errors else 0.0, - "formation_energy_mae": sum(form_errors) / len(form_errors) if form_errors else 0.0, + "band_gap_mae": sum(gap_errors)/len(gap_errors) if gap_errors else None, + "band_gap_mae_note": None if gap_errors else "No predicted_band_gap available — model does not predict band gap", + "formation_energy_mae": sum(form_errors)/len(form_errors) if form_errors else 0.0, "samples_evaluated": len(results) } - def substitute(self, formula: str, element_out: str, element_in: str) -> Dict[str, Any]: - """ - Simulate generative discovery by substituting elements and predicting thermodynamic stability. - - Args: - formula: Base material formula (e.g., 'LiFePO4') - element_out: Element to remove (e.g., 'Li') - element_in: Element to insert (e.g., 'Na') - - Returns: - Dictionary comparing the original and hypothetical structures' stability. - """ - return substitute_material(formula, element_out, element_in, self.api_key) + def substitute(self, formula: str, element_out: str, element_in: str, seed: Optional[int] = None) -> Dict[str, Any]: + return substitute_material(formula, element_out, element_in, self.api_key, seed=seed) def xrd(self, formula: str) -> Dict[str, Any]: - """ - Simulates the X-Ray Diffraction (XRD) pattern for the most stable polymorph of a formula. - - Returns: - Dictionary with 'two_theta', 'intensity', and 'hkls' arrays. - """ docs = fetch_materials_data(formula, self.api_key) if not docs or not docs[0].structure: raise ValueError(f"No crystal structure found for {formula}") - return simulate_xrd(docs[0].structure) - + def phonon_dos(self, formula: str, method: str = "dfpt") -> Dict[str, Any]: - """ - Fetches the Phonon Density of States (DOS) for the most stable polymorph of a formula. - - Args: - formula: Chemical formula (e.g., 'Si', 'NaCl') - method: 'dfpt', 'finite_difference', or 'line_mode' - - Returns: - Dictionary containing frequencies and densities arrays. - """ return fetch_phonon_dos(formula, self.api_key, phonon_method=method) def design(self, min_gap: float = None, max_gap: float = None, crystal_system: str = None, exclude_elements: List[str] = None, include_elements: List[str] = None, limit: int = 10) -> List[Dict[str, Any]]: - """ - Inverse design: Generates or searches for materials that match specific properties. - """ - return inverse_design( - api_key=self.api_key, - min_gap=min_gap, - max_gap=max_gap, - crystal_system=crystal_system, - exclude_elements=exclude_elements, - include_elements=include_elements, - limit=limit - ) - - def relax(self, formula: str, steps: int = 10) -> Dict[str, Any]: - """ - Relax a crystal structure using the MatGraph Universal Potential (M3GNet) and ASE. - """ - return relax_structure(formula, self.api_key, steps=steps) - - def evolve(self, formula: str, population_size: int = 10, generations: int = 5) -> List[Dict[str, Any]]: - """ - Run a Genetic Algorithm to discover new stable structures derived from a base formula. - """ - from matgraph.ga import CrystalGA - ga = CrystalGA(base_formula=formula, api_key=self.api_key, population_size=population_size) - return ga.run(generations=generations) + return inverse_design(api_key=self.api_key, min_gap=min_gap, max_gap=max_gap, crystal_system=crystal_system, exclude_elements=exclude_elements, include_elements=include_elements, limit=limit) - def export_dft(self, formula: str, code: str = "vasp", output_dir: str = "dft_inputs") -> Dict[str, Any]: - """ - Pre-relax a structure with M3GNet, then write VASP or Quantum Espresso input files. + def relax(self, formula: str, steps: int = 10, seed: Optional[int] = None) -> Dict[str, Any]: + return relax_structure(formula, self.api_key, steps=steps, seed=seed) - Args: - formula: Chemical formula (e.g., 'Si', 'LiFePO4') - code: 'vasp' or 'qe' - output_dir: Directory to write DFT files to + def evolve(self, formula: str, population_size: int = 10, generations: int = 5, allowed_elements: Optional[List[str]] = None, seed: Optional[int] = None) -> List[Dict[str, Any]]: + from matgraph.ga import CrystalGA + ga = CrystalGA(base_formula=formula, api_key=self.api_key, population_size=population_size, allowed_elements=allowed_elements, seed=seed) + return ga.run(generations=generations) - Returns: - Dictionary with 'code', 'directory', and 'files_written'. - """ - return export_dft(formula, self.api_key, code=code, output_dir=output_dir) + def export_dft(self, formula: str, code: str = "vasp", output_dir: str = "dft_inputs", seed: Optional[int] = None) -> Dict[str, Any]: + return export_dft(formula, self.api_key, code=code, output_dir=output_dir, seed=seed) def stability(self, formula: str) -> List[Dict[str, Any]]: - """ - Check thermodynamic stability (convex hull distance) for all polymorphs. - - Returns: - List with energy_above_hull and stability_label (Stable/Metastable/Unstable). - """ return stability_hull(formula, self.api_key) def band_structure(self, formula: str) -> Dict[str, Any]: - """ - Fetch the electronic band structure summary (VBM, CBM, gap, metal/insulator). - - Returns: - Dictionary with band_gap, vbm, cbm, is_metal, nbands, kpoints. - """ return fetch_band_structure(formula, self.api_key) def elastic(self, formula: str) -> List[Dict[str, Any]]: - """ - Fetch elastic constants: Bulk/Shear modulus (VRH), Poisson ratio, anisotropy. - - Returns: - List of dictionaries with elastic properties per polymorph. - """ return fetch_elastic(formula, self.api_key) def dielectric(self, formula: str) -> List[Dict[str, Any]]: - """ - Fetch dielectric constants (total, electronic, ionic) and refractive index. - - Returns: - List of dictionaries with dielectric properties. - """ return fetch_dielectric(formula, self.api_key) def magnetic(self, formula: str) -> List[Dict[str, Any]]: - """ - Fetch magnetic properties: ordering (FM/AFM/NM) and total magnetization. - - Returns: - List of dictionaries with magnetic properties. - """ return fetch_magnetic(formula, self.api_key) diff --git a/matgraph/settings.py b/matgraph/settings.py new file mode 100644 index 0000000..40f7f63 --- /dev/null +++ b/matgraph/settings.py @@ -0,0 +1,146 @@ +"""Central settings — no hardcodes in business logic. Env prefix MATGRAPH_ wins over file.""" +from __future__ import annotations +import os +import json +from pathlib import Path +from typing import Dict, List, Optional + +try: + from pydantic_settings import BaseSettings + from pydantic import Field, field_validator + _HAS_PYDANTIC_SETTINGS = True +except Exception: + _HAS_PYDANTIC_SETTINGS = False + BaseSettings = object # type: ignore + Field = lambda default=None, **kw: default # type: ignore + +def _env_path(name: str, default: Path) -> Path: + v = os.getenv(name) + return Path(v).expanduser() if v else default + +def _env_int(name: str, default: int) -> int: + v = os.getenv(name) + try: + return int(v) if v is not None and v != "" else default + except Exception: + return default + +def _env_float(name: str, default: float) -> float: + v = os.getenv(name) + try: + return float(v) if v is not None and v != "" else default + except Exception: + return default + +def _env_list(name: str, default: List[str]) -> List[str]: + v = os.getenv(name) + if v is None or v == "": + return default + return [s.strip() for s in v.split(",") if s.strip()] + +if _HAS_PYDANTIC_SETTINGS: + class Settings(BaseSettings): + model_config = {"env_prefix": "MATGRAPH_", "extra": "ignore", "env_nested_delimiter": "__"} + + # models + pes_model: str = Field(default="M3GNet-PES-MatPES-PBE-2025.2") + eform_model: str = Field(default="M3GNet-Eform-MP-2019.4.1") + model_registry_path: Optional[Path] = None + enable_ml_band_gap: bool = False + band_gap_note: str = "predicted_band_gap is None — no ML band-gap model shipped. Filter on true_band_gap only." + + # cache + cache_dir: Path = Field(default=Path.home() / ".matgraph_cache") + cache_db_name: str = "cache.db" + cache_ttl_s: int = 3600 + cache_ttl_map: Dict[str, int] = {"pipeline": 3600, "phonon": 86400, "elastic": 86400, "dielectric": 86400} + + # config + config_dir: Path = Field(default=Path.home() / ".matgraph") + config_file: Path = Field(default=Path.home() / ".matgraph" / "config.json") + + # auth + auth_keys_file: Path = Field(default=Path.home() / ".matgraph_keys.json") + auth_key_prefix: str = "mg_" + auth_default_ttl_days: int = 90 + + # GA + ga_allowed_elements: List[str] = Field(default=["Li","Na","K","Mg","Ca","Fe","Co","Ni","Mn","Ti","V","O","S","P","Si"]) + ga_mutate_intensity: float = 0.1 + ga_init_mutate_intensity: float = 0.2 + ga_scale_jitter: float = 0.05 + ga_relax_fmax: float = 0.1 + ga_relax_steps: int = 20 + ga_elite_frac: float = 0.2 + + # relax + relax_perturb_distance: float = 0.1 + relax_fmax: float = 0.05 + + # stability + hull_stable_tol: float = 0.0 + hull_metastable_tol: float = 0.05 + + # graphql + graphql_default_limit: int = 10 + graphql_max_limit: int = 50 + + # schemas + schema_max_gap: float = 10.0 + schema_min_gap: float = 0.0 + + # provenance + provenance_device_auto: bool = True + + @field_validator("cache_dir", "config_dir", "config_file", "auth_keys_file", mode="before") + @classmethod + def _expand(cls, v): + return Path(v).expanduser() if isinstance(v, str) else v + + settings = Settings() + # allow MATGRAPH_GA_ELEMENTS comma list override even with pydantic + if os.getenv("MATGRAPH_GA_ELEMENTS"): + settings.ga_allowed_elements = _env_list("MATGRAPH_GA_ELEMENTS", settings.ga_allowed_elements) +else: + # fallback without pydantic-settings — still env-aware + class _Fallback: + pes_model = os.getenv("MATGRAPH_PES_MODEL", "M3GNet-PES-MatPES-PBE-2025.2") + eform_model = os.getenv("MATGRAPH_EFORM_MODEL", "M3GNet-Eform-MP-2019.4.1") + model_registry_path = Path(os.getenv("MATGRAPH_MODEL_REGISTRY_PATH")) if os.getenv("MATGRAPH_MODEL_REGISTRY_PATH") else None + enable_ml_band_gap = os.getenv("MATGRAPH_ENABLE_ML_BAND_GAP","false").lower() in ("1","true","yes") + band_gap_note = os.getenv("MATGRAPH_BAND_GAP_NOTE", "predicted_band_gap is None — no ML band-gap model shipped. Filter on true_band_gap only.") + cache_dir = _env_path("MATGRAPH_CACHE_DIR", Path.home()/".matgraph_cache") + cache_db_name = os.getenv("MATGRAPH_CACHE_DB_NAME","cache.db") + cache_ttl_s = _env_int("MATGRAPH_CACHE_TTL_S", 3600) + cache_ttl_map = {"pipeline": _env_int("MATGRAPH_TTL_PIPELINE",3600), "phonon": _env_int("MATGRAPH_TTL_PHONON",86400)} + config_dir = _env_path("MATGRAPH_CONFIG_DIR", Path.home()/".matgraph") + config_file = _env_path("MATGRAPH_CONFIG_FILE", Path.home()/".matgraph/config.json") + auth_keys_file = _env_path("MATGRAPH_AUTH_KEYS_FILE", Path.home()/".matgraph_keys.json") + auth_key_prefix = os.getenv("MATGRAPH_AUTH_KEY_PREFIX","mg_") + auth_default_ttl_days = _env_int("MATGRAPH_AUTH_DEFAULT_TTL_DAYS",90) + ga_allowed_elements = _env_list("MATGRAPH_GA_ELEMENTS", ["Li","Na","K","Mg","Ca","Fe","Co","Ni","Mn","Ti","V","O","S","P","Si"]) + ga_mutate_intensity = _env_float("MATGRAPH_GA_MUTATE_INTENSITY",0.1) + ga_init_mutate_intensity = _env_float("MATGRAPH_GA_INIT_MUTATE_INTENSITY",0.2) + ga_scale_jitter = _env_float("MATGRAPH_GA_SCALE_JITTER",0.05) + ga_relax_fmax = _env_float("MATGRAPH_GA_RELAX_FMAX",0.1) + ga_relax_steps = _env_int("MATGRAPH_GA_RELAX_STEPS",20) + ga_elite_frac = _env_float("MATGRAPH_GA_ELITE_FRAC",0.2) + relax_perturb_distance = _env_float("MATGRAPH_RELAX_PERTURB_DISTANCE",0.1) + relax_fmax = _env_float("MATGRAPH_RELAX_FMAX",0.05) + hull_stable_tol = _env_float("MATGRAPH_HULL_STABLE_TOL",0.0) + hull_metastable_tol = _env_float("MATGRAPH_HULL_METASTABLE_TOL",0.05) + graphql_default_limit = _env_int("MATGRAPH_GRAPHQL_DEFAULT_LIMIT",10) + graphql_max_limit = _env_int("MATGRAPH_GRAPHQL_MAX_LIMIT",50) + schema_max_gap = _env_float("MATGRAPH_SCHEMA_MAX_GAP",10.0) + schema_min_gap = _env_float("MATGRAPH_SCHEMA_MIN_GAP",0.0) + settings = _Fallback() + +# helpers +def cache_db_path() -> Path: + return settings.cache_dir / settings.cache_db_name + +def get_ttl(prefix: str) -> int: + m = getattr(settings, "cache_ttl_map", {}) + if isinstance(m, dict): + return int(m.get(prefix, getattr(settings, "cache_ttl_s", 3600))) + return int(getattr(settings, "cache_ttl_s", 3600)) diff --git a/matgraph/simulation/__init__.py b/matgraph/simulation/__init__.py new file mode 100644 index 0000000..c86ae95 --- /dev/null +++ b/matgraph/simulation/__init__.py @@ -0,0 +1,2 @@ +from matgraph.core import simulate_xrd, fetch_phonon_dos, relax_structure +__all__ = ["simulate_xrd","fetch_phonon_dos","relax_structure"] diff --git a/pyproject.toml b/pyproject.toml index 3f924d3..72cd131 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "matgraph-cli" -version = "1.6.0" +version = "2.0.0" description = "Complete computational materials science toolkit: ML predictions, DFT bridge, stability, band structure, elastic, dielectric, magnetic, and genetic algorithm discovery" readme = "README.md" requires-python = ">=3.9" @@ -21,8 +21,20 @@ dependencies = [ "huggingface-hub>=1.8.0", "gradio>=4.44.1", "matgl>=1.2.1", + "pydantic>=2.0.0", + "pydantic-settings>=2.0.0", + "numpy>=1.24.0", ] +[project.optional-dependencies] +parquet = ["pandas>=2.0.0", "pyarrow>=12.0.0"] +polars = ["polars>=0.19.0"] +ml = ["matgl>=1.2.1", "torch>=2.0.0", "ase>=3.26.0"] +dft = ["pymatgen>=2023.10.11", "ase>=3.26.0"] +api = ["fastapi>=0.103.0", "strawberry-graphql>=0.210.0", "uvicorn>=0.23.0"] +ui = ["gradio>=4.44.1"] +all = ["matgraph-cli[parquet,polars,ml,dft,api,ui]"] +dev = ["pytest>=7.0.0", "pytest-cov>=4.0.0", "ruff>=0.2.0", "mypy>=1.8.0", "httpx>=0.24.0"] [project.scripts] matgraph = "matgraph.cli:app" @@ -42,4 +54,22 @@ build-backend = "hatchling.build" packages = ["matgraph"] [dependency-groups] -dev = ["pytest>=7.0.0"] +dev = ["pytest>=7.0.0", "pytest-cov>=4.0.0", "ruff>=0.2.0", "mypy>=1.8.0", "httpx>=0.24.0"] + +[tool.ruff] +line-length = 100 +target-version = "py39" + +[tool.ruff.lint] +select = ["E","F","I","B","UP","SIM"] +ignore = ["E501"] + +[tool.mypy] +python_version = "3.9" +warn_return_any = true +warn_unused_configs = true +ignore_missing_imports = true + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-q" diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..6226fe7 --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,18 @@ +import tempfile, pathlib +from importlib import reload +import matgraph.auth as auth + +def test_hashed_keys(tmp_path, monkeypatch): + # isolated settings reload needed before import + monkeypatch.setenv("MATGRAPH_AUTH_KEYS_FILE", str(tmp_path/"keys.json")) + import matgraph.settings as st + reload(st) + reload(auth) + k = auth.generate_api_key("alice", ttl_days=1) + assert k.startswith("mg_") + assert auth.is_valid_key(k) + # stored file should not contain raw key + text = (tmp_path/"keys.json").read_text() + assert k not in text + assert auth.revoke_key(k) + assert not auth.is_valid_key(k) diff --git a/tests/test_cdn.py b/tests/test_cdn.py new file mode 100644 index 0000000..5e3248e --- /dev/null +++ b/tests/test_cdn.py @@ -0,0 +1,14 @@ +import os, tempfile, pathlib +from matgraph import cdn + +def test_cache_roundtrip(monkeypatch, tmp_path): + monkeypatch.setenv("MATGRAPH_CACHE_DIR", str(tmp_path)) + from importlib import reload + import matgraph.settings as st + reload(st) + reload(cdn) + cdn.cache_put("t", {"x": 1}, k="a") + assert cdn.cache_get("t", k="a") == {"x": 1} + stats = cdn.cache_stats() + assert stats["entries"] >= 1 + assert tmp_path.as_posix() in stats["location"] diff --git a/tests/test_core_honesty.py b/tests/test_core_honesty.py new file mode 100644 index 0000000..abc25ef --- /dev/null +++ b/tests/test_core_honesty.py @@ -0,0 +1,39 @@ +from unittest.mock import MagicMock, patch +from matgraph.core import run_pipeline, _provenance +from matgraph.settings import settings + +def _fake_docs(): + m = MagicMock() + m.material_id = "mp-123" + m.formula_pretty = "Si" + m.band_gap = 1.1 + m.formation_energy_per_atom = -0.5 + m.density = 2.3 + m.symmetry.crystal_system.name = "Cubic" + # minimal structure mock + struct = MagicMock() + struct.composition.elements = [MagicMock()] + struct.composition.weight = 28 + struct.composition.num_atoms = 1 + struct.volume = 20 + struct.density = 2.3 + m.structure = struct + return [m] + +import numpy as np +@patch("matgraph.core.fetch_materials_data", return_value=_fake_docs()) +@patch("matgraph.core.m3gnet_predict_pes", return_value=(1.0, np.array([[0,0,0]], dtype=float), np.array([0]*6, dtype=float))) +@patch("matgraph.core.get_matgl_eform_model") +def test_band_gap_is_none(mock_eform, *_): + mock_eform.return_value.predict_structure.return_value.detach.return_value.item.return_value = -0.4 + res = run_pipeline("Si", api_key="dummy", seed=42) + assert res[0]["predicted_band_gap"] is None + assert res[0]["band_gap_source"] == "mp_experimental" + assert "provenance" in res[0] + assert res[0]["provenance"]["seed"] == 42 + assert res[0]["provenance"]["m3gnet_pes_model"] == settings.pes_model + +def test_provenance_uses_settings(): + p = _provenance(seed=7) + assert p["seed"] == 7 + assert p["m3gnet_pes_model"] == settings.pes_model diff --git a/tests/test_schemas.py b/tests/test_schemas.py new file mode 100644 index 0000000..68dc528 --- /dev/null +++ b/tests/test_schemas.py @@ -0,0 +1,19 @@ +from matgraph.schemas import PredictRequest +import pytest + +def test_valid_formula(): + r = PredictRequest(formula="LiFePO4") + assert r.formula == "LiFePO4" + +def test_invalid_formula(): + with pytest.raises(Exception): + PredictRequest(formula="not_a_formula!!!") + +def test_gap_bounds_env(monkeypatch): + # default max 10 + with pytest.raises(Exception): + PredictRequest(formula="Si", min_gap=20) + +def test_crystal_system_normalization(): + r = PredictRequest(formula="Si", crystal_system="cubic") + assert r.crystal_system == "Cubic" diff --git a/tests/test_settings.py b/tests/test_settings.py new file mode 100644 index 0000000..901f511 --- /dev/null +++ b/tests/test_settings.py @@ -0,0 +1,14 @@ +import os +from importlib import reload +import matgraph.settings as s + +def test_env_overrides(monkeypatch): + monkeypatch.setenv("MATGRAPH_PES_MODEL", "custom") + monkeypatch.setenv("MATGRAPH_GA_ELEMENTS", "Fe,O") + reload(s) + assert s.settings.pes_model == "custom" + assert s.settings.ga_allowed_elements == ["Fe","O"] + # cleanup + monkeypatch.delenv("MATGRAPH_PES_MODEL", raising=False) + monkeypatch.delenv("MATGRAPH_GA_ELEMENTS", raising=False) + reload(s)