add fp for deltaspin

dpgen2/exploration/render/traj_render_lammps_spin.py

@@ -63,7 +63,7 @@ def get_confs(
     ) -> dpdata.MultiSystems:
         del conf_filters  # by far does not support conf filters
         ntraj = len(trajs)
-        traj_fmt = "lammps/dump"
+        traj_fmt = "lammps/spin/dump"
         ms = dpdata.MultiSystems(type_map=type_map)
         for ii in range(ntraj):
             if len(id_selected[ii]) > 0:

dpgen2/fp/__init__.py

@@ -19,6 +19,11 @@
     VaspInputs,
 )
 
+from .deltaspin import (
+    PrepDeltaSpin,
+    RunDeltaSpin,
+)
+
 fp_styles = {
     "vasp": {
         "inputs": VaspInputs,
@@ -40,4 +45,9 @@
         "prep": PrepFpOpAbacus,
         "run": RunFpOpAbacus,
     },
+    "deltaspin": {
+        "inputs": VaspInputs,
+        "prep": PrepDeltaSpin,
+        "run": RunDeltaSpin,
+    },
 }

dpgen2/fp/deltaspin.py

@@ -0,0 +1,168 @@
+import logging
+from pathlib import (
+    Path,
+)
+from typing import (
+    Dict,
+    List,
+    Optional,
+    Set,
+    Tuple,
+    Union,
+)
+
+import dpdata
+import numpy as np
+from dargs import (
+    Argument,
+    ArgumentEncoder,
+    Variant,
+    dargs,
+)
+from dflow.python import (
+    OP,
+    OPIO,
+    Artifact,
+    BigParameter,
+    FatalError,
+    OPIOSign,
+    TransientError,
+)
+
+from dpgen2.constants import (
+    fp_default_log_name,
+    fp_default_out_data_name,
+)
+from dpgen2.utils.run_command import (
+    run_command,
+)
+
+from .prep_fp import (
+    PrepFp,
+)
+from .run_fp import (
+    RunFp,
+)
+from .vasp_input import (
+    VaspInputs,
+    make_kspacing_kpoints,
+)
+
+# global static variables
+vasp_conf_name = "POSCAR"
+vasp_input_name = "INCAR"
+vasp_pot_name = "POTCAR"
+vasp_kp_name = "KPOINTS"
+
+class PrepDeltaSpin(PrepFp):
+    def prep_task(
+        self,
+        conf_frame: dpdata.System,
+        vasp_inputs: VaspInputs,
+    ):
+        r"""Define how one DeltaSpin task is prepared.
+
+        Parameters
+        ----------
+        conf_frame : dpdata.System
+            One frame of configuration in the dpdata format.
+        vasp_inputs : VaspInputs
+            The VaspInputs object handels all other input files of the task.
+        """
+
+        conf_frame.to("vasp/deltaspin_inputs", vasp_inputs.incar_template, vasp_conf_name, vasp_input_name)
+        # fix the case when some element have 0 atom, e.g. H0O2
+        tmp_frame = dpdata.System(vasp_conf_name, fmt="vasp/poscar")
+        Path(vasp_pot_name).write_text(vasp_inputs.make_potcar(tmp_frame["atom_names"]))
+        Path(vasp_kp_name).write_text(vasp_inputs.make_kpoints(conf_frame["cells"][0]))  # type: ignore
+
+
+class RunDeltaSpin(RunFp):
+    def input_files(self) -> List[str]:
+        r"""The mandatory input files to run a DeltaSpin task.
+
+        Returns
+        -------
+        files: List[str]
+            A list of madatory input files names.
+
+        """
+        return [vasp_conf_name, vasp_input_name, vasp_pot_name, vasp_kp_name]
+
+    def optional_input_files(self) -> List[str]:
+        r"""The optional input files to run a DeltaSpin task.
+
+        Returns
+        -------
+        files: List[str]
+            A list of optional input files names.
+
+        """
+        return []
+
+    def run_task(
+        self,
+        command: str,
+        out: str,
+        log: str,
+    ) -> Tuple[str, str]:
+        r"""Defines how one FP task runs
+
+        Parameters
+        ----------
+        command : str
+            The command of running vasp task
+        out : str
+            The name of the output data file.
+        log : str
+            The name of the log file
+
+        Returns
+        -------
+        out_name: str
+            The file name of the output data in the dpdata.LabeledSystem format.
+        log_name: str
+            The file name of the log.
+        """
+
+        log_name = log
+        out_name = out
+        # run vasp
+        command = " ".join([command, ">", log_name])
+        ret, out, err = run_command(command, shell=True)
+        if ret != 0:
+            logging.error(
+                "".join(
+                    ("DeltaSpin failed\n", "out msg: ", out, "\n", "err msg: ", err, "\n")
+                )
+            )
+            raise TransientError("DeltaSpin failed")
+        # convert the output to deepmd/npy format
+        sys = dpdata.LabeledSystem("deltaspin_outputs")
+        sys.to("deepmd/npy", out_name)
+        return out_name, log_name
+
+    @staticmethod
+    def args():
+        r"""The argument definition of the `run_task` method.
+
+        Returns
+        -------
+        arguments: List[dargs.Argument]
+            List of dargs.Argument defines the arguments of `run_task` method.
+        """
+
+        doc_deltaspin_cmd = "The command of DeltaSpin"
+        doc_deltaspin_log = "The log file name of DeltaSpin"
+        doc_deltaspin_out = "The output dir name of labeled data. In `deepmd/spin/npy` format provided by `dpdata`."
+        return [
+            Argument(
+                "command", str, optional=True, default="vasp_deltaspin", doc=doc_deltaspin_cmd
+            ),
+            Argument(
+                "out", str, optional=True, default=fp_default_out_data_name, doc=doc_deltaspin_out
+            ),
+            Argument(
+                "log", str, optional=True, default=fp_default_log_name, doc=doc_deltaspin_log
+            ),
+        ]