docs: add CALYPSO model deviation arginfo

dpgen/generator/arginfo.py

@@ -621,16 +621,190 @@ def model_devi_amber_args() -> list[Argument]:
     ]
 
 
+def model_devi_calypso_args() -> list[Argument]:
+    """CALYPSO engine arguments."""
+    doc_model_devi_jobs = (
+        "Settings for CALYPSO structure generation and model deviation. "
+        "For native CALYPSO mode, each dict in the list specifies the "
+        "structure-generation settings for one or more iterations. In "
+        "calypso_input_path mode, this key is still required by the current "
+        "runtime but may be an empty list."
+    )
+    doc_times = "List of iteration indices when this job should be executed."
+    doc_nameofatoms = "Element symbols of the different chemical species."
+    doc_numberofatoms = "Number of atoms for each chemical species in one formula unit."
+    doc_numberofformula = "Range of formula units per cell as [min, max]."
+    doc_volume = (
+        "Volume per formula unit in angstrom^3. If omitted, CALYPSO "
+        "determines the volume automatically."
+    )
+    doc_distanceofion = (
+        "Minimal distances between atoms of each chemical species in "
+        "angstrom. The matrix shape should match the number of species."
+    )
+    doc_psoratio = "Proportion of structures generated by the PSO algorithm (0.0-1.0)."
+    doc_popsize = "Population size for structure generation."
+    doc_maxstep = "Maximum number of CALYPSO optimization steps."
+    doc_icode = "Interface code for local optimization: 1=VASP, 2=SIESTA, 3=GULP."
+    doc_split = "Whether to split calculations. Use 'T' for true or 'F' for false."
+    doc_vsc = (
+        "Variable Stoichiometry Control. Use 'T' to enable it or 'F' to disable it."
+    )
+    doc_maxnumatom = "Maximum number of atoms in the unit cell when VSC is 'T'."
+    doc_ctrlrange = (
+        "Variation range for each atom type when VSC is 'T'. The matrix shape "
+        "should match the number of species."
+    )
+    doc_pstress = "Target pressure values in GPa."
+    doc_fmax = "Force convergence criterion for local optimization in eV/angstrom."
+    doc_calypso_input_path = (
+        "Directory containing pre-existing CALYPSO input.dat files. When this "
+        "is set, DP-GEN copies the provided input files instead of generating "
+        "them from model_devi_jobs."
+    )
+    doc_model_devi_max_iter = (
+        "Maximum model-deviation iteration index when using calypso_input_path."
+    )
+    doc_vsc_mode = "Enable variable stoichiometry mode for external input files."
+    doc_model_devi_dt = (
+        "Timestep retained for compatibility with existing CALYPSO parameter files."
+    )
+
+    scalar_or_singleton_float = [float, list[float]]
+    scalar_or_singleton_int = [int, list[int]]
+
+    calypso_args = [
+        Argument(
+            "model_devi_jobs",
+            list,
+            optional=False,
+            repeat=True,
+            doc=doc_model_devi_jobs,
+            sub_fields=[
+                Argument("times", list[int], optional=False, doc=doc_times),
+                Argument("NameOfAtoms", list[str], optional=False, doc=doc_nameofatoms),
+                Argument(
+                    "NumberOfAtoms",
+                    list[int],
+                    optional=False,
+                    doc=doc_numberofatoms,
+                ),
+                Argument(
+                    "NumberOfFormula",
+                    list[int],
+                    optional=True,
+                    default=[1, 1],
+                    doc=doc_numberofformula,
+                ),
+                Argument(
+                    "Volume",
+                    scalar_or_singleton_float,
+                    optional=True,
+                    doc=doc_volume,
+                ),
+                Argument(
+                    "DistanceOfIon",
+                    list[list[float]],
+                    optional=False,
+                    doc=doc_distanceofion,
+                ),
+                Argument(
+                    "PsoRatio",
+                    scalar_or_singleton_float,
+                    optional=True,
+                    default=0.6,
+                    doc=doc_psoratio,
+                ),
+                Argument(
+                    "PopSize",
+                    scalar_or_singleton_int,
+                    optional=True,
+                    default=30,
+                    doc=doc_popsize,
+                ),
+                Argument(
+                    "MaxStep",
+                    scalar_or_singleton_int,
+                    optional=True,
+                    default=5,
+                    doc=doc_maxstep,
+                ),
+                Argument(
+                    "ICode",
+                    scalar_or_singleton_int,
+                    optional=True,
+                    default=1,
+                    doc=doc_icode,
+                ),
+                Argument("Split", str, optional=True, default="T", doc=doc_split),
+                Argument("VSC", str, optional=True, default="F", doc=doc_vsc),
+                Argument(
+                    "MaxNumAtom",
+                    scalar_or_singleton_int,
+                    optional=True,
+                    doc=doc_maxnumatom,
+                ),
+                Argument(
+                    "CtrlRange",
+                    list[list[int]],
+                    optional=True,
+                    doc=doc_ctrlrange,
+                ),
+                Argument(
+                    "PSTRESS",
+                    scalar_or_singleton_float,
+                    optional=True,
+                    default=[0.001],
+                    doc=doc_pstress,
+                ),
+                Argument(
+                    "fmax",
+                    scalar_or_singleton_float,
+                    optional=True,
+                    default=0.01,
+                    doc=doc_fmax,
+                ),
+            ],
+        ),
+        Argument("calypso_input_path", str, optional=True, doc=doc_calypso_input_path),
+        Argument(
+            "model_devi_max_iter", int, optional=True, doc=doc_model_devi_max_iter
+        ),
+        Argument("vsc", bool, optional=True, default=False, doc=doc_vsc_mode),
+        Argument("model_devi_dt", float, optional=True, doc=doc_model_devi_dt),
+    ]
+    # CALYPSO uses the same downstream candidate-selection controls as LAMMPS.
+    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 [*calypso_args, *common_args]
+
+
 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 structure-generation engine for crystal prediction."
     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",

