Babu09's picture
Deploy MEIDNet Matter 0.10.1 (4b7339b)
b2e3563 verified
Raw History Blame Contribute Delete
25.1 kB
"""
The ``meidnet`` command.
meidnet init write a starter meidnet.yaml next to your data
meidnet check CONFIG is my data usable? → check_report.html
meidnet train CONFIG learn the latent space → model.pt, training_report.html
meidnet generate CONFIG design candidates → CIFs, generation_report.html
meidnet studio CONFIG interactive design workbench in your browser
meidnet demo 5-minute demo with the published perovskite model
meidnet screen DIR stability screening of CIFs with MACE (optional extra)
meidnet info MODEL describe a checkpoint
meidnet families list material families
meidnet schema JSON Schema of the config (for editors and the Studio)
meidnet download-data fetch the Perov-5 dataset used in the paper
"""
from __future__ import annotations
import argparse
import os
import sys
import textwrap
from meidnet import __version__
PEROV5_URL = "https://raw.githubusercontent.com/txie-93/cdvae/main/data/perov_5/{split}.csv"
CKPT_URL = "https://github.com/ABnano/MEIDNet/raw/main/checkpoints/dual_autoencoder_clip_earlyfusion_propertyaware_2k.pth"
CKPT_NAME = "dual_autoencoder_clip_earlyfusion_propertyaware_2k.pth"
def _cfg(path):
from meidnet.config import load_config
from meidnet.pipeline import load_plugins
cfg = load_config(path)
load_plugins(cfg.plugins, cfg.resolve)
return cfg
def published_checkpoint() -> str:
"""Path to the published Perov-5 checkpoint; downloads it to ~/.meidnet if needed."""
for cand in (os.path.join("checkpoints", CKPT_NAME),
os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "checkpoints", CKPT_NAME)):
if os.path.exists(cand):
return cand
cache = os.path.join(os.path.expanduser("~"), ".meidnet", CKPT_NAME)
if not os.path.exists(cache):
import urllib.request
os.makedirs(os.path.dirname(cache), exist_ok=True)
print(f"downloading the published MEIDNet model (2.8 MB) to {cache} ...")
urllib.request.urlretrieve(CKPT_URL, cache)
return cache
# ───────────────────────── init ─────────────────────────
TEMPLATE_CUSTOM = """\
# MEIDNet configuration — edit the values, keep the structure.
# Every setting is explained at https://babu09-meidnet.hf.space/docs/reference/config.html
name: {name}
description: ""
output_dir: runs/{{name}}
data:
table: {table} # CSV/Excel with one row per material
id_column: {id_column}
cif_column: {cif_column} # column holding CIF text ... or use structures_dir
# structures_dir: structures/ # folder of <id>.cif files (set cif_column: null)
properties:
{properties}
val_fraction: 0.1
max_sites: 20
align_to_prototype: true # re-order atoms to the family prototype (needed for generation)
model:
latent_dim: 128
training:
epochs: 200
batch_size: 16
device: auto
generation:
family: {family}
variant: {variant}
objectives:
{objectives}
targets:
{targets}
per_target: 4
population: 48
rounds: 20
steps: 800
exclude_elements: [] # e.g. [Pb, Cd]
# only_elements: {{B: [Ti, Zr, Hf]}}
# overrides: {{tolerance_factor: {{min: 0.8, max: 1.0}}}}
"""
TEMPLATE_GENERATION = """\
# MEIDNet configuration for family-free generation (the symmetry decoder), set up like the configuration behind
# MEIDNet Matter's live generator (checkpoints/configs/mp20_wyck.yaml). Needs the symmetry side-car files
# (wyckoff_<split>.json.gz) next to the tables; `meidnet train` builds them when they are missing.
# Every setting is explained at https://babu09-meidnet.hf.space/docs/reference/config.html
name: {name}
description: "family-free generation with the symmetry decoder"
output_dir: runs/{{name}}
data:
table: {table} # the intake's train.csv (one row per material, CIF text in cif_column)
val_table: {val_table}
id_column: {id_column}
cif_column: {cif_column}
properties:
{properties}
max_sites: 20
site_order: roles
align_to_prototype: false # no family: atoms are ordered by role, not aligned to a prototype
model:
latent_dim: 128
decoder_coordinate_input: zeros
periodic_encoder: true
element_features: cgcnn
decoder_geometry: wyckoff # the symmetry decoder: a space group and the symmetry-distinct sites
training:
epochs: 200
batch_size: 16
learning_rate: 0.001
seed: 0
device: auto
save_every: 50
loss_weights: {{structure_reconstruction: 1.0, structure_property: 1.0, symmetry: 1.0}} # labels read from the structure need all three
# then: python -m meidnet_eval.generate_to_target --ckpt runs/{name}/out/model.pt --intake <intake> --gap <column> --targets 1 2 3 --tag mytag
"""
TEMPLATE_PEROV5 = """\
# MEIDNet configuration reproducing the published Perov-5 experiment.
# Data: `meidnet download-data` fetches the CDVAE Perov-5 split into data/perov5/.
name: perov5
description: Cubic ABX3 perovskites, band gap + formation enthalpy (Perov-5)
output_dir: runs/{name}
data:
table: data/perov5/train.csv
id_column: material_id
cif_column: cif
properties:
- {column: heat_all, label: Formation enthalpy, unit: eV/atom, normalize: false}
- {column: dir_gap, label: Direct band gap, unit: eV, normalize: false}
val_table: data/perov5/val.csv
max_sites: 20
align_to_prototype: false # the published model used atoms in file order
training:
epochs: 200
batch_size: 16
contrastive_warmup_epochs: 1200
device: auto
generation:
family: perovskite_abx3
variant: halide
objectives:
- {property: dir_gap, loss: l2, weight: 10000, select_weight: 1.0}
- {property: heat_all, loss: l1, weight: 6000, select_weight: 0.4}
targets:
- {dir_gap: 1.5, heat_all: -0.10}
- {dir_gap: 2.5, heat_all: -0.10}
- {dir_gap: 3.5, heat_all: -0.10}
per_target: 4
population: 48
rounds: 20
steps: 800
"""
def cmd_init(a):
path = a.output
if os.path.exists(path) and not a.force:
sys.exit(f"{path} already exists (use --force to overwrite)")
if a.template == "perov5":
text = TEMPLATE_PEROV5
elif a.template == "generation":
props = a.properties or ["band_gap"]
pl = "\n".join(f" - {{column: {p}, unit: \"\", normalize: true}}" for p in props)
table = a.table
val_table = os.path.join(os.path.dirname(table), "val.csv") if os.path.basename(table) == "train.csv" else "null"
text = TEMPLATE_GENERATION.format(name=a.name, table=table, val_table=val_table, id_column=a.id_column,
cif_column=a.cif_column, properties=pl)
else:
props = a.properties or ["band_gap"]
pl = "\n".join(f" - {{column: {p}, unit: \"\"}}" for p in props)
ol = "\n".join(f" - {{property: {p}, loss: l2, weight: 10000}}" for p in props)
tl = (" - {" + ", ".join(f"{p}: 0.0" for p in props) + "}"
" # ← set your target values (check_report.html shows the training range)")
text = TEMPLATE_CUSTOM.format(name=a.name, table=a.table, id_column=a.id_column, cif_column=a.cif_column,
properties=pl, family=a.family, variant=a.variant or "null",
objectives=ol, targets=tl)
text, note = _match_structure_source(text, a.table, a.cif_column, a.structures_dir)
with open(path, "w", encoding="utf-8") as f:
f.write(text)
if note:
print(note)
print(f"wrote {path}\nnext: meidnet check {path}")
def _match_structure_source(text: str, table: str, cif_column: str, structures_dir: str | None):
"""The written config names where the structures are. A table without the CIF column, next to a folder of
<id>.cif files, is the common case for a user's own data: the config then points at that folder instead of at a
column that is not there (a first external test ended with 0 usable rows for this reason)."""
sdir = structures_dir
if sdir is None and os.path.exists(table):
try:
from meidnet.data import read_table
cols = list(read_table(table).columns)
except Exception:
cols = None
if cols is not None and cif_column not in cols:
for cand in ("structures", "cifs", "cif", "poscars"):
d = os.path.join(os.path.dirname(os.path.abspath(table)), cand)
if os.path.isdir(d):
sdir = cand
break
if not sdir:
return text, None
text = text.replace(f" cif_column: {cif_column} # column holding CIF text ... or use structures_dir\n"
" # structures_dir: structures/ # folder of <id>.cif files (set cif_column: null)\n",
f" cif_column: null # the table has no CIF column; the structures are files\n"
f" structures_dir: {sdir}/ # folder of <id>.cif files, one per row of the table\n")
return text, (f"the table has no `{cif_column}` column; {sdir}/ next to it holds the structures, so the config points there "
f"(cif_column: null, structures_dir: {sdir}/)")
def cmd_check(a):
from meidnet.pipeline import check
cfg = _cfg(a.config)
info = check(cfg)
rep = info["report"]
print(f"\n{rep.kept} of {rep.rows} materials usable.")
for reason, n in rep.skipped.most_common():
print(f" skipped {n:>6} {reason} e.g. {', '.join(rep.examples.get(reason, []))}")
if rep.reduced_to_primitive:
print(f" {rep.reduced_to_primitive} structures larger than max_sites were read as their primitive cell (the same crystal)")
print(f"report: {info['report_path']}")
if rep.kept == 0:
# a process exit code is what a script or a newcomer reads first: nothing usable is a failure, not a report
sys.exit("no material is usable: fix the problems above (and in the report) before training")
def cmd_train(a):
from meidnet.pipeline import train
train(_cfg(a.config), epochs=a.epochs)
def cmd_generate(a):
from meidnet.pipeline import generate
cfg = _cfg(a.config)
if a.model:
cfg.model_path = os.path.abspath(a.model)
if a.quick:
g = cfg.generation
g.rounds, g.steps, g.population = min(g.rounds, 3), min(g.steps, 150), min(g.population, 24)
generate(cfg)
_annotate_generation(cfg)
def _annotate_generation(cfg):
"""After `meidnet generate`: say which candidates already exist in the training data (with the data's own value),
and print the model's validation error next to the targets, so a predicted hit is read as a model estimate."""
import csv as _csv
import pandas as pd
out = os.path.join(cfg.out_dir, "generation", "candidates.csv")
if not os.path.exists(out):
return
cands = pd.read_csv(out)
if cands.empty:
return
try:
from pymatgen.core import Composition, Structure
table = pd.read_csv(cfg.resolve(cfg.data.table)) if str(cfg.data.table).endswith(".csv") else pd.read_excel(cfg.resolve(cfg.data.table))
if "formula" in table.columns:
keys = table["formula"].astype(str).map(lambda f: Composition(f).reduced_formula)
else:
keys = table[cfg.data.cif_column].astype(str).map(lambda c: Structure.from_str(c, fmt="cif").composition.reduced_formula)
props = [p.column for p in cfg.data.properties]
known = {}
for k, (_, row) in zip(keys, table.iterrows()):
known.setdefault(k, {p: row.get(p) for p in props})
ck = cands["formula"].astype(str).map(lambda f: Composition(f).reduced_formula)
cands["known_in_training"] = ck.map(lambda k: k in known)
for p in props:
cands[f"training_{p}"] = ck.map(lambda k: known.get(k, {}).get(p))
cands.to_csv(out, index=False, quoting=_csv.QUOTE_MINIMAL)
n_known = int(cands["known_in_training"].sum())
print(f"\n{n_known} of {len(cands)} candidates are already in the training table (training_<property> columns give the "
f"data's own values; the pred_ columns are the model's estimates); {len(cands) - n_known} are new compositions.")
for _, r in cands[cands["known_in_training"]].iterrows():
vals = ", ".join(f"{p}: model {r.get('pred_' + p):.2f} vs data {r.get('training_' + p):.2f}" for p in props
if pd.notna(r.get(f"training_{p}")) and pd.notna(r.get(f"pred_{p}")))
print(f" known: {r['formula']:<14} {vals}")
except Exception as e: # the annotation must never undo a finished run
print(f"(known/new annotation skipped: {e})")
try:
from meidnet.checkpoint import load_checkpoint
lm = load_checkpoint(cfg.model_path if getattr(cfg, "model_path", None) else cfg.checkpoint_path, device="cpu")
val = (lm.meta.get("history") or {}).get("val") or []
if val and val[-1].get("mae"):
print("validation error of this model (held-out, from training): " + ", ".join(f"{k} {v:.3g}" for k, v in val[-1]["mae"].items())
+ ". A candidate's predicted value is an estimate with that error.")
gap = next((p.column for p in cfg.data.properties if "gap" in p.column.lower()), None)
intake = os.path.dirname(os.path.abspath(cfg.resolve(cfg.data.table)))
model = cfg.model_path if getattr(cfg, "model_path", None) else cfg.checkpoint_path
check_out = os.path.join(cfg.out_dir, "check")
if gap:
print("the independent check (label read from each cell, qualified judge, two potentials, re-judge, novelty):\n"
f" python -m meidnet_eval.check_candidates {out} --ckpt {model} --intake {intake} --gap {gap} --out {check_out}")
else:
# MEGNet reads band gaps only: for any other property the check runs without an independent judge
print("the check (label read from each cell, two potentials, re-encode after relaxation, novelty), without an "
"independent judge: the MEGNet judge reads band gaps only and none of "
+ ", ".join(p.column for p in cfg.data.properties) + " is one:\n"
f" python -m meidnet_eval.check_candidates {out} --ckpt {model} --intake {intake} "
f"--gap {cfg.data.properties[0].column} --judge none --out {check_out}")
if not os.path.exists(os.path.join(intake, "test.csv")):
print(f" ({intake}/test.csv does not exist: the check qualifies its judge and measures novelty on an intake with "
f"train/val/test splits; make one with\n python -m meidnet_eval.intake <table.csv> <intake_dir> "
f"--id {cfg.data.id_column} --cif {cfg.data.cif_column or 'cif'} --props {' '.join(p.column for p in cfg.data.properties)})")
except Exception as e:
print(f"(validation error not read: {e})")
def cmd_info(a):
from meidnet.checkpoint import describe, load_checkpoint
print(describe(load_checkpoint(a.model)))
def cmd_families(a):
from meidnet.family import list_families, load_family
if a.name:
fam = load_family(a.name, variant=a.variant, default_variant=True)
print(fam.describe())
print("\nvariants:", ", ".join(f"{k} ({v})" for k, v in fam.variants.items()) or "none")
print("\nconstraints:")
from meidnet.constraints import explain
for c in fam.constraints:
t, text = explain(c["name"], c)
print(f" - {t}: {text}")
print(f"\nfile: {fam.source}")
else:
for n in list_families():
print(n)
def cmd_schema(a):
from meidnet.config import json_schema
text = json_schema()
if a.output:
with open(a.output, "w", encoding="utf-8") as f:
f.write(text)
print(f"wrote {a.output}")
else:
print(text)
def cmd_download_data(a):
import urllib.request
os.makedirs(a.dir, exist_ok=True)
for split in ("train", "val", "test"):
dst = os.path.join(a.dir, f"{split}.csv")
if os.path.exists(dst) and not a.force:
print(f"{dst} exists")
continue
print(f"downloading {split}.csv ...")
urllib.request.urlretrieve(PEROV5_URL.format(split=split), dst)
print(f"Perov-5 (CDVAE split, Xie et al. 2022) is in {a.dir}")
def cmd_demo(a):
"""Generate a few perovskites with the published model and open the report."""
from meidnet.config import config_from_dict
from meidnet.pipeline import generate
ckpt = published_checkpoint()
raw = {
"name": "demo", "output_dir": a.out, "model_path": os.path.abspath(ckpt),
"description": "MEIDNet demo: published Perov-5 model",
"generation": {
"family": "perovskite_abx3", "variant": a.family,
"objectives": [{"property": "dir_gap", "loss": "l2", "weight": 10000, "select_weight": 1.0},
{"property": "heat_all", "loss": "l1", "weight": 6000, "select_weight": 0.4}],
"targets": [{"dir_gap": a.band_gap, "heat_all": a.enthalpy}],
"per_target": a.n, "population": 24, "rounds": a.rounds, "steps": a.steps,
"min_cosine_sep": 0.98,
},
}
cfg = config_from_dict(raw, base_dir=os.getcwd())
cfg.training.device = a.device
res = generate(cfg)
rep = os.path.join(cfg.out_dir, "generation_report.html")
print(f"\n{len(res.saved)} candidates. Report: {rep}")
if not a.no_open:
import webbrowser
webbrowser.open("file://" + os.path.abspath(rep))
def cmd_studio(a):
from meidnet.studio.server import export_static, serve
cfg = _cfg(a.config) if a.config else None
if a.export_static:
export_static(cfg, a.export_static, model_path=a.model)
return
serve(cfg, model_path=a.model, port=a.port, open_browser=not a.no_open, host=a.host, public=a.public,
run_root=a.run_root, docs_dir=a.docs_dir)
def cmd_space(a):
from meidnet.checkpoint import load_checkpoint
from meidnet.designspace import enumerate_space, space_to_csv
from meidnet.pipeline import family_for
cfg = _cfg(a.config)
lm = load_checkpoint(a.model or cfg.checkpoint_path)
fam = family_for(cfg, need_variant=True)
space = enumerate_space(fam, lm, max_compositions=a.max)
out = a.output or os.path.join(cfg.out_dir, "design_space.csv")
space_to_csv(space, out)
n_ok = sum(all(r["ok"].values()) for r in space["rows"])
print(f"{len(space['rows'])} compositions ({n_ok} pass every rule) -> {out}")
if a.chemiscope:
import json
from meidnet.studio.chemiscope import space_dataset
with open(a.chemiscope, "w", encoding="utf-8") as f:
json.dump(space_dataset(fam, space, lm), f)
print(f"chemiscope dataset -> {a.chemiscope} (open it at https://chemiscope.org)")
def cmd_screen(a):
from meidnet.screen import screen
screen(a.dir, train_csv=a.train_csv, threshold=a.threshold, device=a.device, model_path=a.model)
def main(argv=None):
for stream in (sys.stdout, sys.stderr): # print ✓ / ⚠ safely on legacy Windows consoles
if hasattr(stream, "reconfigure"):
try:
stream.reconfigure(errors="replace")
except Exception:
pass
p = argparse.ArgumentParser(prog="meidnet", description=textwrap.dedent(__doc__),
formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("--version", action="version", version=f"meidnet {__version__}")
sub = p.add_subparsers(dest="cmd", required=True)
s = sub.add_parser("init", help="write a starter meidnet.yaml")
s.add_argument("-o", "--output", default="meidnet.yaml")
s.add_argument("--template", choices=["custom", "perov5", "generation"], default="custom",
help="custom: a family model (screening and family generation); generation: the symmetry decoder for family-free generation; perov5: the paper's demo")
s.add_argument("--name", default="my_materials")
s.add_argument("--table", default="materials.csv")
s.add_argument("--id-column", default="material_id")
s.add_argument("--cif-column", default="cif")
s.add_argument("--structures-dir", default=None, help="folder of <id>.cif files, when the table holds no CIF text")
s.add_argument("--properties", nargs="*", help="property column names")
s.add_argument("--family", default="perovskite_abx3")
s.add_argument("--variant", default=None)
s.add_argument("--force", action="store_true")
s.set_defaults(fn=cmd_init)
for name, fn, h in (("check", cmd_check, "read the data and write check_report.html"),
("train", cmd_train, "train the model and write training_report.html"),
("generate", cmd_generate, "design candidates and write generation_report.html")):
s = sub.add_parser(name, help=h)
s.add_argument("config", nargs="?", default="meidnet.yaml")
if name == "train":
s.add_argument("--epochs", type=int, default=None, help="override training.epochs")
if name == "generate":
s.add_argument("--model", default=None, help="checkpoint to use instead of the config's model_path")
s.add_argument("--quick", action="store_true", help="few rounds/steps - a smoke test")
s.set_defaults(fn=fn)
s = sub.add_parser("studio", help="interactive design workbench (local web page)")
s.add_argument("config", nargs="?", default=None, help="meidnet.yaml (optional: defaults to the published model)")
s.add_argument("--model", default=None)
s.add_argument("--port", type=int, default=8765)
s.add_argument("--no-open", action="store_true")
s.add_argument("--export-static", metavar="FILE.html", default=None,
help="write a self-contained copy of the Studio (no server; search disabled)")
s.add_argument("--host", default="127.0.0.1", help="0.0.0.0 to accept connections from other machines")
s.add_argument("--public", action="store_true", help="multi-user hosting mode: per-visitor sessions, capped budgets")
s.add_argument("--run-root", default=None, help="folder for search outputs (default runs/studio)")
s.add_argument("--docs-dir", default=None, help="serve a built MkDocs site at /docs/")
s.set_defaults(fn=cmd_studio)
s = sub.add_parser("space", help="every composition of the family with rule values and predictions → CSV")
s.add_argument("config", nargs="?", default="meidnet.yaml")
s.add_argument("--model", default=None)
s.add_argument("-o", "--output", default=None)
s.add_argument("--max", type=int, default=60000)
s.add_argument("--chemiscope", metavar="FILE.json", default=None,
help="also write a chemiscope.org dataset (property map + 3D structures)")
s.set_defaults(fn=cmd_space)
s = sub.add_parser("demo", help="generate perovskites with the published model")
s.add_argument("--family", choices=["halide", "oxide", "chalcogenide", "nitride"], default="halide")
s.add_argument("--band-gap", type=float, default=2.0)
s.add_argument("--enthalpy", type=float, default=-0.10)
s.add_argument("-n", type=int, default=3)
s.add_argument("--rounds", type=int, default=5)
s.add_argument("--steps", type=int, default=300)
s.add_argument("--out", default="runs/demo")
s.add_argument("--device", default="auto")
s.add_argument("--no-open", action="store_true")
s.set_defaults(fn=cmd_demo)
s = sub.add_parser("screen", help="MACE stability screening (pip install meidnet[stability])")
s.add_argument("dir", help="folder with CIF files (searched recursively)")
s.add_argument("--train-csv", default=None, help="table with a formula column, for novelty")
s.add_argument("--threshold", type=float, default=0.10)
s.add_argument("--device", default="auto")
s.add_argument("--model", default="medium")
s.set_defaults(fn=cmd_screen)
s = sub.add_parser("info", help="describe a checkpoint")
s.add_argument("model")
s.set_defaults(fn=cmd_info)
s = sub.add_parser("families", help="list / describe material families")
s.add_argument("name", nargs="?")
s.add_argument("--variant", default=None)
s.set_defaults(fn=cmd_families)
s = sub.add_parser("schema", help="JSON Schema of meidnet.yaml")
s.add_argument("-o", "--output", default=None)
s.set_defaults(fn=cmd_schema)
s = sub.add_parser("download-data", help="fetch the Perov-5 dataset")
s.add_argument("--dir", default="data/perov5")
s.add_argument("--force", action="store_true")
s.set_defaults(fn=cmd_download_data)
a = p.parse_args(argv)
a.fn(a)
if __name__ == "__main__":
main()