diff --git a/dpgen/generator/arginfo.py b/dpgen/generator/arginfo.py index 0f8abb865..0e2607006 100644 --- a/dpgen/generator/arginfo.py +++ b/dpgen/generator/arginfo.py @@ -621,16 +621,240 @@ def model_devi_amber_args() -> list[Argument]: ] +def _is_scalar_or_singleton(value) -> bool: + """Accept legacy one-item lists while rejecting ambiguous CALYPSO values.""" + return not isinstance(value, list) or len(value) == 1 + + +def model_devi_calypso_jobs_args() -> Argument: + """Return native CALYPSO structure-generation settings.""" + return Argument( + "model_devi_jobs", + list, + repeat=True, + optional=False, + doc=( + "CALYPSO generation settings. Each entry applies to the iterations " + "listed in times. An empty list is accepted in external-input mode." + ), + sub_fields=[ + Argument( + "times", + list[int], + optional=False, + doc="Iteration indices that use this generation setup.", + ), + Argument( + "NameOfAtoms", + list[str], + optional=False, + doc="Element symbols for all generated species.", + ), + Argument( + "NumberOfAtoms", + list[int], + optional=False, + doc="Atoms of each species in one formula unit.", + ), + Argument( + "NumberOfFormula", + list[int], + optional=True, + default=[1, 1], + doc="Inclusive minimum and maximum formula units per cell.", + ), + Argument( + "Volume", + [None, float, int, list[float], list[int]], + optional=True, + extra_check=_is_scalar_or_singleton, + extra_check_errmsg="Volume must be a scalar or a one-item list.", + doc="Volume per formula unit in cubic Angstrom.", + ), + Argument( + "DistanceOfIon", + list[list[float]], + optional=False, + doc="Square matrix of minimum inter-species distances in Angstrom.", + ), + Argument( + "PsoRatio", + [float, int, list[float], list[int]], + optional=True, + default=0.6, + extra_check=_is_scalar_or_singleton, + extra_check_errmsg="PsoRatio must be a scalar or a one-item list.", + doc="Fraction of structures generated by particle-swarm optimization.", + ), + Argument( + "PopSize", + [int, list[int]], + optional=True, + default=30, + extra_check=_is_scalar_or_singleton, + extra_check_errmsg="PopSize must be an integer or a one-item list.", + doc="CALYPSO population size.", + ), + Argument( + "MaxStep", + [int, list[int]], + optional=True, + default=5, + extra_check=_is_scalar_or_singleton, + extra_check_errmsg="MaxStep must be an integer or a one-item list.", + doc="Maximum particle-swarm optimization steps.", + ), + Argument( + "ICode", + [int, list[int]], + optional=True, + default=1, + extra_check=_is_scalar_or_singleton, + extra_check_errmsg="ICode must be an integer or a one-item list.", + doc="CALYPSO local-optimization interface code.", + ), + Argument( + "Split", + str, + optional=True, + default="T", + doc="CALYPSO Split flag, written as 'T' or 'F'.", + ), + Argument( + "VSC", + str, + optional=True, + default="F", + doc="Variable-stoichiometry flag, written as 'T' or 'F'.", + ), + Argument( + "MaxNumAtom", + [int, list[int]], + optional=True, + extra_check=_is_scalar_or_singleton, + extra_check_errmsg="MaxNumAtom must be an integer or a one-item list.", + doc="Maximum atoms per cell when VSC is enabled.", + ), + Argument( + "CtrlRange", + list[list[int]], + optional=True, + doc="Per-species atom-count ranges when VSC is enabled.", + ), + Argument( + "PSTRESS", + list[float], + optional=True, + default=[0.001], + doc="Target pressures in GPa; one CALYPSO directory is made per value.", + ), + Argument( + "fmax", + [float, int, list[float], list[int]], + optional=True, + default=0.01, + extra_check=_is_scalar_or_singleton, + extra_check_errmsg="fmax must be a scalar or a one-item list.", + doc="Force convergence threshold in eV/Angstrom.", + ), + Argument( + "task_min", + int, + optional=True, + doc="Per-job override of the minimum number of labeling tasks.", + ), + Argument( + "model_devi_f_trust_lo", + [float, dict], + optional=True, + doc="Per-job lower force-deviation threshold.", + ), + Argument( + "model_devi_f_trust_hi", + [float, dict], + optional=True, + doc="Per-job upper force-deviation threshold.", + ), + Argument( + "model_devi_v_trust_lo", + [float, dict], + optional=True, + doc="Per-job lower virial-deviation threshold.", + ), + Argument( + "model_devi_v_trust_hi", + [float, dict], + optional=True, + doc="Per-job upper virial-deviation threshold.", + ), + ], + ) + + +def model_devi_calypso_args() -> list[Argument]: + """Return model-deviation arguments supported by CALYPSO workflows.""" + common_names = { + "model_devi_skip", + "model_devi_f_trust_lo", + "model_devi_f_trust_hi", + "model_devi_v_trust_lo", + "model_devi_v_trust_hi", + "model_devi_adapt_trust_lo", + "model_devi_numb_candi_f", + "model_devi_numb_candi_v", + "model_devi_perc_candi_f", + "model_devi_perc_candi_v", + "model_devi_clean_traj", + "shuffle_poscar", + } + common_args = [ + argument for argument in model_devi_lmp_args() if argument.name in common_names + ] + return [ + model_devi_calypso_jobs_args(), + *common_args, + Argument( + "model_devi_dt", + float, + optional=True, + doc="Accepted for compatibility; CALYPSO does not run MD timesteps.", + ), + Argument( + "calypso_input_path", + str, + optional=True, + doc="Directory containing user-provided CALYPSO input.dat files.", + ), + Argument( + "model_devi_max_iter", + int, + optional=True, + doc="Last iteration generated when calypso_input_path is used.", + ), + Argument( + "vsc", + bool, + optional=True, + default=False, + doc="Use variable-stoichiometry input files in external-input mode.", + ), + ] + + def model_devi_args() -> list[Variant]: doc_model_devi_engine = "Engine for the model deviation task." doc_amber = "Amber DPRc engine. The command argument in the machine file should be path to sander." + doc_calypso = ( + "CALYPSO crystal-structure generation and model-deviation workflow. " + "It supports native per-iteration settings or external input.dat files." + ) return [ Variant( "model_devi_engine", [ Argument("lammps", dict, model_devi_lmp_args(), doc="LAMMPS"), Argument("amber", dict, model_devi_amber_args(), doc=doc_amber), - Argument("calypso", dict, [], doc="TODO: add doc"), + Argument("calypso", dict, model_devi_calypso_args(), doc=doc_calypso), Argument("gromacs", dict, [], doc="TODO: add doc"), ], default_tag="lammps", diff --git a/dpgen/generator/lib/make_calypso.py b/dpgen/generator/lib/make_calypso.py index 42d5e7f69..b315c6057 100644 --- a/dpgen/generator/lib/make_calypso.py +++ b/dpgen/generator/lib/make_calypso.py @@ -162,6 +162,15 @@ def _make_model_devi_buffet(jdata, calypso_run_opt_path): raise FileNotFoundError("input.dat") +def _unwrap_calypso_scalar(value, name): + """Normalize a legacy one-item CALYPSO list to its scalar value.""" + if not isinstance(value, list): + return value + if len(value) != 1: + raise ValueError(f"{name} must be a scalar or a one-item list") + return value[0] + + def _make_model_devi_native_calypso(iter_index, model_devi_jobs, calypso_run_opt_path): for iiidx, jobbs in enumerate(model_devi_jobs): if iter_index in jobbs.get("times"): @@ -176,12 +185,12 @@ def _make_model_devi_native_calypso(iter_index, model_devi_jobs, calypso_run_opt nameofatoms = cur_job.get("NameOfAtoms") numberofatoms = cur_job.get("NumberOfAtoms") numberofformula = cur_job.get("NumberOfFormula", [1, 1]) - volume = cur_job.get("Volume") + volume = _unwrap_calypso_scalar(cur_job.get("Volume"), "Volume") distanceofion = cur_job.get("DistanceOfIon") - psoratio = cur_job.get("PsoRatio", 0.6) - popsize = cur_job.get("PopSize", 30) - maxstep = cur_job.get("MaxStep", 5) - icode = cur_job.get("ICode", 1) + psoratio = _unwrap_calypso_scalar(cur_job.get("PsoRatio", 0.6), "PsoRatio") + popsize = _unwrap_calypso_scalar(cur_job.get("PopSize", 30), "PopSize") + maxstep = _unwrap_calypso_scalar(cur_job.get("MaxStep", 5), "MaxStep") + icode = _unwrap_calypso_scalar(cur_job.get("ICode", 1), "ICode") split = cur_job.get("Split", "T") # Cluster @@ -192,10 +201,10 @@ def _make_model_devi_native_calypso(iter_index, model_devi_jobs, calypso_run_opt ctrlrange = None vsc = cur_job.get("VSC", "F") if vsc == "T": - maxnumatom = cur_job.get("MaxNumAtom") + maxnumatom = _unwrap_calypso_scalar(cur_job.get("MaxNumAtom"), "MaxNumAtom") ctrlrange = cur_job.get("CtrlRange") # Optimization - fmax = cur_job.get("fmax", 0.01) + fmax = _unwrap_calypso_scalar(cur_job.get("fmax", 0.01), "fmax") # pstress is a List which contains the target stress pstress = cur_job.get("PSTRESS", [0.001]) # pressures diff --git a/tests/generator/test_calypso.py b/tests/generator/test_calypso.py index 5d9abb854..014e0b039 100644 --- a/tests/generator/test_calypso.py +++ b/tests/generator/test_calypso.py @@ -1,10 +1,13 @@ import os import sys +import tempfile import unittest from pathlib import Path import numpy as np +from dpgen.generator.lib.make_calypso import _make_model_devi_native_calypso + sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) __package__ = "generator" @@ -132,6 +135,36 @@ def test_make_calypso_input(self): os.remove("input.dat") break + def test_native_input_normalizes_singleton_scalars(self): + """Legacy singleton lists are rendered as scalar CALYPSO values.""" + job = { + "times": [0], + "NameOfAtoms": ["Mg"], + "NumberOfAtoms": [1], + "NumberOfFormula": [1, 1], + "Volume": [30], + "DistanceOfIon": [[1.4]], + "PsoRatio": [0.6], + "PopSize": [5], + "MaxStep": [3], + "ICode": [1], + "VSC": "T", + "MaxNumAtom": [20], + "CtrlRange": [[1, 20]], + "PSTRESS": [0.0], + "fmax": [0.01], + } + with tempfile.TemporaryDirectory() as tmpdir: + run_path = Path(tmpdir) / "calypso.000" + run_path.mkdir() + _make_model_devi_native_calypso(0, [job], [str(run_path)]) + generated = (run_path / "input.dat").read_text() + + self.assertIn("Volume = 30", generated) + self.assertIn("PsoRatio = 0.6", generated) + self.assertIn("PopSize = 5", generated) + self.assertNotIn("[", generated) + def test_parse_calypso_input(self): ret = make_calypso_input( ["Mg", "Al", "Cu"], diff --git a/tests/test_calypso_arginfo.py b/tests/test_calypso_arginfo.py new file mode 100644 index 000000000..15baefd77 --- /dev/null +++ b/tests/test_calypso_arginfo.py @@ -0,0 +1,94 @@ +"""Validate CALYPSO model-deviation parameter documentation.""" + +import unittest + +from dargs import Argument + +from dpgen.generator.arginfo import model_devi_args + + +class TestCalypsoArginfo(unittest.TestCase): + def setUp(self): + self.arginfo = Argument("model_devi", dict, sub_variants=model_devi_args()) + self.selection = { + "model_devi_skip": 0, + "model_devi_f_trust_lo": 0.05, + "model_devi_f_trust_hi": 0.15, + "model_devi_clean_traj": True, + } + + def test_native_mode(self): + data = { + "model_devi_engine": "calypso", + **self.selection, + "model_devi_dt": 0.002, + "shuffle_poscar": False, + "model_devi_jobs": [ + { + "times": [0, 1], + "NameOfAtoms": ["Mg", "Al"], + "NumberOfAtoms": [1, 1], + "NumberOfFormula": [1, 2], + "Volume": 30.0, + "DistanceOfIon": [[1.4, 1.5], [1.5, 1.4]], + "PsoRatio": 0.6, + "PopSize": 30, + "MaxStep": 5, + "ICode": 1, + "Split": "T", + "VSC": "T", + "MaxNumAtom": 20, + "CtrlRange": [[1, 10], [1, 10]], + "PSTRESS": [0.0, 100.0], + "fmax": 0.01, + "task_min": 1, + } + ], + } + + normalized = self.arginfo.normalize_value(data) + self.arginfo.check_value(normalized, strict=True) + + def test_external_input_mode(self): + data = { + "model_devi_engine": "calypso", + **self.selection, + "model_devi_jobs": [], + "calypso_input_path": "calypso_input", + "model_devi_max_iter": 20, + "vsc": True, + } + + normalized = self.arginfo.normalize_value(data) + self.arginfo.check_value(normalized, strict=True) + + def test_legacy_singleton_scalars(self): + """Checked-in CALYPSO inputs retain their historical list spelling.""" + data = { + "model_devi_engine": "calypso", + **self.selection, + "model_devi_jobs": [ + { + "times": [0], + "NameOfAtoms": ["Mg"], + "NumberOfAtoms": [1], + "Volume": [30], + "DistanceOfIon": [[1.4]], + "PsoRatio": [0.6], + "PopSize": [30], + "MaxStep": [5], + "ICode": [1], + "VSC": "T", + "MaxNumAtom": [20], + "CtrlRange": [[1, 20]], + "fmax": [0.01], + } + ], + } + + normalized = self.arginfo.normalize_value(data) + self.arginfo.check_value(normalized, strict=True) + + +if __name__ == "__main__": + unittest.main()