dpgen/generator/lib/make_calypso.py

@@ -7,6 +7,20 @@
 import numpy as np
 
 
+def _normalize_calypso_scalar(value, name):
+    if isinstance(value, list):
+        if len(value) != 1:
+            raise ValueError(f"{name} should be a scalar or a single-item list")
+        return value[0]
+    return value
+
+
+def _normalize_calypso_pressures(value):
+    if isinstance(value, list):
+        return value
+    return [value]
+
+
 def make_calypso_input(
     nameofatoms,
     numberofatoms,
@@ -177,11 +191,13 @@ def _make_model_devi_native_calypso(iter_index, model_devi_jobs, calypso_run_opt
     numberofatoms = cur_job.get("NumberOfAtoms")
     numberofformula = cur_job.get("NumberOfFormula", [1, 1])
     volume = cur_job.get("Volume")
+    if volume is not None:
+        volume = _normalize_calypso_scalar(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 = _normalize_calypso_scalar(cur_job.get("PsoRatio", 0.6), "PsoRatio")
+    popsize = _normalize_calypso_scalar(cur_job.get("PopSize", 30), "PopSize")
+    maxstep = _normalize_calypso_scalar(cur_job.get("MaxStep", 5), "MaxStep")
+    icode = _normalize_calypso_scalar(cur_job.get("ICode", 1), "ICode")
     split = cur_job.get("Split", "T")
     # Cluster
 
@@ -192,12 +208,12 @@ 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 = _normalize_calypso_scalar(cur_job.get("MaxNumAtom"), "MaxNumAtom")
         ctrlrange = cur_job.get("CtrlRange")
     # Optimization
-    fmax = cur_job.get("fmax", 0.01)
+    fmax = _normalize_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])
+    pstress = _normalize_calypso_pressures(cur_job.get("PSTRESS", [0.001]))
     # pressures
     for press_idx, temp_calypso_run_opt_path in enumerate(calypso_run_opt_path):
         # cur_press

dpgen/generator/run.py

@@ -59,6 +59,7 @@
 from dpgen.generator.lib.make_calypso import (
     _make_model_devi_buffet,
     _make_model_devi_native_calypso,
+    _normalize_calypso_pressures,
 )
 from dpgen.generator.lib.parse_calypso import (
     _parse_calypso_dis_mtx,
@@ -1356,7 +1357,9 @@ def make_model_devi(iter_index, jdata, mdata):
                 if iter_index in jobbs.get("times"):
                     cur_job = model_devi_jobs[iiidx]
 
-            pressures_list = cur_job.get("PSTRESS", [0.0001])
+            pressures_list = _normalize_calypso_pressures(
+                cur_job.get("PSTRESS", [0.0001])
+            )
             for temp_idx in range(len(pressures_list)):
                 calypso_run_opt_path.append(
                     "%s.%03d" % (_calypso_run_opt_path, temp_idx)  # noqa: UP031

tests/generator/test_calypso.py

@@ -1,4 +1,5 @@
 import os
+import shutil
 import sys
 import unittest
 from pathlib import Path
@@ -8,6 +9,8 @@
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 __package__ = "generator"
 
+from dpgen.generator.lib.make_calypso import _make_model_devi_native_calypso
+
 from .context import (
     _parse_calypso_dis_mtx,
     _parse_calypso_input,
@@ -132,6 +135,24 @@ def test_make_calypso_input(self):
                 os.remove("input.dat")
                 break
 
+    def test_make_native_calypso_accepts_single_item_lists(self):
+        work_path = Path("calypso_native_test")
+        run_path = work_path / "gen_stru_analy.000"
+        if work_path.exists():
+            shutil.rmtree(work_path)
+        run_path.mkdir(parents=True)
+        job = model_devi_jobs["model_devi_jobs"].copy()
+
+        _make_model_devi_native_calypso(0, [{**job, "times": [0]}], [str(run_path)])
+
+        text = (run_path / "input.dat").read_text()
+        self.assertIn("Volume = 30", text)
+        self.assertIn("PsoRatio = 0.6", text)
+        self.assertIn("PopSize = 5", text)
+        self.assertIn("PSTRESS = 0.000000", text)
+        self.assertIn("fmax = 0.010000", text)
+        shutil.rmtree(work_path)
+
     def test_parse_calypso_input(self):
         ret = make_calypso_input(
             ["Mg", "Al", "Cu"],

tests/test_calypso_arginfo.py

@@ -0,0 +1,60 @@
+import json
+import unittest
+from pathlib import Path
+
+from dargs import Argument
+
+from dpgen.generator.arginfo import model_devi_args
+
+
+class TestCalypsoArginfo(unittest.TestCase):
+    def test_selection_arguments_are_accepted(self):
+        """CALYPSO exposes the controls consumed by downstream FP selection."""
+        arginfo = Argument("model_devi", dict, sub_variants=model_devi_args())
+        data = {
+            "model_devi_engine": "calypso",
+            "model_devi_jobs": [],
+            "calypso_input_path": "calypso-input",
+            "model_devi_skip": 0,
+            "model_devi_f_trust_lo": 0.05,
+            "model_devi_f_trust_hi": 0.15,
+            "model_devi_clean_traj": True,
+            "model_devi_numb_candi_f": 10,
+            "model_devi_numb_candi_v": 10,
+            "shuffle_poscar": False,
+        }
+
+        normalized = arginfo.normalize_value(data)
+        arginfo.check_value(normalized, strict=True)
+
+    def test_checked_in_example_is_schema_compatible(self):
+        """The maintained example's CALYPSO section passes strict validation."""
+        param_file = (
+            Path(__file__).parent.parent
+            / "examples"
+            / "run"
+            / "dp-calypso-vasp"
+            / "param.json"
+        )
+        with open(param_file) as fp:
+            example = json.load(fp)
+
+        model_devi_keys = {
+            "model_devi_engine",
+            "model_devi_jobs",
+            "model_devi_dt",
+            "model_devi_skip",
+            "model_devi_f_trust_lo",
+            "model_devi_f_trust_hi",
+            "model_devi_clean_traj",
+            "shuffle_poscar",
+            "vsc",
+        }
+        data = {key: example[key] for key in model_devi_keys}
+        arginfo = Argument("model_devi", dict, sub_variants=model_devi_args())
+        normalized = arginfo.normalize_value(data)
+        arginfo.check_value(normalized, strict=True)
+
+
+if __name__ == "__main__":
+    unittest.main()